feat: DH-21522: allow column region optimizations in Predicate Pushdown filtering

engine/chunk/src/main/java/io/deephaven/chunk/WritableBooleanChunk.java

@@ -10,7 +10,9 @@
 import io.deephaven.chunk.attributes.Any;
 import io.deephaven.chunk.util.pools.MultiChunkPool;
 
+import io.deephaven.function.ArraySort;
 import io.deephaven.util.type.TypeUtils;
+import org.jetbrains.annotations.NotNull;
 
 import java.util.Arrays;
 // region FillWithNullValueImports

engine/chunk/src/main/java/io/deephaven/chunk/WritableObjectChunk.java

@@ -11,6 +11,8 @@
 import io.deephaven.chunk.util.pools.MultiChunkPool;
 
 import io.deephaven.function.ArraySort;
+import io.deephaven.util.type.TypeUtils;
+import org.jetbrains.annotations.NotNull;
 
 import java.util.Arrays;
 // region FillWithNullValueImports

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

@@ -27,7 +27,7 @@
 /**
  * Base class for {@link PushdownFilterContext} to help with execution cost tracking.
  */
-public class BasePushdownFilterContext implements PushdownFilterContext {
+public abstract class BasePushdownFilterContext implements PushdownFilterContext {
 
     /**
      * Enum for the behavior of a filter when applied to null values.
@@ -49,12 +49,13 @@ public enum FilterNullBehavior {
         FAILS_ON_NULLS
     }
 
-    protected final WhereFilter filter;
+    private final WhereFilter filter;
+
     private final List<ColumnSource<?>> columnSources;
 
     private final boolean isRangeFilter;
     private final boolean isMatchFilter;
-    private final boolean supportsDictionaryFiltering;
+    private final boolean supportsChunkFiltering;
 
     private long executedFilterCost;
 
@@ -99,7 +100,7 @@ public BasePushdownFilterContext(
         // TODO (DH-19666): Multi column filters are not supported yet
         // Do not use columnSources.size(), multiple logical columns may alias (rename) the same physical column,
         // yielding a single entry.
-        supportsDictionaryFiltering = (isRangeFilter || isMatchFilter
+        supportsChunkFiltering = (isRangeFilter || isMatchFilter
                 || (isConditionFilter && ((ConditionFilter) filter).getNumInputsUsed() == 1))
                 && ((filter instanceof ExposesChunkFilter && ((ExposesChunkFilter) filter).chunkFilter().isPresent())
                         || isConditionFilter);
@@ -117,20 +118,27 @@ public void updateExecutedFilterCost(long executedFilterCost) {
         this.executedFilterCost = executedFilterCost;
     }
 
+    /**
+     * Gets the filter associated with this context.
+     */
+    public final WhereFilter filter() {
+        return filter;
+    }
+
     /**
      * Get the column sources this filter will use.
      */
-    public List<ColumnSource<?>> columnSources() {
+    public final List<ColumnSource<?>> columnSources() {
         return columnSources;
     }
 
     /**
-     * Whether this filter supports parquet dictionary filtering, which necessitates direct chunk filtering, i.e., it
-     * can be applied to a chunk of data rather than a table. This includes any filter that implements {#@link
-     * ExposesChunkFilter} or {@link ConditionFilter} with exactly one column.
+     * Whether this filter supports direct chunk filtering, i.e., it can be applied to a chunk of data rather than a
+     * table. This includes any filter that implements {@link ExposesChunkFilter} or {@link ConditionFilter} with
+     * exactly one column.
      */
-    public final boolean supportsDictionaryFiltering() {
-        return supportsDictionaryFiltering;
+    public final boolean supportsChunkFiltering() {
+        return supportsChunkFiltering;
     }
 
     /**
@@ -205,13 +213,13 @@ private FilterNullBehavior computeFilterNullBehavior() {
     /**
      * Create a {@link UnifiedChunkFilter} for the {@link WhereFilter} that efficiently filters chunks of data. Every
      * thread that uses this should create its own instance and must close it after use. Can only call when
-     * {@link #supportsDictionaryFiltering()} is {@code true}
+     * {@link #supportsChunkFiltering()} is {@code true}
      *
      * @param maxChunkSize the maximum size of the chunk that will be filtered
      * @return the initialized {@link UnifiedChunkFilter}
      */
     public final UnifiedChunkFilter createChunkFilter(final int maxChunkSize) {
-        if (!supportsDictionaryFiltering) {
+        if (!supportsChunkFiltering) {
             throw new IllegalStateException("Filter does not support chunk filtering: " + Strings.of(filter));
         }
         if (filter instanceof ExposesChunkFilter) {

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

@@ -43,14 +43,17 @@ public interface PushdownFilterMatcher {
      *        that the filter cannot be pushed down.
      * @param onError Consumer of any exceptions that occur during the estimate operation
      */
-    void estimatePushdownFilterCost(
+    default void estimatePushdownFilterCost(
             final WhereFilter filter,
             final RowSet selection,
             final boolean usePrev,
             final PushdownFilterContext context,
             final JobScheduler jobScheduler,
             final LongConsumer onComplete,
-            final Consumer<Exception> onError);
+            final Consumer<Exception> onError) {
+        // Default to having no benefit by pushing down.
+        onComplete.accept(Long.MAX_VALUE);
+    }
 
     /**
      * Push down the given filter to the underlying table and pass the result to the consumer. This method is expected
@@ -76,15 +79,18 @@ void estimatePushdownFilterCost(
      * @param onComplete Consumer of the output rowsets for added and modified rows that pass the filter
      * @param onError Consumer of any exceptions that occur during the pushdown operation
      */
-    void pushdownFilter(
+    default void pushdownFilter(
             final WhereFilter filter,
             final RowSet selection,
             final boolean usePrev,
             final PushdownFilterContext context,
             final long costCeiling,
             final JobScheduler jobScheduler,
             final Consumer<PushdownResult> onComplete,
-            final Consumer<Exception> onError);
+            final Consumer<Exception> onError) {
+        // Default to returning all results as "maybe"
+        onComplete.accept(PushdownResult.allMaybeMatch(selection));
+    }
 
     /**
      * Create a pushdown filter context for this entity.
@@ -94,9 +100,11 @@ void pushdownFilter(
      *
      * @return the created filter context
      */
-    PushdownFilterContext makePushdownFilterContext(
+    default PushdownFilterContext makePushdownFilterContext(
             final WhereFilter filter,
-            final List<ColumnSource<?>> filterSources);
+            final List<ColumnSource<?>> filterSources) {
+        return PushdownFilterContext.NO_PUSHDOWN_CONTEXT;
+    }
 
     /**
      * Given a filter and a list of column sources, return the appropriate {@link PushdownFilterMatcher} to use for

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

@@ -28,6 +28,10 @@ public final class PushdownResult implements SafeCloseable {
      * The entire column contains a single value, so a single read is sufficient to determine matches.
      */
     public static final long SINGLE_VALUE_COLUMN_COST = 1_000L;
+    /**
+     * The entire region contains a single value, so a single read is sufficient to determine matches.
+     */
+    public static final long SINGLE_VALUE_REGION_COST = 2_000L;
     /**
      * Only table/row-group statistics are checked, assuming the metadata is already loaded
      */

engine/table/src/main/java/io/deephaven/engine/table/impl/locations/TableLocation.java

@@ -9,7 +9,7 @@
 import io.deephaven.engine.liveness.LivenessReferent;
 import io.deephaven.engine.table.BasicDataIndex;
 import io.deephaven.engine.table.Table;
-import io.deephaven.engine.table.impl.PushdownFilterMatcher;
+import io.deephaven.engine.table.impl.sources.regioned.RegionedPushdownFilterMatcher;
 import io.deephaven.io.log.impl.LogOutputStringImpl;
 import io.deephaven.util.annotations.FinalDefault;
 import io.deephaven.util.type.NamedImplementation;
@@ -24,7 +24,8 @@
  * source table.
  */
 public interface TableLocation
-        extends NamedImplementation, LogOutputAppendable, TableLocationState, LivenessReferent, PushdownFilterMatcher {
+        extends NamedImplementation, LogOutputAppendable, TableLocationState, LivenessReferent,
+        RegionedPushdownFilterMatcher {
 
     /**
      * Listener interface for anything that wants to know about changes to a location.

engine/table/src/main/java/io/deephaven/engine/table/impl/locations/impl/AbstractTableLocation.java

@@ -10,6 +10,8 @@
 import io.deephaven.engine.table.impl.PushdownFilterContext;
 import io.deephaven.engine.table.impl.PushdownResult;
 import io.deephaven.engine.table.impl.select.WhereFilter;
+import io.deephaven.engine.table.impl.sources.regioned.RegionedPushdownAction;
+import io.deephaven.engine.table.impl.sources.regioned.RegionedPushdownFilterContext;
 import io.deephaven.engine.table.impl.util.FieldUtils;
 import io.deephaven.engine.table.impl.util.JobScheduler;
 import io.deephaven.engine.util.string.StringUtils;
@@ -25,10 +27,12 @@
 import java.lang.ref.SoftReference;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Consumer;
 import java.util.function.LongConsumer;
+import java.util.stream.Collectors;
 
 /**
  * Partial TableLocation implementation for use by TableDataService implementations.
@@ -300,6 +304,10 @@ public final BasicDataIndex getDataIndex(@NotNull final String... columns) {
     @Nullable
     public abstract BasicDataIndex loadDataIndex(@NotNull String... columns);
 
+    // ------------------------------------------------------------------------------------------------------------------
+    // PushdownFilterMatcher default implementation
+    // ------------------------------------------------------------------------------------------------------------------
+
     @Override
     public void estimatePushdownFilterCost(
             final WhereFilter filter,
@@ -309,8 +317,31 @@ public void estimatePushdownFilterCost(
             final JobScheduler jobScheduler,
             final LongConsumer onComplete,
             final Consumer<Exception> onError) {
-        // Default to having no benefit by pushing down.
-        onComplete.accept(Long.MAX_VALUE);
+        if (selection.isEmpty()) {
+            // If the selection is empty, we can skip all pushdown filtering.
+            onComplete.accept(Long.MAX_VALUE);
+            return;
+        }
+
+        final RegionedPushdownFilterContext filterCtx = (RegionedPushdownFilterContext) context;
+
+        // Generate a list of all the allowed actions, sorted by minimum cost.
+        final List<RegionedPushdownAction> sorted = supportedActions()
+                .stream()
+                .filter(action -> action.allows(this, filterCtx))
+                .sorted(Comparator.comparingLong(RegionedPushdownAction::filterCost))
+                .collect(Collectors.toList());
+
+        // If no modes are allowed, we can skip all pushdown filtering.
+        if (sorted.isEmpty()) {
+            onComplete.accept(Long.MAX_VALUE);
+            return;
+        }
+
+        // Delegate to TableLocation to determine which of the supported actions applies to this particular location.
+        try (final RegionedPushdownAction.EstimateContext estimateCtx = makeEstimateContext(filter, filterCtx)) {
+            onComplete.accept(estimatePushdownAction(sorted, filter, selection, usePrev, filterCtx, estimateCtx));
+        }
     }
 
     @Override
@@ -323,8 +354,44 @@ public void pushdownFilter(
             final JobScheduler jobScheduler,
             final Consumer<PushdownResult> onComplete,
             final Consumer<Exception> onError) {
-        // Default to returning all results as "maybe"
-        onComplete.accept(PushdownResult.allMaybeMatch(selection));
+        if (selection.isEmpty()) {
+            // If the selection is empty, we can skip all pushdown filtering.
+            onComplete.accept(PushdownResult.allMaybeMatch(selection));
+            return;
+        }
+
+        final RegionedPushdownFilterContext filterCtx = (RegionedPushdownFilterContext) context;
+
+        // Generate a list of all the supported allowed actions, sorted by minimum cost.
+        final List<RegionedPushdownAction> sorted = supportedActions()
+                .stream()
+                .filter(action -> action.allows(this, filterCtx, costCeiling))
+                .sorted(Comparator.comparingLong(RegionedPushdownAction::filterCost))
+                .collect(Collectors.toList());
+
+        // If no modes are allowed, we can skip all pushdown filtering.
+        if (sorted.isEmpty()) {
+            onComplete.accept(PushdownResult.allMaybeMatch(selection));
+            return;
+        }
+
+        // Initialize the pushdown result with the selection rowset as "maybe" rows
+        PushdownResult result = PushdownResult.allMaybeMatch(selection);
+
+        // Delegate to TableLocation and perform each supported action..
+        try (final RegionedPushdownAction.ActionContext actionCtx = makeActionContext(filter, filterCtx)) {
+            for (final RegionedPushdownAction action : sorted) {
+                try (final PushdownResult ignored = result) {
+                    result = performPushdownAction(action, filter, selection, result, usePrev, filterCtx, actionCtx);
+                }
+                if (result.maybeMatch().isEmpty()) {
+                    // No maybe rows remaining, so no reason to continue filtering.
+                    break;
+                }
+            }
+            // Return the final result
+            onComplete.accept(result);
+        }
     }
 
     @Override

engine/table/src/main/java/io/deephaven/engine/table/impl/sources/BooleanSingleValueSource.java

@@ -12,21 +12,13 @@
 
 import io.deephaven.chunk.WritableChunk;
 import io.deephaven.chunk.attributes.Values;
-import io.deephaven.engine.rowset.RowSet;
 import io.deephaven.engine.table.impl.MutableColumnSourceGetDefaults;
 import io.deephaven.engine.rowset.chunkattributes.RowKeys;
 import io.deephaven.chunk.Chunk;
 import io.deephaven.chunk.LongChunk;
 import io.deephaven.engine.rowset.RowSequence;
-import io.deephaven.engine.table.impl.PushdownFilterContext;
-import io.deephaven.engine.table.impl.PushdownResult;
-import io.deephaven.engine.table.impl.select.WhereFilter;
-import io.deephaven.engine.table.impl.util.JobScheduler;
 import org.jetbrains.annotations.NotNull;
 
-import java.util.function.Consumer;
-import java.util.function.LongConsumer;
-
 import static io.deephaven.util.QueryConstants.NULL_BOOLEAN;
 
 /**
@@ -159,31 +151,13 @@ public boolean providesFillUnordered() {
     }
 
     @Override
-    public void estimatePushdownFilterCost(
-            final WhereFilter filter,
-            final RowSet selection,
-            final boolean usePrev,
-            final PushdownFilterContext context,
-            final JobScheduler jobScheduler,
-            final LongConsumer onComplete,
-            final Consumer<Exception> onError) {
-        // Delegate to the shared code for RowKeyAgnosticChunkSource
-        RowKeyAgnosticChunkSource.estimatePushdownFilterCostHelper(
-                filter, selection, usePrev, context, jobScheduler, onComplete, onError);
+    public Chunk<Values> getValueChunk() {
+        return SingleValuePushdownHelper.makeChunk(get(0));
     }
 
     @Override
-    public void pushdownFilter(
-            final WhereFilter filter,
-            final RowSet selection,
-            final boolean usePrev,
-            final PushdownFilterContext context,
-            final long costCeiling,
-            final JobScheduler jobScheduler,
-            final Consumer<PushdownResult> onComplete,
-            final Consumer<Exception> onError) {
-        // Delegate to the shared code for RowKeyAgnosticChunkSource
-        RowKeyAgnosticChunkSource.pushdownFilterHelper(this, filter, selection, usePrev, context, costCeiling,
-                jobScheduler, onComplete, onError);
+    public Chunk<Values> getPrevValueChunk() {
+        // avoid duplicating the current vs prev logic in getPrev
+        return SingleValuePushdownHelper.makeChunk(getPrev(0));
     }
 }

engine/table/src/main/java/io/deephaven/engine/table/impl/sources/BooleanSparseArraySource.java

@@ -54,14 +54,18 @@
 public class BooleanSparseArraySource extends SparseArrayColumnSource<Boolean>
         implements MutableColumnSourceGetDefaults.ForBoolean /* MIXIN_IMPLS */ {
     // region recyclers
-    private static final SoftRecycler<byte[]> recycler = new SoftRecycler<>(SparseArrayColumnSourceConfiguration.BOOLEAN_RECYCLER_CAPACITY,
-            () -> new byte[BLOCK_SIZE], null);
-    private static final SoftRecycler<byte[][]> recycler2 = new SoftRecycler<>(SparseArrayColumnSourceConfiguration.BOOLEAN_RECYCLER_CAPACITY2,
-            () -> new byte[BLOCK2_SIZE][], null);
-    private static final SoftRecycler<ByteOneOrN.Block2[]> recycler1 = new SoftRecycler<>(SparseArrayColumnSourceConfiguration.BOOLEAN_RECYCLER_CAPACITY1,
-            () -> new ByteOneOrN.Block2[BLOCK1_SIZE], null);
-    private static final SoftRecycler<ByteOneOrN.Block1[]> recycler0 = new SoftRecycler<>(SparseArrayColumnSourceConfiguration.BOOLEAN_RECYCLER_CAPACITY0,
-            () -> new ByteOneOrN.Block1[BLOCK0_SIZE], null);
+    private static final SoftRecycler<byte[]> recycler =
+            new SoftRecycler<>(SparseArrayColumnSourceConfiguration.BOOLEAN_RECYCLER_CAPACITY,
+                    () -> new byte[BLOCK_SIZE], null);
+    private static final SoftRecycler<byte[][]> recycler2 =
+            new SoftRecycler<>(SparseArrayColumnSourceConfiguration.BOOLEAN_RECYCLER_CAPACITY2,
+                    () -> new byte[BLOCK2_SIZE][], null);
+    private static final SoftRecycler<ByteOneOrN.Block2[]> recycler1 =
+            new SoftRecycler<>(SparseArrayColumnSourceConfiguration.BOOLEAN_RECYCLER_CAPACITY1,
+                    () -> new ByteOneOrN.Block2[BLOCK1_SIZE], null);
+    private static final SoftRecycler<ByteOneOrN.Block1[]> recycler0 =
+            new SoftRecycler<>(SparseArrayColumnSourceConfiguration.BOOLEAN_RECYCLER_CAPACITY0,
+                    () -> new ByteOneOrN.Block1[BLOCK0_SIZE], null);
     // endregion recyclers
 
     /**