feat: DH-21486: Allow the clock filter to execute in parallel. (#7594)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: cpwright

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

@@ -1395,6 +1395,14 @@ private QueryTable whereInternal(final WhereFilter... filters) {
                             }
                             if (reindexingFilter.requiresSorting()) {
                                 result = (QueryTable) result.sort(reindexingFilter.getSortColumns());
+                                if (!result.isFlat()) {
+                                    // The only reindexing filters are ClockFilters; and the SortedClockFilter is the
+                                    // only one that requires sorting. It makes the assumption that sort flattens the
+                                    // table, which is generally true for a historical sort - but if the input was
+                                    // already sorted, we don't create a new table but rather just return a copy of the
+                                    // table with the sorted columns set.
+                                    result = (QueryTable) result.flatten();
+                                }
                                 reindexingFilter.sortingDone();
                             }
                             // Execute the current filter (which is or wraps a reindexing filter)

engine/table/src/main/java/io/deephaven/engine/table/impl/select/ClockFilter.java

@@ -80,6 +80,19 @@ public final WritableRowSet filter(
         return initial == null ? RowSetFactory.empty() : initial;
     }
 
+    /**
+     * This method is called from the {@link ClockFilter#filter(RowSet, RowSet, Table, boolean)} method to initialize
+     * the filter's internal data structures.
+     *
+     * <p>
+     * The filter permits parallelization, so it may be called from many initialization threads concurrently.
+     * </p>
+     *
+     * @param selection the selection to be filtered
+     * @param fullSet the full set of rows in the source table
+     * @param table the table that is being filtered
+     * @return the rowset that matches the filter at initialization time
+     */
     @Nullable
     protected abstract WritableRowSet initializeAndGetInitialIndex(@NotNull final RowSet selection,
             @NotNull final RowSet fullSet, @NotNull final Table table);
@@ -131,6 +144,13 @@ public final void run() {
         }
     }
 
+    /**
+     * This method is called from the {@link ClockFilter#run()} method as part of the update graph source refresh.
+     *
+     * <p>
+     * This method is not called concurrently, because there are no updates from the source table.
+     * </p>
+     */
     @Nullable
     protected abstract WritableRowSet updateAndGetAddedIndex();
 
@@ -140,7 +160,7 @@ public final void run() {
     protected final static class Range {
 
         long nextKey;
-        private final long lastKey;
+        private long lastKey;
 
         protected Range(final long firstKey, final long lastKey) {
             nextKey = Require.leq(firstKey, "firstRowKey", lastKey, "lastRowKey");
@@ -170,5 +190,39 @@ RowSetBuilderRandom consumeKeysAndAppendAdded(final ColumnSource<Long> nanosColu
             addedBuilder.addRange(firstKeyAdded, lastKeyAdded);
             return addedBuilder;
         }
+
+        /**
+         * Returns true if this range is before the given key.
+         */
+        boolean isBefore(final long key) {
+            return lastKey < key;
+        }
+
+        /**
+         * Merges range if contiguous; returns true if the range was merged into this range
+         */
+        boolean merge(final long firstKey, final long lastKey) {
+            if (isEmpty()) {
+                this.nextKey = firstKey;
+                this.lastKey = lastKey;
+                return true;
+            }
+            if (firstKey == this.lastKey + 1) {
+                this.lastKey = lastKey;
+                return true;
+            }
+            if (lastKey == this.nextKey - 1) {
+                this.nextKey = firstKey;
+                return true;
+            }
+            return false;
+        }
+
+        /**
+         * @return the first key in this range
+         */
+        long firstKey() {
+            return nextKey;
+        }
     }
 }

engine/table/src/main/java/io/deephaven/engine/table/impl/select/SortedClockFilter.java

@@ -12,6 +12,10 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
+import java.util.Comparator;
+import java.util.LinkedList;
+import java.util.List;
+
 /**
  * This will filter a table on an Instant column for all rows greater than "now" according to a supplied clock. It
  * requires sorting of the input table according to the specified timestamp column, leveraging this for a very efficient
@@ -21,7 +25,17 @@
 public class SortedClockFilter extends ClockFilter {
 
     private boolean sortingDone;
+    /**
+     * the currently active range we are working on; this must be the first range in the table (because we are
+     * guaranteed to be sorted).
+     */
     private Range range;
+    /**
+     * A list of ranges that were not immediately mergeable with the first range. They are ordered according to key, and
+     * we pop one off at a time until all the rows are consumed. We can expect at most one range per initialization
+     * thread to be in this list.
+     */
+    private final List<Range> pendingRanges = new LinkedList<>();
 
     public SortedClockFilter(@NotNull final String columnName,
             @NotNull final Clock clock,
@@ -59,31 +73,41 @@ protected WritableRowSet initializeAndGetInitialIndex(@NotNull final RowSet sele
         // because the input table must be historical, and the historical sort implementation uses a
         // ContiguousRowRedirection.
         Require.requirement(table.isFlat(), "table.isFlat()");
-        // This must be the first filter in a where-clause of its own, again because of the sort, hence selection must
-        // be equal to fullSet.
-        // This test as implemented only works because the table is flat.
-        Require.requirement(selection.size() == fullSet.size()
-                && selection.size() == selection.lastRowKey() - selection.firstRowKey() + 1
-                && fullSet.size() == fullSet.lastRowKey() - fullSet.firstRowKey() + 1,
-                "selection.size() == fullSet.size() && selection.size() == selection.lastRowKey() - selection.firstRowKey() + 1 && fullSet.size() == fullSet.lastRowKey() - fullSet.firstRowKey() + 1");
 
-        range = new Range(selection.firstRowKey(), selection.lastRowKey());
-        return updateAndGetAddedIndex();
+        // we must have a contiguous region in our rowset, there may be multiple ranges that are created; but they
+        // should all merge together, with only the first range actually producing any useful result
+        Require.requirement(selection.isContiguous(), "selection.isContiguous()");
+
+        synchronized (this) {
+            if (range == null) {
+                range = new Range(selection.firstRowKey(), selection.lastRowKey());
+            } else if (!range.merge(selection.firstRowKey(), selection.lastRowKey())) {
+                if (range.isBefore(selection.firstRowKey())) {
+                    pendingRanges.add(new Range(selection.firstRowKey(), selection.lastRowKey()));
+                } else {
+                    pendingRanges.add(range);
+                    range = new Range(selection.firstRowKey(), selection.lastRowKey());
+                }
+            }
+            return updateAndGetAddedIndex();
+        }
     }
 
     @Override
     @Nullable
-    protected WritableRowSet updateAndGetAddedIndex() {
+    protected synchronized WritableRowSet updateAndGetAddedIndex() {
         if (range == null || range.isEmpty()) {
             return null;
         }
         final RowSetBuilderRandom addedBuilder =
                 range.consumeKeysAndAppendAdded(nanosColumnSource, clock.currentTimeNanos(), null);
+        if (range.isEmpty()) {
+            pendingRanges.sort(Comparator.comparingLong(Range::firstKey));
+            while (range.isEmpty() && !pendingRanges.isEmpty()) {
+                range = pendingRanges.remove(0);
+                range.consumeKeysAndAppendAdded(nanosColumnSource, clock.currentTimeNanos(), addedBuilder);
+            }
+        }
         return addedBuilder == null ? null : addedBuilder.build();
     }
-
-    @Override
-    public boolean permitParallelization() {
-        return false;
-    }
 }

engine/table/src/main/java/io/deephaven/engine/table/impl/select/UnsortedClockFilter.java

@@ -69,12 +69,16 @@ public int compare(final Range r1, final Range r2) {
     protected WritableRowSet initializeAndGetInitialIndex(@NotNull final RowSet selection,
             @NotNull final RowSet fullSet,
             @NotNull final Table table) {
-        rangesByNextTime = new PriorityQueue<>(INITIAL_RANGE_QUEUE_CAPACITY, new RangeComparator());
-
         if (selection.isEmpty()) {
             return null;
         }
 
+        synchronized (this) {
+            if (rangesByNextTime == null) {
+                rangesByNextTime = new PriorityQueue<>(INITIAL_RANGE_QUEUE_CAPACITY, new RangeComparator());
+            }
+        }
+
         final RowSetBuilderSequential addedBuilder = RowSetFactory.builderSequential();
 
         final long nowNanos = clock.currentTimeNanos();
@@ -98,7 +102,9 @@ protected WritableRowSet initializeAndGetInitialIndex(@NotNull final RowSet sele
                     || QueryLanguageFunctionUtils.less(currentValue, previousValue)) {
                 // Add the current range, as appropriate
                 if (activeRangeIsDeferred) {
-                    rangesByNextTime.add(new Range(activeRangeFirstKey, activeRangeLastKey));
+                    synchronized (this) {
+                        rangesByNextTime.add(new Range(activeRangeFirstKey, activeRangeLastKey));
+                    }
                 } else {
                     addedBuilder.appendRange(activeRangeFirstKey, activeRangeLastKey);
                 }
@@ -113,7 +119,9 @@ protected WritableRowSet initializeAndGetInitialIndex(@NotNull final RowSet sele
 
         // Add the final range, as appropriate
         if (activeRangeIsDeferred) {
-            rangesByNextTime.add(new Range(activeRangeFirstKey, activeRangeLastKey));
+            synchronized (this) {
+                rangesByNextTime.add(new Range(activeRangeFirstKey, activeRangeLastKey));
+            }
         } else {
             addedBuilder.appendRange(activeRangeFirstKey, activeRangeLastKey);
         }
@@ -141,9 +149,4 @@ protected WritableRowSet updateAndGetAddedIndex() {
         }
         return addedBuilder == null ? null : addedBuilder.build();
     }
-
-    @Override
-    public boolean permitParallelization() {
-        return false;
-    }
 }

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

@@ -0,0 +1,45 @@
+//
+// Copyright (c) 2016-2026 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.engine.table.impl;
+
+import io.deephaven.util.thread.ThreadInitializationFactory;
+
+import java.util.Random;
+import java.util.concurrent.Future;
+
+/**
+ * A version of the operation initialization thread pool that sleeps for a random amount of time before executing the
+ * operation initializer; this is intended for unit tests to allow some perturbation for parallel tasks to get better
+ * coverage.
+ */
+public class RandomizedOperationInitializationThreadPool extends OperationInitializationThreadPool {
+    private final Random random = new Random();
+    private final int sleepMillis;
+
+    /**
+     * Create the thread pool.
+     * 
+     * @param factory the thread initialization factory
+     * @param sleepMillis the maximum number of milliseconds to sleep before executing the operation initializer. Values
+     *        are randomly distributed between 0 and this value.
+     */
+    public RandomizedOperationInitializationThreadPool(ThreadInitializationFactory factory, int sleepMillis) {
+        super(factory);
+        this.sleepMillis = sleepMillis;
+    }
+
+    @Override
+    public Future<?> submit(Runnable runnable) {
+        return super.submit(() -> {
+            final int toSleep = random.nextInt(sleepMillis);
+            try {
+                if (toSleep > 0) {
+                    Thread.sleep(toSleep);
+                }
+            } catch (InterruptedException ignored) {
+            }
+            runnable.run();
+        });
+    }
+}