SELECT max(i)FROM tbl GROUP BY j;

    set spark.sql.adaptive.enabled = true;

    -- For demo purpose only.
    -- Not necesary in real-life usage.
 
    set spark.sql.adaptive.coalescePartitions.minPartitionNum = 1;

    CREATE DATABASE IF NOT EXISTS aqe_demo_db;
    USE aqe_demo_db;
    
    DROP TABLE IF EXISTS items;
    DROP TABLE IF EXISTS sales;
    
    -- Create "items" table.
    
    CREATE TABLE items
    USING parquet
    AS
    SELECT id AS i_item_id,
    CAST(rand() * 1000 AS INT) AS i_price
    FROM RANGE(30000000);
    
    -- Create "sales" table with skew.
    -- Item with id 100 is in 80% of all sales.
    
    CREATE TABLE sales
    USING parquet
    AS
    SELECT CASE WHEN rand() < 0.8 THEN 100 ELSE CAST(rand() * 30000000 AS INT) END AS s_item_id,
    CAST(rand() * 100 AS INT) AS s_quantity,
    DATE_ADD(current_date(), - CAST(rand() * 360 AS INT)) AS s_date
    FROM RANGE(1000000000);

    -- Get the sums of sales quantity grouped by sales date.
    -- The grouped result is very small.
    
    SELECT s_date, sum(s_quantity) AS q
    FROM sales
    GROUP BY s_date
    ORDER BY q DESC;

    -- Get total sales amount grouped by sales date for items with a price lower than 10.
    -- The selectivity of the filter by price is not known in static planning, so the initial plan opts for sort merge join.
    -- But in fact, the "items" table after filtering is very small, so the query can do a broadcast hash join instead.
    
    -- Static explain shows the initial plan with sort merge join.
    
    EXPLAIN FORMATTED
    SELECT s_date, sum(s_quantity * i_price) AS total_sales
    FROM sales
    JOIN items ON s_item_id = i_item_id
    WHERE i_price < 10
    GROUP BY s_date
    ORDER BY total_sales DESC;

    +----------------------------------------------------+
    |                        plan                        |
    +----------------------------------------------------+
    | == Physical Plan ==
    AdaptiveSparkPlan (16)
    +- Sort (15)
    +- Exchange (14)
        +- HashAggregate (13)
            +- Exchange (12)
                +- HashAggregate (11)
                +- Project (10)
                    +- SortMergeJoin Inner (9)
                        :- Sort (4)
                        :  +- Exchange (3)
                        :     +- Filter (2)
                        :        +- Scan parquet app.sales (1)
                        +- Sort (8)
                            +- Exchange (7)
                            +- Filter (6)
                                +- Scan parquet app.items (5)


    (1) Scan parquet app.sales
    Output [3]: [s_item_id#87, s_quantity#88, s_date#89]
    Batched: true
    Location: InMemoryFileIndex [xxxx/hive/warehouse/app.db/sales]
    PushedFilters: [IsNotNull(s_item_id)]
    ReadSchema: struct<s_item_id:int,s_quantity:int,s_date:date>

    (2) Filter
    Input [3]: [s_item_id#87, s_quantity#88, s_date#89]
    Condition : isnotnull(s_item_id#87)

    (3) Exchange
    Input [3]: [s_item_id#87, s_quantity#88, s_date#89]
    Arguments: hashpartitioning(cast(s_item_id#87 as bigint), 800), ENSURE_REQUIREMENTS, [id=#238]

    (4) Sort
    Input [3]: [s_item_id#87, s_quantity#88, s_date#89]
    Arguments: [cast(s_item_id#87 as bigint) ASC NULLS FIRST], false, 0

    (5) Scan parquet app.items
    Output [2]: [i_item_id#80L, i_price#81]
    Batched: true
    Location: InMemoryFileIndex [xxxx/user/hive/warehouse/app.db/items]
    PushedFilters: [IsNotNull(i_price), LessThan(i_price,10), IsNotNull(i_item_id)]
    ReadSchema: struct<i_item_id:bigint,i_price:int>

    (6) Filter
    Input [2]: [i_item_id#80L, i_price#81]
    Condition : ((isnotnull(i_price#81) AND (i_price#81 < 10)) AND isnotnull(i_item_id#80L))

    (7) Exchange
    Input [2]: [i_item_id#80L, i_price#81]
    Arguments: hashpartitioning(i_item_id#80L, 800), ENSURE_REQUIREMENTS, [id=#239]

    (8) Sort
    Input [2]: [i_item_id#80L, i_price#81]
    Arguments: [i_item_id#80L ASC NULLS FIRST], false, 0

    (9) SortMergeJoin
    Left keys [1]: [cast(s_item_id#87 as bigint)]
    Right keys [1]: [i_item_id#80L]
    Join condition: None

    (10) Project
    Output [3]: [s_quantity#88, s_date#89, i_price#81]
    Input [5]: [s_item_id#87, s_quantity#88, s_date#89, i_item_id#80L, i_price#81]

    (11) HashAggregate
    Input [3]: [s_quantity#88, s_date#89, i_price#81]
    Keys [1]: [s_date#89]
    Functions [1]: [partial_sum((s_quantity#88 * i_price#81))]
    Aggregate Attributes [1]: [sum#117L]
    Results [2]: [s_date#89, sum#118L]

    (12) Exchange
    Input [2]: [s_date#89, sum#118L]
    Arguments: hashpartitioning(s_date#89, 800), ENSURE_REQUIREMENTS, [id=#246]

    (13) HashAggregate
    Input [2]: [s_date#89, sum#118L]
    Keys [1]: [s_date#89]
    Functions [1]: [sum((s_quantity#88 * i_price#81))]
    Aggregate Attributes [1]: [sum((s_quantity#88 * i_price#81))#116L]
    Results [2]: [s_date#89, sum((s_quantity#88 * i_price#81))#116L AS total_sales#110L]

    (14) Exchange
    Input [2]: [s_date#89, total_sales#110L]
    Arguments: rangepartitioning(total_sales#110L DESC NULLS LAST, 800), ENSURE_REQUIREMENTS, [id=#249]

    (15) Sort
    Input [2]: [s_date#89, total_sales#110L]
    Arguments: [total_sales#110L DESC NULLS LAST], true, 0

    (16) AdaptiveSparkPlan
    Output [2]: [s_date#89, total_sales#110L]
    Arguments: isFinalPlan=false

    -- The runtime join stategy is changed to broadcast hash join.
    
    SELECT s_date, sum(s_quantity * i_price) AS total_sales
    FROM sales
    JOIN items ON s_item_id = i_item_id
    WHERE i_price < 10
    GROUP BY s_date
    ORDER BY total_sales DESC;

    -- Get the total sales amount grouped by sales date.
    -- The partition in the "sales" table containing value "100" as "s_item_id" is much larger than other partitions.
    -- AQE splits the skewed partition into smaller partitions before joining the "sales" table with the "items" table.
    
    SELECT s_date, sum(s_quantity * i_price) AS total_sales
    FROM sales
    JOIN items ON i_item_id = s_item_id
    GROUP BY s_date
    ORDER BY total_sales DESC;

在AQEShuffleRead阶段，

partition data size total (min, med, max )
8.5 GiB (3.3 MiB, 39.5 MiB, 241.1 MiB (driver))
number of coalesced partitions: 67
number of skewed partitions: 1
number of skewed partition splits: 26
number of partitions: 94

    Batch("LocalRelation", fixedPoint,
        ConvertToLocalRelation,
        PropagateEmptyRelation,
        // PropagateEmptyRelation can change the nullability of an attribute from nullable to
        // non-nullable when an empty relation child of a Union is removed
        UpdateAttributeNullability) :+
        // The following batch should be executed after batch "Join Reorder" and "LocalRelation".

    /*
    * 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.spark.sql.catalyst.optimizer

    import org.apache.spark.sql.catalyst.analysis.CastSupport
    import org.apache.spark.sql.catalyst.expressions._
    import org.apache.spark.sql.catalyst.expressions.Literal.FalseLiteral
    import org.apache.spark.sql.catalyst.plans._
    import org.apache.spark.sql.catalyst.plans.logical._
    import org.apache.spark.sql.catalyst.rules._
    import org.apache.spark.sql.catalyst.trees.TreePattern.{LOCAL_RELATION, TRUE_OR_FALSE_LITERAL}

    /**
    * The base class of two rules in the normal and AQE Optimizer. It simplifies query plans with
    * empty or non-empty relations:
    *  1. Binary-node Logical Plans
    *     - Join with one or two empty children (including Intersect/Except).
    *     - Left semi Join
    *       Right side is non-empty and condition is empty. Eliminate join to its left side.
    *     - Left anti join
    *       Right side is non-empty and condition is empty. Eliminate join to an empty
    *       [[LocalRelation]].
    *  2. Unary-node Logical Plans
    *     - Limit/Repartition with all empty children.
    *     - Aggregate with all empty children and at least one grouping expression.
    *     - Generate(Explode) with all empty children. Others like Hive UDTF may return results.
    */
    abstract class PropagateEmptyRelationBase extends Rule[LogicalPlan] with CastSupport {
    protected def isEmpty(plan: LogicalPlan): Boolean = plan match {
        case p: LocalRelation => p.data.isEmpty
        case _ => false
    }

    protected def nonEmpty(plan: LogicalPlan): Boolean = plan match {
        case p: LocalRelation => p.data.nonEmpty
        case _ => false
    }

    protected def empty(plan: LogicalPlan): LocalRelation =
        LocalRelation(plan.output, data = Seq.empty, isStreaming = plan.isStreaming)

    // Construct a project list from plan's output, while the value is always NULL.
    private def nullValueProjectList(plan: LogicalPlan): Seq[NamedExpression] =
        plan.output.map{ a => Alias(cast(Literal(null), a.dataType), a.name)(a.exprId) }

    protected def commonApplyFunc: PartialFunction[LogicalPlan, LogicalPlan] = {
        // Joins on empty LocalRelations generated from streaming sources are not eliminated
        // as stateful streaming joins need to perform other state management operations other than
        // just processing the input data.
        case p @ Join(_, _, joinType, conditionOpt, _)
            if !p.children.exists(_.isStreaming) =>
        val isLeftEmpty = isEmpty(p.left)
        val isRightEmpty = isEmpty(p.right)
        val isFalseCondition = conditionOpt match {
            case Some(FalseLiteral) => true
            case _ => false
        }
        if (isLeftEmpty || isRightEmpty || isFalseCondition) {
            joinType match {
            case _: InnerLike => empty(p)
            // Intersect is handled as LeftSemi by `ReplaceIntersectWithSemiJoin` rule.
            // Except is handled as LeftAnti by `ReplaceExceptWithAntiJoin` rule.
            case LeftOuter | LeftSemi | LeftAnti if isLeftEmpty => empty(p)
            case LeftSemi if isRightEmpty | isFalseCondition => empty(p)
            case LeftAnti if isRightEmpty | isFalseCondition => p.left
            case FullOuter if isLeftEmpty && isRightEmpty => empty(p)
            case LeftOuter | FullOuter if isRightEmpty =>
                Project(p.left.output ++ nullValueProjectList(p.right), p.left)
            case RightOuter if isRightEmpty => empty(p)
            case RightOuter | FullOuter if isLeftEmpty =>
                Project(nullValueProjectList(p.left) ++ p.right.output, p.right)
            case LeftOuter if isFalseCondition =>
                Project(p.left.output ++ nullValueProjectList(p.right), p.left)
            case RightOuter if isFalseCondition =>
                Project(nullValueProjectList(p.left) ++ p.right.output, p.right)
            case _ => p
            }
        } else if (joinType == LeftSemi && conditionOpt.isEmpty && nonEmpty(p.right)) {
            p.left
        } else if (joinType == LeftAnti && conditionOpt.isEmpty && nonEmpty(p.right)) {
            empty(p)
        } else {
            p
        }

        case p: UnaryNode if p.children.nonEmpty && p.children.forall(isEmpty) => p match {
        case _: Sort => empty(p)
        case _: GlobalLimit if !p.isStreaming => empty(p)
        case _: LocalLimit if !p.isStreaming => empty(p)
        case _: Repartition => empty(p)
        case _: RepartitionByExpression => empty(p)
        // An aggregate with non-empty group expression will return one output row per group when the
        // input to the aggregate is not empty. If the input to the aggregate is empty then all groups
        // will be empty and thus the output will be empty. If we're working on batch data, we can
        // then treat the aggregate as redundant.
        //
        // If the aggregate is over streaming data, we may need to update the state store even if no
        // new rows are processed, so we can't eliminate the node.
        //
        // If the grouping expressions are empty, however, then the aggregate will always produce a
        // single output row and thus we cannot propagate the EmptyRelation.
        //
        // Aggregation on empty LocalRelation generated from a streaming source is not eliminated
        // as stateful streaming aggregation need to perform other state management operations other
        // than just processing the input data.
        case Aggregate(ge, _, _) if ge.nonEmpty && !p.isStreaming => empty(p)
        // Generators like Hive-style UDTF may return their records within `close`.
        case Generate(_: Explode, _, _, _, _, _) => empty(p)
        case _ => p
        }
    }
    }

    /**
    * This rule runs in the normal optimizer and optimizes more cases
    * compared to [[PropagateEmptyRelationBase]]:
    * 1. Higher-node Logical Plans
    *    - Union with all empty children.
    * 2. Unary-node Logical Plans
    *    - Project/Filter/Sample with all empty children.
    *
    * The reason why we don't apply this rule at AQE optimizer side is: the benefit is not big enough
    * and it may introduce extra exchanges.
    */
    object PropagateEmptyRelation extends PropagateEmptyRelationBase {
    private def applyFunc: PartialFunction[LogicalPlan, LogicalPlan] = {
        case p: Union if p.children.exists(isEmpty) =>
        val newChildren = p.children.filterNot(isEmpty)
        if (newChildren.isEmpty) {
            empty(p)
        } else {
            val newPlan = if (newChildren.size > 1) Union(newChildren) else newChildren.head
            val outputs = newPlan.output.zip(p.output)
            // the original Union may produce different output attributes than the new one so we alias
            // them if needed
            if (outputs.forall { case (newAttr, oldAttr) => newAttr.exprId == oldAttr.exprId }) {
            newPlan
            } else {
            val outputAliases = outputs.map { case (newAttr, oldAttr) =>
                val newExplicitMetadata =
                if (oldAttr.metadata != newAttr.metadata) Some(oldAttr.metadata) else None
                Alias(newAttr, oldAttr.name)(oldAttr.exprId, explicitMetadata = newExplicitMetadata)
            }
            Project(outputAliases, newPlan)
            }
        }

        case p: UnaryNode if p.children.nonEmpty && p.children.forall(isEmpty) && canPropagate(p) =>
        empty(p)
    }

    // extract the pattern avoid conflict with commonApplyFunc
    private def canPropagate(plan: LogicalPlan): Boolean = plan match {
        case _: Project => true
        case _: Filter => true
        case _: Sample => true
        case _ => false
    }

    override def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
        _.containsAnyPattern(LOCAL_RELATION, TRUE_OR_FALSE_LITERAL), ruleId) {
        applyFunc.orElse(commonApplyFunc)
    }
    }

    PropagateEmptyRelation can change the nullability of an attribute from nullable to non-nullable when an empty relation child of a Union is removed