[GitHub] [doris] morrySnow commented on a diff in pull request #17968: [Feature](Nereids)add rule to change scalar sub-query to window function

["morrySnow (via GitHub)" <gi...@apache.org>]

morrySnow commented on code in PR #17968:
URL: https://github.com/apache/doris/pull/17968#discussion_r1146099814


##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(

Review Comment:
   ```suggestion
       private static final Set<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
   ```



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class

Review Comment:
   ```suggestion
       private static final Set<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
               LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
       );
       private static final Set<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
               LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
   ```



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t

Review Comment:
   do not use sql, please use plan tree to explain this rule



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class

Review Comment:
   should not restirct on OLAP SCAN, all scan could be applied this rule



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate

Review Comment:
   it is not nessesary to check apply.left() != null



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }

Review Comment:
   why must exist a project?



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()

Review Comment:
   `checkPlanType()`, `checkRelation` and `checkAggType()` why put here? they not process `logicalFilter` at all



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()
+                        && checkRelation(apply.getCorrelationSlot())
+                        && checkPredicate(Sets.newHashSet(outerFilter.getConjuncts()),
+                        Sets.newHashSet(logicalFilter.getConjuncts()))
+        ).isPresent();
+    }
+
+    private Plan trans(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<Plan, LogicalAggregate<Plan>> apply = node.child().child();
+        LogicalAggregate<Plan> agg = apply.right();
+
+        // for example, in tpc-h Q17
+        // window filter conjuncts is cast(l_quantity#id1 as decimal(27, 9)) < `0.2 * avg(l_quantity)`#id2 and
+        // 0.2 * avg(l_quantity#id3) as `0.2 * l_quantity`#id2 of agg's output expression
+        // we change it to cast(l_quantity#id1 as decimal(27, 9)) < 0.2 * `avg(l_quantity#id1) over(window)`#id4
+        // and avg(l_quantity#id1) over(window) as `avg(l_quantity#id1) over(window)`#id4
+        // first we get the slots of the two sides of apply
+
+        Preconditions.checkArgument(apply.getSubCorrespondingConject().get() instanceof ComparisonPredicate);
+        ComparisonPredicate windowFilterConjunct = PlanUtils.maybeCommuteComparisonPredicate(
+                ((ComparisonPredicate) apply.getSubCorrespondingConject().get()), apply.left());
+
+        // build window function, replace the slot
+        List<Expression> windowAggSlots = windowFilterConjunct.left().collectToList(Slot.class::isInstance);
+        Preconditions.checkArgument(windowAggSlots.size() == 1);
+
+        // adjust agg function's nullable.
+        AggregateFunction function = functions.get(0);
+        if (function instanceof NullableAggregateFunction) {
+            function = ((NullableAggregateFunction) function).withAlwaysNullable(false);
+        }
+
+        WindowExpression windowFunction = createWindowFunction(apply.getCorrelationSlot(),
+                function.withChildren(windowAggSlots));
+        NamedExpression windowFunctionAlias = new Alias(windowFunction, "wf");
+
+        // build filter conjunct, get the alias of the agg output and extract its child.
+        // then replace the agg to window function, then build conjunct
+        NamedExpression aggOut = agg.getOutputExpressions().get(0);
+        Expression aggOutExpr = aggOut.child(0);
+        aggOutExpr = new FunctionReplacer().replace(aggOutExpr, windowFunctionAlias.toSlot());
+
+        windowFilterConjunct = ((ComparisonPredicate) windowFilterConjunct
+                .withChildren(windowFilterConjunct.child(0), aggOutExpr));
+
+        LogicalFilter newFilter = ((LogicalFilter) node.withChildren(apply.left()));
+        LogicalWindow newWindow = new LogicalWindow<>(ImmutableList.of(windowFunctionAlias), newFilter);
+        LogicalFilter windowFilter = new LogicalFilter<>(ImmutableSet.of(windowFilterConjunct), newWindow);
+        return windowFilter;
+    }
+
+    // check children's nodes because query process will be changed
+    private boolean checkPlanType() {
+        return outerPlans.stream().allMatch(p -> LEFT_SUPPORTED_PLAN.contains(p.getClass()))
+                && innerPlans.stream().allMatch(p -> RIGHT_SUPPORTED_PLAN.contains(p.getClass()));
+    }
+
+    // check aggregation of inner scope
+    private boolean checkAggType() {
+        List<LogicalAggregate> aggSet = innerPlans.stream().filter(LogicalAggregate.class::isInstance)
+                .map(LogicalAggregate.class::cast)
+                .collect(Collectors.toList());
+        if (aggSet.size() > 1) {
+            // window functions don't support nesting.
+            return false;
+        }
+        aggOp = aggSet.get(0);
+        functions = ((List<AggregateFunction>) ExpressionUtils.<AggregateFunction>collectAll(
+                aggOp.getOutputExpressions(), AggregateFunction.class::isInstance));
+        Preconditions.checkArgument(functions.size() == 1);
+        return functions.stream().allMatch(f -> SUPPORTED_FUNCTION.contains(f.getClass()) && !f.isDistinct());
+    }
+
+    // check if the relations of the outer's includes the inner's
+    private boolean checkRelation(List<Expression> correlatedSlots) {
+        List<LogicalRelation> outerTables = outerPlans.stream().filter(LogicalRelation.class::isInstance)
+                .map(LogicalRelation.class::cast)
+                .collect(Collectors.toList());
+        List<LogicalRelation> innerTables = innerPlans.stream().filter(LogicalRelation.class::isInstance)
+                .map(LogicalRelation.class::cast)
+                .collect(Collectors.toList());
+
+        Set<Long> outerIds = outerTables.stream().map(node -> node.getTable().getId()).collect(Collectors.toSet());
+        Set<Long> innerIds = innerTables.stream().map(node -> node.getTable().getId()).collect(Collectors.toSet());
+
+        if (outerTables.size() != outerIds.size() || innerTables.size() != innerIds.size()
+                || outerIds.size() != innerIds.size() + 1) {

Review Comment:
   why we need them ? `outerTables.size() != outerIds.size() || innerTables.size() != innerIds.size()`



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()
+                        && checkRelation(apply.getCorrelationSlot())
+                        && checkPredicate(Sets.newHashSet(outerFilter.getConjuncts()),
+                        Sets.newHashSet(logicalFilter.getConjuncts()))
+        ).isPresent();
+    }
+
+    private Plan trans(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<Plan, LogicalAggregate<Plan>> apply = node.child().child();
+        LogicalAggregate<Plan> agg = apply.right();
+
+        // for example, in tpc-h Q17
+        // window filter conjuncts is cast(l_quantity#id1 as decimal(27, 9)) < `0.2 * avg(l_quantity)`#id2 and
+        // 0.2 * avg(l_quantity#id3) as `0.2 * l_quantity`#id2 of agg's output expression
+        // we change it to cast(l_quantity#id1 as decimal(27, 9)) < 0.2 * `avg(l_quantity#id1) over(window)`#id4
+        // and avg(l_quantity#id1) over(window) as `avg(l_quantity#id1) over(window)`#id4
+        // first we get the slots of the two sides of apply
+
+        Preconditions.checkArgument(apply.getSubCorrespondingConject().get() instanceof ComparisonPredicate);

Review Comment:
   why check here? where to ensure `apply.getSubCorrespondingConject()` is not empty and it  must be a `ComparisonPredicate`?
   if no where to ensure this.
   we should add this to check and return false if it is not satisfy the restriction 



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()
+                        && checkRelation(apply.getCorrelationSlot())
+                        && checkPredicate(Sets.newHashSet(outerFilter.getConjuncts()),
+                        Sets.newHashSet(logicalFilter.getConjuncts()))
+        ).isPresent();
+    }
+
+    private Plan trans(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<Plan, LogicalAggregate<Plan>> apply = node.child().child();
+        LogicalAggregate<Plan> agg = apply.right();
+
+        // for example, in tpc-h Q17
+        // window filter conjuncts is cast(l_quantity#id1 as decimal(27, 9)) < `0.2 * avg(l_quantity)`#id2 and
+        // 0.2 * avg(l_quantity#id3) as `0.2 * l_quantity`#id2 of agg's output expression
+        // we change it to cast(l_quantity#id1 as decimal(27, 9)) < 0.2 * `avg(l_quantity#id1) over(window)`#id4
+        // and avg(l_quantity#id1) over(window) as `avg(l_quantity#id1) over(window)`#id4
+        // first we get the slots of the two sides of apply
+
+        Preconditions.checkArgument(apply.getSubCorrespondingConject().get() instanceof ComparisonPredicate);
+        ComparisonPredicate windowFilterConjunct = PlanUtils.maybeCommuteComparisonPredicate(
+                ((ComparisonPredicate) apply.getSubCorrespondingConject().get()), apply.left());
+
+        // build window function, replace the slot
+        List<Expression> windowAggSlots = windowFilterConjunct.left().collectToList(Slot.class::isInstance);
+        Preconditions.checkArgument(windowAggSlots.size() == 1);
+
+        // adjust agg function's nullable.
+        AggregateFunction function = functions.get(0);
+        if (function instanceof NullableAggregateFunction) {
+            function = ((NullableAggregateFunction) function).withAlwaysNullable(false);
+        }
+
+        WindowExpression windowFunction = createWindowFunction(apply.getCorrelationSlot(),
+                function.withChildren(windowAggSlots));
+        NamedExpression windowFunctionAlias = new Alias(windowFunction, "wf");
+
+        // build filter conjunct, get the alias of the agg output and extract its child.
+        // then replace the agg to window function, then build conjunct
+        NamedExpression aggOut = agg.getOutputExpressions().get(0);
+        Expression aggOutExpr = aggOut.child(0);
+        aggOutExpr = new FunctionReplacer().replace(aggOutExpr, windowFunctionAlias.toSlot());
+
+        windowFilterConjunct = ((ComparisonPredicate) windowFilterConjunct
+                .withChildren(windowFilterConjunct.child(0), aggOutExpr));
+
+        LogicalFilter newFilter = ((LogicalFilter) node.withChildren(apply.left()));
+        LogicalWindow newWindow = new LogicalWindow<>(ImmutableList.of(windowFunctionAlias), newFilter);
+        LogicalFilter windowFilter = new LogicalFilter<>(ImmutableSet.of(windowFilterConjunct), newWindow);
+        return windowFilter;
+    }
+
+    // check children's nodes because query process will be changed
+    private boolean checkPlanType() {
+        return outerPlans.stream().allMatch(p -> LEFT_SUPPORTED_PLAN.contains(p.getClass()))
+                && innerPlans.stream().allMatch(p -> RIGHT_SUPPORTED_PLAN.contains(p.getClass()));
+    }
+
+    // check aggregation of inner scope
+    private boolean checkAggType() {
+        List<LogicalAggregate> aggSet = innerPlans.stream().filter(LogicalAggregate.class::isInstance)
+                .map(LogicalAggregate.class::cast)
+                .collect(Collectors.toList());
+        if (aggSet.size() > 1) {
+            // window functions don't support nesting.
+            return false;
+        }
+        aggOp = aggSet.get(0);
+        functions = ((List<AggregateFunction>) ExpressionUtils.<AggregateFunction>collectAll(
+                aggOp.getOutputExpressions(), AggregateFunction.class::isInstance));
+        Preconditions.checkArgument(functions.size() == 1);
+        return functions.stream().allMatch(f -> SUPPORTED_FUNCTION.contains(f.getClass()) && !f.isDistinct());
+    }
+
+    // check if the relations of the outer's includes the inner's
+    private boolean checkRelation(List<Expression> correlatedSlots) {

Review Comment:
   current code in check relation is too tight. we could loose it in future



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()
+                        && checkRelation(apply.getCorrelationSlot())
+                        && checkPredicate(Sets.newHashSet(outerFilter.getConjuncts()),
+                        Sets.newHashSet(logicalFilter.getConjuncts()))
+        ).isPresent();
+    }
+
+    private Plan trans(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<Plan, LogicalAggregate<Plan>> apply = node.child().child();
+        LogicalAggregate<Plan> agg = apply.right();
+
+        // for example, in tpc-h Q17
+        // window filter conjuncts is cast(l_quantity#id1 as decimal(27, 9)) < `0.2 * avg(l_quantity)`#id2 and
+        // 0.2 * avg(l_quantity#id3) as `0.2 * l_quantity`#id2 of agg's output expression
+        // we change it to cast(l_quantity#id1 as decimal(27, 9)) < 0.2 * `avg(l_quantity#id1) over(window)`#id4
+        // and avg(l_quantity#id1) over(window) as `avg(l_quantity#id1) over(window)`#id4
+        // first we get the slots of the two sides of apply
+
+        Preconditions.checkArgument(apply.getSubCorrespondingConject().get() instanceof ComparisonPredicate);

Review Comment:
   getSubCorrespondingConject fix this typo



##########
fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunctionTest.java:
##########
@@ -0,0 +1,46 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.datasets.tpch.TPCHTestBase;
+import org.apache.doris.nereids.datasets.tpch.TPCHUtils;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.util.MemoPatternMatchSupported;
+import org.apache.doris.nereids.util.PlanChecker;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class AggScalarSubQueryToWindowFunctionTest extends TPCHTestBase implements MemoPatternMatchSupported {

Review Comment:
   i think we need use plan constructor and plan checker to do rule UT, and we need to verify all check could work well in UT. in other word u need to construct the plan that cannot be rewritten and check the rule do the right thing



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########


Review Comment:
   we must use Precodition.check very carefully, rewrite rule should not throw exception when the rule cannot be applied. we can use precodition only when we want to check a bug



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########


Review Comment:
   fix warning hints by idea as mush as possible.



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class

Review Comment:
   i think filter should in left plans



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()
+                        && checkRelation(apply.getCorrelationSlot())
+                        && checkPredicate(Sets.newHashSet(outerFilter.getConjuncts()),
+                        Sets.newHashSet(logicalFilter.getConjuncts()))
+        ).isPresent();
+    }
+
+    private Plan trans(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<Plan, LogicalAggregate<Plan>> apply = node.child().child();
+        LogicalAggregate<Plan> agg = apply.right();
+
+        // for example, in tpc-h Q17
+        // window filter conjuncts is cast(l_quantity#id1 as decimal(27, 9)) < `0.2 * avg(l_quantity)`#id2 and
+        // 0.2 * avg(l_quantity#id3) as `0.2 * l_quantity`#id2 of agg's output expression
+        // we change it to cast(l_quantity#id1 as decimal(27, 9)) < 0.2 * `avg(l_quantity#id1) over(window)`#id4
+        // and avg(l_quantity#id1) over(window) as `avg(l_quantity#id1) over(window)`#id4
+        // first we get the slots of the two sides of apply
+
+        Preconditions.checkArgument(apply.getSubCorrespondingConject().get() instanceof ComparisonPredicate);
+        ComparisonPredicate windowFilterConjunct = PlanUtils.maybeCommuteComparisonPredicate(
+                ((ComparisonPredicate) apply.getSubCorrespondingConject().get()), apply.left());
+
+        // build window function, replace the slot
+        List<Expression> windowAggSlots = windowFilterConjunct.left().collectToList(Slot.class::isInstance);
+        Preconditions.checkArgument(windowAggSlots.size() == 1);
+
+        // adjust agg function's nullable.
+        AggregateFunction function = functions.get(0);
+        if (function instanceof NullableAggregateFunction) {
+            function = ((NullableAggregateFunction) function).withAlwaysNullable(false);
+        }
+
+        WindowExpression windowFunction = createWindowFunction(apply.getCorrelationSlot(),
+                function.withChildren(windowAggSlots));
+        NamedExpression windowFunctionAlias = new Alias(windowFunction, "wf");
+
+        // build filter conjunct, get the alias of the agg output and extract its child.
+        // then replace the agg to window function, then build conjunct
+        NamedExpression aggOut = agg.getOutputExpressions().get(0);
+        Expression aggOutExpr = aggOut.child(0);
+        aggOutExpr = new FunctionReplacer().replace(aggOutExpr, windowFunctionAlias.toSlot());
+
+        windowFilterConjunct = ((ComparisonPredicate) windowFilterConjunct
+                .withChildren(windowFilterConjunct.child(0), aggOutExpr));
+
+        LogicalFilter newFilter = ((LogicalFilter) node.withChildren(apply.left()));
+        LogicalWindow newWindow = new LogicalWindow<>(ImmutableList.of(windowFunctionAlias), newFilter);
+        LogicalFilter windowFilter = new LogicalFilter<>(ImmutableSet.of(windowFilterConjunct), newWindow);
+        return windowFilter;
+    }
+
+    // check children's nodes because query process will be changed
+    private boolean checkPlanType() {
+        return outerPlans.stream().allMatch(p -> LEFT_SUPPORTED_PLAN.contains(p.getClass()))
+                && innerPlans.stream().allMatch(p -> RIGHT_SUPPORTED_PLAN.contains(p.getClass()));
+    }
+
+    // check aggregation of inner scope
+    private boolean checkAggType() {
+        List<LogicalAggregate> aggSet = innerPlans.stream().filter(LogicalAggregate.class::isInstance)
+                .map(LogicalAggregate.class::cast)
+                .collect(Collectors.toList());
+        if (aggSet.size() > 1) {
+            // window functions don't support nesting.
+            return false;
+        }
+        aggOp = aggSet.get(0);
+        functions = ((List<AggregateFunction>) ExpressionUtils.<AggregateFunction>collectAll(
+                aggOp.getOutputExpressions(), AggregateFunction.class::isInstance));
+        Preconditions.checkArgument(functions.size() == 1);
+        return functions.stream().allMatch(f -> SUPPORTED_FUNCTION.contains(f.getClass()) && !f.isDistinct());
+    }
+
+    // check if the relations of the outer's includes the inner's
+    private boolean checkRelation(List<Expression> correlatedSlots) {
+        List<LogicalRelation> outerTables = outerPlans.stream().filter(LogicalRelation.class::isInstance)
+                .map(LogicalRelation.class::cast)
+                .collect(Collectors.toList());
+        List<LogicalRelation> innerTables = innerPlans.stream().filter(LogicalRelation.class::isInstance)
+                .map(LogicalRelation.class::cast)
+                .collect(Collectors.toList());
+
+        Set<Long> outerIds = outerTables.stream().map(node -> node.getTable().getId()).collect(Collectors.toSet());
+        Set<Long> innerIds = innerTables.stream().map(node -> node.getTable().getId()).collect(Collectors.toSet());
+
+        if (outerTables.size() != outerIds.size() || innerTables.size() != innerIds.size()
+                || outerIds.size() != innerIds.size() + 1) {
+            return false;
+        }
+
+        outerIds.removeAll(innerIds);
+        if (outerIds.size() != 1) {
+            return false;
+        }

Review Comment:
   only need to check outer table contains all inner table, and outer table that do not appear in inner have no join condition with the table correlated with inner.



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/util/PlanUtils.java:
##########
@@ -38,4 +42,13 @@ public static Optional<LogicalFilter<? extends Plan>> filter(Set<Expression> pre
     public static Plan filterOrSelf(Set<Expression> predicates, Plan plan) {
         return filter(predicates, plan).map(Plan.class::cast).orElse(plan);
     }
+
+    public static ComparisonPredicate maybeCommuteComparisonPredicate(ComparisonPredicate expression, Plan left) {

Review Comment:
   add a java doc on it



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()
+                        && checkRelation(apply.getCorrelationSlot())
+                        && checkPredicate(Sets.newHashSet(outerFilter.getConjuncts()),
+                        Sets.newHashSet(logicalFilter.getConjuncts()))
+        ).isPresent();
+    }
+
+    private Plan trans(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<Plan, LogicalAggregate<Plan>> apply = node.child().child();
+        LogicalAggregate<Plan> agg = apply.right();
+
+        // for example, in tpc-h Q17
+        // window filter conjuncts is cast(l_quantity#id1 as decimal(27, 9)) < `0.2 * avg(l_quantity)`#id2 and
+        // 0.2 * avg(l_quantity#id3) as `0.2 * l_quantity`#id2 of agg's output expression
+        // we change it to cast(l_quantity#id1 as decimal(27, 9)) < 0.2 * `avg(l_quantity#id1) over(window)`#id4
+        // and avg(l_quantity#id1) over(window) as `avg(l_quantity#id1) over(window)`#id4
+        // first we get the slots of the two sides of apply
+
+        Preconditions.checkArgument(apply.getSubCorrespondingConject().get() instanceof ComparisonPredicate);
+        ComparisonPredicate windowFilterConjunct = PlanUtils.maybeCommuteComparisonPredicate(
+                ((ComparisonPredicate) apply.getSubCorrespondingConject().get()), apply.left());
+
+        // build window function, replace the slot
+        List<Expression> windowAggSlots = windowFilterConjunct.left().collectToList(Slot.class::isInstance);
+        Preconditions.checkArgument(windowAggSlots.size() == 1);

Review Comment:
   we should not check, unless u have verify it before or `windowAggSlots.size() != 1` represent a bug. if so, u need add some error msg



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()
+                        && checkRelation(apply.getCorrelationSlot())
+                        && checkPredicate(Sets.newHashSet(outerFilter.getConjuncts()),
+                        Sets.newHashSet(logicalFilter.getConjuncts()))
+        ).isPresent();
+    }
+
+    private Plan trans(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<Plan, LogicalAggregate<Plan>> apply = node.child().child();
+        LogicalAggregate<Plan> agg = apply.right();
+
+        // for example, in tpc-h Q17
+        // window filter conjuncts is cast(l_quantity#id1 as decimal(27, 9)) < `0.2 * avg(l_quantity)`#id2 and
+        // 0.2 * avg(l_quantity#id3) as `0.2 * l_quantity`#id2 of agg's output expression
+        // we change it to cast(l_quantity#id1 as decimal(27, 9)) < 0.2 * `avg(l_quantity#id1) over(window)`#id4
+        // and avg(l_quantity#id1) over(window) as `avg(l_quantity#id1) over(window)`#id4
+        // first we get the slots of the two sides of apply
+
+        Preconditions.checkArgument(apply.getSubCorrespondingConject().get() instanceof ComparisonPredicate);
+        ComparisonPredicate windowFilterConjunct = PlanUtils.maybeCommuteComparisonPredicate(
+                ((ComparisonPredicate) apply.getSubCorrespondingConject().get()), apply.left());
+
+        // build window function, replace the slot
+        List<Expression> windowAggSlots = windowFilterConjunct.left().collectToList(Slot.class::isInstance);
+        Preconditions.checkArgument(windowAggSlots.size() == 1);
+
+        // adjust agg function's nullable.
+        AggregateFunction function = functions.get(0);
+        if (function instanceof NullableAggregateFunction) {
+            function = ((NullableAggregateFunction) function).withAlwaysNullable(false);
+        }
+
+        WindowExpression windowFunction = createWindowFunction(apply.getCorrelationSlot(),
+                function.withChildren(windowAggSlots));
+        NamedExpression windowFunctionAlias = new Alias(windowFunction, "wf");
+
+        // build filter conjunct, get the alias of the agg output and extract its child.
+        // then replace the agg to window function, then build conjunct
+        NamedExpression aggOut = agg.getOutputExpressions().get(0);
+        Expression aggOutExpr = aggOut.child(0);
+        aggOutExpr = new FunctionReplacer().replace(aggOutExpr, windowFunctionAlias.toSlot());
+
+        windowFilterConjunct = ((ComparisonPredicate) windowFilterConjunct
+                .withChildren(windowFilterConjunct.child(0), aggOutExpr));
+
+        LogicalFilter newFilter = ((LogicalFilter) node.withChildren(apply.left()));
+        LogicalWindow newWindow = new LogicalWindow<>(ImmutableList.of(windowFunctionAlias), newFilter);
+        LogicalFilter windowFilter = new LogicalFilter<>(ImmutableSet.of(windowFilterConjunct), newWindow);
+        return windowFilter;
+    }
+
+    // check children's nodes because query process will be changed
+    private boolean checkPlanType() {
+        return outerPlans.stream().allMatch(p -> LEFT_SUPPORTED_PLAN.contains(p.getClass()))
+                && innerPlans.stream().allMatch(p -> RIGHT_SUPPORTED_PLAN.contains(p.getClass()));
+    }
+
+    // check aggregation of inner scope
+    private boolean checkAggType() {
+        List<LogicalAggregate> aggSet = innerPlans.stream().filter(LogicalAggregate.class::isInstance)
+                .map(LogicalAggregate.class::cast)
+                .collect(Collectors.toList());
+        if (aggSet.size() > 1) {
+            // window functions don't support nesting.
+            return false;
+        }
+        aggOp = aggSet.get(0);
+        functions = ((List<AggregateFunction>) ExpressionUtils.<AggregateFunction>collectAll(
+                aggOp.getOutputExpressions(), AggregateFunction.class::isInstance));
+        Preconditions.checkArgument(functions.size() == 1);
+        return functions.stream().allMatch(f -> SUPPORTED_FUNCTION.contains(f.getClass()) && !f.isDistinct());
+    }
+
+    // check if the relations of the outer's includes the inner's
+    private boolean checkRelation(List<Expression> correlatedSlots) {
+        List<LogicalRelation> outerTables = outerPlans.stream().filter(LogicalRelation.class::isInstance)
+                .map(LogicalRelation.class::cast)
+                .collect(Collectors.toList());
+        List<LogicalRelation> innerTables = innerPlans.stream().filter(LogicalRelation.class::isInstance)
+                .map(LogicalRelation.class::cast)
+                .collect(Collectors.toList());
+
+        Set<Long> outerIds = outerTables.stream().map(node -> node.getTable().getId()).collect(Collectors.toSet());
+        Set<Long> innerIds = innerTables.stream().map(node -> node.getTable().getId()).collect(Collectors.toSet());
+
+        if (outerTables.size() != outerIds.size() || innerTables.size() != innerIds.size()
+                || outerIds.size() != innerIds.size() + 1) {

Review Comment:
   why `outerIds.size() != innerIds.size() + 1`? paper do not has this restirct



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()
+                        && checkRelation(apply.getCorrelationSlot())
+                        && checkPredicate(Sets.newHashSet(outerFilter.getConjuncts()),
+                        Sets.newHashSet(logicalFilter.getConjuncts()))
+        ).isPresent();
+    }
+
+    private Plan trans(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<Plan, LogicalAggregate<Plan>> apply = node.child().child();
+        LogicalAggregate<Plan> agg = apply.right();
+
+        // for example, in tpc-h Q17
+        // window filter conjuncts is cast(l_quantity#id1 as decimal(27, 9)) < `0.2 * avg(l_quantity)`#id2 and
+        // 0.2 * avg(l_quantity#id3) as `0.2 * l_quantity`#id2 of agg's output expression
+        // we change it to cast(l_quantity#id1 as decimal(27, 9)) < 0.2 * `avg(l_quantity#id1) over(window)`#id4
+        // and avg(l_quantity#id1) over(window) as `avg(l_quantity#id1) over(window)`#id4
+        // first we get the slots of the two sides of apply
+
+        Preconditions.checkArgument(apply.getSubCorrespondingConject().get() instanceof ComparisonPredicate);
+        ComparisonPredicate windowFilterConjunct = PlanUtils.maybeCommuteComparisonPredicate(
+                ((ComparisonPredicate) apply.getSubCorrespondingConject().get()), apply.left());
+
+        // build window function, replace the slot
+        List<Expression> windowAggSlots = windowFilterConjunct.left().collectToList(Slot.class::isInstance);
+        Preconditions.checkArgument(windowAggSlots.size() == 1);
+
+        // adjust agg function's nullable.
+        AggregateFunction function = functions.get(0);
+        if (function instanceof NullableAggregateFunction) {
+            function = ((NullableAggregateFunction) function).withAlwaysNullable(false);
+        }
+
+        WindowExpression windowFunction = createWindowFunction(apply.getCorrelationSlot(),
+                function.withChildren(windowAggSlots));
+        NamedExpression windowFunctionAlias = new Alias(windowFunction, "wf");
+
+        // build filter conjunct, get the alias of the agg output and extract its child.
+        // then replace the agg to window function, then build conjunct
+        NamedExpression aggOut = agg.getOutputExpressions().get(0);
+        Expression aggOutExpr = aggOut.child(0);
+        aggOutExpr = new FunctionReplacer().replace(aggOutExpr, windowFunctionAlias.toSlot());
+
+        windowFilterConjunct = ((ComparisonPredicate) windowFilterConjunct
+                .withChildren(windowFilterConjunct.child(0), aggOutExpr));
+
+        LogicalFilter newFilter = ((LogicalFilter) node.withChildren(apply.left()));
+        LogicalWindow newWindow = new LogicalWindow<>(ImmutableList.of(windowFunctionAlias), newFilter);
+        LogicalFilter windowFilter = new LogicalFilter<>(ImmutableSet.of(windowFilterConjunct), newWindow);
+        return windowFilter;
+    }
+
+    // check children's nodes because query process will be changed
+    private boolean checkPlanType() {
+        return outerPlans.stream().allMatch(p -> LEFT_SUPPORTED_PLAN.contains(p.getClass()))
+                && innerPlans.stream().allMatch(p -> RIGHT_SUPPORTED_PLAN.contains(p.getClass()));
+    }
+
+    // check aggregation of inner scope
+    private boolean checkAggType() {
+        List<LogicalAggregate> aggSet = innerPlans.stream().filter(LogicalAggregate.class::isInstance)
+                .map(LogicalAggregate.class::cast)
+                .collect(Collectors.toList());
+        if (aggSet.size() > 1) {
+            // window functions don't support nesting.
+            return false;
+        }
+        aggOp = aggSet.get(0);
+        functions = ((List<AggregateFunction>) ExpressionUtils.<AggregateFunction>collectAll(
+                aggOp.getOutputExpressions(), AggregateFunction.class::isInstance));
+        Preconditions.checkArgument(functions.size() == 1);
+        return functions.stream().allMatch(f -> SUPPORTED_FUNCTION.contains(f.getClass()) && !f.isDistinct());
+    }
+
+    // check if the relations of the outer's includes the inner's
+    private boolean checkRelation(List<Expression> correlatedSlots) {
+        List<LogicalRelation> outerTables = outerPlans.stream().filter(LogicalRelation.class::isInstance)
+                .map(LogicalRelation.class::cast)
+                .collect(Collectors.toList());
+        List<LogicalRelation> innerTables = innerPlans.stream().filter(LogicalRelation.class::isInstance)
+                .map(LogicalRelation.class::cast)
+                .collect(Collectors.toList());
+
+        Set<Long> outerIds = outerTables.stream().map(node -> node.getTable().getId()).collect(Collectors.toSet());
+        Set<Long> innerIds = innerTables.stream().map(node -> node.getTable().getId()).collect(Collectors.toSet());
+
+        if (outerTables.size() != outerIds.size() || innerTables.size() != innerIds.size()
+                || outerIds.size() != innerIds.size() + 1) {
+            return false;
+        }
+
+        outerIds.removeAll(innerIds);
+        if (outerIds.size() != 1) {
+            return false;
+        }
+
+        LogicalRelation correlatedRelation = outerTables.stream()
+                .filter(node -> node.getTable().getId() == outerIds.iterator().next()).findFirst().get();
+        return ExpressionUtils.collect(correlatedSlots, NamedExpression.class::isInstance).stream()
+                .map(NamedExpression.class::cast)
+                .allMatch(e -> correlatedRelation.getOutputExprIdSet().contains(e.getExprId()));
+    }
+
+    private boolean checkPredicate(Set<Expression> outerConjuncts, Set<Expression> innerConjuncts) {
+        Iterator<Expression> innerIter = innerConjuncts.iterator();
+        // inner predicate should be the sub-set of outer predicate.
+        while (innerIter.hasNext()) {
+            Expression innerExpr = innerIter.next();
+            Iterator<Expression> outerIter = outerConjuncts.iterator();
+            while (outerIter.hasNext()) {
+                Expression outerExpr = outerIter.next();
+                if (new ExpressionIdenticalChecker().check(innerExpr, outerExpr)) {

Review Comment:
   use static instance instead new 



##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/AggScalarSubQueryToWindowFunction.java:
##########
@@ -0,0 +1,371 @@
+// 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.nereids.rules.rewrite.logical;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+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.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.NullableAggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
+import org.apache.doris.nereids.trees.plans.GroupPlan;
+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.LogicalApply;
+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.LogicalLimit;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanVisitor;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.nereids.util.PlanUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * change scalar sub query containing agg to window function. such as:
+ * SELECT SUM(l_extendedprice) / 7.0 AS avg_yearly
+ *  FROM lineitem, part
+ *  WHERE p_partkey = l_partkey AND
+ *  p_brand = 'Brand#23' AND
+ *  p_container = 'MED BOX' AND
+ *  l_quantity<(SELECT 0.2*avg(l_quantity)
+ *  FROM lineitem
+ *  WHERE l_partkey = p_partkey);
+ * to:
+ * SELECT SUM(l_extendedprice) / 7.0 as avg_yearly
+ *  FROM (SELECT l_extendedprice, l_quantity,
+ *    0.2 * avg(l_quantity)over(partition by p_partkey)
+ *    AS avg_l_quantity
+ *    FROM lineitem, part
+ *    WHERE p_partkey = l_partkey and
+ *    p_brand = 'Brand#23' and
+ *    p_container = 'MED BOX') t
+ * WHERE l_quantity < avg_l_quantity;
+ */
+
+public class AggScalarSubQueryToWindowFunction extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+    private static final ImmutableSet<Class<? extends AggregateFunction>> SUPPORTED_FUNCTION = ImmutableSet.of(
+            Min.class, Max.class, Count.class, Sum.class, Avg.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> LEFT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalLimit.class, LogicalJoin.class, LogicalProject.class
+    );
+    private static final ImmutableSet<Class<? extends LogicalPlan>> RIGHT_SUPPORTED_PLAN = ImmutableSet.of(
+            LogicalOlapScan.class, LogicalJoin.class, LogicalProject.class, LogicalAggregate.class, LogicalFilter.class
+    );
+    private List<LogicalPlan> outerPlans = null;
+    private List<LogicalPlan> innerPlans = null;
+    private LogicalAggregate aggOp = null;
+    private List<AggregateFunction> functions = null;
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext context) {
+        return plan.accept(this, context);
+    }
+
+    @Override
+    public Plan visitLogicalFilter(LogicalFilter filter, JobContext context) {
+        if (!checkPattern(filter)) {
+            return filter;
+        }
+        LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node
+                = ((LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>>) filter);
+        if (!check(node)) {
+            return filter;
+        }
+        return trans(node);
+    }
+
+    private boolean checkPattern(LogicalFilter filter) {
+        if (!(filter.child() instanceof LogicalProject)) {
+            return false;
+        }
+        LogicalProject project = (LogicalProject) filter.child();
+        if (project.child() == null || !(project.child() instanceof LogicalApply)) {
+            return false;
+        }
+        // filter(apply()) may be also ok.
+        // but nereids will match the pattern filter(project(apply()))
+        LogicalApply apply = ((LogicalApply<?, ?>) project.child());
+        if (!apply.isScalar() || !apply.isCorrelated() || !apply.getSubCorrespondingConject().isPresent()) {
+            return false;
+        }
+        return apply.left() != null && apply.right() instanceof LogicalAggregate
+                && ((LogicalAggregate<?>) apply.right()).child() != null;
+    }
+
+    private boolean check(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<?, ?> apply = node.child().child();
+        LogicalPlan outer = ((LogicalPlan) apply.child(0));
+        LogicalPlan inner = ((LogicalPlan) apply.child(1));
+        outerPlans = new PlanCollector().collect(outer);
+        innerPlans = new PlanCollector().collect(inner);
+        LogicalFilter outerFilter = node;
+        Optional<LogicalFilter> innerFilter = innerPlans.stream()
+                .filter(LogicalFilter.class::isInstance)
+                .map(LogicalFilter.class::cast).findFirst();
+        return innerFilter.filter(
+                logicalFilter -> checkPlanType() && checkAggType()
+                        && checkRelation(apply.getCorrelationSlot())
+                        && checkPredicate(Sets.newHashSet(outerFilter.getConjuncts()),
+                        Sets.newHashSet(logicalFilter.getConjuncts()))
+        ).isPresent();
+    }
+
+    private Plan trans(LogicalFilter<LogicalProject<LogicalApply<Plan, LogicalAggregate<Plan>>>> node) {
+        LogicalApply<Plan, LogicalAggregate<Plan>> apply = node.child().child();
+        LogicalAggregate<Plan> agg = apply.right();
+
+        // for example, in tpc-h Q17
+        // window filter conjuncts is cast(l_quantity#id1 as decimal(27, 9)) < `0.2 * avg(l_quantity)`#id2 and
+        // 0.2 * avg(l_quantity#id3) as `0.2 * l_quantity`#id2 of agg's output expression
+        // we change it to cast(l_quantity#id1 as decimal(27, 9)) < 0.2 * `avg(l_quantity#id1) over(window)`#id4
+        // and avg(l_quantity#id1) over(window) as `avg(l_quantity#id1) over(window)`#id4
+        // first we get the slots of the two sides of apply
+
+        Preconditions.checkArgument(apply.getSubCorrespondingConject().get() instanceof ComparisonPredicate);
+        ComparisonPredicate windowFilterConjunct = PlanUtils.maybeCommuteComparisonPredicate(
+                ((ComparisonPredicate) apply.getSubCorrespondingConject().get()), apply.left());
+
+        // build window function, replace the slot
+        List<Expression> windowAggSlots = windowFilterConjunct.left().collectToList(Slot.class::isInstance);
+        Preconditions.checkArgument(windowAggSlots.size() == 1);
+
+        // adjust agg function's nullable.
+        AggregateFunction function = functions.get(0);
+        if (function instanceof NullableAggregateFunction) {
+            function = ((NullableAggregateFunction) function).withAlwaysNullable(false);
+        }
+
+        WindowExpression windowFunction = createWindowFunction(apply.getCorrelationSlot(),
+                function.withChildren(windowAggSlots));
+        NamedExpression windowFunctionAlias = new Alias(windowFunction, "wf");

Review Comment:
   u need use a unique name generate by column name generator
   refer to `statementContext.generateColumnName()`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org