Zoltán Borók-Nagy created IMPALA-15521:
------------------------------------------
Summary: Planner adds a hash exchange that re-partitions a join
input on its existing partitioning (TPC-DS q78)
Key: IMPALA-15521
URL: https://issues.apache.org/jira/browse/IMPALA-15521
Project: IMPALA
Issue Type: Improvement
Components: Frontend
Reporter: Zoltán Borók-Nagy
In TPC-DS q78 (master 1f34ea915e, default options), join 18 joins two
aggregations that are already hash-partitioned compatibly with the join keys:
{noformat}
F03:PLAN FRAGMENT [HASH(d_year,ws_item_sk,ws_bill_customer_sk)]
18:HASH JOIN [RIGHT OUTER JOIN, PARTITIONED]
| hash predicates: d_year = d_year, ws_bill_customer_sk = ss_customer_sk,
ws_item_sk = ss_item_sk
|--29:EXCHANGE [HASH(d_year,ss_item_sk,ss_customer_sk)]
| F07:PLAN FRAGMENT [HASH(d_year,ss_item_sk,ss_customer_sk)]
| 28:AGGREGATE [FINALIZE]
{noformat}
Exchange 29 re-hashes F07's 48.41M rows on exactly the partition exprs F07
already has. With the query-wide hash seed, every sender instance sends all of
its rows to one receiver.
*Why:* DistributedPlanner.isCompatPartition() (DistributedPlanner.java:842)
does not accept this join.
* Condition 2 compares join exprs with partition exprs using List.equals. It
fails because the join conjuncts are in a different order (year, customer,
item) than the partition exprs (year, item, customer).
* Condition 3 needs value transfer in both directions, which an outer join does
not have.
The fallback (:536-546) then uses getCompatPartition() to build an rhs
partition that matches the lhs. That partition is identical to F07's existing
one, and an exchange is added on top. The TODO at :506 already mentions
relaxing the check.
*Impact:* the 1:1 exchange has one channel per sender and is bound by round
trips.
* MT_DOP=0: q78 takes 27.9 s on master and 12.4 s with the exchange removed.
* MT_DOP=8: it is still about 2 s of q78's 7.6 s.
*Proposed fix* (prototyped; results matched master for all 104 queries):
# Do not add a hash exchange on a join input whose data partition is identical
to the one the exchange would produce: same exprs, same order, same types.
Merge the input into the join's fragment instead. Only do this when the
partition exprs do not reference a tuple that is nullable at the input's root.
This rule alone removes exchange 29.
# In isCompatPartition(), accept permutations. Each partition position i must
match one join conjunct j on both sides (lhsPart[i] = lhsJoin[j] and rhsPart[i]
= rhsJoin[j]), with identical types. Do not apply this to joins that return
unmatched build rows (RIGHT OUTER, FULL OUTER, RIGHT ANTI).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
