System and method for distributed SQL join processing in shared-nothing relational database clusters using stationary tables

Abstract

System and method embodiments are provided for improving the performance of query processing in a massively parallel processing (MPP) database system by selecting one or more stationary tables for processing the query plan. In an embodiment method, a join query associated with a plurality of tables of the MPP database system is received at a coordinator process. Stationary tables are then selected from the tables according to table partition keys in the join query. The stationary tables are indicated and the join query are sent to a plurality of data nodes communicating with the coordinator process. At each of the data nodes, an execution plan for the join query is generated. Each of the data nodes then executes the execution plan to obtain query results without sharing information of the stationary tables between the data nodes.

What is claimed is: 
     
       1. A method for query processing in a massively parallel processing (MPP) database system, the method comprising:
 receiving, at a coordinator process that runs on the MPP, a join query associated with a plurality of tables of the MPP database system; 
 designating, from the tables to a plurality of processing nodes of the MPP, stationary tables and non-stationary tables, wherein the join query is processed by passing at least some information stored in the non-stationary tables between the processing nodes without passing any information stored in the stationary tables between the processing nodes; and 
 executing, at each of the processing nodes, an execution plan to obtain query results, wherein executing the execution plan comprises passing at least some data stored in the non-stationary tables between the processing nodes without passing any information stored in the stationary tables between the processing nodes. 
 
     
     
       2. The method of  claim 1  further comprising generating, at each of the processing nodes, the execution plan for the join query taking into consideration the stationary tables. 
     
     
       3. The method of  claim 1  further comprising combining, at the coordinator process, the query results from the processing nodes to obtain a final result. 
     
     
       4. The method of  claim 1 , wherein the stationary tables are selected according to partition keys in the join query. 
     
     
       5. The method of  claim 4  further comprising:
 identifying groups of tables in the join query that join on respective partition keys; 
 calculating a combined weight of each group; and 
 designate tables in a group with highest weight as stationary. 
 
     
     
       6. The method of  claim 1  further comprising aggregating, at the coordinator process, the query results from each of the processing nodes. 
     
     
       7. The method of  claim 1 , wherein the execution plan comprises an organized task for handling the join query. 
     
     
       8. A method for query processing in a massively parallel processing (MPP) database system, the method comprising:
 receiving, at a first processing node of the MPP from a coordinator process that runs on the MPP, a join query associated with a plurality of tables of the MPP database system; 
 receiving an indication specifying that at least a first stationary data partition from stationary tables has been assigned to the first processing node, wherein the stationary tables are selected from the plurality of tables by the coordinator process according to table partition keys in the join query, wherein plurality of tables further includes non-stationary tables, and wherein the join query is configured to be processed by passing at least some data stored in the non-stationary tables between processing nodes of the MPP without passing any information stored in the stationary tables between the processing nodes; 
 generating an execution plan for the join query; and 
 executing, at the first processing node, the execution plan to obtain query results, wherein the first processing node executes the execution plan by sharing at least some information in the non-stationary tables with other processing nodes without sharing information of the first stationary data partition with other processing nodes of the MPP that are processing the same join query. 
 
     
     
       9. The method of  claim 8  further comprising:
 forwarding the query results from the first processing node to the coordinator process to combine, at the coordinator process, the query results. 
 
     
     
       10. The method of  claim 9 , wherein a plurality of exclusive partitions of the tables are allocated to the first processing node and the other processing nodes, wherein the exclusive partitions for different processing nodes do not comprise common data rows, and wherein the execution plan is executed to obtain query results on the exclusive partitions allocated for the first processing node. 
     
     
       11. The method of  claim 10 , wherein data at different exclusive partitions of the stationary tables at the different processing nodes are not moved between the different processing nodes. 
     
     
       12. The method of  claim 10 , wherein data at different exclusive partitions of the tables that are not stationary tables at the different processing nodes are moved between the different processing nodes to execute the execution plan at the processing node. 
     
     
       13. The method of  claim 12 , wherein the data is moved between the different processing nodes upon request from the first processing node when the data is needed to complete the execution plan for the join query and is not available locally. 
     
     
       14. The method of  claim 10 , wherein the same execution plan is executed at each of the processing nodes on the different exclusive partitions of the tables. 
     
     
       15. An apparatus for query processing in a massively parallel processing (MPP) database system, the apparatus comprising:
 a plurality of data nodes configured to process a join query on partitions of data tables of the MPP database system; 
 a processor; and 
 a computer readable storage medium storing programming for execution by the processor, the programming including instructions to:
 receive, at a coordinator process communicating with the data nodes, a join query associated with the tables; 
 designate stationary tables and non-stationary tables from the tables, wherein the join query is processed by passing at least some information stored in the non-stationary tables between the data nodes without passing any information stored in the stationary tables between the data nodes; 
 indicate the stationary tables and send the join query to the data nodes, wherein the data nodes execute execution plans to generate query results by sharing at least some data stored in the non-stationary tables with other data nodes without sharing any information stored in the stationary tables with other data nodes; and 
 combine the query results generated by the data nodes. 
 
 
     
     
       16. The apparatus of  claim 15 , wherein the data nodes are further configured to forward the query results to the coordinator process. 
     
     
       17. The apparatus of  claim 16 , wherein the data nodes generate the execution plan for the join query taking into consideration the stationary tables. 
     
     
       18. The apparatus of  claim 15 , wherein the programming includes further instructions to allocate a plurality of exclusive partitions of the tables to the data nodes, wherein the exclusive partitions for different data nodes do not comprise common data rows. 
     
     
       19. The apparatus of  claim 18 , wherein the data nodes execute the same execution plan on the different exclusive partitions to obtain query results. 
     
     
       20. The apparatus of  claim 19 , wherein the data nodes do not share data in the different exclusive partitions of the stationary tables with other data nodes. 
     
     
       21. The apparatus of  claim 19 , wherein the data nodes share data in the non-stationary tables upon request from the data nodes.