Make TableUpdateValidator composable in a sequence of operations.

engine/table/src/main/java/io/deephaven/engine/table/impl/TableUpdateValidator.java

@@ -19,6 +19,7 @@
 import io.deephaven.util.SafeCloseableList;
 import org.apache.commons.lang3.mutable.MutableInt;
 
+import javax.sound.midi.Track;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -48,8 +49,9 @@ public static TableUpdateValidator make(final String description, final QueryTab
     private final ModifiedColumnSet validationMCS;
     private ColumnInfo[] columnInfos;
 
-    private WritableRowSet rowSet;
+    private TrackingWritableRowSet rowSet;
     private QueryTable resultTable;
+    private ModifiedColumnSet.Transformer transformer;
     private SharedContext sharedContext;
     private final String description;
 
@@ -97,11 +99,10 @@ public String getLogPrefix() {
 
     @Override
     public Result initialize(boolean usePrev, long beforeClock) {
-        rowSet = usePrev ? tableToValidate.getRowSet().copyPrev() : tableToValidate.getRowSet().copy();
+        rowSet = (usePrev ? tableToValidate.getRowSet().copyPrev() : tableToValidate.getRowSet().copy()).toTracking();
 
-        resultTable = new QueryTable(RowSetFactory.empty().toTracking(),
-                Collections.emptyMap());
-        resultTable.setFlat();
+        resultTable = new QueryTable(rowSet, tableToValidate.getColumnSourceMap());
+        transformer = tableToValidate.newModifiedColumnSetIdentityTransformer(resultTable);
 
         final TableUpdateListener listener;
         try (final SafeCloseable ignored1 = maybeOpenSharedContext();
@@ -146,7 +147,7 @@ public void deepValidation() {
     }
 
     private void onUpdate(final TableUpdate upstream) {
-        if (resultTable.size() >= MAX_ISSUES) {
+        if (issues.size() >= MAX_ISSUES) {
             return;
         }
 
@@ -212,7 +213,13 @@ private void onUpdate(final TableUpdate upstream) {
                 }
                 result.append("\n");
                 resultTable.notifyListenersOnError(new RuntimeException(result.toString()), null);
+                return;
             }
+
+            final TableUpdateImpl downstream = TableUpdateImpl.copy(upstream);
+            downstream.modifiedColumnSet = resultTable.getModifiedColumnSetForUpdates();
+            transformer.transform(upstream.modifiedColumnSet(), downstream.modifiedColumnSet);
+            resultTable.notifyListeners(downstream);
         }
     }
 
@@ -242,6 +249,15 @@ private void noteIssue(Supplier<String> issue) {
         }
     }
 
+    /**
+     * Has an update validation failed on this table?
+     * 
+     * @return true if an update validation has failed on this table.
+     */
+    public boolean hasFailed() {
+        return !issues.isEmpty();
+    }
+
     private void validateValues(final String what, final ModifiedColumnSet columnsToCheck, final RowSet toValidate,
             final boolean usePrev, final boolean invertMCS) {
         try (final RowSequence.Iterator it = toValidate.getRowSequenceIterator()) {

engine/table/src/test/java/io/deephaven/engine/table/impl/ErrorListener.java

@@ -16,9 +16,7 @@ public ErrorListener(Table table) {
     }
 
     @Override
-    public void onUpdate(final TableUpdate upstream) {
-        TestCase.fail("Should not have gotten an update!");
-    }
+    public void onUpdate(final TableUpdate upstream) {}
 
     @Override
     public void onFailureInternal(Throwable originalException, Entry sourceEntry) {

engine/table/src/test/java/io/deephaven/engine/table/impl/QueryTableTest.java

@@ -1464,6 +1464,9 @@ public void testSnapshotDependencies() {
             // Now we should flush the second snapshot
             flushed = updateGraph.flushOneNotificationForUnitTests();
             TestCase.assertTrue(flushed);
+            // Which also generates a result notification as a pass-through
+            flushed = updateGraph.flushOneNotificationForUnitTests();
+            TestCase.assertTrue(flushed);
             TestCase.assertTrue(
                     snappedFirst.satisfied(ExecutionContext.getContext().getUpdateGraph().clock().currentStep()));
             TestCase.assertTrue(
@@ -1474,6 +1477,9 @@ public void testSnapshotDependencies() {
             // This should flush the second TUV
             flushed = updateGraph.flushOneNotificationForUnitTests();
             TestCase.assertTrue(flushed);
+            // Which also generates a result notification as a pass-through
+            flushed = updateGraph.flushOneNotificationForUnitTests();
+            TestCase.assertTrue(flushed);
 
             // And now we should be done
             flushed = updateGraph.flushOneNotificationForUnitTests();

engine/test-utils/src/main/java/io/deephaven/engine/testutil/EvalNugget.java

@@ -52,10 +52,7 @@ public EvalNugget(String description) {
     private Throwable exception = null;
 
     // We should listen for failures on the table, and if we get any, the test case is no good.
-    class FailureListener extends InstrumentedTableUpdateListener {
-        FailureListener() {
-            super("Failure Listener");
-        }
+    class FailureListener extends ValidationFailureListener {
 
         @Override
         public void onUpdate(final TableUpdate upstream) {
@@ -64,6 +61,17 @@ public void onUpdate(final TableUpdate upstream) {
                 System.out.println(upstream);
             }
         }
+    }
+
+    class ValidationFailureListener extends InstrumentedTableUpdateListener {
+        ValidationFailureListener() {
+            super("Failure Listener");
+        }
+
+        @Override
+        public void onUpdate(TableUpdate upstream) {
+            // do nothing
+        }
 
         @Override
         public void onFailureInternal(Throwable originalException, Entry sourceEntry) {
@@ -88,10 +96,11 @@ public void onFailureInternal(Throwable originalException, Entry sourceEntry) {
     }
 
     private final TableUpdateValidator validator;
+    private final TableUpdateListener validationFailureListener = new ValidationFailureListener();
     {
         if (originalValue instanceof QueryTable && ((QueryTable) originalValue).isRefreshing()) {
             validator = TableUpdateValidator.make((QueryTable) originalValue);
-            validator.getResultTable().addUpdateListener(failureListener);
+            validator.getResultTable().addUpdateListener(validationFailureListener);
         } else {
             validator = null;
         }

engine/test-utils/src/main/java/io/deephaven/engine/testutil/UpdateValidatorNugget.java

@@ -10,6 +10,7 @@
 import io.deephaven.engine.table.impl.QueryTable;
 import io.deephaven.engine.table.impl.TableUpdateValidator;
 import io.deephaven.engine.util.TableTools;
+import io.deephaven.util.annotations.ReferentialIntegrity;
 import junit.framework.TestCase;
 import org.junit.Assert;
 
@@ -27,7 +28,7 @@ public UpdateValidatorNugget(final QueryTable table) {
         this.validator = TableUpdateValidator.make(originalValue);
 
         originalValue.addUpdateListener(failureListener);
-        validator.getResultTable().addUpdateListener(failureListener);
+        validator.getResultTable().addUpdateListener(validatorFailureListener);
     }
 
     private final QueryTable originalValue;
@@ -36,26 +37,34 @@ public UpdateValidatorNugget(final QueryTable table) {
     private Throwable exception = null;
 
     // We should listen for failures on the table, and if we get any, the test case is no good.
-    private final TableUpdateListener failureListener =
-            new InstrumentedTableUpdateListener("Failure Listener") {
-                @Override
-                public void onUpdate(TableUpdate update) {}
-
-                @Override
-                public void onFailureInternal(Throwable originalException, Entry sourceEntry) {
-                    exception = originalException;
-                    final StringWriter errors = new StringWriter();
-                    originalException.printStackTrace(new PrintWriter(errors));
-                    TestCase.fail(errors.toString());
-                }
-            };
+    @ReferentialIntegrity
+    private final TableUpdateListener failureListener = new FailureListener();
+    @ReferentialIntegrity
+    private final TableUpdateListener validatorFailureListener = new FailureListener();
 
     public void validate(final String msg) {
         Assert.assertNull(exception);
-        Assert.assertEquals(0, validator.getResultTable().size());
+        Assert.assertFalse(validator.hasFailed());
     }
 
     public void show() {
         TableTools.showWithRowSet(originalValue, 100);
     }
+
+    private class FailureListener extends InstrumentedTableUpdateListener {
+        public FailureListener() {
+            super("Failure Listener");
+        }
+
+        @Override
+        public void onUpdate(TableUpdate update) {}
+
+        @Override
+        public void onFailureInternal(Throwable originalException, Entry sourceEntry) {
+            exception = originalException;
+            final StringWriter errors = new StringWriter();
+            originalException.printStackTrace(new PrintWriter(errors));
+            TestCase.fail(errors.toString());
+        }
+    }
 }