Fang-Yu Rao created IMPALA-14008:
------------------------------------
Summary: Consider distributing the results of hash join to
multiple Impala hosts when only one single Impala host is executing the probe
side
Key: IMPALA-14008
URL: https://issues.apache.org/jira/browse/IMPALA-14008
Project: IMPALA
Issue Type: Improvement
Reporter: Fang-Yu Rao
Currently, when the probe/left side of a hash join is only executed by one
single Impala host, the results of the hash join will only be distributed to
one single host for further processing. In some scenarios, it would be
beneficial to distribute the results of a hash join to multiple Impala hosts
for execution, e.g., the results of hash join have to be further
aggregated/processed.
For instance, for the query "{{{}select straight_join
count(functional.tinyinttable.int_col) from functional.tinyinttable,
functional.alltypes where functional.tinyinttable.int_col =
functional.alltypes.id{}}}", we have the following distributed query plan where
there is only one Impala host working on the aggregation, if the probe/left
side of the join is only executed by one Impala host. Note that we added
"{{{}straight_join{}}}" to force Impala's frontend to preserve the join order.
{code:java}
Operator #Hosts #Inst Avg Time Max Time #Rows Est. #Rows
Peak Mem Est. Peak Mem Detail
---------------------------------------------------------------------------------------------------------------------------------
F02:ROOT 1 1 0.000ns 0.000ns
0 0
06:AGGREGATE 1 1 0.000ns 0.000ns 1 1
16.00 KB 16.00 KB FINALIZE
05:EXCHANGE 1 1 0.000ns 0.000ns 1 1
16.00 KB 16.00 KB UNPARTITIONED
F00:EXCHANGE SENDER 1 1 0.000ns 0.000ns
0 48.00 KB
03:AGGREGATE 1 1 0.000ns 0.000ns 1 1
29.00 KB 16.00 KB
02:HASH JOIN 1 1 0.000ns 0.000ns 10 5
1.98 MB 1.94 MB INNER JOIN, BROADCAST
|--04:EXCHANGE 1 1 0.000ns 0.000ns 7.30K 7.30K
336.00 KB 52.52 KB BROADCAST
| F01:EXCHANGE SENDER 3 3 0.000ns 0.000ns
9.62 KB 32.00 KB
| 01:SCAN HDFS 3 3 26.669ms 32.003ms 7.30K 7.30K
324.00 KB 160.00 MB functional.alltypes
00:SCAN HDFS 1 1 4.000ms 4.000ms 10 5
29.00 KB 32.00 MB functional.tinyinttable
{code}
On the other hand, for the same query without the query hint of
"{{{}straight_join{}}}", there will be multiple Impala hosts executing the
probe/left side of the hash join in "{{{}select
count(functional.tinyinttable.int_col) from functional.tinyinttable,
functional.alltypes where functional.tinyinttable.int_col =
functional.alltypes.id{}}}" and there will be multiple Impala hosts working on
the aggregation as shown in the following.
{code:java}
Operator #Hosts #Inst Avg Time Max Time #Rows Est. #Rows
Peak Mem Est. Peak Mem Detail
---------------------------------------------------------------------------------------------------------------------------------
F02:ROOT 1 1 0.000ns 0.000ns
0 0
06:AGGREGATE 1 1 0.000ns 0.000ns 1 1
16.00 KB 16.00 KB FINALIZE
05:EXCHANGE 1 1 0.000ns 0.000ns 3 3
32.00 KB 16.00 KB UNPARTITIONED
F00:EXCHANGE SENDER 3 3 0.000ns 0.000ns
31.00 B 48.00 KB
03:AGGREGATE 3 3 0.000ns 0.000ns 3 3
64.00 KB 16.00 KB
02:HASH JOIN 3 3 1.333ms 4.000ms 10 7.30K
1.98 MB 1.94 MB INNER JOIN, BROADCAST
|--04:EXCHANGE 3 3 0.000ns 0.000ns 10 5
16.00 KB 16.00 KB BROADCAST
| F01:EXCHANGE SENDER 1 1 0.000ns 0.000ns
118.00 B 32.00 KB
| 00:SCAN HDFS 1 1 0.000ns 0.000ns 10 5
29.00 KB 32.00 MB functional.tinyinttable
01:SCAN HDFS 3 3 6.667ms 8.000ms 7.30K 7.30K
342.00 KB 160.00 MB functional.alltypes
{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
