KlustronDB Multi-Level Parallel Query Processing Technology

Background

This article focuses on how KlustronDB executes distributed query tasks in multi-level parallelism across the entire cluster, fully utilizing the abundant CPU cores, memory, and storage resources in Kluscomp instances and multiple Klustore instances to achieve excellent query processing performance. In the ideal scenario, KlustronDB can use all the server resources in the cluster to execute the same query statement, achieving ultimate performance.

Overview of KlustronDB Multi-Level Parallel Query Processing Technology

The KlustronDB multi-level parallel query processing architecture is divided into three levels --- multiple sub-tasks of the same user query can be executed in parallel within a Kluscomp instance, between Kluscomp instances and Klustore instances, and within Klustore instances. The following briefly introduces the parallel query mechanisms of these three levels.

Within Kluscomp instances, KlustronDB inherits PostgreSQL's parallel query architecture and has enhanced and expanded it. PostgreSQL supports executing eligible query tasks in parallel using multiple processes, so the optimizer decides whether to use the parallel query processing mechanism based on the query statement, the characteristics of the data, and the limit on the number of processes available. If it is used, it will create several worker threads and assign them sub-tasks from the query plan that can be executed in parallel, achieving parallel query processing.

The parallel query capability between Kluscomp instances and Klustore instances is newly designed and developed by the KlustronDB team. It can operate independently of the Kluscomp instance parallel query framework, or it can collaborate with it.

If the Kluscomp instances of KlustronDB insert, update, or delete multiple rows of data, and these rows are distributed across table shards located on multiple shards, the Kluscomp instances will asynchronously send data read and write statements to the target Klustore instances, rather than sending data read and write statements to one Klustore instance sequentially and waiting for the result before sending to the next Klustore instance. In this way, multiple Klustore instances can execute data read and write statements in parallel.

The KlustronDB team has developed a series of new features in the Klustore instance kunlun-storage to support KlustronDB's multi-level parallel query processing technology, enabling multiple read-only query sub-tasks of the same query to be executed in multiple connections of the same Klustore instance. At the same time, we have implemented a new high-performance incremental execution technology in kunlun-storage to improve the fetch() performance of prepared statements, which can significantly enhance the execution performance of queries containing limit clauses and subqueries (when they cannot be converted to semi-joins).

The three parts are detailed below.

KlustronDB Kluscomp instances parallel query processing

In PostgreSQL's multi-process parallel query processing architecture, the process that handles user connections is called the backend process. It is responsible for receiving the user's SQL requests, executing the requests, and ultimately returning the query results to the client. As part of executing the requests, it performs query optimization and serial query execution.

If a query plan has a parallel executable sub-plan (PESP), the backend process is also called a leader process. As part of query execution, the leader process will create several worker subprocesses to execute different data segments of this PESP according to the parallel query resource configuration of the instance, with each data segment referred to as a partial plan.

The Leader process divides sub-tasks (partial plans) according to the range of data that PESP needs to read, and assigns them to worker processes for execution. Each worker process executes the PESP sub-task (i.e., partial plan) with exactly the same tree structure, but the data intervals are different. Each returns non-overlapping sub-result sets to the Gather or Gather Merge nodes to collect the results from the workers.

For example, during query optimization, the leader process converts a seqscan serial execution node that needs to scan 100,000 rows of data into four Parallel_seqscan subtasks of 25,000 rows each, which are assigned to four worker processes for execution during query execution. Each worker executes a Parallel_seqscan, but with different parameters and scanning different data ranges. The leader process also acts as a worker process to execute one of the subtasks. After completing the subtasks, the leader process executes a Gather or Gather Merge node to collect the sub-results from each worker process and merge them into the result of this PESP, passing it to the upper-level node.

(Parallel Query Example 1)

The above diagram is an example of a distributed parallel query plan in KlustronDB. The following content will gradually cover its components and provide a detailed analysis and explanation.

Not all Plan nodes can be executed in parallel. The query optimizer generates a query plan by obtaining the parallel execution properties of the Plan. Parallel execution properties include three types: Parallel Safe, Parallel Restricted, and Parallel Unsafe. Parallel Safe nodes can be located in a parallel execution plan subtree or other parts of the query plan that include a parallel execution plan. Parallel Restricted nodes can only be located above the Gather node, meaning they cannot be part of a parallel execution plan. Parallel Unsafe nodes prohibit their associated query plan from being executed in parallel; a query plan containing Parallel Unsafe nodes cannot be executed in parallel at all. Common nodes that can be executed in parallel include Parallel Append, Partial Aggregate, Parallel Join, Parallel Index/Bitmap/Sequential Scan. However, the last three types of Plan nodes are not used in KlustronDB to query user data because user data is stored in Klustore instances and these Plan nodes are not needed.

A plan subtree can only be considered for conversion into a parallel execution plan if all its plan nodes can be executed in parallel (Parallel Safe). However, not all nodes in a parallel execution plan are split into multiple sub-tasks for execution. For example, the inner nodes of hash/nested loop/merge joins are fully executed in each parallel worker, while only the query workload of the outer nodes is divided into multiple sub-tasks and distributed to multiple worker processes. Only the inner nodes of a Parallel Hash Join are split among all workers, executed in parallel by them, and their data is shared.

Finally, during query optimization and query execution, parallel execution is also restricted by resource limitation variables, including max_worker_processes, max_parallel_workers_per_gather, and max_parallel_workers. These variables determine how many sub-tasks the query optimizer will divide a parallelizable subtree of a query plan into, as well as how many worker processes will actually execute in parallel during execution.

(Function switch and resource limit variables for parallel queries)

The above image shows the configuration variables related to parallel execution in the KlustronDB Kluscomp instance Kluscomp instance, including the parallel feature switch and the limit on the number of parallel execution processes. Among them, enable_parallel_remotescan is specific to Kluscomp instance, while the other variables are inherited from PostgreSQL.

The above content is functionality that PostgreSQL-11.5 originally possesses. In Kluscomp (the Kluscomp instance of KlustronDB), we fully inherit, develop, and utilize the above capabilities. At the same time, we have extended these capabilities to distributed scenarios in the optimizer of Kluscomp instance, as detailed in the next section.

Execute Remote Plan in Parallel

Remote Plan is a newly added query plan node in KlustronDB, used to retrieve user data from Klustore instances. When the enable_parallel_remotescan switch is turned on, the optimizer of Kluscomp can not only assign Remote Plans that scan different table partitions to multiple partial plans to allow multiple worker processes to execute in parallel, but also, based on statistical information, split Remote Plan nodes that scan the same table partition or an unpartitioned single table into multiple Remote Plan sub-tasks according to the range of rows that need to be scanned. This way, even if the Remote Plan needs to scan a single table, it can still be executed in parallel.

(Detailed version of Parallel Query Example 1)

The above image is a distributed parallel query plan generated by Kluscomp instance, from which it can be seen:

1. Each worker executes a RemotePlan node to scan a partition of the t2 table, and the leader process also executes a RemotePlan.

2. The three table partitions of T2 are located on the same shard 3. The leader process and the three worker processes each take responsibility for fetching data from one table partition. They each connect to the shard's primary node to retrieve data for each table partition, and these connections can use the same snapshot, thereby returning data with transactional consistency.

3. The left node of the Hash join is executed in parallel (Parallel Append), whereas the right node is executed by every worker. Therefore, worker processes do not share data, which also leads to some resource waste.

4. The Partial Aggregate node performs an Aggregate on the rows returned by each worker's own Hash Join node. Each worker process executes the subtree with the Partial Aggregate as the root. The dataset scanned by each worker's subtree is determined by the Remote Plan of each partial plan.

5. The Gather node collects the results returned by each worker process. The leader process executes the Gather node and the portion above it. The Gather node needs to send the result set to Finalize Aggregate in order to generate the final Aggregate result using the partial aggregate results from each worker.

Additionally, in parallel query processing scenarios, Kluscomp can still use read-write separation technology to execute read-only queries on the shard replicas, which is especially suitable for OLAP scenarios and can avoid placing excessive load on the primary node.

The Remote Plans executed in parallel will each connect to Klustore instances within the worker processes to perform query tasks. To obtain consistent query results, these worker processes must use the same snapshot in their respective connections. Therefore, we have added connection snapshot sharing capability in kunlun-storage to cooperate with Kluscomp in executing parallel distributed query plans. This part will be explained below.

5 Parallel Query Processing Between KlustronDB Kluscomp instances and Klustore instances

The parallel query processing within the above Kluscomp instances can only execute read-only query subtrees. KlustronDB implements parallel query processing between Kluscomp instances and Klustore instances. Insert/update/delete statements sent to multiple shards are sent asynchronously, so multiple shards will receive their own insert/update/delete statements almost simultaneously and begin execution, thereby achieving parallel execution of insert/update/delete statements.

Support features of KlustronDB Klustore instances

The parallel query capability of MySQL Community Edition is very limited, and KlustronDB does not use it. We have developed a series of technologies in kunlun-storage to support the performance improvement of KlustronDB queries. Some of these technologies are used in KlustronDB's parallel query processing, while others can be used not only in parallel query processing but also in serial query processing.

Transaction Snapshot Sharing Technology

KlustronDB's newly designed and developed Transaction Snapshot Sharing Technology allows multiple worker threads on Klustore instances to execute multiple Remote Plans of the same query plan from the Kluscomp instances, enabling both Kluscomp instances and Klustore instances to execute a SELECT query in parallel. Therefore, this technology is a fundamental condition for Kluscomp instances to execute Remote Plans in parallel. Without this technology, the data seen by multiple worker processes may not match the data seen by a single snapshot if executed serially, which would result in incorrect query results.

When a Kluscomp instance starts a transaction, the main process (backend process, also the Leader process) commands the Klustore instance to create a snapshot in its main connection. Then, in the connections initiated by parallel worker processes, special commands are executed to replicate and use the snapshot from the main connection, so that the data visible in the worker processes is exactly the same as the data seen in the main connection, ensuring consistency of query results.

Prepared statement

If a prepared statement is executed in the client connection, the Kluscomp instance will also send several prepared statements to the target Klustore instance. After the client binds parameters to its prepared statement, the Kluscomp instance will also send the bind parameter commands to the Klustore instance to bind parameters to the corresponding prepared statement. This can avoid repeatedly parsing query statements that are executed repeatedly.

In particular, in scenarios of incremental execution in the next section where its performance effectiveness can be demonstrated, the performance advantage of using prepared statements will be especially evident.

Incremental On-Demand Execution Technology

In the community edition of MySQL, when a prepared statement is executed using execute(), the entire query is executed and all the query results are stored in a temporary table. Subsequent use of fetch() retrieves result rows continuously from this temporary table. The problem with this approach is that it does not achieve the intended purpose of fetch() — incremental execution on demand, which avoids unnecessary consumption of computing resources. At the same time, executing all the query results and writing them into a temporary table can cause significant temporary I/O bandwidth consumption and storage usage when the result set is large.

There are many common SQL queries, such as those with a LIMIT clause, subqueries using EXISTS/ANY/SOME/IN/NOT IN that cannot be converted to semi-joins, and queries with row count limits on subquery results, where it is often necessary to fetch part of the query results and then stop execution. The execution method of prepared statements in the community edition of MySQL cannot perform well with queries that have these characteristics.

In kunlun-storage, we designed and implemented an incremental on-demand query execution technology --- the client (in the KlustronDB scenario, this is Kluscomp instance) executes as many rows as it fetches(). At this point, kunlun-storage can achieve performance far superior to MySQL using this technology, while completely avoiding the performance and resource overhead issues related to the community version of MySQL.

For example, suppose a partitioned table t1 has 100 million rows of data, distributed across 16 table shards on 4 shards, with the partition key being id. Execute a statement like this:

select*from t1 whereage
between 18 and 36 limit 1000;

After the Kluscomp instance uses prepared statements, the query it sends to the Klustore instances only needs to retrieve a total of 1,000 rows from these 16 table shards, and the rest of the data rows do not contribute to the query result at all; without the aforementioned Kunlun-storage technology, those queries would have been executed completely, resulting in a significant increase in execution time, and all rows in the t1 table that meet the filter conditions would be returned to the Kluscomp instance. The Kluscomp instance could receive millions of rows, but it only returns the first 1,000 rows to the client and discards all the rest. This results in a large waste of computing resources (CPU time slices, network bandwidth, memory bandwidth).

7 Performance and Comparison of KlustronDB OLAP Query Processing

Next, we will test the performance of several typical queries under parallel and serial execution and analyze the reasons. Basic setup: max_parallel_workers=8; max_parallel_workers_per_gather=4

So each query subtree has at most 4 processes working in parallel, including the leader process. Below is the preparation of test data --- defining two data tables for testing and inserting test data.

)

(Data preparation --- define tables and insert a large amount of data)

1. Joining two tables with row filtering (referred to as Query 1, see the figure below for the SQL statement, its query plan, and execution time)

Query result: The time consumption of parallel queries is about one-third that of serial query execution (112ms VS 312ms)

This is because during parallel execution, 3 out of the 4 table partitions are located on the same shard 3. This shard is overloaded during parallel execution, so the performance of parallel execution does not increase to four times that of serial execution. In addition, building the internal table hash table is something that each worker process has to do, and this part of the time consumption is not reduced by parallelism.

(Parallel execution plan and time consumption of query 1)

(Serial execution plan and time consumption of Query 1)

2. After joining the two tables, perform an aggregate query, one with grouping (Query 2) and the other without grouping (Query 3)

The query results are shown in the table below, and the detailed query statements along with their query plans and time consumption are shown in the figure below.

Query	Time Taken for Parallel Execution (ms)	Time Taken for Serial Execution (ms)
Query2	92	250
Query3	72	247

Query 2 takes 20 milliseconds longer than Query 3 because each worker has to perform a hash-based grouping operation and sort all the groups generated on its node. Both Query 2 and Query 3 have a Finalize Aggregate operation that the leader thread needs to perform separately, and each worker independently executes the hash join on the inner nodes and builds the hash table using the returned data rows. Therefore, the time spent on this part of the work cannot be evenly distributed among the four workers, making the time slightly more than a quarter of the serial execution time.

(Query 3 parallel execution plan and time consumption)

(Query 2 parallel execution plan and time consumption)

(Query 2, 3 serial execution plans and time consumption)

Summary

KlustronDB's parallel query processing technology fully utilizes PostgreSQL's single-node parallel query processing capabilities and extends them to distributed query processing scenarios. At the same time, we have enhanced and expanded the query processing capabilities of the community edition of MySQL in kunlun-storage, providing fundamental support for KlustronDB's distributed parallel query processing. Ultimately, kunlun-storage and Kluscomp work seamlessly together to achieve good distributed parallel query processing performance. In the future, we will continue to expand and enhance KlustronDB's parallel query processing capabilities in order to fully leverage the computing power of a large number of hardware resources in distributed database clusters, achieving better query processing performance.