[VL] Exchange reuse not applied due to unnormalized Alias in GenerateExecTransformer's generator

[jiangjiangtian]

Backend

VL (Velox)

Bug description

Suppose we want to execute the following SQL (the SQL is just for showing and doesn't mean anything):

CREATE OR REPLACE TEMP VIEW employees AS
SELECT * FROM VALUES
(1, 'Alice', 'Sales,Computer', 1000),
(2, 'Bob', 'Marketing,Sales', 2000),
(3, 'Charlie', 'Marketing,Trading', 3000)
AS employees(emp_id, emp_name, dept_names, sale);

set spark.sql.adaptive.enabled=false;

EXPLAIN WITH base AS (
SELECT * FROM employees
  LATERAL VIEW explode(split(dept_names, ',')) AS dept_name
)
SELECT * FROM
(
  SELECT emp_id, emp_name, COUNT(*)
  FROM (
    SELECT emp_id, emp_name, dept_names, sale
    FROM base
    GROUP BY 1, 2, 3, 4
  )
  GROUP BY 1, 2
) a
JOIN
(
  SELECT emp_id, SUM(sale)
  FROM (
    SELECT emp_id, emp_name, dept_names, sale
    FROM base
    GROUP BY 1, 2, 3, 4
  )
  GROUP BY 1
) b
ON a.emp_id = b.emp_id;

The plan is:

== Physical Plan ==
VeloxColumnarToRow
+- ^(6) BroadcastHashJoinExecTransformer [emp_id#78], [emp_id#86], Inner, BuildRight, false
   :- ^(6) HashAggregateTransformer(keys=[emp_id#78, emp_name#79], functions=[count(1)], isStreamingAgg=false)
   :  +- ^(6) InputIteratorTransformer[emp_id#78, emp_name#79, count#101L]
   :     +- ColumnarExchange hashpartitioning(emp_id#78, emp_name#79, 2000), ENSURE_REQUIREMENTS, [emp_id#78, emp_name#79, count#101L], [plan_id=1889], [shuffle_writer_type=hash], [OUTPUT] List(emp_id:IntegerType, emp_name:StringType, count:LongType)
   :        +- VeloxResizeBatches 1024, 2147483647, 10485760
   :           +- ^(2) ProjectExecTransformer [hash(emp_id#78, emp_name#79, 42) AS hash_partition_key#129, emp_id#78, emp_name#79, count#101L]
   :              +- ^(2) FlushableHashAggregateTransformer(keys=[emp_id#78, emp_name#79], functions=[partial_count(1)], isStreamingAgg=false)
   :                 +- ^(2) ProjectExecTransformer [emp_id#78, emp_name#79]
   :                    +- ^(2) HashAggregateTransformer(keys=[emp_id#78, emp_name#79, dept_names#80, sale#81], functions=[], isStreamingAgg=false)
   :                       +- ^(2) InputIteratorTransformer[emp_id#78, emp_name#79, dept_names#80, sale#81]
   :                          +- ColumnarExchange hashpartitioning(emp_id#78, emp_name#79, dept_names#80, sale#81, 2000), ENSURE_REQUIREMENTS, [emp_id#78, emp_name#79, dept_names#80, sale#81], [plan_id=1880], [shuffle_writer_type=hash], [OUTPUT] List(emp_id:IntegerType, emp_name:StringType, dept_names:StringType, sale:IntegerType)
   :                             +- VeloxResizeBatches 1024, 2147483647, 10485760
   :                                +- ^(1) ProjectExecTransformer [hash(emp_id#78, emp_name#79, dept_names#80, sale#81, 42) AS hash_partition_key#128, emp_id#78, emp_name#79, dept_names#80, sale#81]
   :                                   +- ^(1) FlushableHashAggregateTransformer(keys=[emp_id#78, emp_name#79, dept_names#80, sale#81], functions=[], isStreamingAgg=false)
   :                                      +- ^(1) ProjectExecTransformer [emp_id#78, emp_name#79, dept_names#80, sale#81]
   :                                         +- ^(1) GenerateExecTransformer explode(split(dept_names#80, ,, -1) AS _pre_0#104), [emp_id#78, emp_name#79, dept_names#80, sale#81], false, [dept_name#95]
   :                                            +- ^(1) ProjectExecTransformer [emp_id#78, emp_name#79, dept_names#80, sale#81, split(dept_names#80, ,, -1) AS _pre_0#104]
   :                                               +- ^(1) InputIteratorTransformer[emp_id#78, emp_name#79, dept_names#80, sale#81]
   :                                                  +- RowToVeloxColumnar
   :                                                     +- LocalTableScan [emp_id#78, emp_name#79, dept_names#80, sale#81]
   +- ^(6) InputIteratorTransformer[emp_id#86, sum(sale)#99L]
      +- ColumnarBroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=1916]
         +- ^(5) HashAggregateTransformer(keys=[emp_id#86], functions=[sum(sale#89)], isStreamingAgg=false)
            +- ^(5) InputIteratorTransformer[emp_id#86, sum#103L]
               +- ColumnarExchange hashpartitioning(emp_id#86, 2000), ENSURE_REQUIREMENTS, [emp_id#86, sum#103L], [plan_id=1911], [shuffle_writer_type=hash], [OUTPUT] List(emp_id:IntegerType, sum:LongType)
                  +- VeloxResizeBatches 1024, 2147483647, 10485760
                     +- ^(4) ProjectExecTransformer [hash(emp_id#86, 42) AS hash_partition_key#131, emp_id#86, sum#103L]
                        +- ^(4) FlushableHashAggregateTransformer(keys=[emp_id#86], functions=[partial_sum(sale#89)], isStreamingAgg=false)
                           +- ^(4) ProjectExecTransformer [emp_id#86, sale#89]
                              +- ^(4) HashAggregateTransformer(keys=[emp_id#86, emp_name#87, dept_names#88, sale#89], functions=[], isStreamingAgg=false)
                                 +- ^(4) InputIteratorTransformer[emp_id#86, emp_name#87, dept_names#88, sale#89]
                                    +- ColumnarExchange hashpartitioning(emp_id#86, emp_name#87, dept_names#88, sale#89, 2000), ENSURE_REQUIREMENTS, [emp_id#86, emp_name#87, dept_names#88, sale#89], [plan_id=1902], [shuffle_writer_type=hash], [OUTPUT] List(emp_id:IntegerType, emp_name:StringType, dept_names:StringType, sale:IntegerType)
                                       +- VeloxResizeBatches 1024, 2147483647, 10485760
                                          +- ^(3) ProjectExecTransformer [hash(emp_id#86, emp_name#87, dept_names#88, sale#89, 42) AS hash_partition_key#130, emp_id#86, emp_name#87, dept_names#88, sale#89]
                                             +- ^(3) FlushableHashAggregateTransformer(keys=[emp_id#86, emp_name#87, dept_names#88, sale#89], functions=[], isStreamingAgg=false)
                                                +- ^(3) ProjectExecTransformer [emp_id#86, emp_name#87, dept_names#88, sale#89]
                                                   +- ^(3) GenerateExecTransformer explode(split(dept_names#88, ,, -1) AS _pre_1#105), [emp_id#86, emp_name#87, dept_names#88, sale#89], false, [dept_name#96]
                                                      +- ^(3) ProjectExecTransformer [emp_id#86, emp_name#87, dept_names#88, sale#89, split(dept_names#88, ,, -1) AS _pre_1#105]
                                                         +- ^(3) InputIteratorTransformer[emp_id#86, emp_name#87, dept_names#88, sale#89]
                                                            +- RowToVeloxColumnar
                                                               +- LocalTableScan [emp_id#86, emp_name#87, dept_names#88, sale#89]

The plan is correct but we lose the opportunity to apply the optimization of reusing exchange. The two ColumnarExchange nodes (plan_id=1880 and plan_id=1902) are structurally identical, but are not reused (no ReusedExchange appears in the plan).
The root cause is the Alias expression in the generator of GenerateExecTransformer. In the plan above, the first branch has:

GenerateExecTransformer explode(split(dept_names#80, ,, -1) AS _pre_0#104), ...

while the second branch has:

GenerateExecTransformer explode(split(dept_names#88, ,, -1) AS _pre_1#105), ...

The Alias (AS _pre_0#104 / AS _pre_1#105) is nested inside the generator expression. Spark's only assigns position-based ExprId to top-level Alias nodes in the plan's expressions. Since the generator (e.g., Explode) is not an Alias, it falls into the case other branch which calls normalizeExpressions. However, normalizeExpressions only replaces the exprId of AttributeReference — it does not modify the exprId of a nested Alias. As a result, the Alias inside the generator retains its original globally-unique exprId (e.g., #104 vs #105). Since compares exprId, the two structurally-identical sub-plans produce different canonicalized forms, preventing ReuseExchange from reusing the exchange.

Gluten version

No response

Spark version

None

Spark configurations

3.5

System information

No response

Relevant logs