[feature](nereids)support unnest function

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisLexer.g4

@@ -390,6 +390,7 @@ OPTIMIZE: 'OPTIMIZE';
 OPTIMIZED: 'OPTIMIZED';
 OR: 'OR';
 ORDER: 'ORDER';
+ORDINALITY: 'ORDINALITY';
 OUTER: 'OUTER';
 OUTFILE: 'OUTFILE';
 OVER: 'OVER';
@@ -561,6 +562,7 @@ UNINSTALL: 'UNINSTALL';
 UNION: 'UNION';
 UNIQUE: 'UNIQUE';
 UNLOCK: 'UNLOCK';
+UNNEST: 'UNNEST';
 UNSET: 'UNSET';
 UNSIGNED: 'UNSIGNED';
 UP: 'UP';

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -1344,6 +1344,17 @@ lateralView
       tableName=identifier AS columnNames+=identifier (COMMA columnNames+=identifier)*
     ;
 
+unnest:
+    LATERAL? UNNEST LEFT_PAREN expression (COMMA expression)* RIGHT_PAREN (
+        WITH ORDINALITY
+    )? (
+        AS? tableName = identifier (
+            LEFT_PAREN columnNames += identifier (
+                COMMA columnNames += identifier
+            )* RIGHT_PAREN
+        )?
+    )?;
+
 queryOrganization
     : sortClause? limitClause?
     ;
@@ -1403,6 +1414,7 @@ relationPrimary
       (properties=propertyItemList)?
       RIGHT_PAREN tableAlias                                                               #tableValuedFunction
     | LEFT_PAREN relations RIGHT_PAREN                                                     #relationList
+    | unnest                                                                               #unnestFunction
     ;
 
 materializedViewName
@@ -2109,6 +2121,7 @@ nonReserved
     | OPEN
     | OPTIMIZE
     | OPTIMIZED
+    | ORDINALITY
     | PARAMETER
     | PARSED
     | PASSWORD
@@ -2221,6 +2234,7 @@ nonReserved
     | TYPES
     | UNCOMMITTED
     | UNLOCK
+    | UNNEST
     | UNSET
     | UP
     | USER

fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableGeneratingFunctions.java

@@ -40,6 +40,7 @@
 import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeVariantArray;
 import org.apache.doris.nereids.trees.expressions.functions.generator.PosExplode;
 import org.apache.doris.nereids.trees.expressions.functions.generator.PosExplodeOuter;
+import org.apache.doris.nereids.trees.expressions.functions.generator.Unnest;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -78,7 +79,8 @@ public class BuiltinTableGeneratingFunctions implements FunctionHelper {
             tableGenerating(ExplodeJsonArrayJsonOuter.class, "explode_json_array_json_outer"),
             tableGenerating(ExplodeVariantArray.class, "explode_variant_array"),
             tableGenerating(PosExplode.class, "posexplode"),
-            tableGenerating(PosExplodeOuter.class, "posexplode_outer")
+            tableGenerating(PosExplodeOuter.class, "posexplode_outer"),
+            tableGenerating(Unnest.class, "unnest")
     );
 
     public static final ImmutableSet<String> RETURN_MULTI_COLUMNS_FUNCTIONS = new ImmutableSortedSet.Builder<String>(

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

@@ -1483,8 +1483,11 @@ public PlanFragment visitPhysicalGenerate(PhysicalGenerate<? extends Plan> gener
                 .flatMap(List::stream)
                 .map(SlotDescriptor::getId)
                 .collect(Collectors.toList());
+        ArrayList<Expr> conjuncts = generate.getConjuncts().stream()
+                .map(e -> ExpressionTranslator.translate(e, context))
+                .collect(Collectors.toCollection(ArrayList::new));
         TableFunctionNode tableFunctionNode = new TableFunctionNode(context.nextPlanNodeId(),
-                currentFragment.getPlanRoot(), tupleDescriptor.getId(), functionCalls, outputSlotIds);
+                currentFragment.getPlanRoot(), tupleDescriptor.getId(), functionCalls, outputSlotIds, conjuncts);
         tableFunctionNode.setNereidsId(generate.getId());
         context.getNereidsIdToPlanNodeIdMap().put(generate.getId(), tableFunctionNode.getId());
         addPlanRoot(currentFragment, tableFunctionNode, generate);

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

@@ -143,6 +143,7 @@
 import org.apache.doris.nereids.rules.rewrite.PushDownTopNThroughJoin;
 import org.apache.doris.nereids.rules.rewrite.PushDownTopNThroughUnion;
 import org.apache.doris.nereids.rules.rewrite.PushDownTopNThroughWindow;
+import org.apache.doris.nereids.rules.rewrite.PushDownUnnestInProject;
 import org.apache.doris.nereids.rules.rewrite.PushDownVectorTopNIntoOlapScan;
 import org.apache.doris.nereids.rules.rewrite.PushDownVirtualColumnsIntoOlapScan;
 import org.apache.doris.nereids.rules.rewrite.PushFilterInsideJoin;
@@ -528,6 +529,10 @@ public class Rewriter extends AbstractBatchJobExecutor {
                                 new SimplifyWindowExpression()
                         )
                 ),
+                topic("Push down Unnest",
+                        topDown(
+                                new PushDownUnnestInProject()
+                        )),
                 topic("Rewrite join",
                         // infer not null filter, then push down filter, and then reorder join(cross join to inner join)
                         topDown(

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

@@ -555,6 +555,7 @@
 import org.apache.doris.nereids.trees.expressions.WindowFrame;
 import org.apache.doris.nereids.trees.expressions.functions.Function;
 import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.generator.Unnest;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.Array;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.ArraySlice;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.Char;
@@ -2636,6 +2637,58 @@ public LogicalPlan visitTableValuedFunction(TableValuedFunctionContext ctx) {
         });
     }
 
+    @Override
+    public LogicalPlan visitUnnestFunction(DorisParser.UnnestFunctionContext ctx) {
+        return withUnnest(ctx.unnest());
+    }
+
+    private LogicalPlan withUnnest(DorisParser.UnnestContext ctx) {
+        String defaultNestedColumnName = "unnest";
+        String defaultOrdinalityColumnName = "ordinality";
+        List<Expression> arguments = ctx.expression().stream()
+                .<Expression>map(this::typedVisit)
+                .collect(ImmutableList.toImmutableList());
+        boolean needOrdinality = ctx.ORDINALITY() != null;
+        int size = arguments.size();
+
+        String generateName = ctx.tableName != null ? ctx.tableName.getText() : defaultNestedColumnName;
+        // do same thing as later view explode map type, we need to add a project to convert map to struct
+        int argumentsSize = size + (needOrdinality ? 1 : 0);
+        List<String> nestedColumnNames = new ArrayList<>(argumentsSize);
+        int columnNamesSize = ctx.columnNames.size();
+        if (!ctx.columnNames.isEmpty()) {
+            for (int i = 0; i < columnNamesSize; ++i) {
+                nestedColumnNames.add(ctx.columnNames.get(i).getText());
+            }
+            for (int i = 0; i < size - columnNamesSize; ++i) {
+                nestedColumnNames.add(defaultNestedColumnName);
+            }
+            if (needOrdinality && columnNamesSize < argumentsSize) {
+                nestedColumnNames.add(defaultOrdinalityColumnName);
+            }
+        } else {
+            if (size == 1) {
+                nestedColumnNames.add(generateName);
+            } else {
+                for (int i = 0; i < size; ++i) {
+                    nestedColumnNames.add(defaultNestedColumnName);
+                }
+            }
+            if (needOrdinality) {
+                nestedColumnNames.add(defaultOrdinalityColumnName);
+            }
+        }
+        String columnName = nestedColumnNames.get(0);
+        Unnest unnest = new Unnest(false, needOrdinality, arguments);
+        // only unnest use LogicalOneRowRelation as LogicalGenerate's child,
+        // so we can check LogicalGenerate's child to know if it's unnest function
+        return new LogicalGenerate<>(ImmutableList.of(unnest),
+                ImmutableList.of(new UnboundSlot(generateName, columnName)),
+                ImmutableList.of(nestedColumnNames),
+                new LogicalOneRowRelation(StatementScopeIdGenerator.newRelationId(),
+                ImmutableList.of(new Alias(Literal.of(0)))));
+    }
+
     /**
      * Create a star (i.e. all) expression; this selects all elements (in the specified object).
      * Both un-targeted (global) and targeted aliases are supported.
@@ -4279,13 +4332,30 @@ private LogicalPlan withJoinRelations(LogicalPlan input, RelationContext ctx) {
                 }
             }
             if (ids == null) {
-                last = new LogicalJoin<>(joinType, ExpressionUtils.EMPTY_CONDITION,
-                        condition.map(ExpressionUtils::extractConjunction)
-                                .orElse(ExpressionUtils.EMPTY_CONDITION),
-                        distributeHint,
-                        Optional.empty(),
-                        last,
-                        plan(join.relationPrimary()), null);
+                LogicalPlan right = plan(join.relationPrimary());
+                if (right instanceof LogicalGenerate
+                        && right.child(0) instanceof LogicalOneRowRelation
+                        && ((LogicalGenerate<?>) right).getGenerators().get(0) instanceof Unnest) {
+                    if (joinType.isLeftJoin() || joinType.isInnerJoin() || joinType.isCrossJoin()) {
+                        LogicalGenerate oldRight = (LogicalGenerate<?>) right;
+                        Unnest oldGenerator = (Unnest) oldRight.getGenerators().get(0);
+                        Unnest newGenerator = joinType.isLeftJoin() ? oldGenerator.withOuter(true) : oldGenerator;
+                        last = new LogicalGenerate<>(ImmutableList.of(newGenerator), oldRight.getGeneratorOutput(),
+                                oldRight.getExpandColumnAlias(), condition.map(ExpressionUtils::extractConjunction)
+                                .orElse(ExpressionUtils.EMPTY_CONDITION), last);
+                    } else {
+                        throw new ParseException("The combining JOIN type must be INNER, LEFT or CROSS for UNNEST",
+                                join);
+                    }
+                } else {
+                    last = new LogicalJoin<>(joinType, ExpressionUtils.EMPTY_CONDITION,
+                            condition.map(ExpressionUtils::extractConjunction)
+                                    .orElse(ExpressionUtils.EMPTY_CONDITION),
+                            distributeHint,
+                            Optional.empty(),
+                            last,
+                            right, null);
+                }
             } else {
                 last = new LogicalUsingJoin<>(joinType, last, plan(join.relationPrimary()), ids, distributeHint);
 
@@ -4507,15 +4577,19 @@ private LogicalPlan withRelations(LogicalPlan inputPlan, List<RelationContext> r
         for (RelationContext relation : relations) {
             // build left deep join tree
             LogicalPlan right = withJoinRelations(visitRelation(relation), relation);
-            left = (left == null) ? right :
-                    new LogicalJoin<>(
-                            JoinType.CROSS_JOIN,
-                            ExpressionUtils.EMPTY_CONDITION,
-                            ExpressionUtils.EMPTY_CONDITION,
-                            new DistributeHint(DistributeType.NONE),
-                            Optional.empty(),
-                            left,
-                            right, null);
+            // check if it's unnest
+            boolean shouldBeParent = right instanceof LogicalGenerate
+                    && right.child(0) instanceof LogicalOneRowRelation;
+            left = (left == null) ? right
+                    : shouldBeParent ? ((LogicalGenerate) right).withChildren(ImmutableList.of(left))
+                            : new LogicalJoin<>(
+                                    JoinType.CROSS_JOIN,
+                                    ExpressionUtils.EMPTY_CONDITION,
+                                    ExpressionUtils.EMPTY_CONDITION,
+                                    new DistributeHint(DistributeType.NONE),
+                                    Optional.empty(),
+                                    left,
+                                    right, null);
             // TODO: pivot and lateral view
         }
         return left;

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

@@ -374,6 +374,8 @@ public enum RuleType {
     PUSH_DOWN_LIMIT_DISTINCT_THROUGH_JOIN(RuleTypeClass.REWRITE),
     PUSH_DOWN_LIMIT_DISTINCT_THROUGH_PROJECT_JOIN(RuleTypeClass.REWRITE),
     PUSH_DOWN_LIMIT_DISTINCT_THROUGH_UNION(RuleTypeClass.REWRITE),
+    // push down unnest
+    PUSH_DOWN_UNNEST_IN_PROJECT(RuleTypeClass.REWRITE),
     // adjust nullable
     ADJUST_NULLABLE(RuleTypeClass.REWRITE),
     ADJUST_CONJUNCTS_RETURN_TYPE(RuleTypeClass.REWRITE),

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

@@ -283,7 +283,7 @@ private LogicalPlan bindGenerate(MatchingContext<LogicalGenerate<Plan>> ctx) {
             if (!(boundGenerator instanceof TableGeneratingFunction)) {
                 throw new AnalysisException(boundGenerator.toSql() + " is not a TableGeneratingFunction");
             }
-            Function generator = (Function) boundGenerator;
+            Function generator = ExpressionUtils.convertUnnest((Function) boundGenerator);
             boundGenerators.add(generator);
 
             Slot boundSlot = new SlotReference(slot.getNameParts().get(1), generator.getDataType(),
@@ -312,18 +312,61 @@ boundSlot, new StringLiteral(fields.get(idx).getName())),
                 }
             }
         }
-        LogicalGenerate<Plan> ret = new LogicalGenerate<>(
-                boundGenerators.build(), outputSlots.build(), generate.child());
+        /*
+         * SELECT
+         *     id,
+         *     tags
+         * FROM
+         *     items
+         *     LEFT JOIN lateral unnest(tags) AS t(tag) ON t.tag = name;
+         *
+         * t.tag is unnest's output, so the conjunct t.tag = name may reference slot from child and its own output
+         *
+         */
+        int conjunctSize = generate.getConjuncts().size();
+        List<Expression> newConjuncts = new ArrayList<>(conjunctSize);
+        if (conjunctSize > 0) {
+            List<Slot> childOutputs = generate.child().getOutput();
+            List<Slot> conjunctsScopeSlots = new ArrayList<>(expandAlias.size() + childOutputs.size());
+            for (Alias alias : expandAlias) {
+                conjunctsScopeSlots.add(alias.toSlot());
+            }
+
+            conjunctsScopeSlots.addAll(childOutputs);
+            Scope conjunctsScope = toScope(cascadesContext, conjunctsScopeSlots);
+            ExpressionAnalyzer conjunctsAnalyzer = new ExpressionAnalyzer(
+                    generate, conjunctsScope, cascadesContext, true, false);
+            Map<Slot, Expression> replaceMap = ExpressionUtils.generateReplaceMap(expandAlias);
+            for (Expression expression : generate.getConjuncts()) {
+                expression = conjunctsAnalyzer.analyze(expression);
+                Expression newExpression = expression.rewriteDownShortCircuit(
+                        e -> replaceMap.getOrDefault(e, e));
+                newConjuncts.add(newExpression);
+            }
+        }
+
+        LogicalGenerate<Plan> logicalGenerate = new LogicalGenerate<>(
+                boundGenerators.build(), outputSlots.build(), ImmutableList.of(), newConjuncts, generate.child());
         if (!expandAlias.isEmpty()) {
+            // project should contain: generator.child slot + expandAlias
+            List<NamedExpression> allProjectSlots = new ArrayList<>(generate.child().getOutput().size()
+                    + expandAlias.size());
+            if (!(generate.child() instanceof LogicalOneRowRelation)) {
+                // project should contain: generator.child slot + expandAlias except:
+                allProjectSlots.addAll(generate.child().getOutput().stream()
+                        .map(NamedExpression.class::cast)
+                        .collect(Collectors.toList()));
+            } else {
+                // unnest with literal argument as unnest([1,2,3])
+                // we should not add LogicalOneRowRelation's output slot in this case
+                // so do nothing
+            }
             // we need a project to deal with explode(map) to struct with field alias
-            // project should contains: generator.child slot + expandAlias
-            List<NamedExpression> allProjectSlots = generate.child().getOutput().stream()
-                    .map(NamedExpression.class::cast)
-                    .collect(Collectors.toList());
             allProjectSlots.addAll(expandAlias);
-            return new LogicalProject<>(allProjectSlots, ret);
+            return new LogicalProject<>(allProjectSlots, logicalGenerate);
+        } else {
+            return logicalGenerate;
         }
-        return ret;
     }
 
     private LogicalSetOperation bindSetOperation(LogicalSetOperation setOperation) {

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

@@ -24,6 +24,7 @@
 import org.apache.doris.nereids.trees.expressions.WindowExpression;
 import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
 import org.apache.doris.nereids.trees.expressions.functions.generator.TableGeneratingFunction;
+import org.apache.doris.nereids.trees.expressions.functions.generator.Unnest;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.GroupingScalarFunction;
 import org.apache.doris.nereids.trees.expressions.typecoercion.TypeCheckResult;
 import org.apache.doris.nereids.trees.plans.Plan;
@@ -117,7 +118,7 @@ private void checkUnexpectedExpressions(Plan plan) {
         for (Expression expr : plan.getExpressions()) {
             expr.foreachUp(e -> {
                 for (Class<? extends Expression> type : unexpectedExpressionTypes) {
-                    if (type.isInstance(e)) {
+                    if (type.isInstance(e) && !(e instanceof Unnest)) {
                         throw new AnalysisException(plan.getType() + " can not contains "
                                 + type.getSimpleName() + " expression: " + ((Expression) e).toSql());
                     }

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

@@ -37,6 +37,7 @@
 import org.apache.doris.nereids.trees.expressions.WindowExpression;
 import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
 import org.apache.doris.nereids.trees.expressions.functions.agg.AnyValue;
+import org.apache.doris.nereids.trees.expressions.functions.generator.Unnest;
 import org.apache.doris.nereids.trees.expressions.literal.Literal;
 import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
 import org.apache.doris.nereids.trees.plans.Plan;
@@ -175,7 +176,7 @@ private LogicalPlan normalizeAgg(LogicalAggregate<Plan> aggregate, Optional<Logi
                     if (arg instanceof Literal) {
                         continue;
                     }
-                    if (arg.containsType(SubqueryExpr.class, WindowExpression.class)) {
+                    if (arg.containsType(SubqueryExpr.class, WindowExpression.class, Unnest.class)) {
                         needPushDownSelfExprs.add(arg);
                     } else {
                         needPushDownInputs.add(arg);

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

@@ -30,6 +30,7 @@ public Rule build() {
         return logicalGenerate().then(generate -> new PhysicalGenerate<>(
                 generate.getGenerators(),
                 generate.getGeneratorOutput(),
+                generate.getConjuncts(),
                 generate.getLogicalProperties(),
                 generate.child())
         ).toRule(RuleType.LOGICAL_GENERATE_TO_PHYSICAL_GENERATE);