You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@drill.apache.org by "Boaz Ben-Zvi (JIRA)" <ji...@apache.org> on 2018/09/07 22:25:00 UTC
[jira] [Commented] (DRILL-6735) Enhance the Hash-Join Operator to
perform Semi and Anti-Semi joins
[ https://issues.apache.org/jira/browse/DRILL-6735?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16607732#comment-16607732 ]
Boaz Ben-Zvi commented on DRILL-6735:
-------------------------------------
Note for the planner side: While this change is good for most cases, there still could be few cases where the current implementation would work better. Mainly when very many (join key) duplicates are expected on the build side.
If the planner can predict such cases (based on statistics), then the planner should decide which of the two options to use.
> Enhance the Hash-Join Operator to perform Semi and Anti-Semi joins
> ------------------------------------------------------------------
>
> Key: DRILL-6735
> URL: https://issues.apache.org/jira/browse/DRILL-6735
> Project: Apache Drill
> Issue Type: Improvement
> Components: Execution - Relational Operators, Query Planning & Optimization
> Affects Versions: 1.14.0
> Reporter: Boaz Ben-Zvi
> Assignee: Boaz Ben-Zvi
> Priority: Major
> Fix For: 1.15.0
>
>
> Currently Drill implements Semi-Join (see DRILL-402) by using a regular join, with a DISTINCT operator under the build upstream side to eliminate duplicates. Typically a physical plan for the Semi uses a hash-join, with a hash-aggr performing the DISTINCT (see example below).
> This effectively builds the same hash table(s) twice - a big waste of time and memory.
> +Improvement+: Eliminate the Hash-Aggr from the plan, and notify the Hash-Join to perform a Semi-join. The HJ then would just skip the duplicates in its hash table(s), thus performing a Semi -Join.
> Example:
> {code}
> select c.c_first_name, c.c_last_name from dfs.`/data/json/s1/customer` c where c.c_customer_sk in (select s.ss_customer_sk from dfs.`/data/json/s1/store_sales` s) limit 4;
> {code}
> And the result plan (see the HJ at 01-03, and the Hash Agg at 01-05):
> {code}
> 00-00 Screen : rowType = RecordType(ANY c_first_name, ANY c_last_name): rowcount = 4.0, cumulative cost = {4693752.96 rows, 2.3095576720000003E7 cpu, 0.0 io, 2.1598011392E9 network, 3.5895861760000005E7 memory}, id = 1320
> 00-01 Project(c_first_name=[$1], c_last_name=[$2]) : rowType = RecordType(ANY c_first_name, ANY c_last_name): rowcount = 4.0, cumulative cost = {4693752.56 rows, 2.3095576320000004E7 cpu, 0.0 io, 2.1598011392E9 network, 3.5895861760000005E7 memory}, id = 1319
> 00-02 Project(c_customer_sk=[$1], c_first_name=[$2], c_last_name=[$3], ss_customer_sk=[$0]) : rowType = RecordType(ANY c_customer_sk, ANY c_first_name, ANY c_last_name, ANY ss_customer_sk): rowcount = 4.0, cumulative cost = {4693748.56 rows, 2.3095568320000004E7 cpu, 0.0 io, 2.1598011392E9 network, 3.5895861760000005E7 memory}, id = 1318
> 00-03 SelectionVectorRemover : rowType = RecordType(ANY ss_customer_sk, ANY c_customer_sk, ANY c_first_name, ANY c_last_name): rowcount = 4.0, cumulative cost = {4693744.56 rows, 2.3095552320000004E7 cpu, 0.0 io, 2.1598011392E9 network, 3.5895861760000005E7 memory}, id = 1317
> 00-04 Limit(fetch=[4]) : rowType = RecordType(ANY ss_customer_sk, ANY c_customer_sk, ANY c_first_name, ANY c_last_name): rowcount = 4.0, cumulative cost = {4693740.56 rows, 2.3095548320000004E7 cpu, 0.0 io, 2.1598011392E9 network, 3.5895861760000005E7 memory}, id = 1316
> 00-05 UnionExchange : rowType = RecordType(ANY ss_customer_sk, ANY c_customer_sk, ANY c_first_name, ANY c_last_name): rowcount = 4.0, cumulative cost = {4693736.56 rows, 2.3095532320000004E7 cpu, 0.0 io, 2.1598011392E9 network, 3.5895861760000005E7 memory}, id = 1315
> 01-01 SelectionVectorRemover : rowType = RecordType(ANY ss_customer_sk, ANY c_customer_sk, ANY c_first_name, ANY c_last_name): rowcount = 4.0, cumulative cost = {4693732.56 rows, 2.3095500320000004E7 cpu, 0.0 io, 2.1597356032E9 network, 3.5895861760000005E7 memory}, id = 1314
> 01-02 Limit(fetch=[4]) : rowType = RecordType(ANY ss_customer_sk, ANY c_customer_sk, ANY c_first_name, ANY c_last_name): rowcount = 4.0, cumulative cost = {4693728.56 rows, 2.3095496320000004E7 cpu, 0.0 io, 2.1597356032E9 network, 3.5895861760000005E7 memory}, id = 1313
> 01-03 HashJoin(condition=[=($1, $0)], joinType=[inner]) : rowType = RecordType(ANY ss_customer_sk, ANY c_customer_sk, ANY c_first_name, ANY c_last_name): rowcount = 90182.8, cumulative cost = {4693724.56 rows, 2.3095480320000004E7 cpu, 0.0 io, 2.1597356032E9 network, 3.5895861760000005E7 memory}, id = 1312
> 01-05 HashAgg(group=[{0}]) : rowType = RecordType(ANY ss_customer_sk): rowcount = 18036.56, cumulative cost = {4509140.0 rows, 2.1824237600000005E7 cpu, 0.0 io, 1.4775549952E9 network, 3.4918780160000004E7 memory}, id = 1309
> 01-06 Project(ss_customer_sk=[$0]) : rowType = RecordType(ANY ss_customer_sk): rowcount = 180365.6, cumulative cost = {4328774.4 rows, 2.0381312800000004E7 cpu, 0.0 io, 1.4775549952E9 network, 3.17443456E7 memory}, id = 1308
> 01-07 HashToRandomExchange(dist0=[[$0]]) : rowType = RecordType(ANY ss_customer_sk, ANY E_X_P_R_H_A_S_H_F_I_E_L_D): rowcount = 180365.6, cumulative cost = {4148408.8000000003 rows, 2.0200947200000003E7 cpu, 0.0 io, 1.4775549952E9 network, 3.17443456E7 memory}, id = 1307
> 02-01 UnorderedMuxExchange : rowType = RecordType(ANY ss_customer_sk, ANY E_X_P_R_H_A_S_H_F_I_E_L_D): rowcount = 180365.6, cumulative cost = {3968043.2 rows, 1.73150976E7 cpu, 0.0 io, 0.0 network, 3.17443456E7 memory}, id = 1306
> 04-01 Project(ss_customer_sk=[$0], E_X_P_R_H_A_S_H_F_I_E_L_D=[hash32AsDouble($0, 1301011)]) : rowType = RecordType(ANY ss_customer_sk, ANY E_X_P_R_H_A_S_H_F_I_E_L_D): rowcount = 180365.6, cumulative cost = {3787677.6 rows, 1.7134732E7 cpu, 0.0 io, 0.0 network, 3.17443456E7 memory}, id = 1305
> 04-02 HashAgg(group=[{0}]) : rowType = RecordType(ANY ss_customer_sk): rowcount = 180365.6, cumulative cost = {3607312.0 rows, 1.6232904E7 cpu, 0.0 io, 0.0 network, 3.17443456E7 memory}, id = 1304
> 04-03 Scan(groupscan=[EasyGroupScan [selectionRoot=file:/data/json/s1/store_sales, numFiles=1, columns=[`ss_customer_sk`], files=[file:/data/json/s1/store_sales/0_0_0.json]]]) : rowType = RecordType(ANY ss_customer_sk): rowcount = 1803656.0, cumulative cost = {1803656.0 rows, 1803656.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 1303
> 01-04 BroadcastExchange : rowType = RecordType(ANY c_customer_sk, ANY c_first_name, ANY c_last_name): rowcount = 55516.0, cumulative cost = {111032.0 rows, 610676.0 cpu, 0.0 io, 6.82180608E8 network, 0.0 memory}, id = 1311
> 03-01 Scan(groupscan=[EasyGroupScan [selectionRoot=file:/data/json/s1/customer, numFiles=1, columns=[`c_customer_sk`, `c_first_name`, `c_last_name`], files=[file:/data/json/s1/customer/0_0_0.json]]]) : rowType = RecordType(ANY c_customer_sk, ANY c_first_name, ANY c_last_name): rowcount = 55516.0, cumulative cost = {55516.0 rows, 166548.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 1310
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)