Understanding GBase 8a MPP Cluster Distributed Execution Plans
Note
GBase 8a MPP Cluster is an in-house developed, large-scale distributed analytical database from General Data Technology Co., Ltd. (GBase). Tables in the 8a database fall into three categories: replicated tables, randomly distributed tables, and hash-distributed tables. Analyzing the execution plan of SQL statements mainly involves examining the distributed execution plan steps at the cluster’s gcluster layer. You can use EXPLAIN in the SQL command line to print a statement’s execution plan, and then consider optimization strategies based on that plan.
Thanks to IT Xiao Chen (alias chenoracle) for compiling and organizing this content!
Original article on ITPUB blog: http://blog.itpub.net/29785807/viewspace-2690364/
The content has been tested in the following environment:
Database version: GBase8a_MPP_Cluster-NoLicense-FREE-8.6.2_build43-R7
Keywords: GBase 8a MPP Cluster, Execution Plan, Database Optimization
This article is divided into three parts. Please read through patiently.
1. Multi-Table Join Execution Plan Analysis
2. GROUP BY Execution Plan Analysis
3. Optimization Case Studies
1. Multi-Table Join Execution Plan Analysis
1.1 Static Hash Join Execution Plan
Scenario: Two hash-distributed tables are joined, and the join column is the hash distribution key column.
In theory, static hash join is the most efficient because the distributed join operator can be executed independently on each node without pulling replicated tables or performing dynamic hash redistribution.
However, if the data distributed by the hash key is severely skewed, meaning most data concentrates on a single node, performance may suffer because the overall execution time depends on the slowest node.
Example:
create table t1 (aid int,gid int) distributed by ('gid');
create table t2 (bid int,gid int) distributed by ('gid');
insert into t1 values(1,100),(1,200),(2,100),(3,50),(2,20),(6,80),(9,10),(6,0),(3,12),(1,18),(9,1);
insert into t2 values(1,0),(300,12),(1,6),(20,50),(50,10),(1,80),(3,10),(9,15),(20,12),(13,18),(2,1);
gbase> explain select t1.aid,t2.bid from t1 inner join t2 on t1.gid=t2.gid and t1.gid=100;
Note:
● ID: SQL execution step, order from bottom to top
● MOTION: How the results of a step are handled
● OPERATION: The specific operation performed within a step
● TABLE: The table involved in an operation
● CONDITION: Conditions involved in an operation.
● t1[gid], t2[gid]: Indicates t1 and t2 are hash-distributed tables with the distribution key column gid
1.2 Distributed Table Joining a Replicated Table Execution Plan
Scenario: A randomly distributed table or a hash-distributed table is joined with a replicated table.
The distributed join operator can also be executed independently on each node, without needing to pull replicated tables or perform dynamic hash redistribution. When a randomly distributed table is joined with a replicated table, data is evenly spread across nodes, achieving high efficiency. The efficiency of joining a hash-distributed table with a replicated table depends on the evenness of the hash distribution column data.
Example:
create table t3(bid int,gid int) replicated;
insert into t3 select * from t2;
explain select t1.aid,t3.bid from t1 inner join t3 on t1.gid=t3.gid;
Note:
● t1[gid]: Indicates t1 is a hash-distributed table with the distribution key column gid
● t3[REP]: REP is the abbreviation for replicated table, indicating t3 is a replicated table
1.3 Small Table as Replicated Copy Join Execution Plan
Scenario: Two hash-distributed tables t1 and t2 are joined, but the join column in t2 is not its hash distribution column. When the data volume of t2 is relatively small, a replicated copy of t2 can be generated on each node, effectively turning the execution plan into a distributed table joining a replicated table.
Alternatively, if t1 is a randomly distributed table and t2 is a hash-distributed or randomly distributed table with very little data, a replicated copy of t2 can be generated on each node, leading to a “distributed table join replicated table” plan.
How small should t2 be for replicating instead of dynamic redistribution? There are corresponding parameters that govern this.
Example:
create table t4 (aid int,gid int);
create table t5 (bid int,gid int);
insert into t4 values(1,100),(1,200),(2,100),(3,50),(2,20),(6,80),(9,10),(6,0),(3,12),(1,18),(9,1);
insert into t4 select * from t4;
insert into t4 select * from t4;
insert into t4 select * from t4;
insert into t4 select * from t4;
insert into t4 select * from t4;
insert into t5 values(1,0),(300,12),(1,6),(20,50),(50,10),(1,80),(3,10),(9,15),(20,12),(13,18),(2,1);
gbase> show variables like '%hash_redist_threshold_row%';
gbase> explain select t4.aid,t5.bid from t4 inner join t5 on t4.gid=t5.gid;
Note:
● t5[DIS], t4[DIS]: DIS is the abbreviation for randomly distributed table, indicating both t4 and t5 are randomly distributed tables
● [BROADCASE]: The results are broadcast into a replicated table. Step 00 broadcasts the t5 random distribution table into a replicated table
1.4 Dynamic Redistribution Hash Join Execution Plan
Scenario: Two hash-distributed tables t1 and t2 are joined, and the join column in t2 is not its hash distribution column. T2 is redistributed based on the join column using hash redistribution, then joined with t1 in a distributed fashion across nodes.
Example:
t6 is hash-redistributed by gid and then joined with t2.
create table t6 (bid int,gid int) distributed by ('bid');
insert into t6 select * from t2;
gbase> explain select t1.aid,t6.bid from t1 inner join t6 on t1.gid=t6.gid;
Note:
● [REDIST(gid)]: Results are redistributed into a hash-distributed table with distribution key gid. Step 00 redistributes the t6 hash-distributed table into a hash-distributed table keyed on gid
Scenario: Two hash-distributed tables t1 and t2 are joined, but neither join column is the hash distribution column of t1 or t2. Both t1 and t2 are hash-redistributed by gid, and then the join is performed distributively across nodes.
Example:
Both T6 and T7 are hash-redistributed by gid, then the join is performed distributively.
create table t7 (aid int,gid int) distributed by ('aid');
insert into t7 select * from t1;
gbase> explain select t7.aid,t6.bid from t7 inner join t6 on t7.gid=t6.gid;
Note:
● [REDIST(gid)]: Results are redistributed into a hash-distributed table with distribution key gid
● Step 00 redistributes the t7 table into a hash-distributed table keyed on gid
● Step 01 redistributes the t6 table into a hash-distributed table keyed on gid
2. GROUP BY Operation Scenario Analysis
2.1 Static Hash GROUP BY Execution Plan
Scenario: When performing a GROUP BY operation on a hash-distributed table, if the grouping columns include the hash distribution column, no dynamic redistribution or two-phase GROUP BY is needed. Each node performs the GROUP BY locally and then aggregates the result set, achieving the highest efficiency.
Example:
gbase> create table t0 (c1 int,c2 varchar(10),c3 int,c4 int) distributed by ('c1');
gbase> insert into t0
values(1,'a',100,1),(2,'b',1,1),(100,'a',12,1),(1,'c',16,1),(3,'d',22,1),(4,'e',1,1),(1,'a',8,1),(200,'e',16,1),(200,'abc',12,1),(8,'x',12,1);
gbase> explain Select c1,c2,c3,sum(c4) from t0 where c3<100 group by 1,2,3;
2.2 Dynamic Redistribution GROUP BY Execution Plan
Scenario: When performing a GROUP BY on a hash-distributed table, if the grouping columns do not include the hash distribution column, dynamic redistribution or two-phase GROUP BY is required. Efficiency is lower, and whether dynamic redistribution or two-phase GROUP BY is used is controlled by the parameter gcluster_hash_redistribute_groupby_optimize. Dynamic redistribution takes the first column after GROUP BY as the new hash distribution column and dynamically generates a hash-distributed table, then performs the GROUP BY.
Example:
gbase> explain Select c2,c3,sum(c4) from t0 where c3<100 group by c2,c3;
gbase> explain Select c2,c3,sum(c4) from t0 where c3<100 group by c3,c2;
2.3 Two-Phase GROUP BY Execution Plan
Scenario: When performing a GROUP BY on a hash-distributed table and the grouping columns do not include the hash distribution column, dynamic redistribution or two-phase GROUP BY is needed, resulting in lower efficiency. As mentioned, the choice is controlled by gcluster_hash_redistribute_groupby_optimize. In two-phase GROUP BY, each node performs a GROUP BY individually, results are gathered at the cluster layer, and then another GROUP BY is performed at the cluster layer. The larger the aggregated intermediate result set, the worse the performance.
Example:
gbase> show variables like 'gcluster_hash_redistribute_groupby_optimize';
gbase> set gcluster_hash_redistribute_groupby_optimize=0;
gbase> explain Select c2,c3,sum(c4) from t0 where c3<100 group by c2,c3;
Note:
● [GATHER]: Results are sent to the aggregation node;
3. Optimization Case Studies
Original SQL:
select distinct a.class_name,b.sno,b.sname
from t_class a,t_student b,t_sex c,sc_course d
where a.class_id = b.class_id
and c.sex_id = b.sex_id
and b.sno = d.sno
and d.grade < 60;
From the execution plan, we can see:
1 t_sex is pulled as a replicated table
2 t_class is pulled as a replicated table
3 sc_course, after filtering by grade condition, is pulled as a replicated table
4 The randomly distributed t_student and the replicated t_sex are joined on sex_id
5 The result of step 4 is joined with the replicated t_class on class_id
6 The result of step 5 is joined with the replicated sc_course on sno
7 The result set from step 6 undergoes dynamic redistribution by the class_name column (due to the DISTINCT operation)
8 The final result set is merged and returned to the client
Analysis of current issues:
All tables above are randomly distributed tables. During multi-table joins, the database automatically pulls t_sex, t_class, and the filtered sc_course into replicated tables and joins them with the randomly distributed t_student. However, even after filtering, sc_course still has a relatively large data volume, making the cost of pulling it as a replicated table very high.
Optimization solution:
Since t_sex and t_class are very small tables, change them from randomly distributed tables to replicated tables.
sc_course and t_student have larger data volumes; change them from randomly distributed tables to hash-distributed tables with the join column as the distribution key.
After optimization:
gbase> select distinct a.class_name,b.sno,b.sname from t_class01 a,t_student01 b,t_sex01 c,sc_course01 d where a.class_id = b.class_id and c.sex_id = b.sex_id and b.sno = d.sno and d.grade < 60;
SQL execution time optimized from 1.03 seconds to 0.09 seconds
1 sc_course01 is a hash-distributed table with sno as the hash key, filtered by the grade<60 condition.
2 t_student01 is a hash-distributed table with sno as the hash key, full table scan.
3 sc_course01 and t_student01 are joined on the sno column.
4 The result of step 3 is joined with t_class01 on class_id.
5 The result of step 4 is joined with t_sex01 on sex_id.
6 Merge the result set and return to the client.
The above is an introduction to distributed execution plan analysis and simple optimization cases. We hope it helps you make better use of the GBase 8a MPP Cluster product.