[feature](nereids)support window function

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

@@ -187,6 +187,10 @@ limitClause
     | (LIMIT offset=INTEGER_VALUE COMMA limit=INTEGER_VALUE)
     ;
 
+partitionClause
+    : PARTITION BY expression (COMMA expression)*
+    ;
+
 joinType
     : INNER?
     | CROSS
@@ -316,7 +320,8 @@ primaryExpression
     | ASTERISK                                                                                 #star
     | qualifiedName DOT ASTERISK                                                               #star
     | functionIdentifier LEFT_PAREN ((DISTINCT|ALL)? arguments+=expression
-      (COMMA arguments+=expression)* (ORDER BY sortItem (COMMA sortItem)*)?)? RIGHT_PAREN      #functionCall
+      (COMMA arguments+=expression)* (ORDER BY sortItem (COMMA sortItem)*)?)? RIGHT_PAREN
+      (OVER windowSpec)?                                                                        #functionCall
     | LEFT_PAREN query RIGHT_PAREN                                                             #subqueryExpression
     | ATSIGN identifier                                                                        #userVariable
     | DOUBLEATSIGN (kind=(GLOBAL | SESSION) DOT)? identifier                                     #systemVariable
@@ -332,6 +337,33 @@ functionIdentifier
     | LEFT | RIGHT
     ;
 
+windowSpec
+    // todo: name for windowRef; we haven't support it
+    // : name=identifier
+    // | LEFT_PAREN name=identifier RIGHT_PAREN
+    : LEFT_PAREN
+        partitionClause?
+        sortClause?
+        windowFrame?
+        RIGHT_PAREN
+    ;
+
+windowFrame
+    : frameUnits start=frameBoundary
+    | frameUnits BETWEEN start=frameBoundary AND end=frameBoundary
+    ;
+
+frameUnits
+    : ROWS
+    | RANGE
+    ;
+
+frameBoundary
+    : UNBOUNDED boundType=(PRECEDING | FOLLOWING)
+    | boundType=CURRENT ROW
+    | expression boundType=(PRECEDING | FOLLOWING)
+    ;
+
 qualifiedName
     : identifier (DOT identifier)*
     ;
@@ -569,7 +601,6 @@ nonReserved
     | OUT
     | OUTER
     | OUTPUTFORMAT
-    | OVER
     | OVERLAPS
     | OVERLAY
     | OVERWRITE

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

@@ -64,7 +64,6 @@
  */
 public class AnalyticExpr extends Expr {
     private static final Logger LOG = LoggerFactory.getLogger(AnalyticExpr.class);
-    private static String NTILE = "NTILE";
 
     private FunctionCallExpr fnCall;
     private final List<Expr> partitionExprs;
@@ -81,6 +80,7 @@ public class AnalyticExpr extends Expr {
     // SQL string of this AnalyticExpr before standardization. Returned in toSqlImpl().
     private String sqlString;
 
+    private static String NTILE = "NTILE";
     private static String LEAD = "LEAD";
     private static String LAG = "LAG";
     private static String FIRSTVALUE = "FIRST_VALUE";
@@ -784,7 +784,7 @@ private void standardize(Analyzer analyzer) throws AnalysisException {
         }
 
         // Reverse the ordering and window for windows ending with UNBOUNDED FOLLOWING,
-        // and and not starting with UNBOUNDED PRECEDING.
+        // and not starting with UNBOUNDED PRECEDING.
         if (window != null
                 && window.getRightBoundary().getType() == BoundaryType.UNBOUNDED_FOLLOWING
                 && window.getLeftBoundary().getType() != BoundaryType.UNBOUNDED_PRECEDING) {
@@ -1002,4 +1002,8 @@ private String exprListToDigest(List<? extends Expr> exprs) {
         }
         return Joiner.on(", ").join(strings);
     }
+
+    @Override
+    public void finalizeImplForNereids() throws AnalysisException {
+    }
 }

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

@@ -41,7 +41,7 @@ public class AnalyticWindow {
             new Boundary(BoundaryType.UNBOUNDED_PRECEDING, null),
             new Boundary(BoundaryType.CURRENT_ROW, null));
 
-    enum Type {
+    public enum Type {
         ROWS("ROWS"),
         RANGE("RANGE");
 
@@ -61,7 +61,7 @@ public TAnalyticWindowType toThrift() {
         }
     }
 
-    enum BoundaryType {
+    public enum BoundaryType {
         UNBOUNDED_PRECEDING("UNBOUNDED PRECEDING"),
         UNBOUNDED_FOLLOWING("UNBOUNDED FOLLOWING"),
         CURRENT_ROW("CURRENT ROW"),
@@ -152,7 +152,7 @@ public Boundary(BoundaryType type, Expr e) {
         }
 
         // c'tor used by clone()
-        private Boundary(BoundaryType type, Expr e, BigDecimal offsetValue) {
+        public Boundary(BoundaryType type, Expr e, BigDecimal offsetValue) {
             Preconditions.checkState(
                     (type.isOffset() && e != null)
                     || (!type.isOffset() && e == null));

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

@@ -297,8 +297,8 @@ public InferPredicateState(InferPredicateState that) {
 
     // state shared between all objects of an Analyzer tree
     // TODO: Many maps here contain properties about tuples, e.g., whether
-    // a tuple is outer/semi joined, etc. Remove the maps in favor of making
-    // them properties of the tuple descriptor itself.
+    //  a tuple is outer/semi joined, etc. Remove the maps in favor of making
+    //  them properties of the tuple descriptor itself.
     private static class GlobalState {
         private final DescriptorTable descTbl = new DescriptorTable();
         private final Env env;

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

@@ -495,6 +495,10 @@ public boolean isAnalyticFn() {
         return isAnalyticFn;
     }
 
+    public void setIsAnalyticFn(boolean isAnalyticFn) {
+        this.isAnalyticFn = isAnalyticFn;
+    }
+
     public boolean isAggregateFn() {
         return isAggregateFn;
     }

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

@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.catalog;
+
+import org.apache.doris.nereids.trees.expressions.functions.window.DenseRank;
+import org.apache.doris.nereids.trees.expressions.functions.window.FirstValue;
+import org.apache.doris.nereids.trees.expressions.functions.window.Lag;
+import org.apache.doris.nereids.trees.expressions.functions.window.LastValue;
+import org.apache.doris.nereids.trees.expressions.functions.window.Lead;
+import org.apache.doris.nereids.trees.expressions.functions.window.Ntile;
+import org.apache.doris.nereids.trees.expressions.functions.window.Rank;
+import org.apache.doris.nereids.trees.expressions.functions.window.RowNumber;
+
+import com.google.common.collect.ImmutableList;
+
+/**
+ * Builtin aggregate functions.
+ *
+ * Note: Please ensure that this class only has some lists and no procedural code.
+ *       It helps to be clear and concise.
+ */
+public class BuiltinWindowFunctions implements FunctionHelper {
+
+    public final ImmutableList<WindowFunc> windowFunctions = ImmutableList.of(
+            window(DenseRank.class, "dense_rank"),
+            window(FirstValue.class, "first_value"),
+            window(Lag.class, "lag"),
+            window(LastValue.class, "last_value"),
+            window(Lead.class, "lead"),
+            window(Ntile.class, "ntile"),
+            window(Rank.class, "rank"),
+            window(RowNumber.class, "row_number")
+    );
+
+    public static final BuiltinWindowFunctions INSTANCE = new BuiltinWindowFunctions();
+
+    // Note: Do not add any code here!
+    private BuiltinWindowFunctions() {}
+}

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

@@ -23,6 +23,7 @@
 import org.apache.doris.nereids.trees.expressions.functions.generator.TableGeneratingFunction;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.ScalarFunction;
 import org.apache.doris.nereids.trees.expressions.functions.table.TableValuedFunction;
+import org.apache.doris.nereids.trees.expressions.functions.window.WindowFunction;
 
 import com.google.common.collect.ImmutableList;
 
@@ -69,6 +70,10 @@ default AggregateFunc agg(Class<? extends AggregateFunction> functionClass) {
         return new AggregateFunc(functionClass, functionName);
     }
 
+    default WindowFunc window(Class<? extends WindowFunction> functionClass, String... functionNames) {
+        return new WindowFunc(functionClass, functionNames);
+    }
+
     /**
      * Resolve AggregateFunction class, convert to FunctionBuilder and wrap to AggregateFunc
      * @param functionClass the AggregateFunction class
@@ -129,4 +134,10 @@ public TableGeneratingFunc(Class<? extends TableGeneratingFunction> functionClas
             super(functionClass, names);
         }
     }
+
+    class WindowFunc extends NamedFunc<WindowFunction> {
+        public WindowFunc(Class<? extends WindowFunction> functionClass, String... names) {
+            super(functionClass, names);
+        }
+    }
 }

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

@@ -88,6 +88,7 @@ private void registerBuiltinFunctions(Map<String, List<FunctionBuilder>> name2Bu
         FunctionHelper.addFunctions(name2Builders, BuiltinAggregateFunctions.INSTANCE.aggregateFunctions);
         FunctionHelper.addFunctions(name2Builders, BuiltinTableValuedFunctions.INSTANCE.tableValuedFunctions);
         FunctionHelper.addFunctions(name2Builders, BuiltinTableGeneratingFunctions.INSTANCE.tableGeneratingFunctions);
+        FunctionHelper.addFunctions(name2Builders, BuiltinWindowFunctions.INSTANCE.windowFunctions);
     }
 
     public String getCandidateHint(String name, List<FunctionBuilder> candidateBuilders) {

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

@@ -96,7 +96,6 @@ public void plan(StatementBase queryStmt, org.apache.doris.thrift.TQueryOptions
         if (explainLevel.isPlanLevel) {
             return;
         }
-
         PhysicalPlan physicalPlan = (PhysicalPlan) resultPlan;
         PhysicalPlanTranslator physicalPlanTranslator = new PhysicalPlanTranslator();
         PlanTranslatorContext planTranslatorContext = new PlanTranslatorContext(cascadesContext);
@@ -161,7 +160,6 @@ public Plan plan(LogicalPlan plan, PhysicalProperties requireProperties, Explain
                     return analyzedPlan;
                 }
             }
-
             // rule-based optimize
             rewrite();
             if (explainLevel == ExplainLevel.REWRITTEN_PLAN || explainLevel == ExplainLevel.ALL_PLAN) {
@@ -170,7 +168,6 @@ public Plan plan(LogicalPlan plan, PhysicalProperties requireProperties, Explain
                     return rewrittenPlan;
                 }
             }
-
             deriveStats();
 
             if (statementContext.getConnectContext().getSessionVariable().isEnableDPHypOptimizer()) {
@@ -182,7 +179,6 @@ public Plan plan(LogicalPlan plan, PhysicalProperties requireProperties, Explain
 
             PhysicalPlan physicalPlan = chooseBestPlan(getRoot(), requireProperties);
 
-            // post-process physical plan out of memo, just for future use.
             physicalPlan = postProcess(physicalPlan);
             if (explainLevel == ExplainLevel.OPTIMIZED_PLAN || explainLevel == ExplainLevel.ALL_PLAN) {
                 optimizedPlan = physicalPlan;

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

@@ -76,6 +76,7 @@
 import org.apache.doris.nereids.trees.expressions.functions.scalar.JsonArray;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.JsonObject;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.ScalarFunction;
+import org.apache.doris.nereids.trees.expressions.functions.window.WindowFunction;
 import org.apache.doris.nereids.trees.expressions.literal.DateTimeLiteral;
 import org.apache.doris.nereids.trees.expressions.literal.DateTimeV2Literal;
 import org.apache.doris.nereids.trees.expressions.literal.Literal;
@@ -282,13 +283,58 @@ public Expr visitCast(Cast cast, PlanTranslatorContext context) {
     @Override
     public Expr visitInPredicate(InPredicate inPredicate, PlanTranslatorContext context) {
         List<Expr> inList = inPredicate.getOptions().stream()
-                .map(e -> translate(e, context))
+                .map(e -> e.accept(this, context))
                 .collect(Collectors.toList());
         return new org.apache.doris.analysis.InPredicate(inPredicate.getCompareExpr().accept(this, context),
                 inList,
                 false);
     }
 
+    @Override
+    public Expr visitWindowFunction(WindowFunction function, PlanTranslatorContext context) {
+        // translate argument types from DataType to Type
+        List<Expr> catalogArguments = function.getArguments()
+                .stream()
+                .map(arg -> arg.accept(this, context))
+                .collect(ImmutableList.toImmutableList());
+        ImmutableList<Type> argTypes = catalogArguments.stream()
+                .map(arg -> arg.getType())
+                .collect(ImmutableList.toImmutableList());
+
+        // translate argument from List<Expression> to FunctionParams
+        List<Expr> arguments = function.getArguments()
+                .stream()
+                .map(arg -> new SlotRef(arg.getDataType().toCatalogDataType(), arg.nullable()))
+                .collect(ImmutableList.toImmutableList());
+        FunctionParams windowFnParams = new FunctionParams(false, arguments);
+
+        // translate isNullable()
+        NullableMode nullableMode = function.nullable()
+                ? NullableMode.ALWAYS_NULLABLE
+                : NullableMode.ALWAYS_NOT_NULLABLE;
+
+        // translate function from WindowFunction to old AggregateFunction
+        boolean isAnalyticFunction = true;
+        org.apache.doris.catalog.AggregateFunction catalogFunction = new org.apache.doris.catalog.AggregateFunction(
+                new FunctionName(function.getName()), argTypes,
+                function.getDataType().toCatalogDataType(),
+                function.getDataType().toCatalogDataType(),
+                function.hasVarArguments(),
+                null, "", "", null, "",
+                null, "", null, false,
+                isAnalyticFunction, false, TFunctionBinaryType.BUILTIN,
+                true, true, nullableMode
+        );
+
+        // generate FunctionCallExpr
+        boolean isMergeFn = false;
+        FunctionCallExpr functionCallExpr =
+                new FunctionCallExpr(catalogFunction, windowFnParams, windowFnParams, isMergeFn, catalogArguments);
+        functionCallExpr.setIsAnalyticFnCall(true);
+        return functionCallExpr;
+
+    }
+
     @Override
     public Expr visitScalarFunction(ScalarFunction function, PlanTranslatorContext context) {
         List<Expression> nereidsArguments = adaptFunctionArgumentsForBackends(function);