Skip to content

[SPARK-37455][SQL] Replace hash with sort aggregate if child is already sorted #34702

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
c21 wants to merge 5 commits into from
Closed

[SPARK-37455][SQL] Replace hash with sort aggregate if child is already sorted #34702

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6448864
Replace hash agg with sort agg if child is already sorted
c21 Nov 24, 2021
a683137
Do not replace hash agg if grouping expression is empty
c21 Nov 26, 2021
e8609fd
Fix unit test failure
c21 Nov 27, 2021
cff1424
Check the approach to check partial agg based on logical plan instead
c21 Dec 1, 2021
8ce7d27
Disable rule by default and back out TPCDS plan change
c21 Dec 3, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Disable rule by default and back out TPCDS plan change
  • Loading branch information
@c21
c21 committed Dec 3, 2021
commit 8ce7d279cec23d3bc50c8d6f870a10b11395523c
2 changes: 1 addition & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1509,7 +1509,7 @@ object SQLConf {
.doc("Whether to replace hash aggregate node with sort aggregate based on children's ordering")
.version("3.3.0")
.booleanConf
.createWithDefault(true)
.createWithDefault(false)

val STATE_STORE_PROVIDER_CLASS =
buildConf("spark.sql.streaming.stateStore.providerClass")
Expand Down
10 changes: 5 additions & 5 deletions sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.sf100/explain.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
+- Exchange (44)
+- * HashAggregate (43)
+- * HashAggregate (42)
+- SortAggregate (41)
+- * HashAggregate (41)
+- * Project (40)
+- * BroadcastHashJoin Inner BuildRight (39)
:- * Project (33)
Expand Down Expand Up @@ -221,21 +221,21 @@ Join condition: None
Output [3]: [cs_order_number#5, cs_ext_ship_cost#6, cs_net_profit#7]
Input [5]: [cs_ship_date_sk#1, cs_order_number#5, cs_ext_ship_cost#6, cs_net_profit#7, d_date_sk#23]

(41) SortAggregate
(41) HashAggregate [codegen id : 11]
Input [3]: [cs_order_number#5, cs_ext_ship_cost#6, cs_net_profit#7]
Keys [1]: [cs_order_number#5]
Functions [2]: [partial_sum(UnscaledValue(cs_ext_ship_cost#6)), partial_sum(UnscaledValue(cs_net_profit#7))]
Aggregate Attributes [2]: [sum(UnscaledValue(cs_ext_ship_cost#6))#26, sum(UnscaledValue(cs_net_profit#7))#27]
Results [3]: [cs_order_number#5, sum#28, sum#29]

(42) HashAggregate [codegen id : 12]
(42) HashAggregate [codegen id : 11]
Input [3]: [cs_order_number#5, sum#28, sum#29]
Keys [1]: [cs_order_number#5]
Functions [2]: [merge_sum(UnscaledValue(cs_ext_ship_cost#6)), merge_sum(UnscaledValue(cs_net_profit#7))]
Aggregate Attributes [2]: [sum(UnscaledValue(cs_ext_ship_cost#6))#26, sum(UnscaledValue(cs_net_profit#7))#27]
Results [3]: [cs_order_number#5, sum#28, sum#29]

(43) HashAggregate [codegen id : 12]
(43) HashAggregate [codegen id : 11]
Input [3]: [cs_order_number#5, sum#28, sum#29]
Keys: []
Functions [3]: [merge_sum(UnscaledValue(cs_ext_ship_cost#6)), merge_sum(UnscaledValue(cs_net_profit#7)), partial_count(distinct cs_order_number#5)]
Expand All @@ -246,7 +246,7 @@ Results [3]: [sum#28, sum#29, count#31]
Input [3]: [sum#28, sum#29, count#31]
Arguments: SinglePartition, ENSURE_REQUIREMENTS, [id=#32]

(45) HashAggregate [codegen id : 13]
(45) HashAggregate [codegen id : 12]
Input [3]: [sum#28, sum#29, count#31]
Keys: []
Functions [3]: [sum(UnscaledValue(cs_ext_ship_cost#6)), sum(UnscaledValue(cs_net_profit#7)), count(distinct cs_order_number#5)]
Expand Down
126 changes: 62 additions & 64 deletions ...core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16.sf100/simplified.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,76 +1,74 @@
WholeStageCodegen (13)
WholeStageCodegen (12)
HashAggregate [sum,sum,count] [sum(UnscaledValue(cs_ext_ship_cost)),sum(UnscaledValue(cs_net_profit)),count(cs_order_number),order count ,total shipping cost ,total net profit ,sum,sum,count]
InputAdapter
Exchange #1
WholeStageCodegen (12)
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]
InputAdapter
SortAggregate [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]
WholeStageCodegen (11)
Project [cs_order_number,cs_ext_ship_cost,cs_net_profit]
BroadcastHashJoin [cs_ship_date_sk,d_date_sk]
Project [cs_ship_date_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit]
BroadcastHashJoin [cs_call_center_sk,cc_call_center_sk]
Project [cs_ship_date_sk,cs_call_center_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit]
BroadcastHashJoin [cs_ship_addr_sk,ca_address_sk]
SortMergeJoin [cs_order_number,cr_order_number]
InputAdapter
WholeStageCodegen (5)
Project [cs_ship_date_sk,cs_ship_addr_sk,cs_call_center_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit]
SortMergeJoin [cs_order_number,cs_order_number,cs_warehouse_sk,cs_warehouse_sk]
InputAdapter
WholeStageCodegen (2)
Sort [cs_order_number]
InputAdapter
Exchange [cs_order_number] #2
WholeStageCodegen (1)
Project [cs_ship_date_sk,cs_ship_addr_sk,cs_call_center_sk,cs_warehouse_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit]
Filter [cs_ship_date_sk,cs_ship_addr_sk,cs_call_center_sk]
ColumnarToRow
InputAdapter
Scan parquet default.catalog_sales [cs_ship_date_sk,cs_ship_addr_sk,cs_call_center_sk,cs_warehouse_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit,cs_sold_date_sk]
InputAdapter
WholeStageCodegen (4)
Sort [cs_order_number]
InputAdapter
Exchange [cs_order_number] #3
WholeStageCodegen (3)
Project [cs_warehouse_sk,cs_order_number]
ColumnarToRow
InputAdapter
Scan parquet default.catalog_sales [cs_warehouse_sk,cs_order_number,cs_sold_date_sk]
InputAdapter
WholeStageCodegen (7)
Sort [cr_order_number]
InputAdapter
Exchange [cr_order_number] #4
WholeStageCodegen (6)
Project [cr_order_number]
ColumnarToRow
InputAdapter
Scan parquet default.catalog_returns [cr_order_number,cr_returned_date_sk]
InputAdapter
BroadcastExchange #5
WholeStageCodegen (8)
Project [ca_address_sk]
Filter [ca_state,ca_address_sk]
ColumnarToRow
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]
Project [cs_ship_date_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit]
BroadcastHashJoin [cs_call_center_sk,cc_call_center_sk]
Project [cs_ship_date_sk,cs_call_center_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit]
BroadcastHashJoin [cs_ship_addr_sk,ca_address_sk]
SortMergeJoin [cs_order_number,cr_order_number]
InputAdapter
WholeStageCodegen (5)
Project [cs_ship_date_sk,cs_ship_addr_sk,cs_call_center_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit]
SortMergeJoin [cs_order_number,cs_order_number,cs_warehouse_sk,cs_warehouse_sk]
InputAdapter
WholeStageCodegen (2)
Sort [cs_order_number]
InputAdapter
Exchange [cs_order_number] #2
WholeStageCodegen (1)
Project [cs_ship_date_sk,cs_ship_addr_sk,cs_call_center_sk,cs_warehouse_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit]
Filter [cs_ship_date_sk,cs_ship_addr_sk,cs_call_center_sk]
ColumnarToRow
InputAdapter
Scan parquet default.catalog_sales [cs_ship_date_sk,cs_ship_addr_sk,cs_call_center_sk,cs_warehouse_sk,cs_order_number,cs_ext_ship_cost,cs_net_profit,cs_sold_date_sk]
InputAdapter
WholeStageCodegen (4)
Sort [cs_order_number]
InputAdapter
Scan parquet default.customer_address [ca_address_sk,ca_state]
Exchange [cs_order_number] #3
WholeStageCodegen (3)
Project [cs_warehouse_sk,cs_order_number]
ColumnarToRow
InputAdapter
Scan parquet default.catalog_sales [cs_warehouse_sk,cs_order_number,cs_sold_date_sk]
InputAdapter
WholeStageCodegen (7)
Sort [cr_order_number]
InputAdapter
Exchange [cr_order_number] #4
WholeStageCodegen (6)
Project [cr_order_number]
ColumnarToRow
InputAdapter
Scan parquet default.catalog_returns [cr_order_number,cr_returned_date_sk]
InputAdapter
BroadcastExchange #6
WholeStageCodegen (9)
Project [cc_call_center_sk]
Filter [cc_county,cc_call_center_sk]
BroadcastExchange #5
WholeStageCodegen (8)
Project [ca_address_sk]
Filter [ca_state,ca_address_sk]
ColumnarToRow
InputAdapter
Scan parquet default.call_center [cc_call_center_sk,cc_county]
Scan parquet default.customer_address [ca_address_sk,ca_state]
InputAdapter
BroadcastExchange #7
WholeStageCodegen (10)
Project [d_date_sk]
Filter [d_date,d_date_sk]
BroadcastExchange #6
WholeStageCodegen (9)
Project [cc_call_center_sk]
Filter [cc_county,cc_call_center_sk]
ColumnarToRow
InputAdapter
Scan parquet default.date_dim [d_date_sk,d_date]
Scan parquet default.call_center [cc_call_center_sk,cc_county]
InputAdapter
BroadcastExchange #7
WholeStageCodegen (10)
Project [d_date_sk]
Filter [d_date,d_date_sk]
ColumnarToRow
InputAdapter
Scan parquet default.date_dim [d_date_sk,d_date]
10 changes: 5 additions & 5 deletions sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q16/explain.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
+- Exchange (44)
+- * HashAggregate (43)
+- * HashAggregate (42)
+- SortAggregate (41)
+- * HashAggregate (41)
+- * Project (40)
+- * BroadcastHashJoin Inner BuildRight (39)
:- * Project (33)
Expand Down Expand Up @@ -221,21 +221,21 @@ Join condition: None
Output [3]: [cs_order_number#5, cs_ext_ship_cost#6, cs_net_profit#7]
Input [5]: [cs_call_center_sk#3, cs_order_number#5, cs_ext_ship_cost#6, cs_net_profit#7, cc_call_center_sk#23]

(41) SortAggregate
(41) HashAggregate [codegen id : 11]
Input [3]: [cs_order_number#5, cs_ext_ship_cost#6, cs_net_profit#7]
Keys [1]: [cs_order_number#5]
Functions [2]: [partial_sum(UnscaledValue(cs_ext_ship_cost#6)), partial_sum(UnscaledValue(cs_net_profit#7))]
Aggregate Attributes [2]: [sum(UnscaledValue(cs_ext_ship_cost#6))#26, sum(UnscaledValue(cs_net_profit#7))#27]
Results [3]: [cs_order_number#5, sum#28, sum#29]

(42) HashAggregate [codegen id : 12]
(42) HashAggregate [codegen id : 11]
Input [3]: [cs_order_number#5, sum#28, sum#29]
Keys [1]: [cs_order_number#5]
Functions [2]: [merge_sum(UnscaledValue(cs_ext_ship_cost#6)), merge_sum(UnscaledValue(cs_net_profit#7))]
Aggregate Attributes [2]: [sum(UnscaledValue(cs_ext_ship_cost#6))#26, sum(UnscaledValue(cs_net_profit#7))#27]
Results [3]: [cs_order_number#5, sum#28, sum#29]

(43) HashAggregate [codegen id : 12]
(43) HashAggregate [codegen id : 11]
Input [3]: [cs_order_number#5, sum#28, sum#29]
Keys: []
Functions [3]: [merge_sum(UnscaledValue(cs_ext_ship_cost#6)), merge_sum(UnscaledValue(cs_net_profit#7)), partial_count(distinct cs_order_number#5)]
Expand All @@ -246,7 +246,7 @@ Results [3]: [sum#28, sum#29, count#31]
Input [3]: [sum#28, sum#29, count#31]
Arguments: SinglePartition, ENSURE_REQUIREMENTS, [id=#32]

(45) HashAggregate [codegen id : 13]
(45) HashAggregate [codegen id : 12]
Input [3]: [sum#28, sum#29, count#31]
Keys: []
Functions [3]: [sum(UnscaledValue(cs_ext_ship_cost#6)), sum(UnscaledValue(cs_net_profit#7)), count(distinct cs_order_number#5)]
Expand Down
Loading