[BugFix] Fix Regex OOM when there are 10+ regex clauses

Signed-off-by: Jialiang Liang <jiallian@amazon.com>

Author: RyanL1997

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

@@ -8,6 +8,7 @@
 import static org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.TYPE_FACTORY;
 
 import java.sql.Connection;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -61,6 +62,12 @@ public class CalcitePlanContext {
 
   @Getter public Map<String, RexLambdaRef> rexLambdaRefMap;
 
+  /** Accumulated filter conditions to prevent deep Filter node chains */
+  private final List<RexNode> pendingFilterConditions = new ArrayList<>();
+
+  /** Flag to indicate if filter accumulation mode is active */
+  @Getter @Setter private boolean filterAccumulationEnabled = false;
+
   private CalcitePlanContext(FrameworkConfig config, SysLimit sysLimit, QueryType queryType) {
     this.config = config;
     this.sysLimit = sysLimit;
@@ -134,4 +141,52 @@ public static boolean isLegacyPreferred() {
   public void putRexLambdaRefMap(Map<String, RexLambdaRef> candidateMap) {
     this.rexLambdaRefMap.putAll(candidateMap);
   }
+
+  /**
+   * Adds a filter condition to the accumulation list instead of creating immediate Filter RelNode.
+   * This prevents deep Filter node chains that cause memory explosion.
+   */
+  public void addFilterCondition(RexNode condition) {
+    pendingFilterConditions.add(condition);
+  }
+
+  /**
+   * Applies all accumulated filter conditions as a single Filter RelNode with AND operations. This
+   * creates a single Filter node instead of a deep chain of Filter nodes.
+   */
+  public void flushFilterConditions() {
+    if (pendingFilterConditions.isEmpty()) {
+      return;
+    }
+
+    if (pendingFilterConditions.size() == 1) {
+      relBuilder.filter(pendingFilterConditions.get(0));
+    } else {
+      // Combine all filter conditions with AND
+      RexNode combinedCondition = relBuilder.and(pendingFilterConditions);
+      relBuilder.filter(combinedCondition);
+    }
+    pendingFilterConditions.clear();
+  }
+
+  /**
+   * Enables filter accumulation mode to prevent deep Filter node chains. Should be called before
+   * processing multiple filter operations.
+   */
+  public void enableFilterAccumulation() {
+    filterAccumulationEnabled = true;
+  }
+
+  /**
+   * Disables filter accumulation mode. Should be called after processing multiple filter
+   * operations.
+   */
+  public void disableFilterAccumulation() {
+    filterAccumulationEnabled = false;
+  }
+
+  /** Returns true if there are pending filter conditions that need to be flushed. */
+  public boolean hasPendingFilterConditions() {
+    return !pendingFilterConditions.isEmpty();
+  }
 }

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

@@ -177,7 +177,20 @@ public CalciteRelNodeVisitor(DataSourceService dataSourceService) {
   }
 
   public RelNode analyze(UnresolvedPlan unresolved, CalcitePlanContext context) {
-    return unresolved.accept(this, context);
+    // Enable filter accumulation if this plan contains multiple filtering operations
+    // that could create deep Filter RelNode chains
+    if (countFilteringOperations(unresolved) >= 2) {
+      context.enableFilterAccumulation();
+      try {
+        unresolved.accept(this, context);
+        context.flushFilterConditions(); // Flush accumulated conditions before returning
+        return context.relBuilder.peek(); // Get the result after flushing
+      } finally {
+        context.disableFilterAccumulation();
+      }
+    } else {
+      return unresolved.accept(this, context);
+    }
   }
 
   @Override
@@ -241,7 +254,12 @@ public RelNode visitFilter(Filter node, CalcitePlanContext context) {
       context.relBuilder.filter(ImmutableList.of(v.get().id), condition);
       context.popCorrelVar();
     } else {
-      context.relBuilder.filter(condition);
+      // Use filter accumulation to prevent deep Filter node chains
+      if (context.isFilterAccumulationEnabled()) {
+        context.addFilterCondition(condition);
+      } else {
+        context.relBuilder.filter(condition);
+      }
     }
     return context.relBuilder.peek();
   }
@@ -290,7 +308,12 @@ public RelNode visitRegex(Regex node, CalcitePlanContext context) {
       regexCondition = context.rexBuilder.makeCall(SqlStdOperatorTable.NOT, regexCondition);
     }
 
-    context.relBuilder.filter(regexCondition);
+    // Use filter accumulation to prevent deep Filter node chains
+    if (context.isFilterAccumulationEnabled()) {
+      context.addFilterCondition(regexCondition);
+    } else {
+      context.relBuilder.filter(regexCondition);
+    }
     return context.relBuilder.peek();
   }
 
@@ -381,6 +404,11 @@ private boolean containsSubqueryExpression(Node expr) {
   public RelNode visitProject(Project node, CalcitePlanContext context) {
     visitChildren(node, context);
 
+    // Flush accumulated filter conditions before schema-changing operations
+    if (context.isFilterAccumulationEnabled() && context.hasPendingFilterConditions()) {
+      context.flushFilterConditions();
+    }
+
     if (isSingleAllFieldsProject(node)) {
       return handleAllFieldsProject(node, context);
     }
@@ -3237,4 +3265,36 @@ private RexNode createOptimizedTransliteration(
       throw new RuntimeException("Failed to optimize sed expression: " + sedExpression, e);
     }
   }
+
+  /**
+   * Counts the number of filtering operations in an UnresolvedPlan tree that would create Filter
+   * RelNodes. This is used to detect queries with multiple regex/filter operations that could cause
+   * deep Filter RelNode chains and memory exhaustion.
+   *
+   * @param plan the UnresolvedPlan to analyze
+   * @return the count of filtering operations found
+   */
+  private int countFilteringOperations(UnresolvedPlan plan) {
+    if (plan == null) {
+      return 0;
+    }
+
+    int count = 0;
+
+    // Count this node if it's a filtering operation
+    if (plan instanceof Regex || plan instanceof Filter) {
+      count = 1;
+    }
+
+    // Recursively count filtering operations in children
+    if (plan.getChild() != null) {
+      for (Node child : plan.getChild()) {
+        if (child instanceof UnresolvedPlan) {
+          count += countFilteringOperations((UnresolvedPlan) child);
+        }
+      }
+    }
+
+    return count;
+  }
 }