SQL tuning methods and practices - PolarDB - Alibaba Cloud Documentation Center

This topic describes SQL tuning methods and practices. After you find a slow SQL query that needs to be tuned, you can execute the EXPLAIN statement to view the execution plan. Then, you can optimize the SQL statement by pushing down more computations to MySQL at the storage layer based on schema optimization, appropriately adding indexes, optimizing the execution plan, and increasing the degree of parallelism.

Push down more computations

PolarDB-X pushes down as many computations as possible to MySQL at the storage layer. This reduces the costs of data transmission, networks, and computations in PolarDB-X. This also enables more efficient SQL query execution. PolarDB-X can push down the following computations:

Filter conditions, such as the conditions in WHERE and HAVING clauses.
Aggregations, such as COUNT and GROUP BY clauses. Aggregations are divided into two phases.
Sortings, such as ORDER BY.
JOIN operations and subqueries. The sharding modes of JOIN Keys on both sides must be the same or one side is a broadcast table.

The following example shows how to push down more computations to MySQL to improve the execution efficiency:

Execute the EXPLAIN statement to view the execution plan.

EXPLAIN SELECT * FROM customer, nation 
        WHERE c_nationkey = n_nationkey AND n_regionkey = 3;

Sample result:

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 ('?'))")

A Batched Key Access (BKA) join is used in the execution plan. Each time the BKA join selects a group of rows from the left table, it performs an IN query to select the correlated rows from the right table. Then, the BKA join combines the rows from both tables. The left table contains a large amount of data. As a result, data must be retrieved many times. This slows down the execution.

The JOIN operation cannot be pushed down because customers are grouped by the c_custkey primary key, while nations are grouped by the n_nationkey primary key. The JOIN Keys for this query are c_nationkey and n_nationkey. The sharding modes of JOIN Keys on both sides are different and the pushdown fails.

The nation table contains a small amount of data and few changes are made to the table. Therefore, you can create the following broadcast table to replace the nation 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;  --- Claim the table as a broadcast table.

View the execution plan again.
```
EXPLAIN SELECT * FROM customer, nation 
        WHERE c_nationkey = n_nationkey AND n_regionkey = 3;
```
Sample result:
```
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`))")
```
After you modify the nation table, joins are no longer used in the execution plan. Most computations are pushed down to LogicalView in MySQL. PolarDB-X needs only to gather the results and then return them to the user. This greatly improves the execution efficiency. For more information about pushdown, see Push down and rewrite queries.

Add indexes

PolarDB-X supports the global secondary index (GSI) feature. The following example shows how to push down more computations by using global secondary indexes:

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;
```
Sample result:
```
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 execution plan, the orders table is sharded based on o_orderkey but the customer table is sharded based on c_custkey. The join cannot be pushed down because the two tables use different sharding modes.
A large number of orders whose order amount is greater than USD 100 were placed on November 11, 2019. It is time-consuming to perform a cross-shard join. Therefore, a global secondary index must be created on the orders table to push down the join computation. The o_orderkey, o_custkey, o_orderdate, and o_totalprice columns of the orders table are included in the query. The o_orderkey column is the shard key of the primary table. The o_custkey column is the shard key of the index table. The o_orderdate and o_totalprice columns are specified as covering columns of the index table to prevent the TABLE ACCESS BY INDEX ROWID issue.

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 execution plan again.

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;

Sample result:

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 the global secondary index is created and then forcibly used by specifying FORCE INDEX(i_o_custkey), the cross-shard join becomes a local join in IndexScan and is performed by MySQL. In addition, the covering columns help prevent the TABLE ACCESS BY INDEX ROWID issue. This improves the query efficiency. For more information about how to use global secondary indexes, see GSI.

Optimize execution plans

In most cases, the query optimizer of PolarDB-X can automatically generate the optimal execution plan. However, an automatically generated execution plan may not be optimal if values are missing or deviations exist. In these cases, hints can be used to optimize the execution plan. The following example shows 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;

Sample result:

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` < ?))")

A small number of orders whose order amount is less than USD 10 were placed on November 15, 2019. In this case, BKA joins are more efficient than hash joins. For more information about BKA joins and hash joins, see Optimize and execute JOIN operations.

Use /*+TDDL:BKA_JOIN(orders, customer)*/ Hint to force the optimizer to use a BKA join. A BKA join is also known as a 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;

Sample result:

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 ('?'))")

The execution plan is optimized after /*+TDDL:BKA_JOIN(orders, customer)*/ is added. The SQL query performance is improved. To make the hint effective, include the hint in the SQL statement in your application. You can also use the plan management feature to bind the execution plan to the SQL statement. We recommend that you use the plan management feature instead of modifying the SQL statement. See the following sample code:

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';

This way, PolarDB-X executes only the preceding execution plan for the specified SQL statement, even if the parameters of the SQL statement are modified. For more information about the plan management feature, see Execution plan management.

Process a query in parallel

You can use HINT /*+TDDL:PARALLELISM=4*/ to specify the degree of parallelism. This enables the system to use multiple CPU cores to accelerate computations. Sample code:

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;

Sample result:

+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| 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 default degree of parallelism is not high. You can specify the degree of parallelism to improve the query efficiency in standalone or parallel computing mode.

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;

Sample result:

+-------------------------------------------------------------------------------------------------------------------------------------------------------------+
| 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))                                                                      |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------+