This topic describes the basic principles of how PolarDB-X processes SQL statements and the concepts of execution plans.
The architecture of a distributed database is different from that of a standalone database. Therefore, in addition to the tuning methods for standalone databases, distributed databases have their own characteristics. PolarDB-X can identify the causes of slow SQL execution based on information such as statistics, execution plans, concurrency policies, and the runtime fed back after execution, and perform targeted tuning.
A slow query may be related to the execution speed of physical SQL statements, the degree of parallelism (DOP), and whether the execution plan and index selection are appropriate. Therefore, in distributed databases, the cost of SQL tuning is generally higher than that in standalone databases.
Basic principles
PolarDB-X is a distributed database product that decouples storage from compute. When a query SQL statement is sent to the PolarDB-X compute node (CN) (this SQL statement is also known as logical SQL), PolarDB-X splits it into two parts: the pushable part and the non-pushable part. The pushable part is also called physical SQL. Non-pushable SQL statements are executed on the CN, and pushable SQL statements are executed on data nodes (DNs).
PolarDB-X optimizes SQL statements based on the following principles:
-
Push down logical SQL statements to DNs for execution whenever possible. In addition to avoiding data network interactions between CNs and DNs, this makes full use of the concurrent execution capability of multiple shards and leverages the resources of each DN to accelerate queries.
-
For the physical operators contained in the non-pushable part, the query optimizer selects the optimal way to execute them, for example, selecting appropriate physical operators, selecting an appropriate DOP policy, and using massively parallel processing (MPP) execution.
NoteThe DOP refers to the maximum number of data executions that are performed in parallel during a query. For a CN, it is the number of threads for multi-threaded computation that uses multi-core capabilities. For a DN, it is the number of pushed-down physical SQL statements that are executed at the same time.
-
PolarDB-X indexes are generally classified into local indexes and global indexes. A local index is an index on a single DN (MySQL index), and a global index is a distributed index built on multiple DNs. Selecting an appropriate index can greatly improve the retrieval speed of PolarDB-X.
Introduction to execution plans
After an SQL statement enters the PolarDB-X distributed database, a runnable execution plan is generated after parsing and optimization. The execution plan is composed based on the dependencies between operators during execution. Generally, through the execution plan tree, you can see how the SQL statement runs efficiently inside the database. Examples:
-
Example 1
Run the following command:
EXPLAIN select count(*) from lineitem group by L_LINESTATUS;The returned execution plan information is as follows:
| HashAgg(group="L_LINESTATUS", count(*)="SUM(count(*))") | | Exchange(distribution=hash[0], collation=[]) | | LogicalView(tables="[000000-000003].lineitem_[00-15]", shardCount=16, sql="SELECT `L_LINESTATUS`, COUNT(*) AS `count(*)` FROM `lineitem` AS `lineitem` GROUP BY `L_LINESTATUS`")Because the GROUP BY column is not aligned with the partition key of the lineitem table, the GROUP BY cannot be fully pushed down to DNs for execution. Therefore, the GROUP BY is split into two phases: partition agg is pushed down to DNs to perform partial aggregation first; then the data is redistributed at the CN layer, and a final aggregation is performed to output the results.
-
LogicalView: Because there are 16 shards, multiple pushed-down physical SQL statements are generated, and each physical SQL statement carries the GROUP BY part to perform pre-aggregation.
-
Exchange: aggregates the data returned by LogicalView, redistributes the data by the L_LINESTATUS field, and outputs it to downstream operators.
-
HashAgg: accepts multiple inputs and performs the final aggregation.
-
-
Example 2
Run the following command:
EXPLAIN select * from lineitem, orders where L_ORDERKEY= O_ORDERKEY;The returned execution plan is as follows:
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | HashJoin(condition="O_ORDERKEY = L_ORDERKEY", type="inner") | | Exchange(distribution=hash[0], collation=[]) | | LogicalView(tables="[000000-000003].lineitem_[00-15]", shardCount=16, sql="SELECT `L_ORDERKEY`, `L_PARTKEY`, `L_SUPPKEY`, `L_LINENUMBER`, `L_QUANTITY`, `L_EXTENDEDPRICE`, `L_DISCOUNT`, `L_TAX`, `L_RETURNFLAG`, `L_LINESTATUS`, `L_SHIPDATE`, `L_COMMITDATE`, `L_RECEIPTDATE`, `L_SHIPINSTRUCT`, `L_SHIPMODE`, `L_COMMENT` FROM `lineitem` AS `lineitem`") | | Exchange(distribution=hash[0], collation=[]) | | LogicalView(tables="[000000-000003].orders_[00-15]", shardCount=16, sql="SELECT `O_ORDERKEY`, `O_CUSTKEY`, `O_ORDERSTATUS`, `O_TOTALPRICE`, `O_ORDERDATE`, `O_ORDERPRIORITY`, `O_CLERK`, `O_SHIPPRIORITY`, `O_COMMENT` FROM `orders` AS `orders`")Example 2 is a typical join between two tables. Because the partition keys of the two tables are not aligned, the join is not pushed down. In the entire execution process, the data of both tables is scanned, and the join computation is performed at the CN layer.
-
LogicalView: scans the table data.
-
Exchange: aggregates the data returned by LogicalView, redistributes the data by the columns in the join condition, and outputs it to the downstream Join operator.
-
HashJoin: accepts inputs from both sides and computes the join result by building a hash table.
-
-
Example 3
Run the following command:
EXPLAIN select * from lineitem, orders where L_LINENUMBER= O_ORDERKEY;The returned execution plan information is as follows:
| Gather(concurrent=true) | | LogicalView(tables="[000000-000003].lineitem_[00-15],orders_[00-15]", shardCount=16, sql="SELECT `lineitem`.`L_ORDERKEY`, `lineitem`.`L_PARTKEY`, `lineitem`.`L_SUPPKEY`, `lineitem`.`L_LINENUMBER`, `lineitem`.`L_QUANTITY`, `lineitem`.`L_EXTENDEDPRICE`, `lineitem`.`L_DISCOUNT`, `lineitem`.`L_TAX`, `lineitem`.`L_RETURNFLAG`, `lineitem`.`L_LINESTATUS`, `lineitem`.`L_SHIPDATE`, `lineitem`.`L_COMMITDATE`, `lineitem`.`L_RECEIPTDATE`, `lineitem`.`L_SHIPINSTRUCT`, `lineitem`.`L_SHIPMODE`, `lineitem`.`L_COMMENT`, `orders`.`O_ORDERKEY`, `orders`.`O_CUSTKEY`, `orders`.`O_ORDERSTATUS`, `orders`.`O_TOTALPRICE`, `orders`.`O_ORDERDATE`, `orders`.`O_ORDERPRIORITY`, `orders`.`O_CLERK`, `orders`.`O_SHIPPRIORITY`, `orders`.`O_COMMENT` FROM `lineitem` AS `lineitem` INNER JOIN `orders` AS `orders` ON (`lineitem`.`L_LINENUMBER` = `orders`.`O_ORDERKEY`)") |Example 3 is also a typical join between two tables. Because the partition keys of the two tables are aligned, the entire join is pushed down to the DNs of each shard for execution. The upper-layer CN only needs to aggregate and output the results returned by the DNs through the Gather operator.
-
Example 4
Run the following command:
EXPLAIN select * from gsi_dml_unique_multi_index_base where integer_test=1;The returned execution plan information is as follows:
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | Project(pk="pk", integer_test="integer_test", varchar_test="varchar_test", char_test="char_test", blob_test="blob_test", tinyint_test="tinyint_test", tinyint_1bit_test="tinyint_1bit_test", smallint_test="smallint_test", mediumint_test="mediumint_test", bit_test="bit_test", bigint_test="bigint_test", float_test="float_test", double_test="double_test", decimal_test="decimal_test", date_test="date_test", time_test="time_test", datetime_test="datetime_test", timestamp_test="timestamp_test", year_test="year_test", mediumtext_test="mediumtext_test") | | BKAJoin(condition="pk = pk", type="inner") | | IndexScan(tables="DRDS_POLARX1_QATEST_APP_000000_GROUP.gsi_dml_unique_multi_index_index1_a0ol_01", sql="SELECT `pk`, `integer_test`, `varchar_test`, `char_test`, `bit_test`, `bigint_test`, `double_test`, `date_test` FROM `gsi_dml_unique_multi_index_index1` AS `gsi_dml_unique_multi_index_index1` WHERE (`integer_test` = ?)") | | Gather(concurrent=true) | | LogicalView(tables="[000000-000003].gsi_dml_unique_multi_index_base_[00-15]", shardCount=16, sql="SELECT `pk`, `blob_test`, `tinyint_test`, `tinyint_1bit_test`, `smallint_test`, `mediumint_test`, `float_test`, `decimal_test`, `time_test`, `datetime_test`, `timestamp_test`, `year_test`, `mediumtext_test` FROM `gsi_dml_unique_multi_index_base` AS `gsi_dml_unique_multi_index_base` WHERE ((`integer_test` = ?) AND (`pk` IN (...)))") | | HitCache:trueIn this example, the SQL statement itself is only a simple query with a predicate, but the execution plan shows a join between two tables (BKAJoin). This is mainly because the gsi_dml_unique_multi_index_base table has a global secondary index on the integer_test column. Hitting the index can reduce the scan cost, but this index is not a covering index, so a table lookup operation is required.
-
IndexScan: scans the data of the index table gsi_dml_unique_multi_index_index1_a0ol_01 based on integer_test=1.
-
BKAJoin: collects the results of IndexScan, and performs a table lookup join with the base table gsi_dml_unique_multi_index_base through this operator to obtain the values of other columns.
NoteGenerally, by querying the execution plan, you can view information such as whether a global secondary index is hit. However, for the pushed-down part of SQL statements, you can also use the explain execute command to obtain the execution status of physical SQL statements on a DN, for example, whether a local index of the DN is hit.
-