Cleanup ConstructSnapshot, add multiple UpdateGraph awareness, and op…

…timize static cases (#4228)

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

@@ -1258,7 +1258,7 @@ public Table setTotalsTable(String directive) {
     public static void initializeWithSnapshot(
             String logPrefix, SwapListener swapListener, ConstructSnapshot.SnapshotFunction snapshotFunction) {
         if (swapListener == null) {
-            snapshotFunction.call(false, ExecutionContext.getContext().getUpdateGraph().clock().currentValue());
+            snapshotFunction.call(false, LogicalClock.NULL_CLOCK_VALUE);
             return;
         }
         ConstructSnapshot.callDataSnapshotFunction(logPrefix, swapListener.makeSnapshotControl(), snapshotFunction);

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

@@ -82,6 +82,7 @@ public SwapListener(final BaseTable<?> sourceTable) {
     public ConstructSnapshot.SnapshotControl makeSnapshotControl() {
         // noinspection AutoBoxing
         return ConstructSnapshot.makeSnapshotControl(
+                sourceTable.getUpdateGraph(),
                 this::start,
                 (final long currentClockValue, final boolean usingPreviousValues) -> isInInitialNotificationWindow(),
                 (final long afterClockValue, final boolean usedPreviousValues) -> end(afterClockValue));

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

@@ -31,6 +31,7 @@ public SwapListenerEx(@NotNull final BaseTable<?> sourceTable, @NotNull final No
     @Override
     public ConstructSnapshot.SnapshotControl makeSnapshotControl() {
         return ConstructSnapshot.makeSnapshotControl(
+                sourceTable.getUpdateGraph(),
                 this::startWithExtra,
                 (final long currentClockValue, final boolean usingPreviousValues) -> isInInitialNotificationWindow()
                         && extra.getLastNotificationStep() == extraLastNotificationStep,

engine/table/src/test/java/io/deephaven/engine/table/impl/remote/TestConstructSnapshot.java

@@ -8,6 +8,7 @@
 import io.deephaven.engine.testutil.ControlledUpdateGraph;
 import io.deephaven.engine.testutil.testcase.RefreshingTableTestCase;
 import io.deephaven.engine.updategraph.LogicalClock;
+import io.deephaven.engine.updategraph.UpdateGraph;
 import io.deephaven.util.SafeCloseable;
 import org.apache.commons.lang3.mutable.MutableLong;
 
@@ -16,6 +17,7 @@ public class TestConstructSnapshot extends RefreshingTableTestCase {
     public void testClockChange() throws InterruptedException {
         final MutableLong changed = new MutableLong(0);
         final ConstructSnapshot.SnapshotControl control = new ConstructSnapshot.SnapshotControl() {
+
             @Override
             public Boolean usePreviousValues(long beforeClockValue) {
                 // noinspection AutoBoxing
@@ -26,6 +28,11 @@ public Boolean usePreviousValues(long beforeClockValue) {
             public boolean snapshotConsistent(final long currentClockValue, final boolean usingPreviousValues) {
                 return true;
             }
+
+            @Override
+            public UpdateGraph getUpdateGraph() {
+                return ExecutionContext.getContext().getUpdateGraph();
+            }
         };
         final ExecutionContext executionContext = ExecutionContext.getContext();
         final Runnable snapshot_test = () -> {

engine/updategraph/src/main/java/io/deephaven/engine/updategraph/LogicalClock.java

@@ -10,6 +10,12 @@
  */
 public interface LogicalClock {
 
+    /**
+     * The "null" value, which encodes {step=-1, state=Idle}. Used as a marker when no clock value is appropriate, e.g.
+     * for snapshots of static data.
+     */
+    long NULL_CLOCK_VALUE = -1L;
+
     /**
      * The state component of a logical timestamp.
      */

engine/updategraph/src/main/java/io/deephaven/engine/updategraph/LogicalClockImpl.java

@@ -29,8 +29,16 @@ public class LogicalClockImpl implements LogicalClock {
 
     private static final Logger log = LoggerFactory.getLogger(LogicalClockImpl.class);
 
-    // {2, Idle}, just in case any code has 0 or 1 as an initializer.
-    private final AtomicLong currentValue = new AtomicLong(5L);
+    /**
+     * Our initial clock value. Equivalent to {step=2, state=Idle}. Uses step 2 in case any code has 0 or 1 as an
+     * initializer.
+     */
+    private static final long INITIAL_CLOCK_VALUE = 5L;
+
+    /**
+     * The current value, encoding both step and state as a single long value.
+     */
+    private final AtomicLong currentValue = new AtomicLong(INITIAL_CLOCK_VALUE);
 
     /**
      * Get the current value of the clock.
@@ -106,6 +114,6 @@ public final void ensureUpdateCycleCompleted(final long updatingCycleValue) {
      */
     @TestUseOnly
     public final void resetForUnitTests() {
-        currentValue.set(5L);
+        currentValue.set(INITIAL_CLOCK_VALUE);
     }
 }

server/src/main/java/io/deephaven/server/barrage/BarrageMessageProducer.java

@@ -2195,6 +2195,11 @@ public boolean snapshotCompletedConsistently(final long afterClockValue, final b
             }
             return success;
         }
+
+        @Override
+        public UpdateGraph getUpdateGraph() {
+            return parent.getUpdateGraph();
+        }
     }
 
     @VisibleForTesting