[
https://issues.apache.org/jira/browse/IMPALA-14008?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Fang-Yu Rao updated IMPALA-14008:
---------------------------------
Description:
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, e.g., aggregation. The performance
could be improved if multiple Impala hosts could participate in the aggregation
in such a case.
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}
was:
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}
> 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
> Priority: Major
>
> 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, e.g., aggregation. The performance
> could be improved if multiple Impala hosts could participate in the
> aggregation in such a case.
>
> 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)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
