Parallel query

Prerequisites

The PolarDB cluster version is PolarDB for MySQL 8.0 and the revision version is 8.0.1.0.5 or later. For how to confirm the cluster version, see Query the kernel version.

Scenarios

Parallel queries are applicable to most SELECT statements, such as queries on large tables, multi-table queries that use JOINs, and queries that require a large amount of computing. The effect of parallel queries is not significant for short queries.

• Lightweight analysis business.

  Report queries are often complex and time-consuming. The parallel query feature can accelerate individual queries.

• More available system resources.

  Parallel queries consume more system resources. You can use the parallel query feature to improve resource usage and query efficiency only when your system has more CPU resources, low I/O loads, and sufficient memory.

• Combine online and offline queries.

  You can use different endpoints to connect services to different database nodes. This ensures that the services do not affect each other. If you enable the parallel query feature for a single-node endpoint, the read-only node that corresponds to the single-node endpoint is accessed over the endpoint that is used for online analytical processing (OLAP).

Methods to use parallel queries

• Use system parameters

  PolarDB allows you to use the global parameter max_parallel_degree to specify the maximum number of threads that each SQL statement can use for parallel processing. The default value is 0. You can modify the parameter when this parameter is being used. After you modify the parameter, you do not need to restart the database. For more information, see Specify cluster parameters.

  We recommend that you configure the following parallel query settings:

  • Gradually increase the value of the max_parallel_degree parameter. We recommend that you set a value that does not exceed a quarter of CPU cores. Enable the parallel query feature only when your cluster has at least eight CPU cores. Do not enable the parallel query feature for small clusters. Set the max_parallel_degree parameter to 2 at the beginning when you use the parallel query feature. Check the CPU usage one day later. If the CPU usage is low, you can gradually increase the value. If the CPU usage is high, stop increasing the value.
  • To disable the parallel query feature, set the max_parallel_degree parameter to 0. If the max_parallel_degree parameter is set to 1, the parallel query feature is enabled and the degree of parallelism (DOP) is 1.
  • In the PolarDB console, the max_parallel_degree parameter is prefixed with loose and changed to loose_max_parallel_degree. This allows the max_parallel_degree parameter to be applicable to programs of other versions, and follows the default settings of the max_parallel_degree parameter in the MySQL configuration files. This also ensures that programs of other versions can recognize the parameter and return no errors. For more information, see Program Option Modifiers.
  • When you enable the parallel query feature, set the innodb_adaptive_hash_index parameter to OFF. If the innodb_adaptive_hash_index parameter is set to ON, the parallel query performance is decreased.
  You can modify the global parameter to adjust the cluster-level DOP. You can also adjust the DOP for SQL statements in a specific session. For example, if you add a session-level environment variable to the setting of the Java Database Connectivity (JDBC) connection string, you can set the DOP for a specific application.

  set max_parallel_degree = n 

• Use hints

  You can optimize a single statement by using hints. For example, if the parallel query feature is disabled by the system, you can use hints to accelerate a slow SQL query that frequently runs.

  You can execute one of the following statements to enable the parallel query feature:

  SELECT /*+PARALLEL(x)*/ ... FROM ...;   -- x >0
  
  SELECT /*+ SET_VAR(max_parallel_degree=n) */  *  FROM ...   // n > 0

  You can execute one of the following statements to disable the parallel query feature:

  SELECT /*+NO_PARALLEL()*/ ... FROM ... ;
  
  SELECT /*+ SET_VAR(max_parallel_degree=0) */  *  FROM ...

  Advanced hint usage

  The parallel query feature supports two hints: PARALLEL and NO_PARALLEL.

  • You can use the PARALLEL hint to enforce parallel queries and specify the DOP and the table on which parallel scans are performed.
  • You can use the NO_PARALLEL hint to enforce serial queries or specify the tables for which parallel scans are prohibited.

  The following hint syntax is provided:

  /*+ PARALLEL [( [query_block] [table_name]  [degree] )] */
  /*+ NO_PARALLEL [( [query_block] [table_name][, table_name] )] */

  The following table describes the parameters.

  Parameter	Description
  query_block	The name of the query block to which the hint is applied.
  table_name	The name of the table to which the hint is applied.
  degree	The DOP.

  Examples:

  SELECT /*+PARALLEL()*/ * FROM t1, t2; 
  -- The force_parallel_mode parameter is set to true. This indicates that parallel queries are forcibly executed even if the table contains less than 20,000 rows.
  -- The default max_parallel_degree parameter is used to specify the DOP. If max_parallel_degree is set to a value greater than 0, 
  -- parallel queries are enabled. If max_parallel_degree is set to 0, parallel queries are disabled.
  
  SELECT /*+PARALLEL(8)*/ * FROM t1, t2; 
  -- This statement enforces parallel queries in a DOP of 8. 
  -- The force_parallel_mode parameter is set to true. This indicates that parallel queries are forcibly executed even if the table contains less than 20,000 rows.
  -- The max_parallel_degree parameter is set to 8.
  
  SELECT /*+ SET_VAR(max_parallel_degree=8) */  *  FROM ...   
  -- The max_parallel_degree parameter is set to 8. 
  -- The force_parallel_mode parameter is set to false. This indicates that parallel queries are automatically disabled if the table contains less than 20,000 rows.
  
  SELECT /*+PARALLEL(t1)*/ * FROM t1, t2; 
  -- This statement enforces parallel queries on table t1 by using the /*+PARALLEL()*/ syntax for table t1.
  
  SELECT /*+PARALLEL(t1 8)*/ * FROM t1, t2; 
  -- This statement enforces parallel queries on table t1 in a DOP of 8 by using the /*+PARALLEL(8)*/ syntax for table t1.
  
  SELECT /*+PARALLEL(@subq1)*/ SUM(t.a) FROM t WHERE t.a = 
    (SELECT /*+QB_NAME(subq1)*/ SUM(t1.a) FROM t1); 
  -- This statement enforces parallel subqueries. The default parameter max_parallel_degree is used to specify the DOP. 
  -- If max_parallel_degree is set to a value greater than 0, parallel queries are enabled. If max_parallel_degree is set to 0, parallel queries are disabled.
  
  SELECT /*+PARALLEL(@subq1 8)*/ SUM(t.a) FROM t WHERE t.a = 
    (SELECT /*+QB_NAME(subq1)*/ SUM(t1.a) FROM t1); 
  -- This statement enforces parallel subqueries. The max_parallel_degree parameter is set to 8.
  
  SELECT SUM(t.a) FROM t WHERE t.a = 
    (SELECT /*+PARALLEL()*/ SUM(t1.a) FROM t1); 
  -- This statement enforces parallel subqueries. 
  -- The default max_parallel_degree parameter is used to specify the DOP. 
  -- If max_parallel_degree is set to a value greater than 0, parallel queries are enabled. If max_parallel_degree is set to 0, parallel queries are disabled.
  
  SELECT SUM(t.a) FROM t WHERE t.a = 
    (SELECT /*+PARALLEL(8)*/ SUM(t1.a) FROM t1); 
  -- This statement enforces parallel subqueries. The max_parallel_degree parameter is set to 8.
  
  SELECT /*+NO_PARALLEL()*/ * FROM t1, t2; 
  -- This statement disables parallel queries.
  
  SELECT /*+NO_PARALLEL(t1)*/ * FROM t1, t2; 
  -- This statement prohibits parallel queries for only table t1. If the system enables parallel queries, the system may perform parallel scans and run parallel queries on table t2.
  
  SELECT /*+NO_PARALLEL(t1, t2)*/ * FROM t1, t2; 
  -- This statement prohibits parallel queries for table t1 and table t2.
  
  SELECT /*+NO_PARALLEL(@subq1)*/ SUM(t.a) FROM t WHERE t.a = 
    (SELECT /*+QB_NAME(subq1)*/ SUM(t1.a) FROM t1); 
  -- This statement prohibits parallel subqueries.
  
  SELECT SUM(t.a) FROM t WHERE t.a = 
    (SELECT /*+NO_PARALLEL()*/ SUM(t1.a) FROM t1); 
   -- This statement prohibits parallel subqueries.

  Note The PARALLEL hint is ineffective for queries that do not support parallel processing or tables that do not support parallel scans.

  You can also use hints to specify the policy to run parallel subqueries. For more information, see the support for subqueries in Parallel execution plans. The following hint syntax is provided:

  /*+ PQ_PUSHDOWN [( [query_block])] */ The subqueries run in parallel based on the pushdown policy.
  /*+ NO_PQ_PUSHDOWN [( [query_block])] */ The subqueries run in parallel based on the shared access policy.

  Examples:

  # The subqueries run in parallel based on the pushdown policy.
  EXPLAIN SELECT /*+ PQ_PUSHDOWN(@qb1) */ * FROM t2 WHERE t2.a =
                   (SELECT /*+ qb_name(qb1) */ a FROM t1);
  
  # The subqueries run in parallel based on the shared access policy.
  EXPLAIN SELECT /*+ NO_PQ_PUSHDOWN(@qb1) */ * FROM t2 WHERE t2.a =
                   (SELECT /*+ qb_name(qb1) */ a FROM t1);
  # No hint is added to the query block.
  EXPLAIN SELECT * FROM t2 WHERE t2.a =
                   (SELECT /*+ NO_PQ_PUSHDOWN() */ a FROM t1);

• Use a threshold to specify whether the optimizer enables parallel queries

  PolarDB provides two thresholds to specify whether the optimizer enables parallel queries. Parallel queries are enabled by the optimizer when SQL statements meet one of these conditions that are specified by these thresholds.

  • records_threshold_for_parallelism

    If the number of scanned rows in the statement exceeds the value of records_threshold_for_parallelism, the optimizer enables parallel queries. The default value is 10000. If your services require a small volume of data or low concurrency of complex queries, we recommend that you set the threshold to a value greater than 2000.

    Note The number of scanned rows is estimated based on the statistics of tables. It may be different from the actual number of rows.
  • cost_threshold_for_parallelism

    If the serial execution cost of query statements exceeds the value of cost_threshold_for_parallelism, the optimizer enables parallel queries. The default value is 50000.

• Force the optimizer to enable parallel queries

  You can run the following command to enable force_parallel_mode and force the optimizer to enable parallel queries.

  set force_parallel_mode = on

  Notice

  • The force_parallel_mode parameter is used only for debugging. We recommend that you do not use this parameter in production environments.
  • After you enable the force_parallel_mode parameter, the PolarDB optimizer performs parallel queries whenever possible. However, parallel queries cannot be executed in certain cases due to the limits of scenarios.

Parameters and variables

Performance metrics

The following tests use 100 GB of data that is generated based on TPC Benchmark™H (TPC-H) to test the performance metrics of a PolarDB for MySQL 8.0 cluster. In the test, the PolarDB cluster has a 32-core CPU and 256 GB of memory. The max_parallel_degree parameter is set to 32 or 0. For more information, see OLAP performance tests.

The test result shows that 95% of the SQL queries in TPC-H are accelerated. For 70% of the accelerated SQL queries, the query speed is more than eight times that of SQL queries that are not executed in parallel.

Parallel execution plans

The following sections describe parallel query information in execution plans that are returned by the EXPLAIN statement.

• Parallel scans

  In a parallel scan, workers simultaneously scan the data of a table. Each worker produces a partial result and returns the partial result to a leader thread. The leader thread gathers the results by using the Gather node and returns the final result to the client.

• Parallel joins on multiple tables

  When parallel queries are enabled, the complete multi-table join operation is sent among workers for parallel processing. The PolarDB optimizer selects only the optimal table for parallel scanning. Non-parallel scanning is performed on all other tables. Each worker produces a partial result and returns the partial result to a leader thread. The leader thread gathers the results by using the Gather node and returns the final result to the client.

• Parallel sorting

  The PolarDB optimizer sends the ORDER BY operation to all workers for parallel processing. Each worker produces a partial result. The leader thread gathers, merges, and sorts all partial results by using the Gather Merge node, and returns the final sorting result to the client.

• Parallel grouping

  The PolarDB optimizer sends the GROUP BY operation to all workers for parallel processing. Each worker is responsible for the GROUP BY operation on a portion of the data. Each worker produces a partial result of GROUP BY. The leader thread gathers the results from all workers by using the Gather node. The PolarDB optimizer checks whether to perform a GROUP BY operation in the leader thread again based on the query plan. For example, if a loose index scan is used to run a GROUP BY query, the leader thread does not perform a GROUP BY operation. If the loose index scan is not used, the leader thread performs a GROUP BY operation and returns the final result to the client.

• Parallel aggregation

  The execution of the aggregate function is sent to all the parallel workers. PolarDB supports parallel aggregation in two stages. Each worker that participates in the parallel portion of the query first performs an aggregation step. In this case, the leader thread gathers results from all workers by using the Gather or Gather Merge node. Then, the leader thread re-aggregates the results from all workers to produce the final result.

• Support for subqueries
  In parallel queries, subqueries can run by using one of the following policies:

  • Serial execution in the leader thread

    If subqueries do not support parallel processing, the subqueries in the leader thread are run in sequence. For example, if two tables are joined by using a JOIN clause that references user-defined functions (UDFs), the subqueries are run in sequence.

  • Parallel execution in the leader thread. This starts another group of workers to run parallel subqueries

    After a parallel plan is generated, the execution plan of the leader thread includes the subqueries that support parallel processing. However, these parallel subqueries cannot run in advance or cannot run in parallel based on the shared access policy. For example, if the subqueries include window functions, the subqueries cannot run in parallel based on the shared access policy.

  • Shared access

    After a parallel plan is generated, if the execution plans of workers reference the subqueries that support parallel processing, the PolarDB optimizer runs these parallel subqueries in advance. This allows the workers to directly access the results of the subqueries.

  • Pushed down

    After a parallel plan is generated, if the execution plans of workers reference correlated subqueries, the workers run these correlated subqueries.

Examples of parallel execution plans

In the following example, the pq_test table is used to test parallel queries.

Schema:

mysql> SHOW CREATE TABLE pq_test\G
*************************** 1. row ***************************
       Table: pq_test
Create Table: CREATE TABLE `pq_test` (
  `id` BIGINT(20) NOT NULL AUTO_INCREMENT,
  `help_topic_id` INT(10) UNSIGNED NOT NULL,
  `name` CHAR(64) NOT NULL,
  `help_category_id` SMALLINT(5) UNSIGNED NOT NULL,
  `description` TEXT NOT NULL,
  `example` TEXT NOT NULL,
  `url` TEXT NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=21495809 DEFAULT CHARSET=utf8
1 row in set (0.00 sec)

Table size:

mysql> SHOW TABLE STATUS\G
*************************** 1. row ***************************
           Name: pq_test
         Engine: InnoDB
        Version: 10
     Row_format: Dynamic
           Rows: 20064988
 Avg_row_length: 1898
    Data_length: 38085328896
Max_data_length: 0
   Index_length: 0
      Data_free: 4194304
 Auto_increment: 21495809
    Create_time: 2019-07-30 01:35:27
    Update_time: NULL
     Check_time: NULL
      Collation: utf8_general_ci
       Checksum: NULL
 Create_options:
        Comment:
1 row in set (0.02 sec)

SQL statement:

SELECT COUNT(*) FROM pq_test;

• The EXPLAIN statement returns the following result when the parallel query is not used:

  SET max_parallel_degree=0; EXPLAIN SELECT COUNT(*) FROM pq_test\G

  The following result is returned:

  *************************** 1. row ***************************
             Id: 1
    Select_type: SIMPLE
          Table: pq_test
    Partitions: NULL
           Type: index
  Possible_keys: NULL
            Key: PRIMARY
        Key_len: 8
            Ref: NULL
           Rows: 20064988
       Filtered: 100.00
          Extra: Using index
  1 row in set, 1 warning (0.03 sec)

• The EXPLAIN statement returns the following result when the parallel query is used:

  EXPLAIN SELECT COUNT(*) FROM pq_test\G

  The following result is returned:

  *************************** 1. row ***************************
             Id: 1
    Select_type: SIMPLE
          Table: <gather2>
     Partitions: NULL
           Type: ALL
  Possible_keys: NULL
            Key: NULL
        Key_len: NULL
            Ref: NULL
           Rows: 20064988
       Filtered: 100.00
          Extra: NULL
  *************************** 2. row ***************************
             Id: 2
    Select_type: SIMPLE
          Table: pq_test
     Partitions: NULL
           Type: index
  Possible_keys: NULL
            Key: PRIMARY
        Key_len: 8
            Ref: NULL
           Rows: 10032494
       Filtered: 100.00
          Extra: Parallel scan (2 workers); Using index
  2 rows in set, 1 warning (0.00 sec)

  Note

  • The EXPLAIN output shows that the parallel plan includes a Gather operation. Gather is implemented to gather the partial results that are produced by all workers.
  • In addition, information in the Extra field shows that a parallel scan is performed on the pq_test table. This plan includes a simultaneous scan. In this case, you can perform a simultaneous scan by using two workers.
• The EXPLAIN statement returns the following result when the parallel query is used for a statement that includes subqueries:

  EXPLAIN SELECT a, (select sum(t2.b) from t2 where t2.a = t1.b) FROM t1 WHERE (a, b) IN (SELECT b, MAX(a) FROM t2 GROUP BY b)\G

  The following result is returned:

  *************************** 1. row ***************************
             id: 1
    select_type: PRIMARY
          table: <gather1>
     partitions: NULL
           type: ALL
  possible_keys: NULL
            key: NULL
        key_len: NULL
            ref: NULL
           rows: 2
       filtered: 100.00
          Extra: NULL
  *************************** 2. row ***************************
             id: 1
    select_type: PRIMARY
          table: t1
     partitions: NULL
           type: ALL
  possible_keys: NULL
            key: NULL
        key_len: NULL
            ref: NULL
           rows: 2
       filtered: 100.00
          Extra: Parallel scan (1 workers); Using where
  *************************** 3. row ***************************
             id: 2
    select_type: DEPENDENT SUBQUERY
          table: t2
     partitions: NULL
           type: ALL
  possible_keys: NULL
            key: NULL
        key_len: NULL
            ref: NULL
           rows: 2
       filtered: 50.00
          Extra: Parallel pushdown; Using where
  *************************** 4. row ***************************
             id: 3
    select_type: SUBQUERY
          table: <gather3>
     partitions: NULL
           type: ALL
  possible_keys: NULL
            key: NULL
        key_len: NULL
            ref: NULL
           rows: 1
       filtered: 100.00
          Extra: Shared access; Using temporary
  *************************** 5. row ***************************
             id: 3
    select_type: SIMPLE
          table: t2
     partitions: NULL
           type: ALL
  possible_keys: NULL
            key: NULL
        key_len: NULL
            ref: NULL
           rows: 2
       filtered: 100.00
          Extra: Parallel scan (1 workers); Using temporary
  5 rows in set, 2 warnings (0.02 sec)

  These results indicate the following information:

  • For the subquery whose select_type is SBUQUERY, the Extra field displays the Parallel pushdown policy. In the Parallel pushdown policy, the subquery is sent to workers for parallel processing.
  • For the subquery whose select_type is DEPENDENT SUBQUERY, the Extra field displays the Shared access policy. In the Shared access policy, the PolarDB optimizer runs the parallel subquery in advance, and the result is shared among workers.