Skip to content

[SPARK-35349][SQL] Add code-gen for left/right outer sort merge join #32476

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 6 commits into from
Closed

[SPARK-35349][SQL] Add code-gen for left/right outer sort merge join #32476

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
982af12
Add codegen for left/right outer sort merge join
c21 May 6, 2021
4a57664
Fix unit test failures
c21 May 8, 2021
44b210f
Fix more unit test failure
c21 May 9, 2021
765b247
Add comment for findNextJoinRows
c21 May 11, 2021
617f89c
Address all comments
c21 May 11, 2021
429edcc
Address all comments
c21 May 12, 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
Next Next commit
Fix more unit test failure
  • Loading branch information
@c21
c21 committed May 11, 2021
commit 44b210fff8dd04e0ed76924fc9bca4b3907af41c
4 changes: 1 addition & 3 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -520,9 +520,7 @@ case class SortMergeJoinExec(
| }
| } while ($streamedRow != null);
| }
|
| throw new IllegalStateException("Executed unreachable code path in findNextJoinRows");
| return false;
| return false; // unreachable
Copy link
Member

@maropu maropu May 10, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(This is not related to this PR though) In this case, could we throw an illegal state exception?

Copy link
Contributor Author

@c21 c21 May 10, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@maropu - sounds good to me. Will update.

Copy link
Contributor Author

@c21 c21 May 11, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@maropu - Interesting, when I tried to add a throw new IllegalStateException before return false, janino compiler is clever enough to figure out the statement is unreachable and throws exception when trying to compile - https://gist.github.com/c21/196166411d5d0406d9a76b37be889194 . So I think we'd better keep this as it is for now?

Copy link
Member

@maropu maropu May 13, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, I missed this comment. it looks interesting. sgtm.

|}
""".stripMargin, inlineToOuterClass = true)

Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q40.sf100/explain.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ TakeOrderedAndProject (36)
: :- * Project (20)
: : +- * BroadcastHashJoin Inner BuildRight (19)
: : :- * Project (13)
: : : +- SortMergeJoin LeftOuter (12)
: : : +- * SortMergeJoin LeftOuter (12)
: : : :- * Sort (5)
: : : : +- Exchange (4)
: : : : +- * Filter (3)
Expand Down Expand Up @@ -86,7 +86,7 @@ Arguments: hashpartitioning(cr_order_number#9, cr_item_sk#8, 5), ENSURE_REQUIREM
Input [3]: [cr_item_sk#8, cr_order_number#9, cr_refunded_cash#10]
Arguments: [cr_order_number#9 ASC NULLS FIRST, cr_item_sk#8 ASC NULLS FIRST], false, 0

(12) SortMergeJoin
(12) SortMergeJoin [codegen id : 8]
Left keys [2]: [cs_order_number#3, cs_item_sk#2]
Right keys [2]: [cr_order_number#9, cr_item_sk#8]
Join condition: None
Expand Down
5 changes: 3 additions & 2 deletions ...core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q40.sf100/simplified.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,8 @@ TakeOrderedAndProject [w_state,i_item_id,sales_before,sales_after]
Project [cs_warehouse_sk,cs_sales_price,cs_sold_date_sk,cr_refunded_cash,i_item_id]
BroadcastHashJoin [cs_item_sk,i_item_sk]
Project [cs_warehouse_sk,cs_item_sk,cs_sales_price,cs_sold_date_sk,cr_refunded_cash]
InputAdapter
SortMergeJoin [cs_order_number,cs_item_sk,cr_order_number,cr_item_sk]
SortMergeJoin [cs_order_number,cs_item_sk,cr_order_number,cr_item_sk]
InputAdapter
WholeStageCodegen (2)
Sort [cs_order_number,cs_item_sk]
InputAdapter
Expand All @@ -25,6 +25,7 @@ TakeOrderedAndProject [w_state,i_item_id,sales_before,sales_after]
Scan parquet default.catalog_sales [cs_warehouse_sk,cs_item_sk,cs_order_number,cs_sales_price,cs_sold_date_sk]
SubqueryBroadcast [d_date_sk] #1
ReusedExchange [d_date_sk,d_date] #3
InputAdapter
WholeStageCodegen (4)
Sort [cr_order_number,cr_item_sk]
InputAdapter
Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q72.sf100/explain.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ TakeOrderedAndProject (80)
+- Exchange (78)
+- * HashAggregate (77)
+- * Project (76)
+- SortMergeJoin LeftOuter (75)
+- * SortMergeJoin LeftOuter (75)
:- * Sort (68)
: +- Exchange (67)
: +- * Project (66)
Expand Down Expand Up @@ -410,7 +410,7 @@ Arguments: hashpartitioning(cr_item_sk#43, cr_order_number#44, 5), ENSURE_REQUIR
Input [2]: [cr_item_sk#43, cr_order_number#44]
Arguments: [cr_item_sk#43 ASC NULLS FIRST, cr_order_number#44 ASC NULLS FIRST], false, 0

(75) SortMergeJoin
(75) SortMergeJoin [codegen id : 20]
Left keys [2]: [cs_item_sk#4, cs_order_number#6]
Right keys [2]: [cr_item_sk#43, cr_order_number#44]
Join condition: None
Expand Down
5 changes: 3 additions & 2 deletions ...core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q72.sf100/simplified.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@ TakeOrderedAndProject [total_cnt,i_item_desc,w_warehouse_name,d_week_seq,no_prom
WholeStageCodegen (20)
HashAggregate [i_item_desc,w_warehouse_name,d_week_seq] [count,count]
Project [w_warehouse_name,i_item_desc,d_week_seq]
InputAdapter
SortMergeJoin [cs_item_sk,cs_order_number,cr_item_sk,cr_order_number]
SortMergeJoin [cs_item_sk,cs_order_number,cr_item_sk,cr_order_number]
InputAdapter
WholeStageCodegen (17)
Sort [cs_item_sk,cs_order_number]
InputAdapter
Expand Down Expand Up @@ -121,6 +121,7 @@ TakeOrderedAndProject [total_cnt,i_item_desc,w_warehouse_name,d_week_seq,no_prom
ColumnarToRow
InputAdapter
Scan parquet default.promotion [p_promo_sk]
InputAdapter
WholeStageCodegen (19)
Sort [cr_item_sk,cr_order_number]
InputAdapter
Expand Down
24 changes: 12 additions & 12 deletions sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q75.sf100/explain.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ TakeOrderedAndProject (133)
: +- * HashAggregate (66)
: +- Union (65)
: :- * Project (26)
: : +- SortMergeJoin LeftOuter (25)
: : +- * SortMergeJoin LeftOuter (25)
: : :- * Sort (18)
: : : +- Exchange (17)
: : : +- * Project (16)
Expand All @@ -38,7 +38,7 @@ TakeOrderedAndProject (133)
: : +- * ColumnarToRow (20)
: : +- Scan parquet default.catalog_returns (19)
: :- * Project (45)
: : +- SortMergeJoin LeftOuter (44)
: : +- * SortMergeJoin LeftOuter (44)
: : :- * Sort (37)
: : : +- Exchange (36)
: : : +- * Project (35)
Expand All @@ -57,7 +57,7 @@ TakeOrderedAndProject (133)
: : +- * ColumnarToRow (39)
: : +- Scan parquet default.store_returns (38)
: +- * Project (64)
: +- SortMergeJoin LeftOuter (63)
: +- * SortMergeJoin LeftOuter (63)
: :- * Sort (56)
: : +- Exchange (55)
: : +- * Project (54)
Expand Down Expand Up @@ -85,7 +85,7 @@ TakeOrderedAndProject (133)
+- * HashAggregate (123)
+- Union (122)
:- * Project (91)
: +- SortMergeJoin LeftOuter (90)
: +- * SortMergeJoin LeftOuter (90)
: :- * Sort (87)
: : +- Exchange (86)
: : +- * Project (85)
Expand All @@ -103,7 +103,7 @@ TakeOrderedAndProject (133)
: +- * Sort (89)
: +- ReusedExchange (88)
:- * Project (106)
: +- SortMergeJoin LeftOuter (105)
: +- * SortMergeJoin LeftOuter (105)
: :- * Sort (102)
: : +- Exchange (101)
: : +- * Project (100)
Expand All @@ -118,7 +118,7 @@ TakeOrderedAndProject (133)
: +- * Sort (104)
: +- ReusedExchange (103)
+- * Project (121)
+- SortMergeJoin LeftOuter (120)
+- * SortMergeJoin LeftOuter (120)
:- * Sort (117)
: +- Exchange (116)
: +- * Project (115)
Expand Down Expand Up @@ -241,7 +241,7 @@ Arguments: hashpartitioning(cr_order_number#19, cr_item_sk#18, 5), ENSURE_REQUIR
Input [4]: [cr_item_sk#18, cr_order_number#19, cr_return_quantity#20, cr_return_amount#21]
Arguments: [cr_order_number#19 ASC NULLS FIRST, cr_item_sk#18 ASC NULLS FIRST], false, 0

(25) SortMergeJoin
(25) SortMergeJoin [codegen id : 7]
Left keys [2]: [cs_order_number#2, cs_item_sk#1]
Right keys [2]: [cr_order_number#19, cr_item_sk#18]
Join condition: None
Expand Down Expand Up @@ -323,7 +323,7 @@ Arguments: hashpartitioning(sr_ticket_number#40, sr_item_sk#39, 5), ENSURE_REQUI
Input [4]: [sr_item_sk#39, sr_ticket_number#40, sr_return_quantity#41, sr_return_amt#42]
Arguments: [sr_ticket_number#40 ASC NULLS FIRST, sr_item_sk#39 ASC NULLS FIRST], false, 0

(44) SortMergeJoin
(44) SortMergeJoin [codegen id : 14]
Left keys [2]: [ss_ticket_number#27, ss_item_sk#26]
Right keys [2]: [sr_ticket_number#40, sr_item_sk#39]
Join condition: None
Expand Down Expand Up @@ -405,7 +405,7 @@ Arguments: hashpartitioning(wr_order_number#61, wr_item_sk#60, 5), ENSURE_REQUIR
Input [4]: [wr_item_sk#60, wr_order_number#61, wr_return_quantity#62, wr_return_amt#63]
Arguments: [wr_order_number#61 ASC NULLS FIRST, wr_item_sk#60 ASC NULLS FIRST], false, 0

(63) SortMergeJoin
(63) SortMergeJoin [codegen id : 21]
Left keys [2]: [ws_order_number#48, ws_item_sk#47]
Right keys [2]: [wr_order_number#61, wr_item_sk#60]
Join condition: None
Expand Down Expand Up @@ -529,7 +529,7 @@ Output [4]: [cr_item_sk#94, cr_order_number#95, cr_return_quantity#96, cr_return
Input [4]: [cr_item_sk#94, cr_order_number#95, cr_return_quantity#96, cr_return_amount#97]
Arguments: [cr_order_number#95 ASC NULLS FIRST, cr_item_sk#94 ASC NULLS FIRST], false, 0

(90) SortMergeJoin
(90) SortMergeJoin [codegen id : 32]
Left keys [2]: [cs_order_number#80, cs_item_sk#79]
Right keys [2]: [cr_order_number#95, cr_item_sk#94]
Join condition: None
Expand Down Expand Up @@ -592,7 +592,7 @@ Output [4]: [sr_item_sk#111, sr_ticket_number#112, sr_return_quantity#113, sr_re
Input [4]: [sr_item_sk#111, sr_ticket_number#112, sr_return_quantity#113, sr_return_amt#114]
Arguments: [sr_ticket_number#112 ASC NULLS FIRST, sr_item_sk#111 ASC NULLS FIRST], false, 0

(105) SortMergeJoin
(105) SortMergeJoin [codegen id : 39]
Left keys [2]: [ss_ticket_number#99, ss_item_sk#98]
Right keys [2]: [sr_ticket_number#112, sr_item_sk#111]
Join condition: None
Expand Down Expand Up @@ -655,7 +655,7 @@ Output [4]: [wr_item_sk#128, wr_order_number#129, wr_return_quantity#130, wr_ret
Input [4]: [wr_item_sk#128, wr_order_number#129, wr_return_quantity#130, wr_return_amt#131]
Arguments: [wr_order_number#129 ASC NULLS FIRST, wr_item_sk#128 ASC NULLS FIRST], false, 0

(120) SortMergeJoin
(120) SortMergeJoin [codegen id : 46]
Left keys [2]: [ws_order_number#116, ws_item_sk#115]
Right keys [2]: [wr_order_number#129, wr_item_sk#128]
Join condition: None
Expand Down
Loading