Practice and optimization in analysis scenarios - PolarDB

PolarDB-X is a database service developed based on a hybrid transaction/analytical processing (HTAP) architecture. Most queries processed by the service are transaction processing (TP) queries. This service can also be used for analysis in some scenarios. In most cases, typical analysis scenarios have the following characteristics:

Databases process a small number of write or update requests and a large number of read requests.
Each query reads a large number of rows and a small number of columns from databases.
Most queries are complex queries. A small number of concurrent queries are performed. The throughput of a single query must be improved.
Simple queries can be delayed for a short period of time.
Distributed transactions may not be required in analysis scenarios.
In most query scenarios, fact tables and dimension tables are joined. The size of fact tables is large and the size of dimension tables is small. These scenarios are typical join scenarios.
The size of the query result is smaller than the size of the source data. This indicates that the queried data is stored in the memory of a single server after the data is filtered or aggregated.
In most cases, the data that you want to analyze is the most recent business data. Historical data can be cleared or archived.

In analysis scenarios, in addition to the methods of optimizing TP databases, specific methods are used to optimize the performance based on the preceding characteristics of analysis scenarios. The specific optimization methods support schema design and query optimization.

Schema design

When you design a schema, select the table type, partition key, primary key, and clustered index for optimal performance.

Design a table as a partitioned table or a broadcast table

A copy of data in a broadcast table is stored on each data node in the cluster. We recommend that you store a small volume of data on a broadcast table. Each broadcast table can store up to 200,000 rows of data. When large tables are joined with broadcast tables, computations can be pushed down. This way, data can be computed for tables joined on data nodes. This prevents data in large tables from being pulled to the compute nodes for computing.
Configure the tables that store other business data in partitioned tables. This way, you can fully use the query capability of the distributed system. Theoretically, the larger number of partitions in a table indicates better query performance. This is because multiple partitions can be scanned in parallel. Horizontal scaling can be performed for the storage layer in an easier manner. The storage layer can store tens of millions of records or hundreds of millions of records. We recommend that you store 5 million to 50 million rows of data in a partitioned table.

Select a suitable partition key

By default, the primary key is used for partitioning in PolarDB-X. This reduces the costs of the distributed database service. You can also use a partition key to create a partitioned table. In analysis scenarios, we recommend that you select partition keys based on the following rules:

Select the fields used in join operations as partition keys if possible. This helps push down join conditions and prevents data from being pulled to the computing layer for computing.
Select the fields for which the values are evenly distributed as partition keys if possible. This prevents long tails caused by unevenly distributed values from occurring. Long tails severely affects query performance.

Design appropriate subpartitions

PolarDB-X database in DRDS mode support subpartitions. If a table contains a large volume of data or data skews occur, configure suitable subpartitions in the table. If subpartitions are not configured for the table or the configured subpartitions are invalid, query performance is affected in a negative manner. If you want to import incremental data for your business and analyze the report based on the most recent data, we recommend that you use the date format to configure subpartitions. This prevents historical expired data from being scanned.

//Specify a value for the col field to create partitions.
PARTITION BY HASH(col) 
SUBPARTITION BY LIST (ds) 
//Change the format of the value of the ds field to the month format and use the month format to partition the table.
SUBPARTITION TEMPLATE (
   PARTITION p1 VALUES LESS THAN ('2021-08-00'),
   PARTITION p2 VALUES LESS THAN ('2021-09-00'),
)

Design appropriate indexes

If the partition key is designed based on the foreign key field for your business, other columns in the table remain joined for some complex queries and join queries cannot be pushed down. In this case, you can create a global secondary index (GSI) on the columns that are not used as the partition key. This way, if data in the table needs to be joined for complex queries, the GSI in the table can be joined for these queries. We recommend that you configure all GSIs as clustered indexes in analysis scenarios. This way, data does not need to be queried from the table.

Query optimization

In analysis scenarios, simple queries can be delayed for a short period of time because a large amount of data needs to be analyzed. We recommend that you use the massively parallel processing (MPP) mode. In this mode, the computing resources of multiple compute nodes can be used for complex computations. By default, the MPP mode is enabled only for read-only instances. If data can be analyzed on the primary instance, contact Alibaba Cloud technical support to enable the MPP mode.

When data is queried, PolarDB-X selects a suitable distributed execution plan by using the optimizer and schedules the plan to each compute node. This way, you can fully use the computing resources of the entire cluster and accelerate queries. The distributed execution plans generated in this scenario are selected based on statistics. The plan that has the minimum cost is used. In this case, statistics must be collected in a timely manner. The execution plans generated by the optimizer are not optimal. The following section provides the experience in SQL writing and optimization.

Collect table statistics

PolarDB-X collects statistics at specific intervals. If a distributed execution plan generated by the PolarDB-X query optimizer is not optimal, you can execute the ANALYZE TABLE statement to manually collect statistics of the table on which the execution plan is used.

SQL writing skills

Remove columns that you no longer require.
We recommend that you remove columns that you no longer require because most analysis scenarios require high throughput. This reduces the bandwidth usage. When you write SQL statements, you must include the columns for your business instead of using asterisks (*) to query all columns.
```
//Invalid written SQL statement
select * from T1 where a1>100 and a2<1000;
//More suitable written SQL statements. Only the columns required for your business are returned.
select a1, a2 from T2 where a1>100 and a2<1000;
```
Filter data based on local secondary indexes (LSIs).
In a number of analysis scenarios, subpartitions created based on time are expected. This way, when big data is scanned, time can be used to filter most historical data.
```
select a1,c2 from T1 where time >='2010-01-01 00:00:00';
```
By default, an LSI is created on the partition key column. This prevents full table scans. In a number of high-throughput scan scenarios, LSIs can be created based on filters.

Prevent inefficient SQL syntax.

If a table contains a large number of records, the table is scanned at a low speed. This results in slow queries. When you write SQL statements, take note of the following items:

1. Make sure that indexes are valid.

Do not perform operations on index columns. For example, if you compute data, call functions, and manually convert the data type for an index column or the system automatically converts the data type of the index column, the index becomes invalid and a full table scan is performed.

mysql> explain execute select * from staffs where name= 'hu';
+----+-------------+--------+------------+------+-----------------------+-----------------------+---------+-------+------+----------+-------+
| id | select_type | table  | partitions | type | possible_keys         | key                   | key_len | ref   | rows | filtered | Extra |
+----+-------------+--------+------------+------+-----------------------+-----------------------+---------+-------+------+----------+-------+
|  1 | SIMPLE      | staffs | NULL       | ref  | idx_staffs_nameAgePos | idx_staffs_nameAgePos | 74      | const |    1 |      100 | NULL  |
+----+-------------+--------+------------+------+-----------------------+-----------------------+---------+-------+------+----------+-------+
1 row in set , 1 warning (0.00 sec)

// Other operations are performed on the index column. As a result, a full table scan is performed.
mysql> explain execute select * from staffs where left(name,4)= 'hu';
+----+-------------+--------+------------+------+---------------+------+---------+------+------+----------+-------------+
| id | select_type | table  | partitions | type | possible_keys | key  | key_len | ref  | rows | filtered | Extra       |
+----+-------------+--------+------------+------+---------------+------+---------+------+------+----------+-------------+
|  1 | SIMPLE      | staffs | NULL       | ALL  | NULL          | NULL | NULL    | NULL |  198 |      100 | Using where |
+----+-------------+--------+------------+------+---------------+------+---------+------+------+----------+-------------+
1 row in set , 1 warning (0.00 sec)

If the != operator or the <> operator is used, indexes cannot be used. As a result, a full table scan is performed.

If the is null operator or the is not null operator is used, indexes cannot be used.

mysql>  explain execute select * from staffs where name is null ;
+----+-------------+-------+------------+------+---------------+------+---------+------+------+----------+------------------+
| id | select_type | table | partitions | type | possible_keys | key  | key_len | ref  | rows | filtered | Extra            |
+----+-------------+-------+------------+------+---------------+------+---------+------+------+----------+------------------+
|  1 | SIMPLE      | NULL  | NULL       | NULL | NULL          | NULL | NULL    | NULL | NULL | NULL     | Impossible WHERE |
+----+-------------+-------+------------+------+---------------+------+---------+------+------+----------+------------------+
1 row in set

If the like operator starts with a wildcard, indexes in MySQL become invalid. This results in full table scans.

mysql>  explain exeucte select * from staffs where name like '%hu' ;
+----+-------------+--------+------------+------+---------------+------+---------+------+------+----------+-------------+
| id | select_type | table  | partitions | type | possible_keys | key  | key_len | ref  | rows | filtered | Extra       |
+----+-------------+--------+------------+------+---------------+------+---------+------+------+----------+-------------+
|  1 | SIMPLE      | staffs | NULL       | ALL  | NULL          | NULL | NULL    | NULL |  198 |    11.11 | Using where |
+----+-------------+--------+------------+------+---------------+------+---------+------+------+----------+-------------+
1 row in set

mysql>  explain execute select * from staffs where name like 'hu%' ;
+----+-------------+--------+------------+-------+-----------------------+-----------------------+---------+------+------+----------+-----------------------+
| id | select_type | table  | partitions | type  | possible_keys         | key                   | key_len | ref  | rows | filtered | Extra                 |
+----+-------------+--------+------------+-------+-----------------------+-----------------------+---------+------+------+----------+-----------------------+
|  1 | SIMPLE      | staffs | NULL       | range | idx_staffs_nameAgePos | idx_staffs_nameAgePos | 74      | NULL |    1 |      100 | Using index condition |
+----+-------------+--------+------------+-------+-----------------------+-----------------------+---------+------+------+----------+-----------------------+
1 row in set

2. We recommend that you minimize the number of like operators. This is because like operators are inefficient. We recommend that you use range conditions such as between...and....

3. When multiple tables are joined, take note of the following items:

We recommend that you specify the partition key column in your SQL statement. If the partition key column is not specified in the SQL statement, we recommend that you use the WHERE clause to filter out redundant data.
The ON and WHERE clauses are used for different purposes in outer join operations. The ON clauses are used to specify join conditions. The WHERE clauses are used to filter join results. We recommend that you include the filter for a join operation in an ON clause or in a subquery performed on tables that you want to join. This reduces the volume of data in tables that you want to join.

Check and handle data skews

If your query is very slow or the resource utilization on each node is uneven, check whether a data skew occurred. You can use the following methods to handle data skews:

Data in a partition of a table is distributed to each node. You can execute the show info from table statement to check the volume of data distributed on each node. If data is unevenly distributed on each node, you can configure the partition key of the table.
If the tables that you want to join contain a number of hot join keys, you can use dedicated logic to process skewed keys. For example, if a data skew occurs because a large number of NULL key values exist in the tables that you want to join, you must filter out the NULL values or replace the NULL values with random numbers before you join the tables. The following code block provides an example on the SQL statements that can be executed to handle data skews caused by a large number of NULL values.
```
SELECT * FROM A JOIN B ON CASE WHEN A.value IS NULL THEN CONCAT('value',RAND() ) ELSE A.value END = B.value;
```
You can execute the following statements to view the key values that cause a data skew:
```
-- Execute the following statement to query the data skew: 
SELECT * FROM a JOIN b ON a.key=b.key;  
-- Execute the following statement to view the distribution of key values and identify the key values that cause the data skew: 
SELECT left.key, left.cnt * right.cnt FROM 
(select key, count(*) AS cnt FROM a GROUP BY key) LEFT 
JOIN
(SELECT key, COUNT(*) AS cnt FROM b GROUP BY key) RIGHT
ON left.key=right.key;
```

If a key specified in a GROUP BY clause is a hot key, you can rewrite the SQL statement, add random numbers, and split the long-tail key. For example, you can execute the following statements to handle the hot key:

SELECT Key,COUNT(*) AS Cnt FROM TableName GROUP BY Key;
//To optimize your SQL statements, split and pre-aggregate the hot key and then aggregate the values.
-- The long-tail key KEY001 is used in the following example. 
SELECT a.Key
  , SUM(a.Cnt) AS Cnt
FROM (
  SELECT Key
    , COUNT(*) AS Cnt
FROM TableName
GROUP BY Key, 
  CASE 
    WHEN Key = 'KEY001' THEN rand() % 50
    ELSE 0
   END
) a
GROUP BY a.Key;

Configure the execution plan

After your SQL statement is modified based on the preceding policy, the query performance cannot meet your business requirements and no bottleneck issues related to the computing and storage resources occur. In this case, you can configure the execution plan by using the following methods:

Increase the degree of parallelism (DOP). You can use the /*+TDDL:MPP_PARALLELISM=4*/ hint to specify the DOP of the MPP executor.

mysql> /*+TDDL:TDDL:MPP_PARALLELISM=4*/ 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 or
der by cnt limit 5, 10;

Use a hint to include specific algorithms. For more information about how to obtain better aggregation and join algorithms, see Aggregation and Join.