tencent cloud

TDSQL Boundless

DocumentaçãoTDSQL BoundlessKernel FeaturesPerformance FeaturesParallel Query

Parallel Query

Download
Modo Foco
Tamanho da Fonte
Última atualização: 2026-07-10 14:50:50

Feature Introduction

TDSQL Boundless parallel query (PQ) is a parallel query framework built into the compute engine. When the volume of data to be queried reaches a certain threshold, the optimizer automatically decomposes the query plan into multiple subtasks, which are then executed in parallel by multiple Worker threads, significantly reducing the response time for complex queries.

Overall Process

1. SQL Parsing and Optimization: A serial execution plan is generated by the MySQL native optimizer from the SQL.
2. Parallel Plan Generation: The PQ optimizer evaluates the serial plan to determine whether it meets the conditions for parallel execution.
3. Plan Decomposition: Eligible table scans are split into multiple range shards, generating a PartialPlan (partial plan).
4. Parallel Execution: Each Worker thread (referred to as a Worker) executes its PartialPlan and independently scans the data shards assigned to it.
5. Result Collection: The Leader thread (referred to as the Leader) aggregates the results from each Worker via the Gather operation, performing merge sorting or deduplication as needed.
6. Return to Client: The Leader returns the final result to the client.

Scenarios

Full Table Scan on Large Tables: The table contains more than ten thousand rows.
Multi-Table JOIN Queries: JOIN queries involve scanning large tables.
Aggregation Analysis: Queries involving GROUP BY + aggregate functions on large datasets.
Sorting Queries: ORDER BY sorting operations are required for large datasets.
Partition Table Queries: Each partition is scanned in parallel based on the partition key.
Complex SQL with Subqueries: Subqueries can be pre-executed in parallel.

Feature Principles

Parallel Plan Generation

parallel query (PQ) operates after the MySQL native optimizer. First, MySQL generates a serial execution plan, and then the PQ optimizer analyzes this plan:
1. Parameter Limit Check: max_parallel_workers must be greater than 0, and max_parallel_degree must be valid or a PARALLEL Hint must be present; otherwise, the process exits immediately.
2. Cost Evaluation: It checks whether the serial execution cost exceeds parallel_plan_cost_threshold. This check is skipped when force=on in parallel_query_switch or when a PARALLEL Hint is present.
3. Parallel Safety Check: All expressions in the query are traversed. If any expression is not supported for parallel execution, parallel processing is rejected.
4. Distribution Policy Selection: For each table, three distribution paths are generated sequentially: final_singleton (final singleton distribution, always generated), replicated (replicated distribution, generated when the table can be pushed down), and partial_scan (partial scan, generated only when the number of table rows is greater than parallel_scan_records_threshold). In a sharded system, an additional hashed (hash distribution) path is generated.
5. JOIN Path Combination: A Cartesian product is performed on the outer/inner side paths. Incompatible combinations (such as replicated and final_singleton) are pruned. A Collector (data aggregation operator) is inserted when necessary to bridge the data.
6. Optimal Path Selection: Paths are selected based on cost, with priority given to the path that matches the most Hints. The parallel_plan_compare_serial_cost parameter controls whether the serial path participates in the comparison.
7. Slice Partitioning: The optimal path is split at the Collector. The root Slice runs on the Leader, and each child Slice corresponds to a PartialPlan.

Slice and PartialPlan

The parallel plan shards the original plan into Slices. Each Slice is an independent execution unit:
Slice 0 (Leader side): It runs on the Leader and is responsible for aggregating results. Its operations include the Gather operation and other operations that require a global perspective.
Slice 1 ~ N (Worker side): They are executed in parallel on Workers. Each slice contains a parallel scan of a single table and its associated operators.
For example, a parallel plan for a simple SELECT * FROM t1 WHERE a > 0:
Slice 1: Four Workers each scan 1/4 of the data range of t1.
Slice 0: The Leader collects the results from the four Workers via the Gather operation.
For a query with a JOIN, such as SELECT * FROM t1 JOIN t2:
Slice 1: Four Workers scan t1 in parallel. Each Worker performs a Hash Join with t2 locally.
Slice 0: The Leader collects the JOIN results from the four Workers via the Gather operation.

Parallel Scanning

PQ supports two parallel scan methods. The optimizer automatically selects the appropriate method based on the table structure and query characteristics.

Dynamic Range Scan

Table data is dynamically sharded into N consecutive shards based on the primary key range. Each Worker is responsible for scanning one shard.
Applicable Scenarios:
Regular tables that are not partitioned or do not require partition-based sharding
For queries with range conditions, such as WHERE id BETWEEN X AND Y, the optimizer can further prune irrelevant shards.
Working Mode:
1. The optimizer estimates the total number of rows and data distribution of the table.
2. The Key space is evenly partitioned into N intervals, where N is less than or equal to max_parallel_degree.
3. Each Worker independently performs a scan after receiving an interval.

Partition Scan

Data is sharded according to physical partitions, with each partition assigned to one Worker.
Applicable Scenarios:
Partition tables that use HASH or RANGE partitioning
Queries that aim to leverage partition pruning, such as WHERE partition_key = X
Working Mode:
1. The optimizer identifies the list of available partitions in the table.
2. Assign each partition evenly to a Worker.
3. The partition assigned to a Worker is scanned.

Recommendations for Selecting Parallel Scanning Methods

For a partition table with a sufficient number of partitions (≥ the degree of parallelism), Partition Scan is prioritized.
For a non-partitioned table or a table with few partitions, Dynamic Range Scan is used.
You can explicitly specify it using a Hint: SELECT /*+ PARALLEL(PARTITION) */ or SELECT /*+ PARALLEL(DYNAMIC_RANGE) */

Gather (Result Collection)

Gather is an operator that collects results from multiple Workers for the Leader thread. It occupies a critical position in the Leader-side plan and is responsible for aggregating the results of parallel execution.
Gather supports multiple data processing modes:
Mode
Applicable Scenarios
Description
Normal Gather
Order not required
Simply collects results from each Worker row by row and passes them to the upper-level operator.
Merge Sort
ORDER BY output required
Performs merge sorting on the sorted results from each Worker during collection. The Worker side performs partial sorting, and the Leader side performs the merge.
Gather also supports streaming or materialized output of results.

Parallel Aggregation

The GROUP BY aggregation operation has multiple strategies for parallel execution. The optimizer automatically selects the optimal plan based on query characteristics.

One-Stage Aggregation

Workers are only responsible for scanning data. After the data is aggregated to the Leader, the Leader performs the aggregation.
Applicable Scenarios:
The fallback policy when aggregate functions or GROUP BY expressions do not meet the push-down criteria (for example, containing unsafe aggregate functions, subqueries, and so on), making it impossible to push down the aggregation to Workers.

Two-Stage Aggregation (Default Policy)

Workers first perform local aggregation, then Gather aggregates the results, and finally the Leader performs the final aggregation.
Process:
1. Each Worker scans data in parallel and performs a local GROUP BY on the data within its assigned range.
2. Workers send the local aggregation results to the Leader.
3. The Leader performs a GROUP BY operation on the intermediate results from all Workers to obtain the final aggregated result.

Fully Pushed-Down Aggregation

When the optimizer can determine that the GROUP BY key values among Workers do not overlap, it pushes down the aggregation entirely to the Workers. In this case, the Leader does not need to perform another GROUP BY and only needs to do a simple aggregation.
Enabling Conditions:
The full_grouping_pushdown option in parallel_query_switch has been enabled (enabled by default).
Aggregate functions and GROUP BY expressions meet the criteria for safe push-down.
The data distribution key is compatible with the GROUP BY expression, meaning that the grouped data on each Worker is complete.

Parallel Sorting

Pushed-Down Sorting (Default)

Workers first sort their respective results, and then the Leader merges them using Gather Merge Sort.
1. Each Worker sorts its own data using the local sorting algorithm (filesort).
2. The Leader initiates Merge Sort and simultaneously reads from the ordered result streams of N Workers.
3. It selects the global minimum value using a min-heap and returns the results row by row.
Advantage: It fully utilizes multi-core parallel sorting capabilities to accelerate sorting of large data volumes.

Non-Pushed-Down Sorting

When sorting columns cannot be pushed down (for example, the sorting expression contains non-parallel-safe functions, or the sorting column is a constant), the data is first gathered to the Leader and then sorted serially on the Leader.

Subquery Parallelism

PQ provides three execution policies for queries:

Pre-Evaluation

The subquery is executed in parallel and completes before the parent query executes . The execution result is then directly read by all Workers.
Applicable Scenarios:
Non-correlated subquery (the subquery does not depend on columns from the parent query)
The subquery result set is relatively small and can be loaded into memory.
Example:
SELECT * FROM t1 WHERE t1.dept_id IN (
    SELECT id FROM departments WHERE region = 'ASIA'
);
The subquery SELECT id FROM departments WHERE region = 'ASIA' is first executed in parallel, and its result is shared with the Workers scanning table t1.

Pushdown to Worker

The correlated subquery is pushed as a whole to each Worker for execution. Each Worker independently executes its assigned portion of the subquery.
Applicable Scenarios:
Correlated subquery (the subquery references columns from the parent query)
The subquery_pushdown option must be enabled (enabled by default).

Inline Evaluation

To maintain MySQL's original logic, the parent query drives the execution of subqueries on demand. You can enforce this mode by using the PQ_INLINE_EVALUATION Hint.

Viewing Parallel Plans with EXPLAIN

Basic Parallelism Indicators


tdsql>  EXPLAIN SELECT DISTINCT (COUNT(b) + 1) AS c FROM t1 GROUP BY a;                                                                                                                                                                                                                   
+----+-------------+-------+------------+------+---------------+------+---------+------+------+----------+--------------------------------------------+                                                                                                                                   
| id | select_type | table | partitions | type | possible_keys | key  | key_len | ref  | rows | filtered | Extra                                      |                                                                                                                                   
+----+-------------+-------+------------+------+---------------+------+---------+------+------+----------+--------------------------------------------+                                                                                                                                   
|  1 | SIMPLE      | t1    | NULL       | ALL  | PRIMARY       | NULL | NULL    | NULL |   43 |   100.00 | Parallel scan (4 workers); Using temporary |                                                                                                                                   
+----+-------------+-------+------------+------+---------------+------+---------+------+------+----------+--------------------------------------------+                                                                                                                                   
                                                                                                                                                                                                                                                
                                                                                                                                                                                                                                                                                          
tdsql> SHOW WARNINGS;                                                                                                                                                                                                                                                                     
+-------+------+--------------------------------------------------------------------------------------------------+                                                                                                                                                                       
| Level | Code | Message                                                                                          |                                                                                                                                                                       
+-------+------+--------------------------------------------------------------------------------------------------+                                                                                                                                                                       
| Note  | 1003 | /* select#1 */ select distinct (count(`b`) + 1) AS `c` from `test`.`t1` group by `test`.`t1`.`a` |                                                                                                                                                                       
| Note  | 1003 | Query is executed in a parallel plan;  explain with tree format to see the plan details.         |                                                                                                                                                                       
+-------+------+--------------------------------------------------------------------------------------------------+                                                                                                                                                                       

If the Extra column displays Parallel scan (N workers), it indicates that parallel execution is active. Concurrently, a WARNING message appears: "Query is executed in a parallel plan; explain with tree format to see the plan details." It is recommended to use the tree format to view the complete parallel plan.

Tree Format (Recommended)


tdsql> EXPLAIN FORMAT=TREE
    -> SELECT 
    ->     t2.a,
    ->     (SELECT SUM(t1.a) FROM t1) as total_sum,
    ->     (SELECT COUNT(t1.a) FROM t1) as total_count
    -> FROM t2
    -> ORDER BY t2.a;
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| EXPLAIN                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| -> Gather (slice: 1, workers: 4)  (cost=2310.16..2312.04 rows=5)
   Merge sort: t2.a
    -> Sort: t2.a  (cost=5.54..5.54 rows=1)
        -> Table scan on t2, with range parallel scan  (cost=2.78..3.48 rows=1)
-> Select #2 (subquery in projection; run only once)
    -> Aggregate: sum(t1.a)  (cost=2309.05..2309.05 rows=1)
        -> Gather (slice: 1, workers: 4)  (cost=2308.65..2308.65 rows=1)
            -> Aggregate: sum(t1.a)  (cost=4.14..4.14 rows=1)
                -> Table scan on t1, with range parallel scan  (cost=2.60..3.25 rows=1)
-> Select #3 (subquery in projection; run only once)
    -> Aggregate: count(t1.a)  (cost=2306.05..2306.05 rows=1)
        -> Gather (slice: 1, workers: 4)  (cost=2304.43..2305.27 rows=4)
            -> Count rows in t1  (cost=0.83..0.83 rows=1)
               Table scan on t1, with pushed projection, with range parallel scan
 |
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

Key Node Descriptions:
Node
Description
Gather (slice: 1, workers: 4)
Four workers execute Slice 1 in parallel, and the Leader collects the results.
with range parallel scan or with partition parallel scan
Scans table data using the Dynamic Range Scan/Partition Scan method.
Merge sort: col_name
Performs merge sorting during collection by Gather to ensure the output is ordered by col_name.
Merge sort with duplicate removal: col
Performs merge sorting and duplicate removal during collection by Gather.
Pre-evaluated subqueries: select #N
Pre-executed subqueries in parallel (subquery results are ready).

Execution Analysis

tdsql> explain analyze verbose select * from t1, t2 where t1.a = t2.a;
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| EXPLAIN                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| -> Inner hash join (t2.a = t1.a)  (cost=6.28 rows=3) (actual time=0.511..0.533 rows=3 loops=1)
   Chunk pair files: 0, memory usage: 16kB
    -> Table scan on t2  (cost=0.88 rows=3) (actual time=0.167..0.185 rows=3 loops=1)
    -> Hash
        -> Table scan on t1  (cost=2.84 rows=3) (actual time=0.262..0.283 rows=3 loops=1)
  RPC statistics: leader
    -> LocalScanRecord=latency(ms): 2,0.266323,0.081208...0.185115, retry_count: 0, retry_interval_all(ms): 0.000000, failure_count: 0 |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
The preceding result indicates:
The memory usage is 16 KB.
The query contains two LocalScanRecord RPCs.
The total RPC duration is 0.266323 ms.
The minimum duration for a single RPC is 0.081208 ms, and the maximum duration is 0.185115 ms.
No retries or failures occurred during the RPC execution.
EXPLAIN ANALYZE VERBOSE is suitable for scenarios that require further analysis of RPC overhead, memory usage, and operator execution details.

Viewing Parallel Decisions with Optimizer Trace

When you need to understand why a specific SQL statement is or is not executed in parallel:
SET optimizer_trace_features = 'parallel_plan_optimization=on';
SET optimizer_trace = 'enabled=on';
EXPLAIN SELECT * FROM t1 WHERE a > 4;
SELECT * FROM information_schema.optimizer_trace \\G
Key information in the Trace:
considering.chosen: true/false — indicates whether parallel execution is selected.
considering.optimization.enumerating_paths — indicates which parallel paths the optimizer evaluated.
cause (when PQ is not selected) — specific reasons, such as plan_cost_less_than_threshold and disabled_by_limit

System Variable

Parallelism Control

Variable Name
Level
Default Value
Description
max_parallel_degree
Session
4
Maximum number of workers for a single query. Set to 0 to disable parallelism.
max_parallel_workers
Global
10000
Maximum total number of workers for the entire node. Set to 0 to disable parallelism globally.

Threshold Control

Variable Name
Level
Default Value
Description
parallel_plan_cost_threshold
Session
50000
Consider generating a parallel plan only when the serial execution cost exceeds this threshold.
parallel_scan_records_threshold
Session
10000
Consider a table as a candidate for parallel scanning only when the number of scanned rows exceeds this threshold.
parallel_scan_ranges_threshold
Session
2
Use a parallel plan only when the number of splittable scan ranges exceeds this threshold.

Feature Switch

Variable Name
Level
Default Value
Description
parallel_query_switch
Session
For details, see Description.
The switch for the PQ feature, in bitmask format, supporting the simultaneous setting of multiple options.
parallel_query_switch Option Description:
Option
Default Value
Description
force
off
Forces the use of parallel plans, skipping all cost and row quantity threshold checks.
subquery_pushdown
on
Supports pushing down correlated subqueries as a whole to workers for parallel execution.
full_grouping_pushdown
on
When GROUP BY keys are non-duplicate across Workers, the aggregation is fully pushed down to Workers.
Other scan mode options, such as group_min_max_pushdown, partition_group_min_max_pushdown, dynamic_range_parallel_scan, and partition_parallel_scan, are also enabled by default.
Warning:
parallel_query_switch is a developer option used for debugging. The content of the switch (including additions or deletions) may be adjusted with each release. Do not use it in management tools or formal maintenance scripts.
Usage:
-- View the current switch value.
SHOW VARIABLES LIKE 'parallel_query_switch';

-- Force parallel execution (bypassing all threshold checks).
SET parallel_query_switch = 'force=on';

-- Disable subquery pushdown (queries containing correlated subqueries will not be executed in parallel).
SET parallel_query_switch = 'subquery_pushdown=off';

Optimizer Hints

For details, see Optimizer Hints.

Use Limits

Limit
Description
Dynamic Range Scan
Can only be executed on the Leader.
Index Merge Scan
Does not support parallel scanning.
Rollup
Exclude
Correlated Subquery in SELECT List
Not supported (for example, when the SELECT list references columns from the outer query).
Non-Parallel Safe Expressions
Non-parallel-safe expressions, such as user variables @a:=1, are not supported for parallel execution.

Parallelism Activation Conditions

Parallel plan generation requires simultaneously meeting the following conditions (or set parallel_query_switch='force=on' to bypass all checks):
1. max_parallel_degree > 0
2. The serial execution cost of the query > parallel_plan_cost_threshold.

FAQs

Why Does EXPLAIN Show No Parallelism?

1. Check whether max_parallel_degree is 0.
2. Use Optimizer Trace to view the specific reason:
plan_cost_less_than_threshold → Decrease parallel_plan_cost_threshold.
3. Quick verification: SET parallel_query_switch='force=on' to force parallel execution.

Why Is PQ Slower Than Serial Execution?

Small data volume: The overhead of parallel startup and inter-thread communication exceeds the benefits of parallelism. Increase the threshold to prevent small queries from using parallel execution.
Too many workers: The overhead of thread context switching is excessive. Appropriately reduce max_parallel_degree.
Unnecessary sorting operations: Check whether the Gather Merge Sort introduces additional sorting overhead.

How to Force Parallelism for a Specific SQL Statement

Bind hints using Statement Outline without modifying the application code:
CALL dbms_admin.statement_outline_add_rule(
    'rule_name',
    'SELECT /*+ PARALLEL(4) */ SUM(c), b FROM t1 GROUP BY b'
);

Which Scan Method Should Be Used for Partition Tables?

Number of partitions ≥ desired degree of parallelism: Use PARALLEL(PARTITION) to scan by partition.
For a table with few partitions or a non-partition table: Use PARALLEL(DYNAMIC_RANGE) to scan by range.
When uncertain: Do not specify a scan method and let the optimizer choose automatically.

Reference Documentation

Optimizer Hints
Understand Execution Plan
SQL Tuning Overview
max_parallel_degree parameter
Anterior: Global IndexPróximo: Overview of the KV Separation Feature

Ajuda e Suporte

Esta página foi útil?

comentários