[feature](nereids) Support SEARCH in joins and per-field analyzers

fe/fe-catalog/src/main/java/org/apache/doris/analysis/SearchDslParser.java

@@ -17,6 +17,8 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.common.Pair;
+
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -35,6 +37,8 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
@@ -276,6 +280,9 @@ private static void validateFieldsList(List<String> fields) {
         }
     }
 
+    // an escaped character, or an unescaped @
+    private static final Pattern ESCAPED_OR_AT = Pattern.compile("\\\\.|@");
+
     private static String buildFieldPath(SearchParser.FieldPathContext ctx) {
         if (ctx == null) {
             throw new RuntimeException("Invalid field query: missing field path");
@@ -289,13 +296,49 @@ private static String buildFieldPath(SearchParser.FieldPathContext ctx) {
             }
             String segment = segments.get(i).getText();
             if (segment.startsWith("\"") && segment.endsWith("\"")) {
-                segment = segment.substring(1, segment.length() - 1);
+                // An @ inside quotes is part of the field name; see splitAnalyzerSelector.
+                segment = ESCAPED_OR_AT.matcher(segment.substring(1, segment.length() - 1))
+                        .replaceAll(m -> Matcher.quoteReplacement(m.group().equals("@") ? "\\@" : m.group()));
             }
             fullPath.append(segment);
         }
         return fullPath.toString();
     }
 
+    /**
+     * Splits a DSL field reference {@code path[@analyzer]} into the field path and the analyzer name (null when
+     * no analyzer is selected). The split is purely syntactic and never depends on the table schema: the last
+     * unescaped {@code @} that follows a non-empty path segment selects the analyzer, while {@code \@}, an
+     * {@code @} inside a quoted segment (escaped by buildFieldPath) and an {@code @} that starts a segment
+     * ({@code v.@timestamp}) belong to the field name.
+     */
+    public static Pair<String, String> splitAnalyzerSelector(String fieldReference) {
+        StringBuilder path = new StringBuilder();
+        int selector = -1;
+        for (int i = 0; i < fieldReference.length(); i++) {
+            char c = fieldReference.charAt(i);
+            if (c == '\\' && i + 1 < fieldReference.length()) {
+                char escaped = fieldReference.charAt(++i);
+                if (escaped != '@') {
+                    path.append(c);
+                }
+                path.append(escaped);
+                continue;
+            }
+            if (c == '@' && i > 0 && fieldReference.charAt(i - 1) != '.') {
+                selector = path.length();
+            }
+            path.append(c);
+        }
+        if (selector < 0) {
+            return Pair.of(path.toString(), null);
+        }
+        if (selector == path.length() - 1) {
+            throw new SearchDslSyntaxException("SEARCH analyzer selector must be field@analyzer: " + fieldReference);
+        }
+        return Pair.of(path.substring(0, selector), path.substring(selector + 1));
+    }
+
     private static String normalizeNestedFieldPath(String fieldPath, @Nullable String nestedPath) {
         if (nestedPath == null || nestedPath.isEmpty()) {
             return fieldPath;
@@ -1303,6 +1346,13 @@ public String getNestedPath() {
             return nestedPath;
         }
 
+        /**
+         * Sets the nested path (used for field name normalization).
+         */
+        public void setNestedPath(String nestedPath) {
+            this.nestedPath = nestedPath;
+        }
+
         /**
          * Returns whether the field was explicitly specified in the DSL syntax.
          */
@@ -1365,6 +1415,17 @@ public static class QsFieldBinding {
         @JsonProperty("slotIndex")
         private final int slotIndex;
 
+        @JsonProperty("analyzerName")
+        private String analyzerName;
+
+        public String getAnalyzerName() {
+            return analyzerName;
+        }
+
+        public void setAnalyzerName(String analyzerName) {
+            this.analyzerName = analyzerName;
+        }
+
         @JsonCreator
         public QsFieldBinding(@JsonProperty("fieldName") String fieldName,
                 @JsonProperty("slotIndex") int slotIndex) {
@@ -1390,7 +1451,7 @@ public int getSlotIndex() {
 
         @Override
         public int hashCode() {
-            return Objects.hash(fieldName, slotIndex);
+            return Objects.hash(fieldName, slotIndex, analyzerName);
         }
 
         @Override
@@ -1403,7 +1464,8 @@ public boolean equals(Object o) {
             }
             QsFieldBinding that = (QsFieldBinding) o;
             return slotIndex == that.slotIndex
-                    && Objects.equals(fieldName, that.fieldName);
+                    && Objects.equals(fieldName, that.fieldName)
+                    && Objects.equals(analyzerName, that.analyzerName);
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java

@@ -250,7 +250,7 @@ public Expr visitMatch(Match match, PlanTranslatorContext context) {
         // column/table reference (e.g., after CTE inlining or join projection remapping),
         // we gracefully fall back to invertedIndex = null. The BE can still evaluate MATCH
         // correctly without inverted index (slow path), or the PushDownProject /
-        // PushDownMatchProjectionAsVirtualColumn rules may have already pushed the expression
+        // PushDownIndexSearchAsVirtualColumn rules may have already pushed the expression
         // down for storage-level index evaluation (fast path).
         Index invertedIndex = null;
         String analyzer = match.getAnalyzer().orElse(null);
@@ -698,19 +698,25 @@ public Expr visitSearchExpression(SearchExpression searchExpression,
 
             // Look up the inverted index for each field (needed for variant subcolumn analyzer)
             Index invertedIndex = null;
+            String analyzer = searchExpression.getQsPlan().getFieldBindings()
+                    .get(fieldIndexes.size()).getAnalyzerName();
             if (slotExpr instanceof SlotReference) {
                 SlotReference slot = (SlotReference) slotExpr;
                 OlapTable olapTbl = getOlapTableDirectly(slot);
                 if (olapTbl != null) {
                     Column column = slot.getOriginalColumn().orElse(null);
                     if (column != null) {
-                        invertedIndex = olapTbl.getInvertedIndex(column, slot.getSubPath());
+                        invertedIndex = olapTbl.getInvertedIndex(column, slot.getSubPath(), analyzer);
                     }
                 }
             }
-            if (invertedIndex == null) {
+            if (invertedIndex == null && analyzer == null) {
                 invertedIndex = getInvertedIndexFromTranslatedSlot(translatedSlot, context);
             }
+            if (analyzer != null && invertedIndex == null) {
+                throw new AnalysisException("No inverted index found for SEARCH analyzer '" + analyzer
+                        + "' on " + slotExpr.toSql());
+            }
             fieldIndexes.add(invertedIndex);
         }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Analyzer.java

@@ -54,6 +54,7 @@
 import org.apache.doris.nereids.rules.analysis.SubqueryToApply;
 import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
 import org.apache.doris.nereids.rules.rewrite.MergeFilters;
+import org.apache.doris.nereids.rules.rewrite.RewriteSearchToSlots;
 import org.apache.doris.nereids.rules.rewrite.SimplifyAggGroupBy;
 import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor;
 import org.apache.doris.nereids.trees.plans.logical.LogicalView;
@@ -169,6 +170,8 @@ private static List<RewriteJob> buildAnalyzerJobs() {
             ),
             // run CheckSearchUsage before CheckAnalysis to detect search() in GROUP BY before it gets optimized
             bottomUp(new CheckSearchUsage()),
+            // Bind SEARCH dependencies before predicate movement and column pruning.
+            bottomUp(new RewriteSearchToSlots()),
             // run CheckAnalysis before EliminateGroupByConstant in order to report error message correctly like bellow
             // select SUM(lo_tax) FROM lineorder group by 1;
             // errCode = 2, detailMessage = GROUP BY expression must not contain aggregate functions: sum(lo_tax)

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java

@@ -132,11 +132,11 @@
 import org.apache.doris.nereids.rules.rewrite.PushDownEncodeSlot;
 import org.apache.doris.nereids.rules.rewrite.PushDownFilterIntoSchemaScan;
 import org.apache.doris.nereids.rules.rewrite.PushDownFilterThroughProject;
+import org.apache.doris.nereids.rules.rewrite.PushDownIndexSearchAsVirtualColumn;
 import org.apache.doris.nereids.rules.rewrite.PushDownJoinOnAssertNumRows;
 import org.apache.doris.nereids.rules.rewrite.PushDownLimit;
 import org.apache.doris.nereids.rules.rewrite.PushDownLimitDistinctThroughJoin;
 import org.apache.doris.nereids.rules.rewrite.PushDownLimitDistinctThroughUnion;
-import org.apache.doris.nereids.rules.rewrite.PushDownMatchProjectionAsVirtualColumn;
 import org.apache.doris.nereids.rules.rewrite.PushDownProjectThroughLimit;
 import org.apache.doris.nereids.rules.rewrite.PushDownScoreTopNIntoOlapScan;
 import org.apache.doris.nereids.rules.rewrite.PushDownTopNDistinctThroughJoin;
@@ -154,7 +154,6 @@
 import org.apache.doris.nereids.rules.rewrite.ReduceAggregateChildOutputRows;
 import org.apache.doris.nereids.rules.rewrite.ReorderJoin;
 import org.apache.doris.nereids.rules.rewrite.RewriteCteChildren;
-import org.apache.doris.nereids.rules.rewrite.RewriteSearchToSlots;
 import org.apache.doris.nereids.rules.rewrite.RewriteSimpleAggToConstantRule;
 import org.apache.doris.nereids.rules.rewrite.SaltJoin;
 import org.apache.doris.nereids.rules.rewrite.SemiJoinCommute;
@@ -797,7 +796,7 @@ public class Rewriter extends AbstractBatchJobExecutor {
                 custom(RuleType.ELIMINATE_UNNECESSARY_PROJECT, EliminateUnnecessaryProject::new),
                 topDown(new PushDownVectorTopNIntoOlapScan()),
                 topDown(new PushDownVirtualColumnsIntoOlapScan()),
-                topDown(new PushDownMatchProjectionAsVirtualColumn()),
+                topDown(new PushDownIndexSearchAsVirtualColumn()),
                 topic("score optimize",
                         topDown(new PushDownScoreTopNIntoOlapScan(),
                                 new CheckScoreUsage())
@@ -929,12 +928,6 @@ private static List<RewriteJob> getWholeTreeRewriteJobs(
                             custom(RuleType.DISTINCT_AGG_STRATEGY_SELECTOR,
                                     () -> DistinctAggStrategySelector.INSTANCE))));
 
-                    // Rewrite search function before VariantSubPathPruning
-                    // so that ElementAt expressions from search can be processed
-                    rewriteJobs.addAll(jobs(
-                            bottomUp(new RewriteSearchToSlots())
-                    ));
-
                     if (needSubPathPushDown) {
                         rewriteJobs.addAll(jobs(
                                 topic("variant element_at push down",

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/CheckAfterRewrite.java

@@ -26,6 +26,7 @@
 import org.apache.doris.nereids.trees.expressions.Cast;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.Match;
+import org.apache.doris.nereids.trees.expressions.SearchExpression;
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.expressions.SlotNotFromChildren;
 import org.apache.doris.nereids.trees.expressions.SubqueryExpr;
@@ -50,6 +51,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.roaringbitmap.RoaringBitmap;
 
+import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -65,10 +67,41 @@ public Rule build() {
             checkUnexpectedExpression(plan);
             checkMetricTypeIsUsedCorrectly(plan);
             checkMatchIsUsedCorrectly(plan);
+            checkSearchIsUsedCorrectly(plan);
             return null;
         }).toRule(RuleType.CHECK_ANALYSIS);
     }
 
+    /**
+     * The execution invariant of SEARCH, checked in this one place for every way a rewrite can move it: BE
+     * evaluates a SearchExpression only with inverted indexes inside an OLAP scan, so it must be a conjunct of
+     * the filter on the scan or a scan virtual column, and every child must still bind an index field.
+     */
+    private void checkSearchIsUsedCorrectly(Plan plan) {
+        // Scan virtual columns are not part of LogicalOlapScan.getExpressions().
+        List<? extends Expression> expressions = plan instanceof LogicalOlapScan
+                ? ((LogicalOlapScan) plan).getVirtualColumns() : plan.getExpressions();
+        boolean evaluatedByScan = plan instanceof LogicalOlapScan
+                || (plan instanceof LogicalFilter && plan.child(0) instanceof LogicalOlapScan);
+        for (Expression expression : expressions) {
+            for (SearchExpression search : expression.<SearchExpression>collect(SearchExpression.class::isInstance)) {
+                if (!search.bindsOnlyFields()) {
+                    // e.g. LEFT JOIN ... ON false, or an outer join converted to an anti join
+                    throw new AnalysisException("SEARCH on the null-generating side of an outer join is not "
+                            + "supported: its field is always NULL in " + search.toSql());
+                }
+                if (!evaluatedByScan) {
+                    throw new AnalysisException("SEARCH must be evaluated by an OLAP scan, but " + search.toSql()
+                            + " remains in " + plan.getType() + ". Unless it is a WHERE conjunct that reaches the "
+                            + "scan, all its fields must come from one DUP_KEYS or merge-on-write UNIQUE_KEYS "
+                            + "table reference that is neither on the null-generating side of an outer join nor "
+                            + "below a LIMIT, TopN, aggregate or window; combine separate SEARCH expressions with "
+                            + "SQL AND/OR to search several tables");
+                }
+            }
+        }
+    }
+
     private void checkUnexpectedExpression(Plan plan) {
         boolean isGenerate = plan instanceof Generate;
         boolean isAgg = plan instanceof LogicalAggregate;

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/CheckSearchUsage.java

@@ -26,6 +26,7 @@
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
 import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
 import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
 
@@ -36,9 +37,13 @@
 import java.util.List;
 
 /**
- * Check search expression usage - search() can only be used in WHERE filters on single-table OLAP scans.
+ * Check that search() is used in WHERE filters over OLAP tables, including joins of them.
  * This rule validates that search() expressions only appear in supported contexts.
  * Must run in analysis phase before search() gets optimized away.
+ *
+ * <p>This is a placement check on the analyzed plan and only a necessary condition: predicate push down may
+ * still move the filter anywhere below. Whether the search can be evaluated by a scan is decided by
+ * PushDownIndexSearchAsVirtualColumn and verified on the final plan by CheckAfterRewrite.
  */
 public class CheckSearchUsage implements AnalysisRuleFactory {
     private static final Logger LOG = LogManager.getLogger(CheckSearchUsage.class);
@@ -66,13 +71,13 @@ private void checkPlanRecursively(Plan plan) {
             for (Expression expr : agg.getGroupByExpressions()) {
                 if (containsSearchExpression(expr)) {
                     throw new AnalysisException("search() cannot appear in GROUP BY expressions; "
-                            + "search predicates are only supported in WHERE filters on single-table scans");
+                            + "search predicates are only supported in WHERE filters over OLAP tables");
                 }
             }
             for (Expression expr : agg.getOutputExpressions()) {
                 if (containsSearchExpression(expr)) {
                     throw new AnalysisException("search() cannot appear in aggregate output expressions; "
-                            + "search predicates are only supported in WHERE filters on single-table scans");
+                            + "search predicates are only supported in WHERE filters over OLAP tables");
                 }
             }
         }
@@ -82,7 +87,6 @@ private void checkPlanRecursively(Plan plan) {
             LogicalProject<?> project = (LogicalProject<?>) plan;
             for (Expression expr : project.getProjects()) {
                 if (containsSearchExpression(expr)) {
-                    // Only allow if it's the project directly above a filter->scan pattern
                     throw new AnalysisException("search() can only appear in WHERE filters on OLAP scans; "
                             + "projection of search() is not supported");
                 }
@@ -99,15 +103,13 @@ private void validateSearchUsage(Plan plan) {
         LOG.debug("validateSearchUsage: {}", plan.treeString());
         if (plan instanceof LogicalFilter) {
             Plan child = plan.child(0);
-            if (!isSingleTableScanPipeline(child)) {
-                throw new AnalysisException("search() predicate only supports filtering directly on a single "
-                        + "table scan; remove joins, subqueries, or additional operators between search() "
-                        + "and the target table");
+            if (!isOlapScanPipeline(child)) {
+                throw new AnalysisException("search() predicates require an OLAP scan pipeline");
             }
         } else if (!(plan instanceof LogicalProject)) {
-            // search() can only appear in LogicalFilter or specific LogicalProject nodes
-            throw new AnalysisException("search() predicates are only supported inside WHERE filters on "
-                    + "single-table scans");
+            // a projection gets its own message in checkPlanRecursively
+            throw new AnalysisException("search() predicates are only supported inside WHERE filters over "
+                    + "OLAP tables");
         }
     }
 
@@ -132,12 +134,17 @@ private boolean containsSearchExpression(Expression expression) {
         return false;
     }
 
-    private boolean isSingleTableScanPipeline(Plan plan) {
+    // Every relation below the filter must be an OLAP scan; unary nodes in between are not restricted here.
+    private boolean isOlapScanPipeline(Plan plan) {
         Plan current = plan;
         while (true) {
             if (current instanceof LogicalOlapScan) {
                 return true;
             }
+            if (current instanceof LogicalJoin) {
+                return isOlapScanPipeline(current.child(0))
+                        && isOlapScanPipeline(current.child(1));
+            }
             if (current.arity() != 1) {
                 return false;
             }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnBE.java

@@ -39,6 +39,7 @@
 import org.apache.doris.nereids.trees.expressions.Cast;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.Match;
+import org.apache.doris.nereids.trees.expressions.SearchExpression;
 import org.apache.doris.nereids.trees.expressions.functions.BoundFunction;
 import org.apache.doris.nereids.trees.expressions.functions.ai.AIFunction;
 import org.apache.doris.nereids.trees.expressions.functions.generator.TableGeneratingFunction;
@@ -275,7 +276,7 @@ private static boolean shouldSkipFold(Expression expr) {
         }
 
         // Search function should always not be folded to constant.
-        if (expr instanceof Search) {
+        if (expr instanceof Search || expr instanceof SearchExpression) {
             return true;
         }