Adaptive handling of skew for distributed joins in a cluster

Abstract

Techniques for detecting data skew while performing a distributed join operation on tables in a cluster of nodes managed by database management system (cDBMS), is disclosed. In an embodiment, heavy hitter values in a join column of a table are determined during the runtime of a distributed join operation of the table with another table. The cDBMS keeps in a datastore a count for each unique value read from the join column of the table. The datastore may be a hash table with the unique values serving as keys and may additionally include a heap or a sorted array for an efficient count based traversal. When a count for a particular value in the datastore exceeds a threshold, then the particular value is identified as a heavy hitter value. The tuples from the joined table that include the heavy hitter value, are kept local at the node that the tuples were originally distributed to, while the other joined table tuples are broadcasted to one or more nodes of the cDBMS that at least include the originally distributed nodes.

What is claimed is: 
     
       1. A method for executing a join operation by determining a distribution of tuples, for the join operation, from a database among a cluster of nodes and that are coupled to a database management system (DBMS), the DBMS managing the database, comprising:
 executing a particular join operation to join a first table with a second table based on a first join key of the first table and a second join key of the second table, the executing further comprising:
 said DBMS distributing first plurality of tuples of the first table across nodes of the cluster; 
 said DBMS distributing second plurality of tuples of the second table across nodes of the cluster; 
 said cluster of nodes generating a respective count for each second join key value of a subset of second join key values in said second join key at a receipt of said second plurality of tuples of the second table; 
 based on the respective count for said each second join key value of said subset, establishing a particular second join key value as a heavy hitter; 
 in response to determining that the particular second join key value is a heavy hitter,
 replicating a first tuple across a set of nodes of the cluster, wherein the first tuple contains a first join key value, from the first join key, that corresponds to said particular second join key value; and 
 each node of said set of nodes locally performing a join operation between plurality of tuples of said first table and said second table based on said first join key value and values of said second join key. 
 
 
 
     
     
       2. The method of  claim 1 , wherein generating a respective count of each second join key value of the subset of second join key values further comprises:
 incrementing the respective count of said each second join key value at a receipt by a node of said cluster of a tuple, from the second plurality of tuples, that includes said each second join key value in the second join key of the second table; 
 updating in a datastore said respective count with the incremented respective count of said each second join key value. 
 
     
     
       3. The method of  claim 2 , wherein the datastore comprises a hash table and a sorted array, and wherein each second join key value of the subset is stored in the hash table as a key of the hash table, and a respective count, of said each second join key value, is stored in an element of the sorted array. 
     
     
       4. The method of  claim 2 , wherein:
 the datastore comprises of a hash table and a heap, and wherein each second join key value of the subset is stored in the hash table as a key of the hash table, and 
 a respective count, of said each second join key value, is stored in a heap node of the heap, wherein a parent heap node of said heap node of the heap contains a count that is less than or equal to said respective count of said each second join key value and a child heap node of said heap node of the heap is greater than or equal to said respective count of said each second join key value. 
 
     
     
       5. The method of  claim 1 , further comprising in response to determining that the particular second join key value is a heavy hitter, replicating the first tuple across all nodes of the cluster. 
     
     
       6. The method of  claim 1 , further comprising in response to determining that the particular second join key value is not a heavy hitter, transferring to a node, that contains the first tuple, one or more tuples, of the second plurality of tuples, that include the particular second join key value in the second join key of the second table. 
     
     
       7. The method of  claim 1 , further comprising:
 determining that a datastore storing heavy hitters does not include the particular second join key value; 
 in response to determining that the datastore storing heavy hitters does not include the particular second join key value:
 storing the particular second join key value into the datastore; 
 associating the particular second join key value with an initial count in the datastore. 
 
 
     
     
       8. The method of  claim 1 , further comprising:
 determining that a datastore storing heavy hitters does not include the particular second join key value and that the datastore is full; 
 in response to determining that the datastore storing heavy hitters does not include the particular second join key value and that the datastore is full:
 retrieving from the datastore the lowest count in the datastore and a second join key value associated with the lowest count in the datastore; 
 replacing, in the datastore, the second join key value associated with the lowest count with the particular second join key value; 
 associating with the particular second join key value in the datastore a count equal at least to the lowest count in the datastore. 
 
 
     
     
       9. The method of  claim 8 , wherein the count associated with the second join key value is assigned to an increment of the lowest count in the datastore. 
     
     
       10. The method of  claim 1 , further comprising:
 generating the first join key value by applying a hashing algorithm to a value from the first join key of the first table; and 
 generating the second join key value by applying the hashing algorithm to a value from the second join key of the second table. 
 
     
     
       11. The method of  claim 1 , wherein the first join key is a primary key of the first table. 
     
     
       12. The method of  claim 1 , wherein said cluster of nodes generating a respective count further comprises storing in a datastore of each node, of said cluster, that receives a tuple with said each second join key value, the respective count of said each second join key value. 
     
     
       13. One or more non-transitory storage media storing instructions which, when executed by one or more computing devices, cause executing a join operation by determining a distribution of tuples, for the join operation, from a database among a cluster of nodes and that are coupled to a database management system (DBMS), the DBMS managing the database, the instructions further include instructions which, when executed by the one or more computing devices, cause:
 executing a particular join operation to join a first table with a second table based on a first join key of the first table and a second join key of the second table, wherein the instructions that cause executing the particular join operation further include instructions which, when executed by the one or more computing devices, cause:
 said DBMS distributing first plurality of tuples of the first table across nodes of the cluster; 
 said DBMS distributing second plurality of tuples of the second table across nodes of the cluster; 
 said cluster of nodes generating a respective count for each second join key value of a subset of second join key values in said second join key at a receipt of said second plurality of tuples of the second table; 
 based on the respective count for said each second join key value of said subset, establishing a particular second join key value as a heavy hitter; 
 in response to determining that the particular second join key value is a heavy hitter,
 replicating a first tuple across a set of nodes of the cluster, wherein the first tuple contains a first join key value, from the first join key, that corresponds to said particular second join key value; and 
 each node of said set of nodes locally performing a join operation between plurality of tuples of said first table and said second table based on said first join key value and values of said second join key. 
 
 
 
     
     
       14. The one or more non-transitory storage media of  claim 13 , wherein the instructions further include instructions which, when executed by the one or more computing devices, cause:
 incrementing a respective count of each second join key value, of the subset, at a receipt by a node of said cluster of a tuple, from the second plurality of tuples, that includes said each second join key value in the second join key of the second table; 
 updating in a datastore said respective count with the incremented respective count of said each second join key value. 
 
     
     
       15. The one or more non-transitory storage media of  claim 14 , wherein the datastore comprises a hash table and a sorted array, and wherein each second join key value of the subset is stored in the hash table as a key of the hash table, and a respective count, of said each second join key value, is stored in an element of the sorted array. 
     
     
       16. The one or more non-transitory storage media of  claim 14 , wherein:
 the datastore comprises of a hash table and a heap, and wherein each second join key value of the subset is stored in the hash table as a key of the hash table, and 
 a respective count, of said each second join key value, is stored in a heap node of the heap, wherein a parent heap node of said heap node of the heap contains a count that is less than or equal to said respective count of said each second join key value and a child heap node of said heap node of the heap is greater than or equal to said respective count of said each second join key value. 
 
     
     
       17. The one or more non-transitory storage media of  claim 13 , wherein the instructions further include instructions which, when executed by the one or more computing devices, cause, in response to determining that the particular second join key value is a heavy hitter, replicating the first tuple across all nodes of the cluster. 
     
     
       18. The one or more non-transitory storage media of  claim 13 , wherein the instructions further include instructions for causing, in response to determining that the particular second join key value is not a heavy hitter, transferring to a node, that contains the first tuple, one or more tuples, of the second plurality of tuples, that include the particular second join key value in the second join key of the second table. 
     
     
       19. The one or more non-transitory storage media of  claim 13 , wherein the instructions further include instructions which, when executed by the one or more computing devices, cause:
 determining that a datastore storing heavy hitters does not include the particular second join key value; 
 in response to determining that the datastore storing heavy hitters does not include the particular second join key value:
 storing the particular second join key value into the datastore; 
 associating the particular second join key value with an initial count in the datastore. 
 
 
     
     
       20. The one or more non-transitory storage media of  claim 13 , wherein the instructions further include instructions which, when executed by the one or more computing devices, cause:
 determining that a datastore storing heavy hitters does not include the particular second join key value and that the datastore is full; 
 in response to determining that the datastore storing heavy hitters does not include the particular second join key value and that the datastore is full:
 retrieving from the datastore the lowest count in the datastore and a second join key value associated with the lowest count in the datastore; 
 replacing, in the datastore, the second join key value associated with the lowest count with the particular second join key value; 
 associating with the particular second join key value in the datastore a count equal at least to the lowest count in the datastore. 
 
 
     
     
       21. The one or more non-transitory storage media of  claim 20 , wherein the count associated with the second join key value is assigned to an increment of the lowest count in the datastore. 
     
     
       22. The one or more non-transitory storage media of  claim 13 , wherein the instructions further include instructions which, when executed by the one or more computing devices, cause:
 generating the first join key value by applying a hashing algorithm to a value from the first join key of the first table; and 
 generating the second join key value by applying the hashing algorithm to a value from the second join key of the second table. 
 
     
     
       23. The one or more non-transitory storage media of  claim 13 , wherein the first join key is a primary key of the first table. 
     
     
       24. The one or more non-transitory storage media of  claim 13 , wherein the instructions further include instructions which, when executed by the one or more computing devices, cause storing in a datastore of each node, of said cluster, that receives a tuple with said each second join key value, the respective count of said each second join key value.