Adaptive distribution method for hash operation
A method, apparatus, and system for join operations of a plurality of relations that are distributed over a plurality of storage locations over a network of computing components.
1. A method of performing an implementation of a join operation, comprising:
receiving a relational join query comprising a join condition and an indication of a first relation and a second relation to be joined, wherein the join query is for an equality join operation and the first relation and the second relation are partitioned over processing nodes of a cluster;
determining an actual size of the second relation;
determining whether to distribute at least one of the first and second relations using a broadcast join based at least in part on the actual size of the second relation and a cost metric, wherein determining whether to distribute the first and second relations takes place after the actual size of the second relation has been determined; and
in response to determining to distribute at least one of the first and second relations using a broadcast join:
setting a first communication link between the first relation and at least one probe operator of the join query to a synchronous state; and
duplicating the second relation to the at least one probe operator.
2. The method of claim 1 wherein the cost metric is a memory cost.
3. The method of claim 1 wherein the cost metric is a computational cost.
4. The method of claim 1 , further comprising distributing the second relation using the broadcasting join, wherein using the broadcasting join comprises assigning at least part of the second relation to two or more processing nodes associated with the first relation.
5. The method of claim 4 , wherein distributing the second relation comprises duplicating the second relation across the two or more processing nodes associated with the first relation, and wherein during the duplication of the second relation, a local partition of the first relation is forwarded to a local instance of a probe operator for performing the join query.
6. The method of claim 5 , wherein duplicating the second relation comprises broadcasting each partition of the second relation to every instance of the probe operator across the two or more processing nodes associated with the first relation.
7. The method of claim 1 , further comprising:
determining, based at least in part on the actual size of the second relation, to distribute the second relation using a re-partitioning join; and
distributing the second relation using the re-partitioning join, wherein using the re-partitioning join comprises assigning at least part of the second relation to a processing node based on a key associated with the join condition.
8. The method of claim 7 , wherein distributing the second relation comprises partitioning the second relation across processing nodes of a cluster, and the re-partitioning join is performed by each of the processing nodes of the cluster.
9. The method of claim 7 , wherein each pair of tuples from the first and second relations that satisfy the join condition is assigned to the same processing node.
10. The method of claim 1 , wherein a partitioning function previously applied to the second relation is performed during or before building a hash index, and wherein partitions of the second relation are not read back at a time of partitioning the first relation.
11. A system, comprising:
a memory to store a plurality of relations; and
a processor, operatively coupled with the memory, the processor configured to:
receive a relational join query comprising a join condition and an indication of a first relation and a second relation to be joined, wherein the join query is for an equality join operation and the first relation and the second relation are partitioned over processing nodes of a cluster;
determine an actual size of the second relation;
determine whether to distribute at least one of the first and second relations using a broadcast join based at least in part on the actual size of the second relation and a cost metric, wherein determining whether to distribute the first and second relations takes place after the actual size of the second relation has been determined; and
in response to determining to distribute at least one of the first and second relations using a broadcast join:
set a first communication link between the first relation and at least one probe operator of the join query to a synchronous state; and
duplicate the second relation to the at least one probe operator.
12. The system of claim 11 wherein the cost metric is a memory cost.
13. The system of claim 11 wherein the cost metric is a computational cost.
14. The system of claim 11 , wherein the processor is further to distribute the second relation using the broadcasting join, wherein to use the broadcasting join the processor is to assign at least part of the second relation to two or more processing nodes associated with the first relation.
15. The system of claim 14 , wherein to distribute the second relation the processor is to duplicate the second relation across the two or more processing nodes associated with the first relation, and wherein during the duplication of the second relation, the processor is to forward a local partition of the first relation to a local instance of a probe operator for performing the join query.
16. The system of claim 15 , wherein to duplicate the second relation the processor is to broadcast each partition of the second relation to every instance of the probe operator across the two or more processing nodes associated with the first relation.
17. The system of claim 11 , wherein the processor is further to:
determine, based at least in part on the actual size of the second relation, to distribute the second relation using a re-partitioning join; and
distribute the second relation using the re-partitioning join, wherein to use the re-partitioning join the processor is to assign at least part of the second relation to a processing node based on a key associated with the join condition.
18. The system of claim 17 , wherein to distribute the second relation the processor is to partition the second relation across processing nodes of a cluster, and the re-partitioning join is performed by each of the processing nodes of the cluster.
19. The system of claim 17 , wherein the processor assigns each pair of tuples from the first and second relations that satisfy the join condition to the same processing node.
20. The system of claim 11 , wherein the processor performs a partitioning function previously applied to the second relation during or before building a hash index, and wherein partitions of the second relation are not read back at a time of partitioning the first relation.
21. A non-transitory computer readable medium having instructions stored thereon that, when executed by a processor, cause the processor to:
receive a relational join query comprising a join condition and an indication of a first relation and a second relation to be joined, wherein the join query is for an equality join operation and the first relation and the second relation are partitioned over processing nodes of a cluster;
determine an actual size of the second relation;
determine whether to distribute at least one of the first and second relations using a broadcast join based at least in part on the actual size of the second relation and a cost metric, wherein determining whether to distribute the first and second relations takes place after the actual size of the second relation has been determined; and
in response to determining to distribute at least one of the first and second relations using a broadcast join:
set a first communication link between the first relation and at least one probe operator of the join query to a synchronous state; and
duplicate the second relation to the at least one probe operator.
22. The non-transitory computer readable medium of claim 21 wherein the cost metric is a memory cost.
23. The non-transitory computer readable medium of claim 21 wherein the cost metric is a computational cost.
24. The non-transitory computer readable medium of claim 21 , wherein the processor is further to distribute the second relation using the broadcasting join, wherein to use the broadcasting join the processor is to assign at least part of the second relation to two or more processing nodes associated with the first relation.
25. The non-transitory computer readable medium of claim 24 , wherein to distribute the second relation the processor is to duplicate the second relation across the two or more processing nodes associated with the first relation, and wherein during the duplication of the second relation, the processor is to forward a local partition of the first relation to a local instance of a probe operator for performing the join query.
26. The non-transitory computer readable medium of claim 25 , wherein to duplicate the second relation the processor is to broadcast each partition of the second relation to every instance of the probe operator across the two or more processing nodes associated with the first relation.
27. The non-transitory computer readable medium of claim 21 , wherein the processor is further to:
determine, based at least in part on the actual size of the second relation, to distribute the second relation using a re-partitioning join; and
distribute the second relation using the re-partitioning join, wherein to use the re-partitioning join the processor is to assign at least part of the second relation to a processing node based on a key associated with the join condition.
28. The non-transitory computer readable medium of claim 27 , wherein to distribute the second relation the processor is to partition the second relation across processing nodes of a cluster, and the re-partitioning join is performed by each of the processing nodes of the cluster.
29. The non-transitory computer readable medium of claim 27 , wherein the processor assigns each pair of tuples from the first and second relations that satisfy the join condition to the same processing node.
30. The non-transitory computer readable medium of claim 21 , wherein the processor performs a partitioning function previously applied to the second relation during or before building a hash index, and wherein partitions of the second relation are not read back at a time of partitioning the first relation.