apache / apache/gluten

Hash join plan with datasource v2 in 3.5 has 2 more exchanges than 3.4

Open
#5,530 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.