fix: DH-20680: Reindexing filters to support barriers (#7547)

ReindexingFilters are treated as "serial", but the existing code did not
handle barriers properly when a RIF was part of a list of filters. This
was due to the independent execution of subsets of filters preceding or
following the RIF.

This code tracks the encountered filter barriers and provides them to
subsequent filter executions so the respected barrier requirements can
be met during execution.

Author: lbooker42

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

@@ -31,10 +31,7 @@
 import io.deephaven.engine.table.hierarchical.TreeTable;
 import io.deephaven.engine.table.impl.MemoizedOperationKey.SelectUpdateViewOrUpdateView.Flavor;
 import io.deephaven.engine.table.impl.by.*;
-import io.deephaven.engine.table.impl.filter.ExtractBarriers;
-import io.deephaven.engine.table.impl.filter.ExtractInnerConjunctiveFilters;
-import io.deephaven.engine.table.impl.filter.ExtractShiftedColumnDefinitions;
-import io.deephaven.engine.table.impl.filter.ExtractRespectedBarriers;
+import io.deephaven.engine.table.impl.filter.*;
 import io.deephaven.engine.table.impl.hierarchical.RollupTableImpl;
 import io.deephaven.engine.table.impl.hierarchical.TreeTableImpl;
 import io.deephaven.engine.table.impl.indexer.DataIndexer;
@@ -1332,6 +1329,21 @@ private void initializeFilters(@NotNull final WhereFilter[] filters) {
         compilationProcessor.compile();
     }
 
+    /**
+     * Create a copy of the array of filters, introducing the declared barriers into the first filter if any are
+     * provided.
+     */
+    private WhereFilter[] cloneFiltersForReindexing(final WhereFilter[] filters,
+            final Collection<Object> declaredBarriers) {
+        if (filters.length == 0 || declaredBarriers.isEmpty()) {
+            return filters;
+        }
+        final WhereFilter[] toUse = filters.clone();
+        // Force the first filter to declare all the barriers we have already processed.
+        toUse[0] = toUse[0].withDeclaredBarriers(declaredBarriers.toArray());
+        return toUse;
+    }
+
     private QueryTable whereInternal(final WhereFilter... filters) {
         if (filters.length == 0) {
             return (QueryTable) prepareReturnThis();
@@ -1346,32 +1358,66 @@ private QueryTable whereInternal(final WhereFilter... filters) {
                 () -> {
                     initializeFilters(extractedFilters);
 
-                    for (int fi = 0; fi < extractedFilters.length; ++fi) {
-                        if (!(extractedFilters[fi] instanceof ReindexingFilter)) {
-                            continue;
-                        }
-                        final ReindexingFilter reindexingFilter = (ReindexingFilter) extractedFilters[fi];
-                        final boolean first = fi == 0;
-                        final boolean last = fi == extractedFilters.length - 1;
-                        if (last && !reindexingFilter.requiresSorting()) {
-                            // If this is the last (or only) filter, we can just run it as normal unless it requires
-                            // sorting.
-                            break;
-                        }
-                        QueryTable result = this;
-                        if (!first) {
-                            result = result.whereInternal(Arrays.copyOf(extractedFilters, fi));
-                        }
-                        if (reindexingFilter.requiresSorting()) {
-                            result = (QueryTable) result.sort(reindexingFilter.getSortColumns());
-                            reindexingFilter.sortingDone();
-                        }
-                        result = result.whereInternal(reindexingFilter);
-                        if (!last) {
-                            result = result.whereInternal(
-                                    Arrays.copyOfRange(extractedFilters, fi + 1, extractedFilters.length));
+                    final boolean hasReindexingFilter = Arrays.stream(extractedFilters)
+                            .anyMatch(filter -> !ExtractReindexingFilters.of(filter).isEmpty());
+                    if (hasReindexingFilter) {
+                        // Filters will be processed in subsets of the provided list. Need to track all the
+                        // declared barriers that we have encountered.
+                        final Set<Object> declaredBarriers = new HashSet<>();
+
+                        for (int fi = 0; fi < extractedFilters.length; ++fi) {
+                            final WhereFilter filter = extractedFilters[fi];
+                            final Collection<ReindexingFilter> reindexingFilters = ExtractReindexingFilters.of(filter);
+                            if (reindexingFilters.isEmpty()) {
+                                continue;
+                            }
+                            if (reindexingFilters.size() > 1) {
+                                throw new IllegalStateException(
+                                        "Expected at most one reindexing filter per component filter: " + filter);
+                            }
+                            final ReindexingFilter reindexingFilter = reindexingFilters.iterator().next();
+                            final boolean first = fi == 0;
+                            final boolean last = fi == extractedFilters.length - 1;
+                            if (last && !reindexingFilter.requiresSorting()) {
+                                // If this is the last (or only) filter, we can just run it as normal unless it requires
+                                // sorting.
+                                break;
+                            }
+                            QueryTable result = this;
+                            if (!first) {
+                                final WhereFilter[] subset = Arrays.copyOf(extractedFilters, fi);
+                                final WhereFilter[] toUse = cloneFiltersForReindexing(subset, declaredBarriers);
+                                result = result.whereInternal(toUse);
+                                // collect all the newly declared barriers
+                                for (final WhereFilter priorFilter : subset) {
+                                    declaredBarriers.addAll(ExtractBarriers.of(priorFilter));
+                                }
+                            }
+                            if (reindexingFilter.requiresSorting()) {
+                                result = (QueryTable) result.sort(reindexingFilter.getSortColumns());
+                                reindexingFilter.sortingDone();
+                            }
+                            // Execute the current filter (which is or wraps a reindexing filter)
+                            // adding all previous encountered barriers as declared barriers.
+                            if (!declaredBarriers.isEmpty()) {
+                                result = result.whereInternal(filter.withDeclaredBarriers(declaredBarriers.toArray()));
+                            } else {
+                                result = result.whereInternal(filter);
+                            }
+                            // Add the barriers from the current filter.
+                            declaredBarriers.addAll(ExtractBarriers.of(filter));
+                            if (!last) {
+                                final WhereFilter[] subset =
+                                        Arrays.copyOfRange(extractedFilters, fi + 1, extractedFilters.length);
+                                final WhereFilter[] toUse = cloneFiltersForReindexing(subset, declaredBarriers);
+                                result = result.whereInternal(toUse);
+                                // collect all the new barriers that may have been processed
+                                for (final WhereFilter priorFilter : subset) {
+                                    declaredBarriers.addAll(ExtractBarriers.of(priorFilter));
+                                }
+                            }
+                            return result;
                         }
-                        return result;
                     }
 
                     boolean hasConstArrayOffsetFilter = false;

engine/table/src/main/java/io/deephaven/engine/table/impl/filter/ExtractReindexingFilters.java

@@ -0,0 +1,68 @@
+//
+// Copyright (c) 2016-2025 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.engine.table.impl.filter;
+
+import io.deephaven.engine.table.impl.select.*;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * This will return the ReindexingFilter instances contained within a WhereFilter, if any exist. Note that this removes
+ * all inversion, barrier and serial wrappers and the returned filters should not be used directly for filtering.
+ */
+public enum ExtractReindexingFilters implements WhereFilter.Visitor<Collection<ReindexingFilter>> {
+    EXTRACT_REINDEXING_FILTERS;
+
+    public static Collection<ReindexingFilter> of(final WhereFilter filter) {
+        return filter.walk(EXTRACT_REINDEXING_FILTERS);
+    }
+
+    @Override
+    public Collection<ReindexingFilter> visitOther(final WhereFilter filter) {
+        if (filter instanceof ReindexingFilter) {
+            return List.of((ReindexingFilter) filter);
+        }
+        return List.of();
+    }
+
+    @Override
+    public Collection<ReindexingFilter> visit(final WhereFilterInvertedImpl filter) {
+        return of(filter.getWrappedFilter());
+    }
+
+    @Override
+    public Collection<ReindexingFilter> visit(final WhereFilterSerialImpl filter) {
+        return of(filter.getWrappedFilter());
+    }
+
+    @Override
+    public Collection<ReindexingFilter> visit(final WhereFilterWithDeclaredBarriersImpl filter) {
+        return of(filter.getWrappedFilter());
+    }
+
+    @Override
+    public Collection<ReindexingFilter> visit(final WhereFilterWithRespectedBarriersImpl filter) {
+        return of(filter.getWrappedFilter());
+    }
+
+    @Override
+    public Collection<ReindexingFilter> visit(final DisjunctiveFilter filter) {
+        // DisjunctiveFilter should disallow reindexing filter components, but visit the children anyway because
+        // this could change.
+        return filter.getFilters().stream()
+                .flatMap(whereFilter -> of(whereFilter).stream())
+                .collect(Collectors.toList());
+    }
+
+    @Override
+    public Collection<ReindexingFilter> visit(final ConjunctiveFilter filter) {
+        // ConjunctiveFilter should disallow reindexing filter components, but visit the children anyway because
+        // this could change.
+        return filter.getFilters().stream()
+                .flatMap(whereFilter -> of(whereFilter).stream())
+                .collect(Collectors.toList());
+    }
+}

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

@@ -12,6 +12,10 @@
 public class WhereFilterInvertedImpl extends WhereFilterDelegatingBase {
 
     public static WhereFilter of(WhereFilter filter) {
+        if (filter instanceof ReindexingFilter) {
+            throw new UnsupportedOperationException(
+                    "WhereFilterInvertedImpl does not support ReindexingFilters: " + filter);
+        }
         return new WhereFilterInvertedImpl(filter);
     }