Adaptive distribution method for hash operations

1. A method of performing a join operation, comprising:

receiving a relational join query comprising a join operation and an indication of a first relation and a second relation to be joined, wherein the first relation and the second relation are partitioned over processing nodes of a cluster;

setting a first communication link between the first relation and a plurality of build operators of a build operation to a partition state, wherein the partition state facilitates a partition move of the first relation to the plurality of build operators;

building, by the plurality of build operators, a hash table from the first relation in accordance with the build operation; and

determining, by a processing device of at least one of the plurality of build operators after completion of the build operation and prior to starting distribution of the first or second relation to a plurality of probe operators of a probe operation, whether to distribute the first relation to the probe operation using a broadcast join or to distribute the second relation to the probe operation using a re-partitioning join, wherein the determining is based at least in part on an estimated size of the second relation and a cost metric.

2. The method of claim 1 , wherein the determining is further based on an actual size of the first relation.

3. The method of claim 2 , wherein the actual size of the first relation is determined during execution of the join operation.

4. The method of claim 2 , further comprising building a hash index for the first relation, wherein the actual size of the first relation is determined based on the building of the hash index.

5. The method of claim 2 , wherein the first relation comprises tuples, the method further comprising:

during the partition move of the first relation to the plurality of build operators, forwarding the tuples to the plurality of build operators in accordance with a partitioning function; and

determining a total number of tuples processed for the partition move;

wherein the actual size of the first relation is based on the total number of tuples processed for the partition move.

6. The method of claim 1 , further comprising, upon determining to distribute the first relation to the probe operation using the broadcast join:

converting links between the plurality of build operators and the plurality of probe operators into broadcast links;

converting links between the second relation and the plurality of probe operators to synchronous links; and

sending the first relation through the broadcast links so that each partition of the first relation is broadcasted to every one of the plurality of probe operators.

7. The method of claim 1 , further comprising, upon determining to distribute the second relation to the probe operation using the re-partitioning join:

converting links between the plurality of build operators and the plurality of probe operators into synchronous links;

converting links between the second relation and the plurality of probe operators to partition links; and

sending the second relation through the partition links to the plurality of probe operators.

8. A system, comprising:

a memory to store a plurality of relations; and

a processing device, operatively coupled with the memory, the processing device to:

receive a relational join query comprising a join operation and an indication of a first relation and a second relation to be joined, wherein the first relation and the second relation are partitioned over processing nodes of a cluster;

set a first communication link between the first relation and a plurality of build operators of a build operation to a partition state, wherein the partition state facilitates a partition move of the first relation to the plurality of build operators, wherein the build operation is to build a hash table from the first relation; and

determine, after completion of the build operation and prior to starting distribution of the first or second relation to a plurality of probe operators of a probe operation, whether to distribute the first relation to the probe operation using a broadcast join or to distribute the second relation to the probe operation using a re-partitioning join, wherein the determination is based at least in part on an estimated size of the second relation and a cost metric.

9. The system of claim 8 , wherein the determination is further based on an actual size of the first relation.

10. The system of claim 9 , wherein the actual size of the first relation is determined during execution of the join operation.

11. The system of claim 9 , wherein the processing device is further to build a hash index for the first relation, wherein the actual size of the first relation is determined based on the build of the hash index.

12. The system of claim 9 , wherein the first relation comprises tuples, and the processing device is further to:

during the partition move of the first relation to the plurality of build operators, forward the tuples to the plurality of build operators in accordance with a partitioning function; and

determine a total number of tuples processed for the partition move;

wherein the actual size of the first relation is based on the total number of tuples processed for the partition move.

13. The system of claim 8 , wherein if the processing device determines to distribute the first relation to the probe operation using the broadcast join, the processing device is further to:

convert links between the plurality of build operators and the plurality of probe operators into broadcast links;

convert links between the second relation and the plurality of probe operators to synchronous links; and

send the first relation through the broadcast links so that each partition of the first relation is broadcasted to every one of the plurality of probe operators.

14. The system of claim 8 , wherein if the processing device determines to distribute the second relation to the probe operation using the re-partitioning join, the processing device is further to:

convert links between the plurality of build operators and the plurality of probe operators into synchronous links;

convert links between the second relation and the plurality of probe operators to partition links; and

send the second relation through the partition links to the plurality of probe operators.

15. A non-transitory computer readable medium having instructions stored thereon that, when executed by a processing device, cause the processing device to:

receive a relational join query comprising a join operation and an indication of a first relation and a second relation to be joined, wherein the first relation and the second relation are partitioned over processing nodes of a cluster;

set a first communication link between the first relation and a plurality of build operators of a build operation to a partition state, wherein the partition state facilitates a partition move of the first relation to the plurality of build operators;

build, by the plurality of build operators, a hash table from the first relation in accordance with the build operation; and

determine, after completion of the build operation and prior to starting distribution of the first or second relation to a plurality of probe operators of a probe operation, whether to distribute the first relation to the probe operation using a broadcast join or to distribute the second relation to the probe operation using a re-partitioning join, wherein the determination is based at least in part on an estimated size of the second relation and a cost metric.

16. The non-transitory computer readable medium of claim 15 , wherein the determination is further based on an actual size of the first relation.

17. The non-transitory computer readable medium of claim 16 , wherein the processing device is further to build a hash index for the first relation, wherein the actual size of the first relation is determined based on the build of the hash index.

18. The non-transitory computer readable medium of claim 16 , wherein the first relation comprises tuples, and the processing device is further to:

during the partition move of the first relation to the plurality of build operators, forward the tuples to the plurality of build operators in accordance with a partitioning function; and

determine a total number of tuples processed for the partition move;

wherein the actual size of the first relation is based on the total number of tuples processed for the partition move.

19. The non-transitory computer readable medium of claim 15 , wherein if the processing device determines to distribute the first relation to the probe operation using the broadcast join, the processing device is further to:

convert links between the plurality of build operators and the plurality of probe operators into broadcast links;

convert links between the second relation and the plurality of probe operators to synchronous links; and

send the first relation through the broadcast links so that each partition of the first relation is broadcasted to every one of the plurality of probe operators.

20. The non-transitory computer readable medium of claim 15 , wherein if the processing device determines to distribute the second relation to the probe operation using the re-partitioning join, the processing device is further to:

convert links between the plurality of build operators and the plurality of probe operators into synchronous links;

convert links between the second relation and the plurality of probe operators to partition links; and

send the second relation through the partition links to the plurality of probe operators.