Introduction

Parallel execution is an optimization strategy for SQL queries. It breaks down a query task into multiple subtasks that run simultaneously on multiple processor cores, thereby accelerating the query.

With the widespread use of multi-core processors, multi-threading, and high-speed network connections in computer systems today, parallel execution has become an efficient query technique. This technique significantly reduces the response time of resource-intensive large queries and applies to business scenarios such as offline data warehouses, real-time reporting, and online big data analysis. It is also effective in batch data transfer and quick index table construction.

Parallel execution significantly enhances the performance of the following SQL query scenarios:

• Large table scans, large table joins, large-scale sorting, and aggregation.
• DDL operations, such as primary key modification, column type change, and index creation, on large tables.
• Table creation (using Create Table As Select) from large amounts of existing data.
• Batch data insertion, deletion, and updating.

Scenarios

Parallel execution applies to not only analytical systems such as offline data warehouses, real-time reporting, and online big data analytics but also OLTP scenarios to speed up DDL operations and batch data processing.

Parallel execution aims to reduce SQL execution time by making full use of multiple CPU and I/O resources. It is more efficient than sequential execution under the following conditions:

• A large amount of data is accessed.
• The concurrency of SQL queries is low.
• The system does not require low latency.
• Adequate hardware resources are available.

Parallel execution allows multiple processors to work on the same task concurrently, thereby improving performance. To achieve this, the system must meet the following criteria:

• Run on a multiprocessor system (SMP) or in a cluster
• Have sufficient I/O bandwidth
• Have abundant memory (for memory-intensive operations such as sorting and hash table construction)
• Have moderate load or exhibit peak-valley characteristics (for example, the system load is usually below 30%)

If the system does not meet the preceding criteria, parallel execution may not significantly improve performance. In highly loaded systems with limited memory or I/O capability, parallel execution may even perform worse.

Parallel execution does not have specific hardware requirements. However, the number of CPU cores, memory size, storage I/O performance, and network bandwidth can affect parallel execution performance. If any of these becomes a bottleneck, the overall performance will be impaired.

Principle

Parallel execution is achieved by decomposing an SQL query task into multiple subtasks and scheduling these subtasks to run concurrently on multiple processors.

Once an SQL query is parsed into a parallel execution plan, the execution process is as follows:

1. The SQL main thread (responsible for receiving and parsing SQL) allocates the required thread resources for parallel execution in advance according to the shape of the plan. These thread resources may involve clusters across multiple machines.
2. The SQL main thread enables the parallel scheduling operator (PX Coordinator).
3. The parallel scheduling operator parses the plan, breaks it down into multiple operation steps, and schedules these steps in a bottom-up order. Each operation is designed to support maximum parallel execution.
4. After all operations complete parallel execution, the parallel scheduling operator receives the calculation results and performs serial calculations (such as the final SUM calculation) on these results.

Granule

In parallel data scanning, the basic unit of work is a granule.

OceanBase Database divides a table scan task into multiple granules, each of which defines the scope of the scan task. Since each granule covers the data of only one partition, the data in each partition corresponds to one independent scan task. In other words, each granule corresponds to a scan task within a partition.

Granules can be classified based on the following criteria:

• Partition granule

  A partition granule describes a complete partition. If a scan task involves m partitions, it is divided into m partition granules, regardless of whether the partitions belong to the primary table or the index. Partition granules are most commonly used in partition-wise joins to ensure that corresponding partitions of the two tables are processed by partition granules.

• Block granule

  A block granule describes a continuous range of data in a partition. In a data scan scenario, block granules are typically used to divide data. A partition is divided into several blocks. These blocks are connected in a queue based on some rules for parallel worker threads to consume.

In a scenario where the degree of parallelism is specified, the optimizer automatically determines whether to divide data into partition granules or block granules to help ensure balanced subtasks. If block granules are chosen, the parallel execution framework dynamically determines the division of blocks during runtime based on the principle that a block is neither too large nor too small. A block that is too large can cause data skew and lead to insufficient work for some threads. A block that is too small results in frequent scan context switching, which causes high overheads.

Once the partition-wise (namely, block) division is completed, a scan task corresponds to each division. The table scan operator sequentially processes these scan tasks one by one until all tasks are completed.

Parallel execution model

Producer-consumer pipeline model

Parallel execution is implemented by using the producer-consumer pipeline model.

After a parallel scheduling operator parses an execution plan, the plan is divided into multiple operation steps. Each operation step is called a DFO (Data Flow Operation).

Generally, the parallel scheduling operator starts two DFOS simultaneously. These two DFOS are connected in a producer-consumer manner to enable parallel execution between them. Each DFO uses a group of threads to execute tasks, which is called intra-DFO parallel execution. The number of threads used by an DFO is called the DOP (Degree of Parallisim).

In the producer-consumer model, the consumer DFO of the current stage becomes the producer DFO of the next stage. Under the coordination of the parallel scheduling operator, the consumer DFO and the producer DFO are started simultaneously. The following figure shows the process of the DFOs in the producer-consumer model.

• The data generated by DFO A is transmitted to DFO B for processing in real time.
• After DFO B completes the processing, the data is stored in the current thread and waits for the upstream DFO C to start.
• After receiving the start completion notification from DFO C, DFO B changes its role to a producer and begins to transmit data to DFO C. After receiving the data, DFO C starts to process it.

In the following query example, the execution plan of the SELECT statement first scans the entire game table, then groups the data by team in the team table to calculate the sums, and finally calculates the total scores of each team.

CREATE TABLE game (round INT PRIMARY KEY, team VARCHAR(10), score INT)
     PARTITION BY HASH(round) PARTITIONS 3;
INSERT INTO game VALUES (1, "CN", 4), (2, "CN", 5), (3, "JP", 3);
INSERT INTO game VALUES (4, "CN", 4), (5, "US", 4), (6, "JP", 4);
SELECT /*+ PARALLEL(3) */ team, SUM(score) TOTAL FROM game GROUP BY team;

obclient> EXPLAIN SELECT /*+ PARALLEL(3) */ team, SUM(score) TOTAL FROM game GROUP BY team;
obclient> EXPLAIN SELECT /*+ PARALLEL(3) */ team, SUM(score) TOTAL FROM game GROUP BY team;
+---------------------------------------------------------------------------------------------------------+
| Query Plan                                                                                              |
+---------------------------------------------------------------------------------------------------------+
| =================================================================                                       |
| |ID|OPERATOR                     |NAME    |EST.ROWS|EST.TIME(us)|                                       |
| -----------------------------------------------------------------                                       |
| |0 |PX COORDINATOR               |        |1       |4           |                                       |
| |1 | EXCHANGE OUT DISTR          |:EX10001|1       |4           |                                       |
| |2 |  HASH GROUP BY              |        |1       |4           |                                       |
| |3 |   EXCHANGE IN DISTR         |        |3       |3           |                                       |
| |4 |    EXCHANGE OUT DISTR (HASH)|:EX10000|3       |3           |                                       |
| |5 |     HASH GROUP BY           |        |3       |2           |                                       |
| |6 |      PX BLOCK ITERATOR      |        |1       |2           |                                       |
| |7 |       TABLE SCAN            |game    |1       |2           |                                       |
| =================================================================                                       |
| Outputs & filters:                                                                                      |
| -------------------------------------                                                                   |
|   0 - output([INTERNAL_FUNCTION(game.team, T_FUN_SUM(T_FUN_SUM(game.score)))]), filter(nil), rowset=256 |
|   1 - output([INTERNAL_FUNCTION(game.team, T_FUN_SUM(T_FUN_SUM(game.score)))]), filter(nil), rowset=256 |
|       dop=3                                                                                             |
|   2 - output([game.team], [T_FUN_SUM(T_FUN_SUM(game.score))]), filter(nil), rowset=256                  |
|       group([game.team]), agg_func([T_FUN_SUM(T_FUN_SUM(game.score))])                                  |
|   3 - output([game.team], [T_FUN_SUM(game.score)]), filter(nil), rowset=256                             |
|   4 - output([game.team], [T_FUN_SUM(game.score)]), filter(nil), rowset=256                             |
|       (#keys=1, [game.team]), dop=3                                                                     |
|   5 - output([game.team], [T_FUN_SUM(game.score)]), filter(nil), rowset=256                             |
|       group([game.team]), agg_func([T_FUN_SUM(game.score)])                                             |
|   6 - output([game.team], [game.score]), filter(nil), rowset=256                                        |
|   7 - output([game.team], [game.score]), filter(nil), rowset=256                                        |
|       access([game.team], [game.score]), partitions(p[0-2])                                             |
|       is_index_back=false, is_global_index=false,                                                       |
|       range_key([game.round]), range(MIN ; MAX)always true                                              |
+---------------------------------------------------------------------------------------------------------+
29 rows in set

The following schematic diagram shows the execution of the preceding query:

As shown in the figure, the query actually uses six threads. The execution process and task distribution are as follows:

• Step 1: The first three threads scan the game table and pre-aggregate the game.team data in each thread.
• Step 2: The last three threads aggregate the pre-aggregated data.
• Step 3: The aggregation result of step 2 is returned to the client by the parallel scheduler.

When transmitting the data from step 1 to step 2, a hash join is performed on the game.team field to determine the threads to which the pre-aggregated data is to be sent.

Data distribution between producers and consumers

Data distribution refers to the method used to send data from a group of worker threads (producers) that execute in parallel to another group of worker threads (consumers). The optimizer selects the optimal data redistribution method to achieve the best performance by using a series of optimization strategies.

Common data distribution methods in parallel execution include the following:

• HASH DISTRIBUTION

  In HASH DISTRIBUTION, the producers hash and modulo the data rows based on the distribution key to determine the consumers to send the data rows to. Generally, HASH DISTRIBUTION can distribute data evenly among multiple consumers.

• PKEY DISTRIBUTION

  In PKEY DISTRIBUTION, the producers calculate the partitions of the destination tables of the data rows and send the data rows to the consumers that process the partitions. PKEY DISTRIBUTION is commonly used in partial partition-wise join scenarios. In this scenario, the data on the consumer side does not need to be redistributed and can be directly joined with the data on the producer side on a partition basis. This reduces the network communication and improves the performance.

• PKEY HASH DISTRIBUTION

  In PKEY HASH DISTRIBUTION, the producers calculate the partitions of the destination tables of the data rows and then hash the data rows based on the distribution key to determine the consumers to send the data rows to.

  PKEY HASH DISTRIBUTION is commonly used in parallel DML scenarios. In such scenarios, multiple threads can concurrently update data in one partition. Therefore, the PKEY HASH DISTRIBUTION method is used to ensure that data rows with the same value are processed by the same thread, and data rows with different values are as evenly distributed as possible among the threads.

• BROADCAST DISTRIBUTION

  In BROADCAST DISTRIBUTION, the producers send each data row to each consumer thread, ensuring that each consumer thread has all the data from the producers. BROADCAST DISTRIBUTION is commonly used to copy data of a small table to all nodes that will perform the join operation and then execute the join operation locally on each node. This reduces the network communication.

• BC2HOST DISTRIBUTION

  In BC2HOST DISTRIBUTION, the producers send each data row to each consumer node, ensuring that each consumer node has all the data from the producers. Then, the consumer threads within the nodes cooperate to process the data.

  BC2HOST DISTRIBUTION is commonly used in NESTED LOOP JOIN and SHARED HASH JOIN scenarios. In the NESTED LOOP JOIN scenario, each consumer thread obtains a part of the shared data as the driving data and uses it to perform the join operation on the target table; in the SHARED HASH JOIN scenario, each consumer thread builds a hash table based on the shared data. This avoids the necessity of building the same hash table for each thread.

• RANGE DISTRIBUTION

  In RANGE DISTRIBUTION, the producers divide the data into segments based on ranges, and assign different segments of data to different consumer threads. RANGE DISTRIBUTION is commonly used in sorting scenarios. Each consumer thread needs to sort only the data it is assigned to keep the global order.

• RANDOM DISTRIBUTION

  In RANDOM DISTRIBUTION, the producers randomize the data and send the data to the consumer threads. This ensures that each consumer thread processes nearly the same amount of data to achieve load balancing. RANDOM DISTRIBUTION is commonly used in multi-threaded parallel UNION ALL scenarios. In such scenarios, it suffices to randomize the data and achieve load balancing without establishing any associations among the data.

• HYBRID HASH DISTRIBUTION

  Hybrid HASH DISTRIBUTION is used in adaptive join algorithms. Based on the collected statistics, OceanBase Database provides a set of parameters to define regular values and frequent values. The hybrid HASH DISTRIBUTION method hashes regular values on both sides of the join and broadcasts frequent values on the left side and randomizes frequent values on the right side.

Data transmission between producers and consumers

At any time, two DFOS started by parallel scheduling operators are connected in a producer-consumer manner for parallel execution. To transmit data between the producer and consumer, a data transmission network is needed to be created.

For example, if the data scan operator uses DOP = 2 and the data aggregation operator uses DOP = 3, each of the two producer threads creates three virtual links to connect to the consumer threads. This creates a total of six virtual links.

The virtual transmission network created is called the data transfer layer (DTL). In the parallel execution framework of OceanBase Database, all control messages and row data are transmitted through the DTL. Each worker thread can establish thousands of virtual links, demonstrating high scalability. In addition, the DTL supports features such as data buffering, batch data transmission, and automatic flow control.

When the DTL is located within a node, it uses memory copying to transmit messages; when the DTL spans multiple nodes, it uses network communication to transmit messages.

Worker threads

In a parallel query, there are two types of threads: a main thread and multiple worker threads. The main thread runs in the same thread pool as a regular TP query, while the worker threads run in a dedicated thread pool.

OceanBase Database uses a dedicated thread pool to allocate worker threads for parallel queries. Each tenant has a dedicated parallel execution thread pool on each node it belongs to. All worker threads for parallel queries are allocated from these thread pools.

Before a parallel scheduling operator schedules a DFO, it requests thread resources from the thread pool. After the DFO is executed, the thread resources are immediately released.

The initial size of the thread pool is 0 and it can grow dynamically without an upper limit. To avoid excessive idle threads, the thread pool has an automatic thread recycling mechanism. For any thread:

• If it has been idle for more than 10 minutes and the thread pool still has more than 8 threads, the thread is recycled and destroyed.
• If it has been idle for more than 60 minutes, the thread is unconditionally destroyed.

Although the upper limit of the thread pool size is not fixed, in most cases, the actual upper limit is determined by the following two mechanisms:

• Before parallel execution starts, you must use the Admission module to reserve thread resources. You can only start execution after the reservation succeeds. This mechanism limits the number of concurrent queries. For more information, see Concurrency control and queuing.
• Each time when you request threads from the thread pool, the maximum number of threads that can be allocated to you at a time is N. Here, N is the result of the MIN_CPU of the tenant unit multiplied by the px_workers_per_cpu_quota tenant-level parameter. If you request more threads than N, at most N threads will be allocated to you. The default value of px_workers_per_cpu_quota is 10. For example, the DOP of a DFO is 100. It needs to request 30 threads from node A and 70 threads from node B. If the MIN_CPU of the tenant unit is 4 and the px_workers_per_cpu_quota is 10, then N = 4 × 10 = 40. As a result, this DFO can actually request 30 threads from node A and 40 threads from node B. The actual DOP is 70.

Optimize performance by load balancing

To achieve the best performance, the tasks assigned to all worker threads should be as equal as possible.

When SQL statements use block granularity to divide tasks, the system dynamically allocates tasks to worker threads. This minimizes workload imbalance, ensuring that no worker thread is obviously overloaded.

When SQL statements use partition granularity to divide tasks, you can set the DOP to an integer multiple of the number of worker threads to optimize performance. This is useful for partition-wise joins and parallel DML.

Assume that a table has 16 partitions with approximately the same amount of data in each partition. In this case, you can use 16 worker threads (DOP = 16) to complete the work in about 1/16 of the time; or use five worker threads (DOP = 5) to complete the work in about 1/5 of the time; or use two worker threads (DOP = 2) to complete the work in about half the time. However, if you use 15 worker threads to process 16 partitions, the first thread will start processing the 16th partition after it completes the first one. The other threads will become idle after they complete their work. If the amount of data in each partition is similar, this configuration results in poor performance. If the amount of data in each partition varies, the actual performance varies accordingly.

Likewise, assume that you use six worker threads to process 16 partitions with similar amounts of data in each partition.

In this case, after each thread completes the work of the first partition, it will start processing the second partition. However, only four threads will process the third partition, and the other two threads will remain idle.

Generally, the time required to perform parallel operations on N partitions by using P worker threads is not equal to N/P. This formula does not consider the situation where some threads need to wait for other threads to complete the last partition. However, you can set an appropriate DOP to minimize workload imbalance and optimize performance.

Scenarios where parallel execution is not recommended

Parallel execution is generally not recommended in the following scenarios:

• Typical SQL queries take milliseconds to execute in the system.

  Parallel queries have scheduling overheads measured in milliseconds. For short queries, the scheduling overheads can offset the benefits of parallel execution.

• The system is heavily loaded.

  The design goal of parallel execution is to make full use of idle system resources. If the system is already fully loaded, parallel execution is unlikely to bring additional benefits and may even compromise the overall system performance.

Serial execution uses a single thread to perform database operations. Serial execution is preferable to parallel execution in the following scenarios:

• The query accesses a small amount of data.
• High concurrency is required.
• The query execution time is less than 100 milliseconds.

Parallel execution cannot be implemented at the top level of a DFO for the following reasons:

• The top-level DFO does not need to be parallelized. It is responsible for interactions with the client and performs operations that do not require parallelization, such as LIMIT and PX COORDINATOR.
• A DFO containing a TABLE UDF can be executed only serially, but the rest of the DFO can be parallelized.
• Parallel execution is not supported for ordinary SELECT and DML statements in OLTP systems.