Directly apply Calcite CoreRules.FILTER_MERGE before VolcanoPlanner plan

Co-authored-by: Peng Huo <penghuo@gmail.com>
Signed-off-by: Jialiang Liang <jiallian@amazon.com>

Author: RyanL1997

core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java

@@ -176,13 +176,7 @@ public CalciteRelNodeVisitor(DataSourceService dataSourceService) {
   }
 
   public RelNode analyze(UnresolvedPlan unresolved, CalcitePlanContext context) {
-    // Build the RelNode tree (may contain deep Filter chains)
-    RelNode relNode = unresolved.accept(this, context);
-
-    // Apply filter merge optimization as post-processing
-    // This merges consecutive LogicalFilter nodes to prevent OOM with deep chains
-    FilterMergeVisitor filterMergeVisitor = new FilterMergeVisitor();
-    return relNode.accept(filterMergeVisitor);
+    return unresolved.accept(this, context);
   }
 
   @Override

core/src/main/java/org/opensearch/sql/calcite/FilterMergeVisitor.java

@@ -16,11 +16,15 @@
 import lombok.extern.log4j.Log4j2;
 import org.apache.calcite.jdbc.CalciteSchema;
 import org.apache.calcite.plan.RelTraitDef;
+import org.apache.calcite.plan.hep.HepPlanner;
+import org.apache.calcite.plan.hep.HepProgram;
+import org.apache.calcite.plan.hep.HepProgramBuilder;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelCollations;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rel.logical.LogicalSort;
+import org.apache.calcite.rel.rules.FilterMergeRule;
 import org.apache.calcite.schema.SchemaPlus;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.calcite.tools.FrameworkConfig;
@@ -100,6 +104,7 @@ public void executeWithCalcite(
                           CalcitePlanContext.create(
                               buildFrameworkConfig(), SysLimit.fromSettings(settings), queryType);
                       RelNode relNode = analyze(plan, context);
+                      relNode = mergeAdjacentFilters(relNode);
                       RelNode optimized = optimize(relNode, context);
                       RelNode calcitePlan = convertToCalcitePlan(optimized);
                       executionEngine.execute(calcitePlan, context, listener);
@@ -145,6 +150,7 @@ public void explainWithCalcite(
                       context.run(
                           () -> {
                             RelNode relNode = analyze(plan, context);
+                            relNode = mergeAdjacentFilters(relNode);
                             RelNode optimized = optimize(relNode, context);
                             RelNode calcitePlan = convertToCalcitePlan(optimized);
                             executionEngine.explain(calcitePlan, format, context, listener);
@@ -259,6 +265,18 @@ public RelNode analyze(UnresolvedPlan plan, CalcitePlanContext context) {
     return getRelNodeVisitor().analyze(plan, context);
   }
 
+  /**
+   * Run Calcite FILTER_MERGE once so adjacent filters created during analysis can collapse before
+   * the rest of optimization.
+   */
+  private RelNode mergeAdjacentFilters(RelNode relNode) {
+    HepProgram program =
+        new HepProgramBuilder().addRuleInstance(FilterMergeRule.Config.DEFAULT.toRule()).build();
+    HepPlanner planner = new HepPlanner(program);
+    planner.setRoot(relNode);
+    return planner.findBestExp();
+  }
+
   /** Analyze {@link UnresolvedPlan}. */
   public LogicalPlan analyze(UnresolvedPlan plan, QueryType queryType) {
     return analyzer.analyze(plan, new AnalysisContext(queryType));