SYSTEM, METHOD, AND COMPUTER-READABLE MEDIUM FOR OPTIMIZATION OF MULTIPLE PARALLEL JOIN OPERATIONS ON SKEWED DATA

• US 20100088298A1
• Filed: 10/06/2008
• Published: 04/08/2010
• Est. Priority Date: 10/06/2008
• Status: Active Grant

• Current Assignee: Teradata Corporation
• Original Assignee: Teradata Corporation
• Inventors: Xu, Yu
• Granted Patent: US 8,195,644 B2
• Time in Patent Office: Days
• US Class Current: 707/705
• CPC Class Codes: G06F 16/2456 Join operations

Abstract

A system, method, and computer-readable medium that facilitate management of data skew during a parallel multiple join operation are provided. Portions of tables involved in the join operation are distributed among a plurality of processing modules, and each of the processing modules is provided with a list of skewed values of a join column of a larger table involved in the join operation. Each of the processing modules scans the rows of first and second tables distributed to the processing modules and compares values of the join columns of both tables with the list of skewed values. Rows of a larger table having non-skewed values in the join column are redistributed, and rows of the larger table having skewed values in the join column are maintained locally at the processing modules. Rows of the smaller table that have non-skewed values in the join column are redistributed, and rows of the smaller table that have skewed values in the join column are duplicated among the processing modules. Rows of a third table involved in the join operation are redistributed based on the join attribute value of the rows. A local join data set is then generated by each of the processing modules, and the parallel join is completed by merging each of the processing module'"'"'s local join data set.

Claims

1. A method of performing a multiple join operation in a parallel processing system, comprising:

distributing a respective set of rows of a first table, a second table, and a third table involved in a join operation to each of a plurality of processing modules, wherein the join operation comprises a join on a column of the first table, a column of the second table, and a column of the third table;

receiving, by each of the plurality of processing modules, a list of skewed values of the column of the first table;

redistributing, by each of the plurality of processing modules, any row of the respective set of rows that has a value of the column of the first table that does not match any of the skewed values, any row of the respective set of rows that has a value of the column of the second table that does not match any of the skewed values, and any row of the respective set of rows of the third table;

locally maintaining, by a first processing module of the plurality of processing modules, any row of the respective set of rows that has a value of the column of the first table that matches any of the skewed values; and

duplicating, by a second processing module of the plurality of processing modules, any row of the respective set of rows that has a value of the column of the second table that matches any of the skewed values.

2. The method of claim 1, wherein redistributing any row of the respective set of rows further comprises:

generating a hash value of the column of the first table for a row of the first table; and

redistributing the row to one of the plurality of processing modules based on the hash value.

3. The method of claim 1, wherein redistributing any row of the respective set of rows further comprises:

generating a hash value of the column of the second table for a row of the second table;

redistributing the row of the second table to one of the plurality of processing modules based on the hash value;

generating a hash value of the column of the third table for a row of the third table; and

redistributing the row of the third table to one of the plurality of processing modules based on the hash value.

4. The method of claim 1, wherein locally maintaining any row comprises maintaining any row of the respective set of rows that has a value of the column of the first table that matches any of the skewed values in a spool allocated to the first processing module for rows of the respective set that are not redistributed.

5. The method of claim 1, wherein duplicating a row of the respective set of rows comprises:

transmitting all rows of the respective set of rows that have a value of the column of the second table that matches any of the skewed values to each of the plurality of processing modules; and

storing, by each of the plurality of processing modules, all the transmitted rows to a respective spool allocated for storing duplicated rows.

6. The method of claim 1, further comprising:

allocating a respective first spool to each of the plurality of processing modules that receives a redistributed row of the first table;

storing, by each of the plurality of processing modules that receives a redistributed row of the first table, redistributed rows of the first table in the respective first spool;

allocating a respective second spool to each of the plurality of processing modules that receives a redistributed row of the second table;

storing, by each of the plurality of processing modules that receives a redistributed row of the second table, redistributed rows of the second table in the respective second spool;

storing, by each of the plurality of processing modules that receives a duplicated row, each duplicated row in a respective third spool allocated for storage of duplicated rows;

storing, by each processing module of the plurality of processing modules that locally maintains any row of the respective set of rows that has a value of the column of the first table that matches any of the skewed values, any rows locally maintained by the processing module in a respective fourth spool allocated for storage of rows of the first table that are locally maintained;

allocating a respective fifth spool to each of the plurality of processing modules that receives a redistributed row of the third table; and

storing, by each of the plurality of processing modules that receives a redistributed row of the third table, redistributed rows of the third table in the respective fifth spool.

7. The method of claim 6, further comprising:

generating, by each of the plurality of processing modules, a respective first data set by joining rows of the respective first spool with rows of the respective second spool;

generating, by each of the plurality of processing modules, a respective second data set by joining rows of the respective third spool with rows of the respective fourth spool;

redistributing, by each of the plurality of processing modules, rows of the respective second data set to one of the plurality of processing modules based on the column of the first table;

storing, by each of the plurality of processing modules, any redistributed row of a second data set received by the processing module in a sixth spool; and

generating, by each of the plurality of processing modules, a respective local join data set by joining rows of the fifth spool with a union of the respective first data set and the respective sixth spool.

8. The method of claim 7, further comprising completing the join operation by generating a union of the respective local join data set of each of the plurality of processing modules.

9. A computer-readable medium having computer-executable instructions for execution by a processing system, the computer-executable instructions for performing a multiple join operation in a parallel processing system, the computer-executable instructions, when executed, cause the processing system to:

distribute a respective set of rows of a first table, a second table, and a third table involved in a join operation to each of a plurality of processing modules, wherein the join operation comprises a join on a column of the first table, a column of the second table, and a column of the third table;

receive, by each of the plurality of processing modules, a list of skewed values of the column of the first table;

redistribute, by each of the plurality of processing modules, any row of the respective set of rows that has a value of the column of the first table that does not match any of the skewed values, any row of the respective set of rows that has a value of the column of the second table that does not match any of the skewed values, and any row of the respective set of rows of the third table;

locally maintain, by a first processing module of the plurality of processing modules, any row of the respective set of rows that has a value of the column of the first table that matches any of the skewed values; and

duplicate, by a second processing module of the plurality of processing modules, any row of the respective set of rows that has a value of the column of the second table that matches any of the skewed values.

10. The computer-readable medium of claim 9, wherein the instructions that redistribute any row of the respective set of rows further comprise instructions that, when executed, cause the processing system to:

generate a hash value of the column of the first table for a row of the first table; and

redistribute the row to one of the plurality of processing modules based on the hash value.

11. The computer-readable medium of claim 9, wherein the instructions that redistribute any row of the respective set of rows further comprise instructions that, when executed, cause the processing system to:

generate a hash value of the column of the second table for a row of the second table;

redistribute the row of the second table to one of the plurality of processing modules based on the hash value;

generate a hash value of the column of the third table for a row of the third table; and

redistribute the row of the third table to one of the plurality of processing modules based on the hash value.

12. The computer-readable medium of claim 9, wherein the instructions that locally maintain any row comprise instructions that, when executed, cause the processing system to maintain any row of the respective set of rows that has a value of the column of the first table that matches any of the skewed values in a spool allocated to the first processing module for rows of the respective set that are not redistributed.

13. The computer-readable medium of claim 9, wherein the instructions that duplicate a row of the respective set of rows comprise instructions that, when executed, cause the processing system to:

transmit all rows of the respective set of rows that have a value of the column of the second table that matches any of the skewed values to each of the plurality of processing modules; and

store, by each of the plurality of processing modules, all the transmitted rows to a respective spool allocated for storing duplicated rows.

14. The computer-readable medium of claim 9, further comprising instructions that, when executed, cause the processing system to:

allocate a respective first spool to each of the plurality of processing modules that receives a redistributed row of the first table;

store, by each of the plurality of processing modules that receives a redistributed row of the first table, redistributed rows of the first table in the respective first spool;

allocate a respective second spool to each of the plurality of processing modules that receives a redistributed row of the second table;

store, by each of the plurality of processing modules that receives a redistributed row of the second table, redistributed rows of the second table in the respective second spool;

store, by each of the plurality of processing modules that receives a duplicated row, each duplicated row in a respective third spool allocated for storage of duplicated rows;

store, by each processing module of the plurality of processing modules that locally maintains any row of the respective set of rows that has a value of the column of the first table that matches any of the skewed values, any rows locally maintained by the processing module in a respective fourth spool allocated for storage of rows of the first table that are locally maintained;

allocate a respective fifth spool to each of the plurality of processing modules that receives a redistributed row of the third table; and

store, by each of the plurality of processing modules that receives a redistributed row of the third table, redistributed rows of the third table in the respective fifth spool.

15. The computer-readable medium of claim 14, further comprising instructions that, when executed, cause the processing system to:

generate, by each of the plurality of processing modules, a respective first data set by joining rows of the respective first spool with rows of the respective second spool;

generate, by each of the plurality of processing modules, a respective second data set by joining rows of the respective third spool with rows of the respective fourth spool;

redistribute, by each of the plurality of processing modules, rows of the respective second data set to one of the plurality of processing modules based on the column of the first table;

store, by each of the plurality of processing modules, any redistributed row of a second data set received by the processing module in a sixth spool; and

generate, by each of the plurality of processing modules, a respective local join data set by joining rows of the fifth spool with a union of the respective first data set and the respective sixth spool.

16. The computer-readable medium of claim 15, further comprising instructions that, when executed, cause the processing system to complete the join operation by generating a union of the respective local join data set of each of the plurality of processing modules.

17. A database management system, comprising:

a plurality of processing modules;

at least one storage device to which a respective set of rows of a first table, a second table, and a third table involved in a multiple join operation are allocated to each of the plurality of processing modules, wherein the join operation comprises a join on a column of the first table, a column of the second table, and a column of the third table; and

an optimizer that distributes a list of skewed values of the column of the first table to each of the plurality of processing modules, wherein each processing module of the plurality of processing modules redistributes any row of the respective set of rows that has a value of the column of the first table that does not match any of the skewed values, any row of the respective set of rows that has a value of the column of the second table that does not match any of the skewed values, and any row of the respective set of rows of the third table, locally maintains any row of the respective set of rows that has a value of the column of the first table that matches any of the skewed values, and duplicates any row of the respective set of rows that has a value of the column of the second table that matches any of the skewed values.

18. The system of claim 17, wherein each of the plurality of processing modules that receives a redistributed row of the first table stores the redistributed rows of the first table in a respective first spool allocated to the processing module, stores any redistributed rows of the second table received by the processing module in a respective second spool allocated to the processing module, stores any duplicated rows received by the processing module in a respective third spool allocated to the processing module, stores any rows locally maintained by the processing module in a respective fourth spool allocated to the processing module, and stores any redistributed rows of the third table in a respective fifth spool allocated to the processing module.

19. The system of claim 18, wherein each of the plurality of processing modules generates a respective first data set by joining rows of the respective first spool with rows of the respective second spool, generates a respective second data set by joining rows of the respective third spool with rows of the respective fourth spool, redistributes rows of the respective second data set to one of the plurality of processing modules based on the column of the first table, stores any redistributed row of a second data set received by the processing module in a sixth spool, and generates a respective local join data set by joining rows of the fifth spool with a union of the respective first data set and the respective sixth spool.

20. The system of claim 19, wherein the join operation is completed by generating a union of the respective local join data set of each of the plurality of processing modules.