[Improve)(Variant) do not allow fall back to legacy planner (#30430)

fe/fe-core/src/main/java/org/apache/doris/analysis/Analyzer.java

@@ -1029,6 +1029,11 @@ public class Analyzer {
 
         LOG.debug("register column ref table {}, colName {}, col {}", tblName, colName, col.toSql());
         if (col.getType().isVariantType() || (subColNames != null && !subColNames.isEmpty())) {
+            if (!Config.enable_variant_access_in_original_planner
+                    && (subColNames != null && !subColNames.isEmpty())) {
+                ErrorReport.reportAnalysisException("Variant sub-column access is disabled in original planner,"
+                        + "set enable_variant_access_in_original_planner = true in session variable");
+            }
             if (!col.getType().isVariantType()) {
                 ErrorReport.reportAnalysisException(ErrorCode.ERR_ILLEGAL_COLUMN_REFERENCE_ERROR,
                         Joiner.on(".").join(tblName.getTbl(), colName));

fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java

@@ -186,7 +186,7 @@ public class StatementContext {
      * Add a slot ref attached with paths in context to avoid duplicated slot
      */
     public void addPathSlotRef(Slot root, List<String> paths, SlotReference slotRef, Expression originalExpr) {
-        Comparator<List<String>> pathsComparator = new Comparator<List<String>>() {
+        subColumnSlotRefMap.computeIfAbsent(root, k -> Maps.newTreeMap(new Comparator<List<String>>() {
             @Override
             public int compare(List<String> lst1, List<String> lst2) {
                 Iterator<String> it1 = lst1.iterator();
@@ -199,8 +199,7 @@ public class StatementContext {
                 }
                 return Integer.compare(lst1.size(), lst2.size());
             }
-        };
-        subColumnSlotRefMap.computeIfAbsent(root, k -> Maps.newTreeMap(pathsComparator));
+        }));
         subColumnSlotRefMap.get(root).put(paths, slotRef);
         subColumnOriginalExprMap.put(slotRef, originalExpr);
         originalExprToRewrittenSubColumn.put(originalExpr, slotRef);

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

@@ -219,6 +219,7 @@ public class ExpressionTranslator extends DefaultExpressionVisitor<Expr, PlanTra
         String invertedIndexParser = InvertedIndexUtil.INVERTED_INDEX_PARSER_UNKNOWN;
         String invertedIndexParserMode = InvertedIndexUtil.INVERTED_INDEX_PARSER_COARSE_GRANULARITY;
         Map<String, String> invertedIndexCharFilter = new HashMap<>();
+        // Get the first slot from match's left expr
         SlotRef left = (SlotRef) match.left().getInputSlots().stream().findFirst().get().accept(this, context);
         OlapTable olapTbl = Optional.ofNullable(getOlapTableFromSlotDesc(left.getDesc()))
                                     .orElse(getOlapTableDirectly(left));

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/PushDownFilterThroughProject.java

@@ -42,7 +42,7 @@ public class PushDownFilterThroughProject extends PlanPostProcessor {
         }
 
         PhysicalProject<? extends Plan> project = (PhysicalProject<? extends Plan>) child;
-        if (project.isPulledUpProjectFromScan()) {
+        if (project.hasPushedDownToProjectionFunctions()) {
             // ignore project which is pulled up from LogicalOlapScan
             return filter;
         }

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/Validator.java

@@ -59,7 +59,7 @@ public class Validator extends PlanPostProcessor {
 
         Plan child = filter.child();
         // Forbidden filter-project, we must make filter-project -> project-filter.
-        if (child instanceof PhysicalProject && !((PhysicalProject<?>) child).isPulledUpProjectFromScan()) {
+        if (child instanceof PhysicalProject && !((PhysicalProject<?>) child).hasPushedDownToProjectionFunctions()) {
             throw new AnalysisException(
                     "Nereids generate a filter-project plan, but backend not support:\n" + filter.treeString());
         }

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

@@ -49,10 +49,9 @@ public enum RuleType {
     BINDING_UNBOUND_TVF_RELATION_FUNCTION(RuleTypeClass.REWRITE),
     BINDING_SET_OPERATION_SLOT(RuleTypeClass.REWRITE),
     BINDING_INLINE_TABLE_SLOT(RuleTypeClass.REWRITE),
-    COUNT_LITERAL_REWRITE(RuleTypeClass.REWRITE),
-    BINDING_SLOT_WITH_PATHS_PROJECT(RuleTypeClass.REWRITE),
 
     BINDING_SLOT_WITH_PATHS_SCAN(RuleTypeClass.REWRITE),
+    COUNT_LITERAL_REWRITE(RuleTypeClass.REWRITE),
 
     REPLACE_SORT_EXPRESSION_BY_CHILD_OUTPUT(RuleTypeClass.REWRITE),
 

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

@@ -184,7 +184,7 @@ public class CheckAfterRewrite extends OneAnalysisRuleFactory {
             if (plan instanceof LogicalFilter && (plan.child(0) instanceof LogicalOlapScan
                     || plan.child(0) instanceof LogicalDeferMaterializeOlapScan
                     || plan.child(0) instanceof LogicalProject
-                        && ((LogicalProject<?>) plan.child(0)).isPulledUpProjectFromScan())) {
+                        && ((LogicalProject<?>) plan.child(0)).hasPushedDownToProjectionFunctions())) {
                 return;
             } else {
                 throw new AnalysisException(String.format(

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PullUpProjectUnderTopN.java

@@ -38,6 +38,7 @@ public class PullUpProjectUnderTopN extends OneRewriteRuleFactory {
     @Override
     public Rule build() {
         return logicalTopN(logicalProject().whenNot(p -> p.isAllSlots()))
+                .whenNot(topN -> topN.child().hasPushedDownToProjectionFunctions())
                 .then(topN -> {
                     LogicalProject<Plan> project = topN.child();
                     Set<Slot> outputSet = project.child().getOutputSet();

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownFilterThroughProject.java

@@ -44,7 +44,7 @@ public class PushDownFilterThroughProject implements RewriteRuleFactory {
                 logicalFilter(logicalProject())
                         .whenNot(filter -> filter.child().getProjects().stream().anyMatch(
                                 expr -> expr.anyMatch(WindowExpression.class::isInstance)))
-                        .whenNot(filter -> filter.child().isPulledUpProjectFromScan())
+                        .whenNot(filter -> filter.child().hasPushedDownToProjectionFunctions())
                         .then(PushDownFilterThroughProject::pushdownFilterThroughProject)
                         .toRule(RuleType.PUSH_DOWN_FILTER_THROUGH_PROJECT),
                 // filter(project(limit)) will change to filter(limit(project)) by PushdownProjectThroughLimit,
@@ -52,7 +52,7 @@ public class PushDownFilterThroughProject implements RewriteRuleFactory {
                 logicalFilter(logicalLimit(logicalProject()))
                         .whenNot(filter -> filter.child().child().getProjects().stream()
                                 .anyMatch(expr -> expr.anyMatch(WindowExpression.class::isInstance)))
-                        .whenNot(filter -> filter.child().child().isPulledUpProjectFromScan())
+                        .whenNot(filter -> filter.child().child().hasPushedDownToProjectionFunctions())
                         .then(filter -> {
                             LogicalLimit<LogicalProject<Plan>> limit = filter.child();
                             LogicalProject<Plan> project = limit.child();

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/PushDownToProjectionFunction.java

@@ -19,13 +19,22 @@ package org.apache.doris.nereids.trees.expressions.functions.scalar;
 
 import org.apache.doris.nereids.trees.expressions.Expression;
 
+import java.util.stream.Collectors;
+
 /**
  * Function that could be rewritten and pushed down to projection
  */
 public interface PushDownToProjectionFunction {
-    // check if specified function could be pushed down to project
+    /**
+     * check if specified function could be pushed down to project
+     * @param pushDownExpr expr to check
+     * @return if it is valid to push down input expr
+     */
     static boolean validToPushDown(Expression pushDownExpr) {
-        // Currently only Variant type could be pushed down
-        return pushDownExpr instanceof PushDownToProjectionFunction && pushDownExpr.getDataType().isVariantType();
+        // Currently only element at for variant type could be pushed down
+        return !pushDownExpr.collectToList(
+                    PushDownToProjectionFunction.class::isInstance).stream().filter(
+                            x -> ((Expression) x).getDataType().isVariantType()).collect(
+                    Collectors.toList()).isEmpty();
     }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/Project.java

@@ -71,8 +71,9 @@ public interface Project {
     /**
      * Check if it is a project that is pull up from scan in analyze rule
      * e.g. BindSlotWithPaths
+     * And check if contains PushDownToProjectionFunction that can pushed down to project
      */
-    default boolean isPulledUpProjectFromScan() {
+    default boolean hasPushedDownToProjectionFunctions() {
         return ConnectContext.get() != null
                 && ConnectContext.get().getSessionVariable() != null
                 && ConnectContext.get().getSessionVariable().isEnableRewriteElementAtToSlot()

fe/fe-core/src/main/java/org/apache/doris/nereids/types/VariantType.java

@@ -47,7 +47,7 @@ public class VariantType extends PrimitiveType {
 
     @Override
     public String simpleString() {
-        return "map";
+        return "variant";
     }
 
     @Override