Hash join plan with datasource v2 in 3.5 has 2 more exchanges than 3.4
- Dominant language
- Scala
- Stars
- 1.6k
- Forks
- 657
- Avg merge
- 2d 14h
- Merged PRs (30d)
- 80
Description
_Originally posted by @ayushi-agarwal in https://github.com/apache/incubator-gluten/pull/5445#discussion_r1570045497_
This is the plan in 3.5 for the query
"select l_partkey from lineitem join part join partsupp on l_partkey = p_partkey and l_suppkey = ps_suppkey"
*(9) Project [l_partkey#200L]
+- *(9) SortMergeJoin [l_suppkey#201L], [ps_suppkey#156L], Inner
:- *(6) Sort [l_suppkey#201L ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(l_suppkey#201L, 5), ENSURE_REQUIREMENTS, [plan_id=300]
: +- *(5) Project [l_partkey#200L, l_suppkey#201L]
: +- *(5) SortMergeJoin [l_partkey#200L], [p_partkey#123L], Inner
: :- *(2) Sort [l_partkey#200L ASC NULLS FIRST], false, 0
: : +- Exchange hashpartitioning(l_partkey#200L, 5), ENSURE_REQUIREMENTS, [plan_id=283]
: : +- *(1) Filter (isnotnull(l_partkey#200L) AND isnotnull(l_suppkey#201L))
: : +- *(1) ColumnarToRow
: : +- BatchScan parquet file:/root/workspace/apache_1/backends-velox/target/scala-2.12/test-classes/tpch-data-parquet-velox/lineitem[l_partkey#200L, l_suppkey#201L] ParquetScan DataFilters: [isnotnull(l_partkey#200L), isnotnull(l_suppkey#201L)], Format: parquet, Location: InMemoryFileIndex(1 paths)[file:/root/workspace/apache_1/backends-velox/target/scala-2.12/test-cl..., PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(l_partkey), IsNotNull(l_suppkey)], PushedGroupBy: [], ReadSchema: struct RuntimeFilters: []
: +- *(4) Sort [p_partkey#123L ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(p_partkey#123L, 5), ENSURE_REQUIREMENTS, [plan_id=292]
: +- *(3) Filter isnotnull(p_partkey#123L)
: +- *(3) ColumnarToRow
: +- BatchScan parquet file:/root/workspace/apache_1/backends-velox/target/scala-2.12/test-classes/tpch-data-parquet-velox/part[p_partkey#123L] ParquetScan DataFilters: [isnotnull(p_partkey#123L)], Format: parquet, Location: InMemoryFileIndex(1 paths)[file:/root/workspace/apache_1/backends-velox/target/scala-2.12/test-cl..., PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(p_partkey)], PushedGroupBy: [], ReadSchema: struct RuntimeFilters: []
+- *(8) Sort [ps_suppkey#156L ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(ps_suppkey#156L, 5), ENSURE_REQUIREMENTS, [plan_id=309]
+- *(7) Filter isnotnull(ps_suppkey#156L)
+- *(7) ColumnarToRow
+- BatchScan parquet file:/root/workspace/apache_1/backends-velox/target/scala-2.12/test-classes/tpch-data-parquet-velox/partsupp[ps_suppkey#156L] ParquetScan DataFilters: [isnotnull(ps_suppkey#156L)], Format: parquet, Location: InMemoryFileIndex(1 paths)[file:/root/workspace/apache_1/backends-velox/target/scala-2.12/test-cl..., PartitionFilters: [], PushedAggregation: [], PushedFilters: [IsNotNull(ps_suppkey)], PushedGroupBy: [], ReadSchema: struct RuntimeFilters: []
There was no exchange after part table and lineitem table scan in 3.4
Contributor guide
Research direction
Reproduce the stated three-table query with datasource v2 on Spark 3.4 and 3.5, then compare the physical plans and their Exchange nodes. Trace why the 3.5 plan adds exchanges after the lineitem and part scans; done means the extra exchanges are removed or the version difference is otherwise resolved and verified.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala, spark, sql
- Domain
- databases
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100