Detailed SQL optimization requirements: parallel execution framework and execution plan

Abstract:In a relational database, the optimizer is one of the core components of the database. Since a series of factors will affect the execution of the statement, the optimizer weighs all the factors comprehensively and chooses the best execution plan among many execution plans.

This article is shared from the Huawei cloud community “Huawei Cloud GaussDB(for OpenGauss) special broadcast phase 5: SQL optimization interpretation”, the original author: Huge.

1. Introduction

In a relational database, the optimizer is one of the core components of the database. Since a series of factors will affect the execution of the statement, the optimizer weighs all the factors comprehensively and chooses the best execution plan among many execution plans. With the arrival of the era of big data, such as e-commerce, games, telecommunications and other industries are large-scale applications, a single database node is difficult to cope with the continuous growth of data scale and ensure the need of performance, the business is faced with the problem of “no storage, slow calculation, inaccurate calculation”. GAUSSDB (for OpenGauss) adopts a horizontally extensible distributed architecture, which can well meet the storage and calculation needs of large-scale and massive data. It selects the execution path with the smallest cost value from the target SQL execution plan through the CBO cost of the target SQL execution plan. The cost of each execution path is calculated based on the statistics of tables, indexes, columns and other related objects involved in the target SQL. The actual value reflects an estimate of the I/O, CPU, and network resources consumed to execute the target SQL.

• I/O resources: The cost of reading table data from disk into memory
• CPU resources: The cost of processing table data in memory
• Network resources: Distributed SQL requiring data interaction between DN. In actual execution, the data needed is not in the local DN (data needs to be fetched from other DN), so the network resource consumption will be converted into a pair of equal I/O resource consumption and then estimated.

Based on the content of the 5th live broadcast, this paper introduces the distributed parallel execution framework, distributed execution plan and other aspects.

2. Distributed parallel execution framework

2.1 Actuator: PIPELINE model

The feature of GaussDB(for OpenGauss) executor is that it executes from bottom to top according to the query plan tree, based on the volcano model, that is, each node executes and returns a row of records to the parent node.

The biggest advantage of the volcano model is that it can only take out one tuple at a time on demand. After this tuple is processed, the system will take out the next tuple that meets the conditions until all tuples that meet the conditions are taken out. As you can see from the way this works, it requires very little system resources each time it is executed.

2.2 High performance distributed query engine

GAUSSDB (for OpenGauss) makes full use of the current multi-core characteristics and improves the system throughput through concurrent execution of multi-threads. It is well known that in traditional distributed MPP databases, the cost of data shuffling, or redistribution of data, is very high, which limits the scope of user scenarios.

GAUSSDB (for OpenGauss) can make full use of the current multi-core characteristics, using parallel execution mechanism, has many years of accumulation in SQL execution optimization, and provides three streams (broadcast stream, aggregation stream and redistribution stream) to reduce the flow of data between DN nodes. It breaks through the limitation of user usage scenarios caused by the high cost of data shuffle in traditional distributed MPP database, and even complex mixed SQL and transaction analysis (HTAP) scenarios can be executed best.

The general execution of GaussDB(for OpenGauss) :

• A business application sends SQL to the Coordinator. SQL can contain CRUD operations on data;
• The Coordinator uses the optimizer of the database to generate an execution plan. Each DN processes data according to the execution plan.
• Data is distributed in each DN based on the consistent Hash algorithm, so in the process of data processing, DN may need to obtain data from other DN. GAUSSDB provides three streams (broadcast stream, aggregation stream and redistribution stream) to realize data flow among DN, so that join does not need to be extracted to CN for execution.
• DN returns the result set to Coordinate for summary;
• The Coordinator returns the aggregated results to the business application.

3. Distributed execution plan

CN determines whether SQL statements can be directly executed on each DN without data exchange according to the distributed column information and associated column information of the table. If so, CN adopts LIGHT_QUERY or FQS_QUERY process and maintains an attitude that it is none of its business. I will issue whatever you send to me. The entire Query command is directly sent to DN for execution, and output directly after execution. If data interaction between various DN is needed, the stream operator will be chosen. If you find that you can’t use the stream operator, you go back to the original PGXC process.

3.1 LIGHT_QUERY

• Scenario: Statements can be executed directly in a DN (single shard statement, click the scene).
• Principle: CN directly issues statement QPBE packet to the corresponding DN. The advantages of this approach are high execution efficiency and good linear expansion ratio.

create table t1 ( col1 int, col2 varchar ) distribute by hash(col1);
create table t2 ( col1 int, col2 varchar ) distribute by hash(col1);

3.2 FQS_QUERY

• Scenario: When statements can be completely pushed down to execute on multiple DNs, and no data interaction is required between DNs.
• Principle: CN does not go through the optimizer, but directly generates RemoteQuery plan, which is sent to DN by executor logic. Each DN generates execution plan according to push-down statement and executes it, and the execution results are summarized on CN.

create table t1 ( col1 int, col2 varchar ) distribute by hash(col1);
create table t2 ( col1 int, col2 varchar ) distribute by hash(col1);

The biggest difference between LIGHT_QUERY and FQS_QUERY is that although CN both send the received query directly to DN for processing after decision, LIGHT_QUERY only involves a single DN operation, while FQS_QUERY involves multiple DN operations respectively. None of them involve data interaction between DNs.

3.3 the STREAM GATHER

• Scenario: Data interaction between DNs is required.
• Principle: CN generates an execution plan with Stream operator through the optimizer according to the original statement and sends it to DN for execution. During the execution of DN, there is data interaction (Stream node). The Stream operator establishes a connection between DN for data interaction, and CN summarizes the execution results and assumes most of the calculation.

create table t1 ( col1 int, col2 varchar ) distribute by hash(col1);
create table t2 ( col1 int, col2 varchar ) distribute by hash(col2);

3.4 the STREAM REDISTRIBUTE

• Scenario: Data interaction between DNs is required.
• Principle: CN generates an execution plan with stream operator through the optimizer according to the original statement and sends it to DN for execution. Data interaction (stream node) exists in the execution process of each DN. The stream operator establishes a connection between DN for data interaction.

create table t1 ( col1 int, col2 varchar ) distribute by hash(col1);
create table t2 ( col1 int, col2 varchar ) distribute by hash(col2);

3.5 the STREAM BROADCAST

• Scenario: Data interaction between DNs is required.
• Principle: CN generates an execution plan with stream operator through the optimizer according to the original statement and sends it to DN for execution. Data interaction (stream node) exists in the execution process of each DN. The stream operator establishes a connection between DN for data interaction.

create table t1 ( col1 int, col2 varchar ) distribute by hash(col1);
create table t2 ( col1 int, col2 varchar ) distribute by hash(col2);

When the REDISTRIBUTE operator is used, the data can be redistributed to take full advantage of the computing power of multiple nodes, while the BROADCAST operator is mainly used when the stream’s subplans produce a small amount of data, where the cost of BROADCAST is low.

3.6 PGXC

• Scenarios: Extreme scenarios that do not meet the previous processing methods, with very poor performance.
• Principle: CN generates a remoteQuery plan from some statements in the original statement through the optimizer, and sends each remoteQuery to DN. After DN executes, the intermediate result data is sent to CN. After CN collects the remaining execution plan, CN carries out most of the calculation.

conclusion

To sum up, GAUSSDB (for OpenGauss), as a new generation of self-developed financial level distributed relational database, adopts a horizontally extensible distributed architecture and generates distributed operators and distributed execution plans through SQL optimizer. Three streams (broadcast, aggregated, and redistributed) are provided to reduce the flow of data between DN nodes. The execution engine is a distributed parallel execution framework, which supports inter-node parallelism and intra-node parallelism. It makes full use of the current multi-core characteristics, improves system throughput through concurrent execution, and has high performance query ability under big data.

Ps: more wonderful content, please click into the link to view: https://bbs.huaweicloud.com/l…

Click on the attention, the first time to understand Huawei cloud fresh technology ~