CN114925093A - Method and device for processing database query statement - Google Patents

Abstract

The disclosure discloses a method and a device for processing a database query statement, wherein the method comprises the following steps: receiving a database query statement, wherein the database query statement comprises a main query and a sub query, the query condition of the sub query is a relevant condition of the main query, the main query comprises a filtering condition for filtering based on the query result of the sub query, and the partition table data of the main query and the sub query are distributed on a plurality of servers; querying target data from the partition table data of the main query, wherein the target data comprises first data relevant to the main query and second data relevant to the sub-queries; the first data and the second data are distributed to a target server, and the target server can obtain all sub-queried partition table data; based on the second data, executing the query of the sub-query partition table data in parallel to obtain a first query result; and performing a filtering operation on the first data in parallel by using the filtering condition according to the first query result.

Description

Method and device for processing database query statement

technical field

The present disclosure relates to the field of databases, and in particular, to a method and device for processing database query statements.

Background technique

When a distributed database executes a structured query language (SQL) statement containing correlated sub-queries, the traditional execution plan needs to pull the data involved in the main query and sub-queries back to the filter conditions of the local single-threaded sub-query execution. Each row of the main query needs to be rescanned for subqueries. This execution plan is difficult to parallelize, resulting in poor execution performance.

SUMMARY OF THE INVENTION

In view of this, the present disclosure provides a method and apparatus for processing database query statements, which can improve the execution performance of a distributed database when executing SQL statements including correlated subqueries.

A first aspect provides a method for processing a database query statement, comprising: receiving a database query statement, wherein the database query statement includes a main query and a sub-query, the query conditions of the sub-query are related conditions of the main query, the The main query includes filter conditions for filtering based on the query results of the sub-query, the partition table data of the main query and the sub-query are distributed on multiple servers; the target data is queried from the partition table data of the main query , the target data includes first data related to the main query and second data related to the sub-query; distribute the first data and the second data to a target server, and the target server can Acquire all partition table data of the sub-query; perform a query on the partition table data of the sub-query in parallel based on the second data to obtain a first query result; use the first query result according to the first query result The filter condition performs the filtering operation on the first data in parallel.

Optionally, the distributing the first data and the second data to the target server includes: distributing the first data and the second data to the target server by adaptively selecting a distribution method of hash or random. the target server.

Optionally, the method further includes: collecting the distribution situation of the first data and the second data; according to the distribution situation, judging to use a hash or random distribution method to distribute the first data and the second data. The second data is distributed to the target server.

Optionally, the target server can obtain all the partition table data of the sub-query, including: the target server obtains all the partition table data of the sub-query through the DISTRIBUTED TABLE SCAN operator.

In a second aspect, an apparatus for processing a database query statement is provided, comprising: a receiving module configured to receive a database query statement, wherein the database query statement includes a main query and a subquery, and the query condition of the subquery is the same as that of the main query. Relevant conditions, the main query includes filter conditions for filtering based on the query results of the sub-queries, and the partition table data of the main query and the sub-queries are distributed on multiple servers; The target data is queried in the partition table data of the main query, and the target data includes first data related to the main query and second data related to the sub-query; a distribution module is used for distributing the first data and the sub-query. The second data is distributed to the target server, and the target server can obtain all partition table data of the sub-query; the query module is further configured to: based on the second data, perform parallel execution of the sub-query. The query of the partition table data obtains the first query result; the filtering module is configured to perform the filtering operation on the first data in parallel by using the filtering condition according to the first query result.

Optionally, the distribution module is further configured to distribute the first data and the second data to the target server by adaptively selecting a distribution manner of hash or random.

Optionally, the device further includes: a collection module for collecting the distribution of the first data and the second data; a judgment module for judging the use of hash or random according to the distribution. The distribution method distributes the first data and the second data to the target server.

Optionally, the target server obtains all data table data of the subquery through the DISTRIBUTED TABLE SCAN operator.

In a third aspect, a computer-readable storage medium is provided, and the computer storage medium stores a computer program, which implements the method according to the first aspect when the computer program is executed.

In a fourth aspect, a computer program product is provided, comprising executable code, which, when executed, can implement the method of the first aspect.

The solution of the embodiment of the present disclosure distributes the first data related to the main query and the second data related to the subquery to a target server, where the target server is a server that can obtain all partition table data of the subquery. Since the target server can execute tasks in parallel, the target server can execute the sub-query in parallel based on the second data, and execute the filtering operation on the first data in parallel, that is, the filtering conditions of the sub-query can be executed in parallel, so that the SQL can be improved. Statement execution performance.

Description of drawings

FIG. 1 is a system example diagram of a distributed database provided by an embodiment of the present disclosure.

FIG. 2 is a schematic flowchart of a method for processing a database query statement provided by an embodiment of the present disclosure.

FIG. 3 is an exemplary flowchart of a method for processing a database query statement provided by another embodiment of the present disclosure.

FIG. 4 is an exemplary flowchart of a method for processing a database query statement provided by another embodiment of the present disclosure.

FIG. 5 is a schematic structural diagram of an apparatus for processing a database query statement provided by an embodiment of the present disclosure.

FIG. 6 is a schematic structural diagram of an apparatus for processing a database query statement provided by another embodiment of the present disclosure.

Detailed ways

The technical solutions in the embodiments of the present disclosure will be clearly and completely described below. Obviously, the described embodiments are only a part of the embodiments of the present disclosure, but not all of the embodiments.

With the development of the Internet and the continuous surge in data volume, distributed databases have gradually replaced stand-alone databases. In database use, read operations account for the majority of database operations, and it is reflected in SQL that data query statements are often used. In an SQL statement, the structure composed of "select-from-where" is called query block. When a query block is nested within it as a query condition of another query block, it is called a subquery.

An SQL statement can consist of a main query and sub-queries. The main query can also be called the outer query, and the sub query can also be called the inner query. A subquery is a non-correlated subquery if its execution does not depend on the query results of the main query. Uncorrelated subqueries are executed from the inside out, and each query needs to be executed only once. That is to say, a non-correlated subquery executes the subquery first, caches the execution result of the subquery, and then executes the main query.

If the execution of a subquery depends on the query results of the main query, the subquery is called a correlated subquery. In SQL containing correlated subqueries, the subqueries need to be traversed once for each row executed in the main query. This execution method causes the execution time of SQL statements containing correlated subqueries to increase as the number of rows in the main query increases. For correlated subqueries with complex nesting levels, the execution time of SQL increases exponentially.

Specifically, a correlated subquery indicates that there is a certain conditional association between two queries. The following describes the related concepts of correlated subqueries based on the following SQL statements.

SELECT T2.v2

FROM T2

WHERE T2.v2=(SELECT Sum(v2)

FROM T1

WHERE T2.v1<T1.v1);

The above SQL statement contains a main query and a subquery. The main query statement is "SELECT T2.v2FROMT2WHERE T2.v2=()", and the sub-query statement is "(SELECT Sum(v2)FROM T1WHERE T2.v1<T1.v1)". The subquery statement includes the expression "WHERE T2.v1<T1.v1", which obviously refers to the field v2 in the T2 table of the main query object. That is, the query result of the sub-query object T1 table depends on the main query object T2 table. Therefore, the subquery is a correlated subquery.

In addition, the main query includes a filter condition for filtering based on the query result of the sub-query, and the filter condition may also be referred to as a sub-query filter condition. Still taking the above SQL statement as an example, the filter condition may refer to "WHERE T2.v2=()".

In a correlated subquery, the statement of the main query is quoted, so that the expression associated with the subquery and the main query is a correlated expression. As an example, the expression "where T2.v1<T1.v1" in the above SQL is an expression that associates the subquery object T1 with the main query object T2, therefore, "where T2.v1<T1.v1" is The correlation expression of the correlated subquery, the correlation expression is also called the correlation condition.

The following describes the execution process of the correlated subquery of the above SQL statement:

The first step is to take a tuple (ie, a row) from the main query object table T2, and pass the value of the tuple-related column v1 (ie, T2.v1) to the subquery.

In the second step, the subquery executes the value of the v1 column of the table T2 as a condition, and executes a complete subquery on the T1 table to obtain the query result, and transmit the query result to the main query. A subquery executes a complete query process called a loop.

The third step is to perform the filtering operation. Compare the query result of the subquery with the value of the related column v2 of the tuple to determine whether the value of the v2 column of the table T2 satisfies the filter condition.

In the fourth step, the next tuple is taken out from the main query object table T2, and the first to the third step are executed cyclically until all the tuples in the external query object table T2 are processed.

When the database processes an SQL statement, it generates an execution plan for the execution of the SQL statement. It needs to be explained that the execution plan is the execution path or algorithm of the SQL statement. The execution plan affects the execution performance of the SQL statement. In some scenarios, the optimizer will, for example, find an execution method that uses less resources among many execution plans to display.

For SQL statements with dependent subqueries (ie, correlated subqueries), the database generates a subquery execution plan (subplan). In a distributed database, the execution logic of subplan is usually to first pull the data involved in the main query (such as T2.v2 and T2.v1 above) back to the local, and then execute a complete subquery for each row of the data involved in the main query, And pull the execution result back to the local. The calculation of the sub-query related filter conditions is performed locally, which is also called the calculation of the sub-query filter conditions. It should be noted that the relevant filter condition is an SQL statement for filtering the sub-query result, for example, a where statement in which the sub-query result matches the main query.

The following describes the execution process of the SQL statement including the correlated subquery in the distributed database with reference to FIG. 1 .

FIG. 1 is a structural example diagram of a distributed database provided by an embodiment of the present application. The distributed database 100 in FIG. 1 may include a client node 110 , a coordinator node 120 and a computer node 130 .

For example, the user node 110 can be a client, including a computer, a tablet computer, a mobile phone, etc., for sending the SQL statement required for the query.

The coordinator node 120, also called the coordinator, interacts with the user node 110, is responsible for providing a query interface to the user node 110, receives the SQL statement sent by the user node 110, and converts the SQL statement into a plurality of executable sub-query plans for distribution to compute node 130.

The computing node 130 is responsible for receiving the query task sent by the coordinating node 120 and executing the sub-query plan. The computing node 130 has computing and data storage functions. As an example, the distributed database may further include storage nodes for storing data. The present disclosure does not limit the structure of the distributed database.

Specifically, in the distributed database, the execution process of the SQL statement is as follows: the coordinating node 120 converts the received SQL statement into multiple sub-query plans that can be executed in parallel on different computing nodes 130, and according to the For information such as CPU utilization, free memory size, etc., an appropriate computing node 130 is selected for transmission. Compute node 130 executes the subquery plan and returns the result to the coordinator node. The coordinating node 120 sends the calculation result to the user node 110 again.

It should be noted that the process of the computing node 130 sending the sub-query execution result to the coordinating node 120 may be referred to as pulling back locally. That is to say, the traditional sub-query execution plan needs to pull the query result of the main query and the query result of the sub-query back to the local, and perform the calculation of the filter condition of the correlated sub-query locally.

In some application scenarios, the computing node 130 may be deployed on multiple servers, and the computing nodes 1 to N shown in FIG. 1 are respectively deployed on N servers. The coordinator node 120 can only be deployed on one server. In other words, any computing node 130 can serve as the coordinator node 120, but there can only be one coordinator node 120.

It should be noted that, SQL semantics requires that when the filter condition of the subquery is executed, the subquery data can only return one loop result at a time. If the subquery returns multiple loop results at a time, the system will provide an error. In the traditional execution plan, the sub-query filter conditions are placed locally and executed by a single thread, so there is a problem of poor execution efficiency.

The specific implementation process of the sub-query execution plan in the distributed database is described below with reference to the SQL (1) in the example (1).

The SQL(1) statement is as follows:

creat table T1(v1 int, v2 int) partition by hash(v1) partitions 4;

creat table T2(v1 int, v2 int);

creat table index T1_idx1 on T1(v2)local;

SELECT T2.v2

FROM T2

WHERE T2.v2=(SELECT Sum(v2)

FROM T1

WHERE T2.v1<T1.v1);

In the above SQL(1) statement, the partition table T2 of the main query and the partition table T1 of the sub-query are distributed on multiple servers (servers). Each row in T2 needs to recalculate the aggregate calculation of the subquery (ie, the sum calculation). Then, match the aggregate calculation result with the v2 value of the row in T2 to see if it is satisfied, and return the query result if it is satisfied. As an example, the subquery filter condition in SQL(1) is "WHERE T2.v2=(...)".

When executing SQL(1) in a distributed database, the following execution plan(1) is generated:

ID OPERATOR NAME 0 SUBPLAN FILTER 1 PX COORDINATOR 2 EXCHANGE OUT DISTR : EX10000 3 PX PARTITION ITERATOR 4 TABLE SCAN T2 5 SCALAR GROUP BY 6 PX COORDINATOR 7 EXCHANGE OUT DISTR : EX20000 8 MERGE GROUP BY 9 PX PARTITION ITERATOR 10 TABLE SCAN T1

In some application scenarios, the execution plan can be divided into multiple partial sub-plans, and a partial sub-plan can contain one or more operators. For example, local sub-plans can be divided with the "EXCHANGE" operator as the boundary. For example, in the execution plan (1), operators No. 2 to No. 4 can form a partial sub-plan of the main query (also called a data flow operation (DFO)); No. 7 operator Sub to No. 10 operators can form partial sub-plans of sub-queries.

As an example, a local subplan may be executed by multiple threads. For example, when the main query partition table T2 is distributed on multiple servers, the partition table T2 on these multiple servers can be scanned in parallel by multiple threads.

The specific execution process of the execution plan (1) is as follows:

The No. 1 operator "PX COORDINATOR" (ie the coordinator) initiates multiple threads to scan the main query partition table T2, and sends the scan results to the coordinator.

The coordinator starts the sub-query execution plan and initiates multiple threads to scan the sub-query partition table T1. As an implementation, the worker thread of each subquery will scan the T1 table with the relevant query results (eg T2.v1) of the previous main query.

Each worker thread of the subquery gets a query result, and each thread returns its query result to the coordinator. In some embodiments, the query result may be an aggregated result, that is, each worker thread may aggregate the query result of the partition table data queried by itself. In other embodiments, the query result may be a data set, for example, the query result may be a set of T1.v1.

After the coordinator receives the query result of each worker thread of the subquery, it can perform a global aggregation on the query result of each thread to obtain the execution result of the subquery.

After the coordinator obtains the execution result of the sub-query, it executes the calculation of the filter condition of the correlated sub-query, that is, the calculation of the "WHERET2.v2=(...)" condition. It can be seen from the foregoing that the calculation of the filter condition of the sub-query is completed by a single thread. .

Therefore, it can be seen that in the above execution plan (1), the aggregation calculation of the subquery and the calculation of the filter condition of the subquery cannot be parallelized.

In addition, when SQL(1) is used as an execution condition of a higher-level SQL statement or needs to be connected with other tables, and the execution result of SQL(1) needs to be sent to other SQL query statements, the execution plan (1) will be executed in the subquery The place of filter produces a single point of execution, so that the sending process cannot be executed in parallel.

To sum up, in the process of executing SQL statements containing correlated sub-queries in a distributed database, sub-query calculation, sub-query filter condition calculation and other processes cannot be performed in parallel, resulting in poor execution performance. Moreover, there is no good optimization method at present. Especially when the main query returns a large number of rows or the subquery is a more complex distributed plan, its execution performance will be worse.

Based on the above-mentioned problems, the embodiments of the present disclosure provide a new execution mode, in which the calculation of the filter conditions of the sub-query is not executed by the coordinator node, but by the target server in the calculation node. The coordinating node can send the relevant data in the data of the partition table of the main query to the target server, and the target server executes the calculation of the filter conditions of the sub-query in parallel, thereby improving the execution performance of the SQL statement.

FIG. 2 is a schematic flowchart of a method 200 for processing a database query statement provided by an embodiment of the present disclosure. It should be understood that the steps or operations of the method 200 are shown in FIG. 2, but these steps or operations are merely examples, or the steps may be performed in other orders. The method 200 may include steps S210 to S250, as shown below.

In step S210, a database query statement is received. The database query statement includes a main query and a subquery, the query conditions of the subquery are related conditions of the main query, and the partition table data of the main query and the subquery are distributed on multiple servers.

The database query statement can be any of the aforementioned SQL statements. Since the database query statement includes a main query and a subquery, and the query conditions of the subquery are related conditions of the main query, the database query statement may be an SQL statement including a related subquery.

In a distributed database, the partitioned table data for the main query and subqueries can be distributed across multiple servers. As one example, data for one or more partitions of the main query and/or sub-queries may be stored on each server.

Referring to example (1), the partition table data T2 of the main query can be distributed, for example, on two servers server 1 and server 2, and the partition table data T1 of the subquery can be distributed, for example, on three servers server 3, server 4 and server 5. superior.

In some embodiments, the main query may include filter conditions for filtering based on query results of sub-queries. The filter condition may be the subquery filter condition described above. Taking the above SQL(1) as an example, the filter condition is a "WHERE T2.v2=(...)" statement.

In step S220, the target data is queried from the partition table data of the main query. The target data may include first data related to the main query and second data related to the sub-query.

Taking the above SQL(1) as an example, the first data may be T2.v2, and the second data may be T2.v1. The second data can be understood as data associated with the sub-query in the partition table data of the main query.

The coordinator initiates multiple worker threads to query the partition table data of the main query according to the distribution of the data in the partition table of the main query, and obtains the target data. The target data can be, for example, a row of data in the main query partition table data, or one or more values in this row of data. For example, the target data may include the v1 column and v2 column data in the main query partition table data. For another example, the target data may be the entire T2 table.

As an example, since both server 1 and server 2 store the data of the T2 table, the coordinator can send worker threads to both server 1 and server 2 to execute the query of the T2 table. For example, the coordinator initiates 3 threads on server 1 and server 2 respectively to scan the partition table data T2 of the main query, and obtain the values of T2.v1 and T2.v2 in each row of data in T2.

The above-mentioned step S220 may be performed by the computing node of the main query. The computing node may be, for example, a computing node that stores partition table data of the main query. There can be one or more worker threads on a compute node. If a computing node has multiple worker threads, the multiple worker threads can execute the query of the partition table data of the main query in parallel.

In step S230, the first data and the second data are distributed to the target server. The target server can get all the partition table data of the subquery. The number of the target server can be multiple.

In some embodiments, this process may be performed by the coordinating node described above. The worker thread of the main query can send the first data and the second data to the coordinator node, and then the coordinator node distributes the first data and the second data to the target server. In other embodiments, the process may be performed by a worker thread of the main query. After querying the first data and the second data, the worker thread of the main query can directly distribute the first data and the second data to the target server.

The target server is a server where the partition table data of the subquery is distributed, and the target server may include, for example, multiple servers, such as server 3, server 4, and server 5 described above.

When distributing the first data and the second data to the target server, the first data and the second data of the same row may be distributed to the same server. For example, T2.v1 and T2.v2 of the first row in the T2 table can be distributed to server 3, the second row T2.v1 and T2.v2 of the T2 table can be distributed to server 4, and the second row of the T2 table can be distributed to server 4. The three lines T2.v1 and T2.v2 are distributed to server 5.

There may be one or more worker threads on a target server, and distributing the first data and the second data to the target server may be understood as a worker thread distributing the first data and the second data to the target server. The coordinator node can initiate one or more worker threads on the target server according to the data distribution of the subquery partition table. The coordinating node may distribute the first data and the second data to each worker thread. In some embodiments, the coordinator node may distribute the first data and the second data of the same row to the same worker thread.

In step S240, based on the second data, the query of the partition table data of the sub-query is executed in parallel to obtain the first query result. This process may be performed by the target server.

The target server may use the second data as a query condition, and execute the query of the partition table data of the sub-query. The coordinating node may send different second data to different target servers, and different target servers may execute query operations in parallel based on the received second data.

Still taking the above server 3, server 4 and server 5 as examples, server 3 receives T2.v1 in the first row of the T2 table, then server 3 can use T2.v1 in the first row as a query condition, and execute the query Query of T1 table. What server4 receives is T2.v1 in the second row in the T2 table, then server 4 can use T2.v1 in the second row as a query condition to perform a query on the T1 table. The server 5 receives T2.v1 in the third row in the T2 table, and the server 5 can use the T2.v1 in the third row as a query condition to query the T1 table. Since server 3, server 4, and server 5 receive T2.v1 from different rows, server 3, server 4, and server 5 can query the T1 table in parallel.

The first query result is the query result of the subquery. Taking the above SQL(1) as an example, the first query result is the execution result of "(SELECT Sum(v2)FROM T1WHERE T2.v1<T1.v1)". The first query result may include query results of T2.v1 for different rows.

In step S250, according to the first query result, the filtering operation on the first data is performed in parallel using the filtering conditions.

In the embodiment of the present disclosure, the filtering conditions of the subquery are executed on the target server.

In the traditional solution, the query result of the subquery needs to be returned to the coordinator node, and the coordinator node performs the filtering operation. Different from the traditional solution, in the method 200, the coordinating node will directly distribute the query results (eg target data) of the main query to the sub-queries. For example, the coordinator node initiates multiple worker threads on the target server according to the data distribution of the sub-query partition table. After receiving the target data of the main query, the multiple worker threads scan the data of the sub-query partition table in parallel.

Still taking the above server 3, server 4, and server 5 as examples, server 3 can use T2.v1 in the first row of the T2 table as a query condition to query the T1 table to obtain query result 1. Further, server 3 can perform filtering The operation is to compare the query result 1 with T2.v2 in the first row of the T2 table to determine whether T2.v2 in the first row of the T2 table is equal to the query result 1. If equal, return T2.v2 of the first row of the T2 table; if not, return no data. Server 4 can use T2.v1 in the second row of the T2 table as a query condition, and query the T1 table to obtain query result 2. Further, server 4 can perform a filtering operation, that is, the query result 2 and the second row of the T2 table. Compare T2.v2 to determine whether T2.v2 in the second row of the T2 table is equal to query result 2. If equal, return T2.v2 of the second row of the T2 table; if not, return no data. The server 5 can use T2.v1 in the third row of the T2 table as the query condition, and query the T1 table to obtain the query result 3. Further, the server 5 can perform the filtering operation, that is, the query result 3 and the third row of the T2 table. Compare T2.v2 to determine whether T2.v2 in the third row of the T2 table is equal to query result 3. If equal, return T2.v2 in the third row of the T2 table; if not, return no data.

In some embodiments, multiple worker threads may run on a target server. For example, 2 threads can run on server 3, server 4, and server 5. In this case, the coordinator node can respectively initiate two threads on server 3, server 4 and server 5 to perform the scan of the sub-query partition table data T1, that is to say, the coordinator will scan the target data of the main query in step S220 Distributed to these 6 threads to implement parallel scanning of sub-query partitioned table data T1. Wherein, the execution process of each worker thread is similar to the execution process of server 3, server 4, and server 5 described above, and is not repeated here for indirectness.

In some embodiments, the coordinator distributes the first scan result to the target server by selecting an adaptive random (random) or hash (hash) distribution method. The adaptive random or hash distribution method can refer to the method of dynamically selecting random or hash to distribute data according to the distribution of data during execution.

As an example, the random distribution method may be, for example, a round robin distribution method. It should be noted that the round robin distribution method can evenly distribute the data to each thread of the subquery, so the round robin distribution method can ensure that the data obtained by each thread is uniform and prevent data skew.

The hash distribution method can also distribute the data evenly to each thread of the subquery. But for the hash distribution method, if the same row data appears in the main query partition table data, the hash distribution method will distribute the same row data to the same thread. Since the calculation result of the subquery before the data is cached in the thread, when the same row data is received, the thread does not need to repeatedly calculate the subquery, thereby improving the execution efficiency.

However, in some application scenarios, if there are many rows of the same data in the main query, if the hash distribution method is adopted, all the rows of the same data will be distributed to the same thread. In this way, the thread will be in a work-saturated state, while other threads may be in an idle state, resulting in skewed data distribution, which affects execution efficiency.

If the random distribution method is adopted, each row of data in the main query will be distributed to different worker threads of the sub-query in turn. Although this can avoid the problem of skewed data distribution, it may not enjoy the need to calculate the sub-query due to the same row of data. performance improvements to come.

In response to this problem, the method 200 provided by the embodiment of the present disclosure adopts an adaptive random or hash method to distribute the first data and the second data of the main query to all worker threads of the sub-query.

Specifically, the first scan result of the main query may first be distributed according to a default data distribution method, for example, the first data and the second data are distributed to the working threads of the sub-query in a hash distribution method.

Then, sample the distribution results of the specified row. If the results of hash distribution are not uniform, switch to random distribution and continue to distribute. If the result of hash distribution is uniform, you can continue to use the hash distribution method.

As an implementation manner, sampling may be that each worker thread of the main query reports its own distribution results to the coordinator node, the coordinator node aggregates the distribution results, and decides which distribution method to use according to the aggregated distribution results. After the coordinating node decides to use the hash or random distribution method, it will synchronize the decision result to all worker threads of the main query.

As an embodiment, the target server can obtain all partition table data of the subquery through the "DISTRIBUTED TABLE SCAN" operator. That is to say, the operator "DISTRIBUTED TABLE SCAN" provides the ability to scan data in a distributed manner, so that the worker threads on the target server can execute the scan of the subquery and the execution of the filter condition of the subquery in parallel.

After the target server obtains the first scan result, that is, the worker thread of the subquery obtains the first scan result of the main query, it can execute the scan of the partition table data of the subquery in parallel. After obtaining the subquery result, execute the subquery filter in parallel. conditional calculation.

It can be seen from the above execution process that the method 200 provided by the present disclosure can implement parallel scanning of sub-queries, so the calculation of sub-queries can be executed in parallel, and the calculation of filter conditions of sub-queries can also be executed in parallel. In addition, if the query result obtained by calculation on the target server needs to be sent to other SQL statements, the process can also be executed in parallel. Therefore, the new execution plan provided by the embodiments of the present disclosure can greatly improve the execution performance of the SQL statement.

The method provided by the embodiment of the present disclosure will be further described below with reference to FIG. 3 and example (2). FIG. 3 provides a flowchart of a method for executing an SQL(1) statement according to an embodiment of the present disclosure. Example (2) is an execution plan (2) generated by executing the SQL (1) statement in example (1) according to the method provided by the embodiment of the present disclosure, and the execution plan (2) is specifically as follows:

The execution logic of the above execution plan (2) is as follows:

Step S310, parallel execution (parallel execution, PX) schedules the execution of the DFO where the main query and the sub-query are located.

Operators No. 3 to No. 6 in the execution plan (2) are local sub-plans of the main query, such as the DFO where the main query is located, which are executed on the servers where the main query partition table data T2 is distributed. Operator No. 7 and No. 8 are sub-query local sub-plans, such as the DFO where the sub-query is located, which are executed on the server where the sub-query partition table data T1 is distributed, such as the target server described above.

PX is the No. 0 operator "PX COORDINATOR" in the execution plan (2), which is executed on the coordinator. That is, the coordinator schedules the DFO where the main query and sub-queries are located through the PX operator to start scanning.

In step S320, the multiple working threads where the main query is located perform scanning T2 data, and are distributed to the working threads where the sub-query is located by default in a hash manner.

The DFO where the main query is located scans the partition table data T2 of the main query through multiple threads to obtain a row of data in the partition table data T2. Then, this row of data is distributed to multiple threads where the subquery is located by means of hash distribution. The distribution process can be implemented, for example, by executing the No. 4 operator "EXCHANGE OUT DISTR (Adaptive)" in the execution plan (2).

Step S330, the adaptive sampling decision is distributed in a hash or random manner.

FIG. 4 is an exemplary flowchart of a data distribution process provided by an embodiment of the present disclosure. As shown in Figure 4, PX samples the distribution results of the main query, and synchronizes the decision distribution method to the main query to continue distribution.

Specifically, the PX samples the data distribution situation of multiple threads of the main query, for example, the distribution situation of the first 1000 rows of the main query can be sampled. As an implementation manner, sampling may be that each thread of the main query reports its own data distribution status to the PX, and the PX summarizes the distribution results.

If the distribution results are uniform, or the uniformity of the distribution results is within the preset range, or the distribution results do not cause skewed data distribution, continue to distribute according to the original hash distribution method. At the same time, PX synchronizes the decision of adopting the hash distribution method to all the worker threads of the main query, so that the worker threads of the main query continue to use the hash distribution method to send the scan results to the worker threads of the sub-query.

If the distribution results are not uniform, or the uniformity of the distribution results is not within the preset range, or the distribution results have caused the data distribution to be skewed, PX will change the original hash distribution method to random distribution to distribute data. At the same time, PX will synchronize the decision in the random distribution method to all the worker threads of the main query, so that the worker thread of the main query will then use the random distribution method to send the scan results of the main query to the worker threads of the sub-query.

When the data distribution of the main query reaches the specified number of rows (that is, 1000 rows), PX decides again to use random or hash distribution according to the summary distribution results to continue distributing the scan results of the main query, and synchronizes the decision results to all the work of the main query. Thread, repeating the above process can effectively avoid the problem of skewed data distribution, so that the data distribution is within a controllable range.

Step S340, after receiving the row data of the main query (such as the first data and the second data described above), the working thread where the sub-query is located scans the sub-query data in parallel, and calculates the filter conditions.

After receiving the scan results distributed by the main query, multiple threads of the DFO where the sub-query is located will execute the scan of the sub-query partitioned table data T1 in parallel, in which the No. 8 operator "DISTRIBUTED TABLE SCAN" provides the ability to scan data in a distributed manner.

The scan result of the subquery is matched with the scan result of the main query obtained before, and the rows that satisfy the filter conditions of the subquery are returned. This process is executed by multiple threads in parallel.

Step S350, the rows satisfying the filtering conditions are returned to the client.

In step S340, the row data satisfying the filtering condition has been obtained, and the row data is returned to the client.

To sum up, the embodiment of the present disclosure proposes a new execution mode, which distributes the scan results of the main query to the sub-queries through an adaptive distribution method, and supports the parallelism of the sub-query filter conditions by splitting the main query data in the execution mode. Execution, so that the execution performance of the related execution plan can be linearly improved with the increase of parallelism. On this basis, the embodiment of the present disclosure also creatively proposes to adaptively select a hash or random distribution mode to distribute the main query result, thereby optimizing the problem of poor performance caused by skewed data distribution in the parallel process.

The method embodiments provided by the present disclosure are described in detail above with reference to FIGS. 1 to 4 . The device embodiments provided by the present disclosure will be described below with reference to FIG. 5 to FIG. 6 . It should be understood that the descriptions of the apparatus embodiments correspond to the descriptions of the method embodiments, and therefore, for the parts not described in detail, reference may be made to the foregoing method embodiments.

FIG. 5 is a schematic structural diagram of an apparatus for processing a database query statement according to an embodiment of the present disclosure. The apparatus may be a database, such as a distributed database. In some scenarios, the device may be a native distributed database, where the native distributed database may be a self-developed distributed database, which is not a secondary development or secondary development of an existing distributed database. packaged. Certainly, the apparatus may also be other databases, which are not limited in this embodiment of the present disclosure. The following describes an apparatus to which an embodiment of the present disclosure is applied with reference to FIG. 5 .

The apparatus 500 in FIG. 5 may include a receiving module 510 , a querying module 520 , a distributing module 530 and a filtering module 540 .

The receiving module 510 is configured to receive a database query statement, where the database query statement includes a main query and a subquery, the query conditions of the subquery are related conditions of the main query, and the main query includes a query based on the subquery The result is a filter condition for filtering, and the partition table data of the main query and the subquery are distributed on multiple servers.

The query module 520 is configured to query target data from the partition table data of the main query, where the target data includes first data related to the main query and second data related to the sub-query.

The distribution module 530 is configured to distribute the first data and the second data to a target server, where the target server can obtain all partition table data of the sub-query.

The query module 520 is further configured to: based on the second data, execute the query on the partition table data of the sub-query in parallel to obtain the first query result.

The filtering module 540 is configured to perform the filtering operation on the first data in parallel by using the filtering condition according to the first query result.

Optionally, the distribution module is further configured to distribute the first data and the second data to the target server by adaptively selecting a distribution manner of hash or random.

Optionally, the apparatus 500 further includes: a collection module, configured to collect the distribution situation of the first data and the second data; a judgment module, used to determine whether to use hash or random according to the distribution situation The distribution method distributes the first data and the second data to the target server.

Optionally, the target server obtains all partition table data of the subquery through the DISTRIBUTED TABLE SCAN operator.

FIG. 6 is a schematic structural diagram of an apparatus for processing a database query statement provided by another embodiment of the present disclosure. The apparatus 600 shown in FIG. 6 may be a database or a server. Apparatus 600 may include memory 610 and processor 620 .

Memory 610 may be used to store executable code. The processor 620 may be used to execute executable code stored in the memory 610 to implement the steps in the various methods described above.

In some embodiments, the apparatus 600 may further include a network interface 630 through which data exchange between the processor 620 and the external device may be implemented.

In the above embodiments, it may be implemented in whole or in part by software, hardware, firmware or any other combination. When implemented in software, it can be implemented in whole or in part in the form of a computer program product. The computer program product includes one or more computer instructions. When the computer program instructions are loaded and executed on a computer, the procedures or functions described in accordance with the embodiments of the present disclosure are produced in whole or in part. The computer may be a general purpose computer, special purpose computer, computer network, or other programmable device. The computer instructions may be stored in or transmitted from one computer readable storage medium to another computer readable storage medium, for example, the computer instructions may be downloaded from a website site, computer, server or data center The transmission is carried out to another website site, computer, server or data center by wire (eg coaxial cable, optical fiber, Digital Subscriber Line, DSL) or wireless (eg infrared, wireless, microwave, etc.). The computer-readable storage medium may be any available medium that can be accessed by a computer or a data storage device such as a server, data center, etc. that includes an integration of one or more available media. The usable media may be magnetic media (eg, floppy disks, hard disks, magnetic tapes), optical media (eg, Digital Video Disc (DVD)), or semiconductor media (eg, Solid State Disk (SSD)), etc. .

Those skilled in the art can realize that the units and algorithm steps of each example described in conjunction with the embodiments of the present disclosure can be implemented by electronic hardware, or a combination of computer software and electronic hardware. Whether these functions are performed in hardware or software depends on the specific application and design constraints of the technical solution. Skilled artisans may implement the described functionality using different methods for each particular application, but such implementations should not be considered beyond the scope of this disclosure.

In the several embodiments provided by the present disclosure, it should be understood that the disclosed system, apparatus and method may be implemented in other manners. For example, the apparatus embodiments described above are only illustrative. For example, the division of the units is only a logical function division. In actual implementation, there may be other division methods. For example, multiple units or components may be combined or Can be integrated into another system, or some features can be ignored, or not implemented. On the other hand, the shown or discussed mutual coupling or direct coupling or communication connection may be through some interfaces, indirect coupling or communication connection of devices or units, and may be in electrical, mechanical or other forms.

The units described as separate components may or may not be physically separated, and components displayed as units may or may not be physical units, that is, may be located in one place, or may be distributed to multiple network units. Some or all of the units may be selected according to actual needs to achieve the purpose of the solution in this embodiment.

In addition, each functional unit in each embodiment of the present disclosure may be integrated into one processing unit, or each unit may exist physically alone, or two or more units may be integrated into one unit.

The above are only specific embodiments of the present disclosure, but the protection scope of the present disclosure is not limited to this. should be included within the scope of protection of the present disclosure. Therefore, the protection scope of the present disclosure should be based on the protection scope of the claims.

Claims (9)

1. A method of processing a database query statement, comprising:

receiving a database query statement, wherein the database query statement comprises a main query and a sub query, the query condition of the sub query is the relevant condition of the main query, the main query comprises a filtering condition for filtering based on the query result of the sub query, and the partition table data of the main query and the sub query are distributed on a plurality of servers;

querying target data from partition table data of the main query, wherein the target data comprises first data relevant to the main query and second data relevant to the sub-queries;

the first data and the second data are distributed to a target server, and the target server can obtain all partition table data of the sub-query;

based on the second data, executing the query of the partition table data of the sub-query in parallel to obtain a first query result;

and according to the first query result, performing the filtering operation on the first data in parallel by using the filtering condition.

2. The method of claim 1, the distributing the first data and the second data to a target server, comprising:

and distributing the first data and the second data to the target server by adaptively selecting a hash or random distribution mode.

3. The method of claim 1, further comprising:

collecting the distribution condition of the first data and the second data;

and according to the distribution condition, judging to adopt a hash or random distribution mode to distribute the first data and the second data to the target server.

4. The method of claim 1, wherein the target server is capable of obtaining all partition table data of the sub-query, comprising:

and the target server acquires all partition TABLE data of the sub-query through a DISTRIBUTED TABLE SCAN operator.

5. An apparatus for processing a database query statement, comprising:

the system comprises a receiving module, a processing module and a processing module, wherein the receiving module is used for receiving a database query statement, the database query statement comprises a main query and a sub query, the query condition of the sub query is the relevant condition of the main query, the main query comprises a filtering condition for filtering based on the query result of the sub query, and the partition table data of the main query and the sub query are distributed on a plurality of servers;

the query module is used for querying target data from the partition table data of the main query, wherein the target data comprises first data relevant to the main query and second data relevant to the sub-queries;

the distribution module is used for distributing the first data and the second data to a target server, and the target server can obtain all the partition table data of the sub-query;

the query module is further configured to: based on the second data, executing the query of the partition table data of the sub-query in parallel to obtain a first query result;

and the filtering module is used for executing the filtering operation on the first data in parallel by using the filtering condition according to the first query result.

6. The apparatus of claim 5, wherein the distribution module is further configured to distribute the first data and the second data to the target server by adaptively selecting a distribution manner of hash or random.

7. The apparatus of claim 5, the apparatus further comprising:

the acquisition module is used for acquiring the distribution conditions of the first data and the second data;

and the judging module is used for judging that a hash or random distribution mode is adopted to distribute the first data and the second data to the target server according to the distribution condition.

8. The apparatus of claim 5, the target server to obtain all partition table data of the sub-query through a DISTRIBUTEDTABLE SCAN operator.

9. A computer readable storage medium having stored thereon executable code which, when executed, is capable of implementing the method of any one of claims 1 to 4.

Images