Methods and practice cases for SQL tuning-PolarDB(PolarDB)-阿里云帮助中心

This topic describes the methods and practice cases for SQL tuning. After identifying slow SQL statements, you can use EXPLAIN to view the execution plan and then optimize the statements by pushing down more computations to the storage layer MySQL, adding appropriate indexes, optimizing execution plans, and increasing parallelism.

Push down more computations

PolarDB-X pushes down as many computations as possible to the storage layer MySQL. Pushing down computations reduces data transfer and the overhead of the network layer and the PolarDB-X computation layer, improving the execution efficiency of SQL statements. The computations that PolarDB-X supports pushing down are as follows:

Filter conditions, such as WHERE or HAVING clauses.
Aggregate computations, such as COUNT and GROUP BY, which are processed in two phases.
Sorting computations, such as ORDER BY.
JOINs and subqueries. The shard key strategy on both sides of a JOIN must be the same, or one side must be a broadcast table.

The following example describes how to push down more computations to MySQL to accelerate execution.

Execute the EXPLAIN statement to view the execution plan:
```
EXPLAIN SELECT * FROM customer, nation
        WHERE c_nationkey = n_nationkey AND n_regionkey = 3;
```
The following execution plan is returned:
```
Project(c_custkey="c_custkey", c_name="c_name", c_address="c_address", c_nationkey="c_nationkey", c_phone="c_phone", c_acctbal="c_acctbal", c_mktsegment="c_mktsegment", c_comment="c_comment", n_nationkey="n_nationkey", n_name="n_name", n_regionkey="n_regionkey", n_comment="n_comment")
  BKAJoin(condition="c_nationkey = n_nationkey", type="inner")
    Gather(concurrent=true)
      LogicalView(tables="nation", shardCount=2, sql="SELECT * FROM `nation` AS `nation` WHERE (`n_regionkey` =?)")
    Gather(concurrent=true)
      LogicalView(tables="customer_[0-7]", shardCount=8, sql="SELECT * FROM `customer` AS `customer` WHERE (`c_nationkey` IN ('?'))")
```
In the preceding execution plan, a BKAJoin operator appears. BKAJoin means that for each batch of data fetched from the left table, an IN query is constructed to retrieve the associated rows from the right table, and the JOIN operation is performed at the end. Because the left table has a large amount of data, multiple data fetches are required, resulting in slow execution.

The reason the JOIN cannot be pushed down is that the customer table is sharded by the primary key c_custkey, while the nation table is sharded by the primary key n_nationkey. In this query, the JOIN keys are c_nationkey and n_nationkey. Because the shard key strategies on the two sides are different, the pushdown fails.

Given that the nation table is not large and has almost no update operations, you can use the following statement to rebuild it as a broadcast table:

CREATE TABLE `nation` (
  `n_nationkey` int(11) NOT NULL,
  `n_name` varchar(25) NOT NULL,
  `n_regionkey` int(11) NOT NULL,
  `n_comment` varchar(152) DEFAULT NULL,
  PRIMARY KEY (`n_nationkey`)
) BROADCAST;  --- Declare as a broadcast table

View the execution plan again.
```
EXPLAIN SELECT * FROM customer, nation
        WHERE c_nationkey = n_nationkey AND n_regionkey = 3;
```
The following execution plan is returned:
```
Gather(concurrent=true)
  LogicalView(tables="customer_[0-7],nation", shardCount=8, sql="SELECT * FROM `customer` AS `customer` INNER JOIN `nation` AS `nation` ON ((`nation`.`n_regionkey` =?) AND (`customer`.`c_nationkey` = `nation`.`n_nationkey`))")
```
The JOIN computation no longer appears in the execution plan. Almost all computations are pushed down to the storage layer MySQL (LogicalView operator), and the upper layer only collects and returns the results to the user (Gather operator). The execution performance is significantly improved. For more information about pushdown principles and optimization, see Query rewrite and pushdown.

Add indexes

PolarDB-X supports Global Secondary Index (GSI). The following example describes how to push down more operators by creating a GSI:

Execute the EXPLAIN statement to view the execution plan:
```
EXPLAIN SELECT o_orderkey, c_custkey, c_name FROM orders, customer
        WHERE o_custkey = c_custkey AND o_orderdate = '2019-11-11' AND o_totalprice > 100;
```
The following execution plan is returned:
```
Project(o_orderkey="o_orderkey", c_custkey="c_custkey", c_name="c_name")
  HashJoin(condition="o_custkey = c_custkey", type="inner")
    Gather(concurrent=true)
      LogicalView(tables="customer_[0-7]", shardCount=8, sql="SELECT `c_custkey`, `c_name` FROM `customer` AS `customer`")
    Gather(concurrent=true)
      LogicalView(tables="orders_[0-7]", shardCount=8, sql="SELECT `o_orderkey`, `o_custkey` FROM `orders` AS `orders` WHERE ((`o_orderdate` =?) AND (`o_totalprice` >?))")
```
In the preceding execution plan, the orders table is sharded by o_orderkey and the customer table is sharded by c_custkey. Because the shard dimensions are different, the JOIN computation cannot be pushed down.

Because there are a very large number of orders with a total price above 100 on November 11, 2019, the cross-shard JOIN has high latency. You need to create a GSI on the orders table so that the JOIN computation can be pushed down. The query uses four columns from the orders table: o_orderkey, o_custkey, o_orderdate, and o_totalprice. Among them, o_orderkey and o_custkey are the shard keys of the primary table and the index table respectively, and o_orderdate and o_totalprice are included as covering columns in the index to avoid row lookups.

Create a global secondary index.

CREATE GLOBAL INDEX i_o_custkey ON orders(`o_custkey`)
COVERING (`o_orderdate`, `o_totalprice`)
DBPARTITION BY HASH(`o_custkey`) TBPARTITION BY HASH(`o_custkey`) TBPARTITIONS 4;

Execute the EXPLAIN statement to view the new execution plan:

EXPLAIN SELECT o_orderkey, c_custkey, c_name FROM orders FORCE INDEX(i_o_custkey), customer
        WHERE o_custkey = c_custkey AND o_orderdate = '2019-11-11' AND o_totalprice > 100;

The following execution plan is returned:

Gather(concurrent=true)
  IndexScan(tables="i_o_custkey_[0-7],customer_[0-7]", shardCount=8, sql="SELECT `i_o_custkey`.`o_orderkey`, `customer`.`c_custkey`, `customer`.`c_name` FROM `i_o_custkey` AS `i_o_custkey` INNER JOIN `customer` AS `customer` ON (((`i_o_custkey`.`o_orderdate` =?) AND (`i_o_custkey`.`o_custkey` = `customer`.`c_custkey`)) AND (`i_o_custkey`.`o_totalprice` >?))")

After adding the GSI and using FORCE INDEX(i_o_custkey) to force the use of the index, the cross-shard JOIN becomes a local JOIN on MySQL (inside the IndexScan), and the covering columns eliminate row lookups. The query performance is improved. For more information about global secondary indexes, see Global secondary index.

Execution plan optimization

In most cases, the query optimizer of PolarDB-X can automatically generate optimal execution plans. However, in a few cases, the generated execution plan may not be good enough due to missing or inaccurate statistics. In such cases, you can use Hint statements to influence the optimizer behavior and generate a better execution plan. The following example describes how to optimize an execution plan.

Execute the EXPLAIN statement to view the execution plan:

EXPLAIN SELECT o_orderkey, c_custkey, c_name FROM orders, customer
        WHERE o_custkey = c_custkey AND o_orderdate = '2019-11-15' AND o_totalprice < 10;

The following execution plan is returned:

Project(o_orderkey="o_orderkey", c_custkey="c_custkey", c_name="c_name")
  HashJoin(condition="o_custkey = c_custkey", type="inner")
    Gather(concurrent=true)
      LogicalView(tables="customer_[0-7]", shardCount=8, sql="SELECT `c_custkey`, `c_name` FROM `customer` AS `customer`")
    Gather(concurrent=true)
      LogicalView(tables="orders_[0-7]", shardCount=8, sql="SELECT `o_orderkey`, `o_custkey` FROM `orders` AS `orders` WHERE ((`o_orderdate` =?) AND (`o_totalprice` <?))")

In fact, on November 15, 2019, the number of orders with a total price below 10 CNY was very small, with only a few records. In this case, BKAJoin is a better choice than Hash Join. For more information about BKAJoin and Hash Join, see JOIN optimization and execution.

Use the /*+TDDL:BKA_JOIN(orders, customer)*/ Hint statement to force the optimizer to use BKAJoin (Lookup Join):

EXPLAIN /*+TDDL:BKA_JOIN(orders, customer)*/ SELECT o_orderkey, c_custkey, c_name FROM orders, customer
        WHERE o_custkey = c_custkey AND o_orderdate = '2019-11-15' AND o_totalprice < 10;

The following execution plan is returned:

Project(o_orderkey="o_orderkey", c_custkey="c_custkey", c_name="c_name")
  BKAJoin(condition="o_custkey = c_custkey", type="inner")
    Gather(concurrent=true)
      LogicalView(tables="orders_[0-7]", shardCount=8, sql="SELECT `o_orderkey`, `o_custkey` FROM `orders` AS `orders` WHERE ((`o_orderdate` =?) AND (`o_totalprice` <?))")
    Gather(concurrent=true)
      LogicalView(tables="customer_[0-7]", shardCount=8, sql="SELECT `c_custkey`, `c_name` FROM `customer` AS `customer` WHERE (`c_custkey` IN ('?'))")

From the execution plan, you can see that after adding the /*+TDDL:BKA_JOIN(orders, customer)*/ statement, the execution plan is optimized and the SQL query speed is improved. In addition to adding Hint statements directly in SQL, a more convenient approach is to use the Plan Management feature to bind the execution plan for this SQL. The specific operation is as follows:

BASELINE FIX SQL /*+TDDL:BKA_JOIN(orders, customer)*/ SELECT o_orderkey, c_custkey, c_name FROM orders, customer WHERE o_custkey = c_custkey AND o_orderdate = '2019-11-15';

For this SQL (parameters may differ), PolarDB-X always uses the fixed execution plan as described above. For more information about execution plan management, see Execution plan management.

Parallel execution

You can use the HINT /*+TDDL:PARALLELISM=4*/ statement to specify the degree of parallelism and take full advantage of multi-core capabilities to accelerate computation. The following example describes how to accelerate queries by optimizing the degree of parallel execution.

EXPLAIN PHYSICAL SELECT a.k, count(*) cnt FROM sbtest1 a, sbtest1 b
        WHERE a.id = b.k AND a.id > 1000 GROUP BY k HAVING cnt > 1300 ORDER BY cnt limit 5, 10;

The following execution plan is returned:

+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| PLAN                                                                                                                                                              |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| ExecutorType: AP_LOCAL                                                                                                                                                 |
| The Query's MaxConcurrentParallelism: 2                                                                                                                           |
| Fragment 1                                                                                                                                                        |
|     Shuffle Output layout: [BIGINT, BIGINT] Output layout: [BIGINT, BIGINT]                                                                                       |
|     Output partitioning: SINGLE [] Parallelism: 1                                                                                                                 |
|     TopN(sort="cnt ASC", offset=?2, fetch=?3)                                                                                                                     |
|   Filter(condition="cnt >?1")                                                                                                                                    |
|     HashAgg(group="k", cnt="COUNT()")                                                                                                                             |
|       BKAJoin(condition="k = id", type="inner")                                                                                                                   |
|         RemoteSource(sourceFragmentIds=[0], type=RecordType(INTEGER_UNSIGNED id, INTEGER_UNSIGNED k))                                                             |
|         Gather(concurrent=true)                                                                                                                                   |
|           LogicalView(tables="[000000-000003].sbtest1_[00-15]", shardCount=16, sql="SELECT `k` FROM `sbtest1` AS `sbtest1` WHERE ((`k` >?) AND (`k` IN (...)))") |
| Fragment 0                                                                                                                                                        |
|     Shuffle Output layout: [BIGINT, BIGINT] Output layout: [BIGINT, BIGINT]                                                                                       |
|     Output partitioning: SINGLE [] Parallelism: 1 Splits: 16                                                                                                      |
|     LogicalView(tables="[000000-000003].sbtest1_[00-15]", shardCount=16, sql="SELECT `id`, `k` FROM `sbtest1` AS `sbtest1` WHERE (`id` >?)")                     |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+

The execution plan shows that the default degree of parallelism is not high. You can force a specific degree of parallelism and use the single-node or multi-node parallel mode to accelerate queries.

EXPLAIN PHYSICAL /*+TDDL:PARALLELISM=8*/SELECT a.k, count(*) cnt FROM sbtest1 a, sbtest1 b
        WHERE a.id = b.k AND a.id > 1000 GROUP BY k HAVING cnt > 1300 ORDER BY cnt limit 5, 10;

The following execution plan is returned:

+-------------------------------------------------------------------------------------------------------------------------------------------------------------+
| ExecutorMode: AP_LOCAL                                                                                                                                      |
| Fragment 0 dependency: [] parallelism: 8                                                                                                                    |
| BKAJoin(condition="k = id", type="inner")                                                                                                                   |
|   Gather(concurrent=true)                                                                                                                                   |
|     LogicalView(tables="[000000-000003].sbtest1_[00-15]", shardCount=16, sql="SELECT `id`, `k` FROM `sbtest1` AS `sbtest1` WHERE (`id` >?)")               |
|   Gather(concurrent=true)                                                                                                                                   |
|     LogicalView(tables="[000000-000003].sbtest1_[00-15]", shardCount=16, sql="SELECT `k` FROM `sbtest1` AS `sbtest1` WHERE ((`k` >?) AND (`k` IN (...)))") |
| Fragment 1 dependency: [] parallelism: 8                                                                                                                    |
| LocalBuffer                                                                                                                                                 |
|   RemoteSource(sourceFragmentIds=[0], type=RecordType(INTEGER_UNSIGNED id, INTEGER_UNSIGNED k, INTEGER_UNSIGNED k0))                                        |
| Fragment 2 dependency: [0, 1] parallelism: 8                                                                                                                |
| Filter(condition="cnt >?1")                                                                                                                                |
|   HashAgg(group="k", cnt="COUNT()")                                                                                                                         |
|     RemoteSource(sourceFragmentIds=[1], type=RecordType(INTEGER_UNSIGNED id, INTEGER_UNSIGNED k, INTEGER_UNSIGNED k0))                                      |
| Fragment 3 dependency: [0, 1] parallelism: 1                                                                                                                |
| LocalBuffer                                                                                                                                                 |
|   RemoteSource(sourceFragmentIds=[2], type=RecordType(INTEGER_UNSIGNED k, BIGINT cnt))                                                                      |
| Fragment 4 dependency: [2, 3] parallelism: 1                                                                                                                |
| TopN(sort="cnt ASC", offset=?2, fetch=?3)                                                                                                                   |
|   RemoteSource(sourceFragmentIds=[3], type=RecordType(INTEGER_UNSIGNED k, BIGINT cnt))                                                                      |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------+