You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/04/14 11:56:25 UTC
(spark) branch master updated: [SPARK-55979][SQL] Required input attributes are missing from PartialMerge / Final BaseAggregateExec.references
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new e7229c7b4bd8 [SPARK-55979][SQL] Required input attributes are missing from PartialMerge / Final BaseAggregateExec.references
e7229c7b4bd8 is described below
commit e7229c7b4bd8da80a8a159d7db43ad24bdc7e538
Author: Hongze Zhang <ho...@gmail.com>
AuthorDate: Tue Apr 14 19:56:09 2026 +0800
[SPARK-55979][SQL] Required input attributes are missing from PartialMerge / Final BaseAggregateExec.references
### What changes were proposed in this pull request?
The patch updates BaseAggregateExec.producedAttributes, from
```scala
override def producedAttributes: AttributeSet =
AttributeSet(aggregateAttributes) ++
AttributeSet(resultExpressions.diff(groupingExpressions).map(_.toAttribute)) ++
AttributeSet(aggregateBufferAttributes) ++
AttributeSet(inputAggBufferAttributes.filterNot(child.output.contains))
```
to
```scala
override def producedAttributes: AttributeSet =
AttributeSet(aggregateAttributes) ++
AttributeSet(resultExpressions.diff(groupingExpressions).map(_.toAttribute)) ++
AttributeSet(aggregateBufferAttributes) ++
AttributeSet(inputAggBufferAttributes) -- child.outputSet
```
The patch fixes bug described below by adding `-- child.outputSet` to exclude child's attributes from producedAttributes, ensuring producedAttributes doesn't include anything from the aggregate's input.
### Why are the changes needed?
The current implementation has bug causing a AggregateExec in `Final` mode gives fewer attributes in `references` than it really needs. Our downstream project has a rule to prune the unused inputs, relying on the `references` call. This bug causes some needed inputs to be pruned accidentally.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Tested by newly added test case.
### Was this patch authored or co-authored using generative AI tooling?
The test case was co-authored by AI and reviewed by PR author.
Closes #54778 from zhztheplayer/wip-55979.
Authored-by: Hongze Zhang <ho...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../execution/aggregate/BaseAggregateExec.scala | 2 +-
.../ss_max.sf100/simplified.txt | 4 +--
.../approved-plans-modified/ss_max/simplified.txt | 4 +--
.../approved-plans-v1_4/q16.sf100/simplified.txt | 4 +--
.../approved-plans-v1_4/q16/simplified.txt | 4 +--
.../approved-plans-v1_4/q28.sf100/simplified.txt | 24 +++++++--------
.../approved-plans-v1_4/q28/simplified.txt | 24 +++++++--------
.../approved-plans-v1_4/q70.sf100/simplified.txt | 2 +-
.../approved-plans-v1_4/q70/simplified.txt | 2 +-
.../approved-plans-v1_4/q94.sf100/simplified.txt | 4 +--
.../approved-plans-v1_4/q94/simplified.txt | 4 +--
.../approved-plans-v1_4/q95.sf100/simplified.txt | 4 +--
.../approved-plans-v1_4/q95/simplified.txt | 4 +--
.../approved-plans-v2_7/q70a.sf100/simplified.txt | 2 +-
.../approved-plans-v2_7/q70a/simplified.txt | 2 +-
.../apache/spark/sql/execution/PlannerSuite.scala | 36 ++++++++++++++++++++++
16 files changed, 81 insertions(+), 45 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/BaseAggregateExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/BaseAggregateExec.scala
index 7ae00467a011..f72343df9629 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/BaseAggregateExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/BaseAggregateExec.scala
@@ -87,7 +87,7 @@ trait BaseAggregateExec extends UnaryExecNode with PartitioningPreservingUnaryEx
// it's not empty when the inputAggBufferAttributes is not equal to the aggregate buffer
// attributes of the child Aggregate, when the child Aggregate contains the subquery in
// AggregateFunction. See SPARK-31620 for more details.
- AttributeSet(inputAggBufferAttributes.filterNot(child.output.contains))
+ AttributeSet(inputAggBufferAttributes) -- child.outputSet
override def output: Seq[Attribute] = resultExpressions.map(_.toAttribute)
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-modified/ss_max.sf100/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-modified/ss_max.sf100/simplified.txt
index d1fff7ba25c8..cdc14af7b7c6 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-modified/ss_max.sf100/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-modified/ss_max.sf100/simplified.txt
@@ -3,8 +3,8 @@ WholeStageCodegen (3)
InputAdapter
Exchange #1
WholeStageCodegen (2)
- HashAggregate [ss_sold_date_sk] [count(1),count(ss_sold_date_sk),max(ss_sold_date_sk),max(ss_sold_time_sk),max(ss_item_sk),max(ss_customer_sk),max(ss_cdemo_sk),max(ss_hdemo_sk),max(ss_addr_sk),max(ss_store_sk),max(ss_promo_sk),count(ss_sold_date_sk),count,count,max,max,max,max,max,max,max,max,max,count,count,count,max,max,max,max,max,max,max,max,max,count]
- HashAggregate [ss_sold_date_sk] [count(1),count(ss_sold_date_sk),max(ss_sold_date_sk),max(ss_sold_time_sk),max(ss_item_sk),max(ss_customer_sk),max(ss_cdemo_sk),max(ss_hdemo_sk),max(ss_addr_sk),max(ss_store_sk),max(ss_promo_sk),count,count,max,max,max,max,max,max,max,max,max,count,count,max,max,max,max,max,max,max,max,max]
+ HashAggregate [count,count,max,max,max,max,max,max,max,max,max,ss_sold_date_sk] [count(1),count(ss_sold_date_sk),max(ss_sold_date_sk),max(ss_sold_time_sk),max(ss_item_sk),max(ss_customer_sk),max(ss_cdemo_sk),max(ss_hdemo_sk),max(ss_addr_sk),max(ss_store_sk),max(ss_promo_sk),count(ss_sold_date_sk),count,count,count,max,max,max,max,max,max,max,max,max,count]
+ HashAggregate [ss_sold_date_sk,count,count,max,max,max,max,max,max,max,max,max] [count(1),count(ss_sold_date_sk),max(ss_sold_date_sk),max(ss_sold_time_sk),max(ss_item_sk),max(ss_customer_sk),max(ss_cdemo_sk),max(ss_hdemo_sk),max(ss_addr_sk),max(ss_store_sk),max(ss_promo_sk),count,count,max,max,max,max,max,max,max,max,max]
InputAdapter
Exchange [ss_sold_date_sk] #2
WholeStageCodegen (1)
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-modified/ss_max/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-modified/ss_max/simplified.txt
index d1fff7ba25c8..cdc14af7b7c6 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-modified/ss_max/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-modified/ss_max/simplified.txt
@@ -3,8 +3,8 @@ WholeStageCodegen (3)
InputAdapter
Exchange #1
WholeStageCodegen (2)
- HashAggregate [ss_sold_date_sk] [count(1),count(ss_sold_date_sk),max(ss_sold_date_sk),max(ss_sold_time_sk),max(ss_item_sk),max(ss_customer_sk),max(ss_cdemo_sk),max(ss_hdemo_sk),max(ss_addr_sk),max(ss_store_sk),max(ss_promo_sk),count(ss_sold_date_sk),count,count,max,max,max,max,max,max,max,max,max,count,count,count,max,max,max,max,max,max,max,max,max,count]
- HashAggregate [ss_sold_date_sk] [count(1),count(ss_sold_date_sk),max(ss_sold_date_sk),max(ss_sold_time_sk),max(ss_item_sk),max(ss_customer_sk),max(ss_cdemo_sk),max(ss_hdemo_sk),max(ss_addr_sk),max(ss_store_sk),max(ss_promo_sk),count,count,max,max,max,max,max,max,max,max,max,count,count,max,max,max,max,max,max,max,max,max]
+ HashAggregate [count,count,max,max,max,max,max,max,max,max,max,ss_sold_date_sk] [count(1),count(ss_sold_date_sk),max(ss_sold_date_sk),max(ss_sold_time_sk),max(ss_item_sk),max(ss_customer_sk),max(ss_cdemo_sk),max(ss_hdemo_sk),max(ss_addr_sk),max(ss_store_sk),max(ss_promo_sk),count(ss_sold_date_sk),count,count,count,max,max,max,max,max,max,max,max,max,count]
+ HashAggregate [ss_sold_date_sk,count,count,max,max,max,max,max,max,max,max,max] [count(1),count(ss_sold_date_sk),max(ss_sold_date_sk),max(ss_sold_time_sk),max(ss_item_sk),max(ss_customer_sk),max(ss_cdemo_sk),max(ss_hdemo_sk),max(ss_addr_sk),max(ss_store_sk),max(ss_promo_sk),count,count,max,max,max,max,max,max,max,max,max]
InputAdapter
Exchange [ss_sold_date_sk] #2
WholeStageCodegen (1)
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.sf100/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.sf100/simplified.txt
index def1677f9440..4b656606e58d 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.sf100/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.sf100/simplified.txt
@@ -3,8 +3,8 @@ WholeStageCodegen (12)
InputAdapter
Exchange #1
WholeStageCodegen (11)
- HashAggregate [cs_order_number] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),count(cs_order_number),sum,sum,count,sum,sum,count]
- HashAggregate [cs_order_number] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),sum,sum,sum,sum]
+ HashAggregate [sum,sum,cs_order_number] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),count(cs_order_number),count,sum,sum,count]
+ HashAggregate [cs_order_number,sum,sum] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),sum,sum]
HashAggregate [cs_order_number,cs_ext_ship_cost,cs_net_profit] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),sum,sum,sum,sum]
Project [cs_order_number,cs_ext_ship_cost,cs_net_profit]
BroadcastHashJoin [cs_ship_date_sk,d_date_sk]
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16/simplified.txt
index 0ae1584bd9cc..89eda335dd4a 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16/simplified.txt
@@ -3,8 +3,8 @@ WholeStageCodegen (12)
InputAdapter
Exchange #1
WholeStageCodegen (11)
- HashAggregate [cs_order_number] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),count(cs_order_number),sum,sum,count,sum,sum,count]
- HashAggregate [cs_order_number] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),sum,sum,sum,sum]
+ HashAggregate [sum,sum,cs_order_number] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),count(cs_order_number),count,sum,sum,count]
+ HashAggregate [cs_order_number,sum,sum] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),sum,sum]
HashAggregate [cs_order_number,cs_ext_ship_cost,cs_net_profit] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),sum,sum,sum,sum]
Project [cs_order_number,cs_ext_ship_cost,cs_net_profit]
BroadcastHashJoin [cs_call_center_sk,cc_call_center_sk]
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.sf100/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.sf100/simplified.txt
index 25f9d020f58f..bb7b3d5690eb 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.sf100/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28.sf100/simplified.txt
@@ -8,8 +8,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #1
WholeStageCodegen (2)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #2
WholeStageCodegen (1)
@@ -26,8 +26,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #4
WholeStageCodegen (4)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #5
WholeStageCodegen (3)
@@ -44,8 +44,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #7
WholeStageCodegen (7)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #8
WholeStageCodegen (6)
@@ -62,8 +62,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #10
WholeStageCodegen (10)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #11
WholeStageCodegen (9)
@@ -80,8 +80,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #13
WholeStageCodegen (13)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #14
WholeStageCodegen (12)
@@ -98,8 +98,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #16
WholeStageCodegen (16)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #17
WholeStageCodegen (15)
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28/simplified.txt
index 25f9d020f58f..bb7b3d5690eb 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q28/simplified.txt
@@ -8,8 +8,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #1
WholeStageCodegen (2)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #2
WholeStageCodegen (1)
@@ -26,8 +26,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #4
WholeStageCodegen (4)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #5
WholeStageCodegen (3)
@@ -44,8 +44,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #7
WholeStageCodegen (7)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #8
WholeStageCodegen (6)
@@ -62,8 +62,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #10
WholeStageCodegen (10)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #11
WholeStageCodegen (9)
@@ -80,8 +80,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #13
WholeStageCodegen (13)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #14
WholeStageCodegen (12)
@@ -98,8 +98,8 @@ WholeStageCodegen (18)
InputAdapter
Exchange #16
WholeStageCodegen (16)
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),sum,count,count,count,sum,count,count,count]
- HashAggregate [ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count,sum,count,count]
+ HashAggregate [sum,count,count,ss_list_price] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),count(ss_list_price),count,sum,count,count,count]
+ HashAggregate [ss_list_price,sum,count,count] [avg(UnscaledValue(ss_list_price)),count(ss_list_price),sum,count,count]
InputAdapter
Exchange [ss_list_price] #17
WholeStageCodegen (15)
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q70.sf100/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q70.sf100/simplified.txt
index f98948a8cf59..1c496f1fef26 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q70.sf100/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q70.sf100/simplified.txt
@@ -50,7 +50,7 @@ TakeOrderedAndProject [lochierarchy,s_state,rank_within_parent,total_sum,s_count
WindowGroupLimit [s_state,_w0]
WholeStageCodegen (5)
Sort [s_state,_w0]
- HashAggregate [sum] [sum(UnscaledValue(ss_net_profit)),_w0,s_state,sum]
+ HashAggregate [s_state,sum] [sum(UnscaledValue(ss_net_profit)),_w0,sum]
InputAdapter
Exchange [s_state] #6
WholeStageCodegen (4)
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q70/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q70/simplified.txt
index e61fb9ef8d78..80390af6790f 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q70/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q70/simplified.txt
@@ -50,7 +50,7 @@ TakeOrderedAndProject [lochierarchy,s_state,rank_within_parent,total_sum,s_count
WindowGroupLimit [s_state,_w0]
WholeStageCodegen (5)
Sort [s_state,_w0]
- HashAggregate [sum] [sum(UnscaledValue(ss_net_profit)),_w0,s_state,sum]
+ HashAggregate [s_state,sum] [sum(UnscaledValue(ss_net_profit)),_w0,sum]
InputAdapter
Exchange [s_state] #6
WholeStageCodegen (4)
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.sf100/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.sf100/simplified.txt
index 230b08abe0a5..644722cbab27 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.sf100/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94.sf100/simplified.txt
@@ -3,8 +3,8 @@ WholeStageCodegen (12)
InputAdapter
Exchange #1
WholeStageCodegen (11)
- HashAggregate [ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),count(ws_order_number),sum,sum,count,sum,sum,count]
- HashAggregate [ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum,sum,sum]
+ HashAggregate [sum,sum,ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),count(ws_order_number),count,sum,sum,count]
+ HashAggregate [ws_order_number,sum,sum] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum]
HashAggregate [ws_order_number,ws_ext_ship_cost,ws_net_profit] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum,sum,sum]
Project [ws_order_number,ws_ext_ship_cost,ws_net_profit]
BroadcastHashJoin [ws_ship_date_sk,d_date_sk]
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94/simplified.txt
index 86981cc8083c..c3e58c5bacbf 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q94/simplified.txt
@@ -3,8 +3,8 @@ WholeStageCodegen (12)
InputAdapter
Exchange #1
WholeStageCodegen (11)
- HashAggregate [ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),count(ws_order_number),sum,sum,count,sum,sum,count]
- HashAggregate [ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum,sum,sum]
+ HashAggregate [sum,sum,ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),count(ws_order_number),count,sum,sum,count]
+ HashAggregate [ws_order_number,sum,sum] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum]
HashAggregate [ws_order_number,ws_ext_ship_cost,ws_net_profit] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum,sum,sum]
Project [ws_order_number,ws_ext_ship_cost,ws_net_profit]
BroadcastHashJoin [ws_web_site_sk,web_site_sk]
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.sf100/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.sf100/simplified.txt
index 8922d43c2aaa..1653cba1f0f3 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.sf100/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95.sf100/simplified.txt
@@ -3,8 +3,8 @@ WholeStageCodegen (21)
InputAdapter
Exchange #1
WholeStageCodegen (20)
- HashAggregate [ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),count(ws_order_number),sum,sum,count,sum,sum,count]
- HashAggregate [ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum,sum,sum]
+ HashAggregate [sum,sum,ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),count(ws_order_number),count,sum,sum,count]
+ HashAggregate [ws_order_number,sum,sum] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum]
HashAggregate [ws_order_number,ws_ext_ship_cost,ws_net_profit] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum,sum,sum]
Project [ws_order_number,ws_ext_ship_cost,ws_net_profit]
BroadcastHashJoin [ws_ship_date_sk,d_date_sk]
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95/simplified.txt
index f628b290944a..9acbb8dfd048 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q95/simplified.txt
@@ -3,8 +3,8 @@ WholeStageCodegen (21)
InputAdapter
Exchange #1
WholeStageCodegen (20)
- HashAggregate [ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),count(ws_order_number),sum,sum,count,sum,sum,count]
- HashAggregate [ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum,sum,sum]
+ HashAggregate [sum,sum,ws_order_number] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),count(ws_order_number),count,sum,sum,count]
+ HashAggregate [ws_order_number,sum,sum] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum]
HashAggregate [ws_order_number,ws_ext_ship_cost,ws_net_profit] [sum(UnscaledValue(ws_ext_ship_cost)),sum(UnscaledValue(ws_net_profit)),sum,sum,sum,sum]
Project [ws_order_number,ws_ext_ship_cost,ws_net_profit]
BroadcastHashJoin [ws_web_site_sk,web_site_sk]
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q70a.sf100/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q70a.sf100/simplified.txt
index 2626c57dd127..942ce8297436 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q70a.sf100/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q70a.sf100/simplified.txt
@@ -57,7 +57,7 @@ TakeOrderedAndProject [lochierarchy,s_state,rank_within_parent,total_sum,s_count
WindowGroupLimit [s_state,_w0]
WholeStageCodegen (5)
Sort [s_state,_w0]
- HashAggregate [sum] [sum(UnscaledValue(ss_net_profit)),_w0,s_state,sum]
+ HashAggregate [s_state,sum] [sum(UnscaledValue(ss_net_profit)),_w0,sum]
InputAdapter
Exchange [s_state] #7
WholeStageCodegen (4)
diff --git a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q70a/simplified.txt b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q70a/simplified.txt
index 8d0f4d8a9a94..e9431c1365eb 100644
--- a/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q70a/simplified.txt
+++ b/sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v2_7/q70a/simplified.txt
@@ -57,7 +57,7 @@ TakeOrderedAndProject [lochierarchy,s_state,rank_within_parent,total_sum,s_count
WindowGroupLimit [s_state,_w0]
WholeStageCodegen (5)
Sort [s_state,_w0]
- HashAggregate [sum] [sum(UnscaledValue(ss_net_profit)),_w0,s_state,sum]
+ HashAggregate [s_state,sum] [sum(UnscaledValue(ss_net_profit)),_w0,sum]
InputAdapter
Exchange [s_state] #7
WholeStageCodegen (4)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
index b926cc192bd6..7c434ebee5f9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
@@ -23,6 +23,7 @@ import org.apache.spark.sql.{execution, DataFrame, Row}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, DeclarativeAggregate, Final}
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan, Range, Repartition, RepartitionOperation, Union}
import org.apache.spark.sql.catalyst.plans.physical._
@@ -1584,6 +1585,41 @@ class PlannerSuite extends SharedSparkSession with AdaptiveSparkPlanHelper {
// The groupBy reuse the output partitioning after DirectShufflePartitionID.
checkShuffleCount(grouped, 3)
}
+
+ test("SPARK-55979: required input attributes are missing from PartialMerge / Final " +
+ "BaseAggregateExec.references") {
+ val scanAggBufferAttr = AttributeReference("buf", IntegerType, nullable = true)()
+ val inputAggBufferAttr = scanAggBufferAttr.withName("renamed_buf")
+
+ case class MyAggregate() extends DeclarativeAggregate {
+ override def children: Seq[Expression] = Nil
+ override def nullable: Boolean = true
+ override def dataType: DataType = IntegerType
+ override def prettyName: String = "test_agg"
+ override val aggBufferAttributes: Seq[AttributeReference] = Seq(scanAggBufferAttr)
+ override lazy val inputAggBufferAttributes: Seq[AttributeReference] =
+ Seq(inputAggBufferAttr)
+ override val initialValues: Seq[Expression] = Seq(Literal(0))
+ override val updateExpressions: Seq[Expression] = Seq(scanAggBufferAttr)
+ override val mergeExpressions: Seq[Expression] = Seq(inputAggBufferAttr)
+ override val evaluateExpression: Expression = scanAggBufferAttr
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[Expression]): Expression = copy()
+ }
+ val aggregateExpression = AggregateExpression(MyAggregate(), Final, isDistinct = false)
+ val aggregate = HashAggregateExec(
+ requiredChildDistributionExpressions = None,
+ isStreaming = false,
+ numShufflePartitions = None,
+ groupingExpressions = Nil,
+ aggregateExpressions = Seq(aggregateExpression),
+ aggregateAttributes = Seq(aggregateExpression.resultAttribute),
+ initialInputBufferOffset = 0,
+ resultExpressions = Seq(Alias(aggregateExpression, "result")()),
+ child = LocalTableScanExec(Seq(scanAggBufferAttr), Nil, None))
+
+ assert(aggregate.references.contains(inputAggBufferAttr))
+ }
}
// Used for unit-testing EnsureRequirements
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org