Push user where filter into scan when blocking dedup pushdown

PPLSimplifyDedupRule correctly produces Dedup -> Filter(user where) -> Scan
when a `where` precedes `dedup`. The Filter between Dedup and Scan blocks
DedupPushdownRule's strict Dedup -> Project -> Scan operand chain, so
Volcano falls back to PPLDedupConvertRule and the plan ends up with an
in-memory ROW_NUMBER window instead of the pushed-down composite +
top_hits aggregation.

Add a WITH_FILTER operand variant to DedupPushdownRule that matches
Dedup -> Filter -> Scan, pushes the filter into the scan, then runs the
standard apply() on the resulting Dedup -> Project -> Scan shape.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Jialiang Liang <jiallian@amazon.com>

Author: RyanL1997

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java

@@ -2154,10 +2154,10 @@ public void testTransposeExplain() throws IOException {
   }
 
   /**
-   * Regression test for #5482: a user {@code where} clause adjacent to dedup's bucket-non-null
-   * filter must still let {@code PPLSimplifyDedupRule} fire even after {@code FilterMergeRule} has
-   * merged the two filters into a single AND. Without the fix, the plan keeps the in-memory {@code
-   * ROW_NUMBER} window over a row-fetching scan and dedup pushdown is silently disabled.
+   * Regression test for #5482: with a user {@code where} clause preceding {@code dedup}, the
+   * physical plan must push both the filter and the dedup-as-aggregation into the OpenSearch
+   * scan. Before the fix, the simplify rule didn't match the merged-AND bucket-non-null shape and
+   * the plan kept an in-memory {@code ROW_NUMBER} window above a row-fetching scan.
    */
   @Test
   public void testDedupAfterWherePushDown() throws IOException {
@@ -2166,12 +2166,13 @@ public void testDedupAfterWherePushDown() throws IOException {
         explainQueryToString(
             "source=opensearch-sql_test_index_account | where age > 25 | dedup gender");
     assertTrue(
-        "Expected LogicalDedup in the plan (simplify rule should fire after FilterMergeRule):\n"
-            + result,
-        result.contains("LogicalDedup"));
+        "Expected user where filter pushed down to the scan:\n" + result,
+        result.contains("FILTER->>($8, 25)"));
+    assertTrue(
+        "Expected dedup pushed down as AGGREGATION (composite + top_hits):\n" + result,
+        result.contains("AGGREGATION->"));
     assertFalse(
-        "Unexpected EnumerableWindow in the plan — dedup fell back to the in-memory ROW_NUMBER"
-            + " form, which means PPLSimplifyDedupRule did not fire:\n"
+        "Unexpected EnumerableWindow — dedup fell back to the in-memory ROW_NUMBER form:\n"
             + result,
         result.contains("EnumerableWindow"));
   }

opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/DedupPushdownRule.java

@@ -13,10 +13,12 @@
 import java.util.stream.IntStream;
 import javax.annotation.Nullable;
 import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rel.AbstractRelNode;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelCollations;
 import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.logical.LogicalAggregate;
+import org.apache.calcite.rel.logical.LogicalFilter;
 import org.apache.calcite.rel.logical.LogicalProject;
 import org.apache.calcite.rel.rules.SubstitutionRule;
 import org.apache.calcite.rex.RexInputRef;
@@ -47,6 +49,24 @@ protected DedupPushdownRule(Config config) {
   @Override
   protected void onMatchImpl(RelOptRuleCall call) {
     final LogicalDedup logicalDedup = call.rel(0);
+    if (call.rels[1] instanceof LogicalFilter filter) {
+      // Push the filter into the scan, then synthesize an identity project so the standard
+      // apply() can run on the resulting Dedup → Project → Scan shape.
+      final CalciteLogicalIndexScan scan = call.rel(2);
+      AbstractRelNode scanWithFilter = scan.pushDownFilter(filter);
+      if (scanWithFilter == null) {
+        return;
+      }
+      CalciteLogicalIndexScan newScan = (CalciteLogicalIndexScan) scanWithFilter;
+      RelBuilder relBuilder = call.builder();
+      relBuilder.push(newScan);
+      // force=true so the identity project is materialized (apply() requires a LogicalProject)
+      relBuilder.project(
+          relBuilder.fields(), newScan.getRowType().getFieldNames(), /* force= */ true);
+      LogicalProject identityProject = (LogicalProject) relBuilder.build();
+      apply(call, logicalDedup, identityProject, newScan);
+      return;
+    }
     final LogicalProject projectWithExpr = call.rel(1);
     final CalciteLogicalIndexScan scan = call.rel(2);
     apply(call, logicalDedup, projectWithExpr, scan);
@@ -226,6 +246,27 @@ public interface Config extends OpenSearchRuleConfig {
                                                 .predicate(Config::tableScanChecker)
                                                 .noInputs())));
 
+    // +- LogicalDedup
+    //    +- LogicalFilter      (e.g. a `where` predicate preserved below dedup by
+    //    +- CalciteLogicalIndexScan   PPLSimplifyDedupRule when an upstream `where` is folded
+    //                                 into the bucket-non-null filter)
+    Config WITH_FILTER =
+        ImmutableDedupPushdownRule.Config.builder()
+            .build()
+            .withDescription("Dedup-to-Aggregate-WithFilter")
+            .withOperandSupplier(
+                b0 ->
+                    b0.operand(LogicalDedup.class)
+                        .predicate(dedup -> !dedup.getKeepEmpty())
+                        .oneInput(
+                            b1 ->
+                                b1.operand(LogicalFilter.class)
+                                    .oneInput(
+                                        b2 ->
+                                            b2.operand(CalciteLogicalIndexScan.class)
+                                                .predicate(Config::tableScanChecker)
+                                                .noInputs())));
+
     /**
      * Project must be not pushed since the name of expression would lose after project pushed. E.g.
      * in query "eval new_a = a + 1 | dedup b", the "new_a" will lose.

opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/OpenSearchIndexRules.java

@@ -50,6 +50,8 @@ public class OpenSearchIndexRules {
       SortIndexScanRule.Config.DEFAULT.toRule();
   private static final DedupPushdownRule DEDUP_PUSH_DOWN =
       DedupPushdownRule.Config.DEFAULT.toRule();
+  private static final DedupPushdownRule DEDUP_PUSH_DOWN_WITH_FILTER =
+      DedupPushdownRule.Config.WITH_FILTER.toRule();
   private static final SortProjectExprTransposeRule SORT_PROJECT_EXPR_TRANSPOSE =
       SortProjectExprTransposeRule.Config.DEFAULT.toRule();
   private static final ExpandCollationOnProjectExprRule EXPAND_COLLATION_ON_PROJECT_EXPR =
@@ -75,6 +77,7 @@ public class OpenSearchIndexRules {
           LIMIT_INDEX_SCAN,
           SORT_INDEX_SCAN,
           DEDUP_PUSH_DOWN,
+          DEDUP_PUSH_DOWN_WITH_FILTER,
           SORT_PROJECT_EXPR_TRANSPOSE,
           SORT_AGGREGATION_METRICS_RULE,
           RARE_TOP_PUSH_DOWN,

ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLDedupTest.java

@@ -449,6 +449,38 @@ public void testDedupAfterWhereProducesLogicalDedupRegardlessOfRuleOrder() {
         optimizedExplain.contains(">($5, 1000)"));
   }
 
+  /**
+   * Mirrors the exact shape of {@code CalciteToolsHelper.HEP_PROGRAM} used in production
+   * ({@code addRuleCollection(List.of(FilterMergeRule, PPLSimplifyDedupRule))}). The two
+   * sequenced-{@code addRuleInstance} tests above prove the rule fires under either ordering,
+   * but production runs both rules in a single collection instruction. This test pins that exact
+   * shape to catch any addRuleCollection-vs-addRuleInstance traversal differences.
+   */
+  @Test
+  public void testDedupAfterWhereProducesLogicalDedupWithProductionHepProgram() {
+    String ppl = "source=EMP | where SAL > 1000 | dedup DEPTNO";
+    RelNode raw = getRelNodeRaw(ppl);
+
+    HepProgram program =
+        new HepProgramBuilder()
+            .addRuleCollection(
+                java.util.List.of(
+                    FilterMergeRule.Config.DEFAULT.toRule(),
+                    PPLSimplifyDedupRule.DEDUP_SIMPLIFY_RULE))
+            .build();
+    RelNode optimized = runHepPlanner(raw, program);
+
+    String optimizedExplain = optimized.explain();
+    Assert.assertTrue(
+        "Production HEP_PROGRAM (addRuleCollection) must still produce LogicalDedup:\n"
+            + optimizedExplain,
+        optimizedExplain.contains("LogicalDedup"));
+    Assert.assertFalse(
+        "ROW_NUMBER window form should be removed under production HEP_PROGRAM:\n"
+            + optimizedExplain,
+        optimizedExplain.contains("ROW_NUMBER"));
+  }
+
   private static RelNode runHepPlanner(RelNode root, HepProgram program) {
     HepPlanner planner = new HepPlanner(program);
     planner.setRoot(root);