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

[ayushi-agarwal]

Originally posted by @ayushi-agarwal in #5445 (comment)

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<l_partkey:bigint,l_suppkey:bigint> 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<p_partkey:bigint> 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<ps_suppkey:bigint> RuntimeFilters: []

There was no exchange after part table and lineitem table scan in 3.4