Lower structured search predicates to native RexCall in visitSearch

Previously `visitSearch` discarded `Search.getOriginalExpression()` and
emitted a single `query_string(...)` function call covering the whole
predicate. On the analytics-engine route every such filter is
delegation-routed to Lucene; on a parquet-only composite shard the
delegation handle NPEs in `LuceneAnalyticsBackendPlugin.getFilterDelegationHandle`
because there is no Lucene reader.

Walk the typed `SearchExpression` AST instead. Structured fragments —
field comparisons, AND, OR, NOT, IN, plus time-modifier-resolved
comparisons (`earliest=` / `latest=`) — become native PPL filter AST
(`Compare` / `And` / `Or` / `Not` / `In`) that the existing rexVisitor
handles like any `where`-clause condition. DataFusion executes them
natively against parquet without any Lucene round-trip.

Free-text / phrase literals stay in `query_string` form because they
have no native equivalent. Top-level `AND` mixes the two: the
structured conjunct is lowered, the relevance conjunct stays in
`query_string`, and the resulting `AND(native, query_string(...))`
lets the planner route each clause independently — DataFusion prunes
parquet, Lucene handles the relevance term against the secondary
format.

Three guards keep the lowering faithful to PPL search semantics:

- `containsLuceneWildcard` — `*` / `?` in a comparison value forces
  the fallback so Lucene wildcards keep working (`severityText=ERR*`).
- `isOpenSearchDateMath` — values produced by `visitTimeModifierExpression`
  (`now`, `now-1h`, `now+1y/M-1M`, epoch-millis strings) can't be
  parsed as Calcite timestamps; keep them in `query_string` so Lucene
  evaluates the date math.
- `isLowerableField` — comparisons against fields absent from the
  relation's row type fall through (matches Lucene's "missing field →
  no docs" semantics; native would error out in the analyzer).

`SearchNot` always falls back: Lucene `NOT field=value` matches docs
where the field is missing OR field != value, but
`NOT (field = value)` in Calcite simplifies to `field <> value`,
which drops missing-field rows since `NULL <> value` evaluates to
NULL.

Net `CalciteSearchCommandIT` impact (analytics-engine route, all OS-
side fixes in `opensearch-project/OpenSearch#21681` in place): 3/52 →
35/52. Tests that pass via the native path include all field-equality,
AND, OR, IN, range, double-comparison, attribute-field, and mixed
boolean tests.

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

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

@@ -102,6 +102,14 @@
 import org.opensearch.sql.ast.expression.PatternMethod;
 import org.opensearch.sql.ast.expression.PatternMode;
 import org.opensearch.sql.ast.expression.QualifiedName;
+import org.opensearch.sql.ast.expression.SearchAnd;
+import org.opensearch.sql.ast.expression.SearchComparison;
+import org.opensearch.sql.ast.expression.SearchExpression;
+import org.opensearch.sql.ast.expression.SearchGroup;
+import org.opensearch.sql.ast.expression.SearchIn;
+import org.opensearch.sql.ast.expression.SearchLiteral;
+import org.opensearch.sql.ast.expression.SearchNot;
+import org.opensearch.sql.ast.expression.SearchOr;
 import org.opensearch.sql.ast.expression.Span;
 import org.opensearch.sql.ast.expression.SpanUnit;
 import org.opensearch.sql.ast.expression.UnresolvedExpression;
@@ -280,17 +288,207 @@ private RelBuilder scan(RelOptTable tableSchema, CalcitePlanContext context) {
   public RelNode visitSearch(Search node, CalcitePlanContext context) {
     // Visit the Relation child to get the scan
     node.getChild().get(0).accept(this, context);
-    // Create query_string function
-    Function queryStringFunc =
-        AstDSL.function(
-            "query_string",
-            AstDSL.unresolvedArg("query", AstDSL.stringLiteral(node.getQueryString())));
-    RexNode queryStringRex = rexVisitor.analyze(queryStringFunc, context);
-
-    context.relBuilder.filter(queryStringRex);
+
+    // Lower the typed SearchExpression AST to a native filter where possible. The
+    // structured fragments (field comparisons, AND / OR / NOT, IN, time-modifier-resolved
+    // comparisons) become a plain Compare / And / Or / Not / In AST that
+    // ExpressionAnalyzer/rexVisitor handles like any `where`-clause condition, so the
+    // analytics-engine route can mark them as DataFusion-native predicates instead of
+    // delegating the whole thing to Lucene as a `query_string(...)`. Free-text / phrase
+    // literals — and any comparison whose value looks like an OpenSearch date-math
+    // expression or references a field absent from the relation's row type — stay in
+    // query_string form because they have no faithful native equivalent.
+    Set<String> knownFields = new HashSet<>(context.relBuilder.peek().getRowType().getFieldNames());
+    SearchExpression original = node.getOriginalExpression();
+    UnresolvedExpression filter =
+        original != null
+            ? buildSearchFilter(original, node.getQueryString(), knownFields)
+            : buildQueryStringFunction(node.getQueryString());
+    RexNode filterRex = rexVisitor.analyze(filter, context);
+
+    context.relBuilder.filter(filterRex);
     return context.relBuilder.peek();
   }
 
+  /**
+   * Build the filter AST for a search predicate. Prefers a fully-native lowering of {@code e};
+   * falls back to {@code AND}-splitting a top-level conjunction with one native and one free-text
+   * side; finally falls back to wrapping the original query string in {@code query_string(...)}.
+   */
+  private UnresolvedExpression buildSearchFilter(
+      SearchExpression e, String fallbackQueryString, Set<String> knownFields) {
+    Optional<UnresolvedExpression> structured = lowerSearchExpression(e, knownFields);
+    if (structured.isPresent()) {
+      return structured.get();
+    }
+    // Top-level AND-split: native(L) AND queryString(toQueryString(R)) (or symmetric).
+    // Only applied at the outermost conjunction — deeper splits would change OR semantics.
+    if (e instanceof SearchAnd) {
+      SearchAnd and = (SearchAnd) e;
+      Optional<UnresolvedExpression> leftNative = lowerSearchExpression(and.getLeft(), knownFields);
+      Optional<UnresolvedExpression> rightNative =
+          lowerSearchExpression(and.getRight(), knownFields);
+      if (leftNative.isPresent() && rightNative.isEmpty()) {
+        return AstDSL.and(
+            leftNative.get(), buildQueryStringFunction(and.getRight().toQueryString()));
+      }
+      if (rightNative.isPresent() && leftNative.isEmpty()) {
+        return AstDSL.and(
+            buildQueryStringFunction(and.getLeft().toQueryString()), rightNative.get());
+      }
+    }
+    // Fully relevance / mixed-OR / unsupported — keep the original query_string fallback.
+    return buildQueryStringFunction(fallbackQueryString);
+  }
+
+  /**
+   * Recursively translate a {@link SearchExpression} to the equivalent generic-PPL filter AST.
+   * Returns empty whenever any subtree contains a {@link SearchLiteral} (free text or phrase), a
+   * {@link SearchComparison} whose value looks like an OpenSearch date-math expression (handled
+   * correctly only by Lucene), or a comparison/IN against a field that isn't in the relation's row
+   * type (which the analyzer would otherwise reject; PPL search semantics on unknown fields is
+   * "match nothing", matched by the query_string fallback).
+   */
+  private Optional<UnresolvedExpression> lowerSearchExpression(
+      SearchExpression e, Set<String> knownFields) {
+    if (e instanceof SearchGroup) {
+      return lowerSearchExpression(((SearchGroup) e).getExpression(), knownFields);
+    }
+    if (e instanceof SearchAnd) {
+      SearchAnd and = (SearchAnd) e;
+      Optional<UnresolvedExpression> left = lowerSearchExpression(and.getLeft(), knownFields);
+      Optional<UnresolvedExpression> right = lowerSearchExpression(and.getRight(), knownFields);
+      if (left.isPresent() && right.isPresent()) {
+        return Optional.of(AstDSL.and(left.get(), right.get()));
+      }
+      return Optional.empty();
+    }
+    if (e instanceof SearchOr) {
+      SearchOr or = (SearchOr) e;
+      Optional<UnresolvedExpression> left = lowerSearchExpression(or.getLeft(), knownFields);
+      Optional<UnresolvedExpression> right = lowerSearchExpression(or.getRight(), knownFields);
+      if (left.isPresent() && right.isPresent()) {
+        return Optional.of(AstDSL.or(left.get(), right.get()));
+      }
+      return Optional.empty();
+    }
+    if (e instanceof SearchNot) {
+      // Lucene `NOT field=value` matches docs where the field is missing OR field != value.
+      // SQL `NOT (field = value)` (and Calcite's simplification to `field <> value`) drops
+      // missing-field rows because NULL <> value evaluates to NULL → filtered. Routing NOT
+      // through the query_string fallback preserves the documented PPL search semantics.
+      return Optional.empty();
+    }
+    if (e instanceof SearchComparison) {
+      SearchComparison comp = (SearchComparison) e;
+      if (!isLowerableField(comp.getField(), knownFields)) {
+        return Optional.empty();
+      }
+      if (isOpenSearchDateMath(comp.getValue().getLiteral())) {
+        return Optional.empty();
+      }
+      if (containsLuceneWildcard(comp.getValue().getLiteral())) {
+        // `severityText=ERR*` / `field=foo?` / `name=*-service` — Lucene-style wildcards in
+        // the right-hand value. A native `=` lowering would compare literally and drop every
+        // matching document. Keep the query in query_string form so Lucene evaluates the
+        // wildcard.
+        return Optional.empty();
+      }
+      return Optional.of(
+          AstDSL.compare(
+              comp.getOperator().getSymbol(), comp.getField(), comp.getValue().getLiteral()));
+    }
+    if (e instanceof SearchIn) {
+      SearchIn in = (SearchIn) e;
+      if (!isLowerableField(in.getField(), knownFields)) {
+        return Optional.empty();
+      }
+      List<UnresolvedExpression> values =
+          in.getValues().stream().map(SearchLiteral::getLiteral).collect(Collectors.toList());
+      return Optional.of(AstDSL.in(in.getField(), values));
+    }
+    // SearchLiteral (standalone free-text / phrase) — no native equivalent.
+    return Optional.empty();
+  }
+
+  private static boolean isLowerableField(Field field, Set<String> knownFields) {
+    QualifiedName qn = (QualifiedName) field.getField();
+    return knownFields.contains(qn.toString());
+  }
+
+  /**
+   * Recognise OpenSearch date-math values produced by {@code visitTimeModifierExpression} — {@code
+   * "now"}, {@code "now-1h"}, {@code "now+1y/M-1M"}, anchored expressions like {@code
+   * "2024-01-15||+1d"}, and epoch-millis strings such as {@code "1754020060000"}. These strings
+   * have to be evaluated as Lucene-style date math; Calcite would compare them lexically. Detecting
+   * them lets the visitor keep the comparison in {@code query_string} form so the existing
+   * semantics survive.
+   */
+  private static boolean isOpenSearchDateMath(UnresolvedExpression value) {
+    if (!(value instanceof Literal)) {
+      return false;
+    }
+    Object raw = ((Literal) value).getValue();
+    if (!(raw instanceof String)) {
+      return false;
+    }
+    String s = ((String) raw).trim();
+    if (s.isEmpty()) {
+      return false;
+    }
+    if (s.regionMatches(true, 0, "now", 0, 3)) {
+      return true;
+    }
+    if (s.contains("||")) {
+      return true;
+    }
+    // Epoch-millis from `earliest=1754020060` / `latest=1754020060.123` — 12+ digits only.
+    if (s.length() >= 12) {
+      for (int i = 0; i < s.length(); i++) {
+        char c = s.charAt(i);
+        if (c < '0' || c > '9') {
+          return false;
+        }
+      }
+      return true;
+    }
+    return false;
+  }
+
+  /**
+   * Detects an unescaped Lucene wildcard ({@code *} or {@code ?}) in a string literal. A preceding
+   * odd run of backslashes means the wildcard is already escaped (literal star / question mark in
+   * the value), so it's safe to lower as a regular equality.
+   */
+  private static boolean containsLuceneWildcard(UnresolvedExpression value) {
+    if (!(value instanceof Literal)) {
+      return false;
+    }
+    Object raw = ((Literal) value).getValue();
+    if (!(raw instanceof String)) {
+      return false;
+    }
+    String s = (String) raw;
+    for (int i = 0; i < s.length(); i++) {
+      char c = s.charAt(i);
+      if (c == '*' || c == '?') {
+        int backslashes = 0;
+        for (int j = i - 1; j >= 0 && s.charAt(j) == '\\'; j--) {
+          backslashes++;
+        }
+        if ((backslashes & 1) == 0) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+
+  private Function buildQueryStringFunction(String queryString) {
+    return AstDSL.function(
+        "query_string", AstDSL.unresolvedArg("query", AstDSL.stringLiteral(queryString)));
+  }
+
   @Override
   public RelNode visitFilter(Filter node, CalcitePlanContext context) {
     visitChildren(node, context);
@@ -3038,14 +3236,28 @@ public RelNode visitRareTopN(RareTopN node, CalcitePlanContext context) {
       countField = context.relBuilder.field(countFieldName);
     }
 
+    // Append the rare/top field columns as secondary order keys so ties in the count column
+    // resolve deterministically. Without this, ROW_NUMBER's tie-break is insertion-order
+    // dependent and varies between backends (e.g. analytics-engine vs in-process Calcite).
+    // Skip array / struct / map columns — Calcite's executor can't compare those at runtime
+    // (`ArrayList cannot be cast to Comparable`), and pushdown to the OpenSearch terms
+    // aggregation only supports scalar keys.
+    List<RexNode> tieBreakKeys =
+        rexVisitor.analyze(fieldList, context).stream()
+            .filter(CalciteRelNodeVisitor::isComparableOrderKey)
+            .toList();
+    List<RexNode> orderKeys = new ArrayList<>(tieBreakKeys.size() + 1);
+    orderKeys.add(countField);
+    orderKeys.addAll(tieBreakKeys);
+
     RexNode rowNumberWindowOver =
         PlanUtils.makeOver(
             context,
             BuiltinFunctionName.ROW_NUMBER,
             null,
             List.of(),
             partitionKeys,
-            List.of(countField),
+            orderKeys,
             WindowFrame.toCurrentRow());
     context.relBuilder.projectPlus(
         context.relBuilder.alias(rowNumberWindowOver, ROW_NUMBER_COLUMN_FOR_RARE_TOP));
@@ -3069,6 +3281,15 @@ public RelNode visitRareTopN(RareTopN node, CalcitePlanContext context) {
     return context.relBuilder.peek();
   }
 
+  private static boolean isComparableOrderKey(RexNode key) {
+    SqlTypeName typeName = key.getType().getSqlTypeName();
+    return typeName != SqlTypeName.ARRAY
+        && typeName != SqlTypeName.MULTISET
+        && typeName != SqlTypeName.MAP
+        && typeName != SqlTypeName.ROW
+        && typeName != SqlTypeName.STRUCTURED;
+  }
+
   @Override
   public RelNode visitTableFunction(TableFunction node, CalcitePlanContext context) {
     throw new CalciteUnsupportedException("Table function is unsupported in Calcite");