Adaptive distribution method for hash operations
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 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;
separating the equality join operation into a build operation and a probe operation;
determining, by a processing device, an actual size of the second relation;
determining, by the processing device, 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;
in response to determining to distribute at least one of the first and second relations using the broadcast join, setting, by the processing device, a 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, the at least one probe operator to join the first relation and the second relation based on the join condition.
2. The method of claim 1 , wherein determining to distribute at least one of the first and second relations using the broadcast join is further based on a cost metric associated with the broadcast join of the first relation and the second relation.
3. The method of claim 2 , wherein the cost metric is a computational cost or a memory cost.
4. The method of claim 2 , wherein determining to distribute at least one of the first and second relations using the broadcast join based on the cost metric comprises:
determining that the broadcast join is a most efficient type of join operation based on the cost metric.
5. The method of claim 1 , further comprising distributing the second relation using the broadcast join, wherein using the broadcast join comprises assigning at least part of the second relation to two or more processing nodes associated with the first relation.
6. The method of claim 5 , 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 the processing device forwards a local partition of the first relation to a local instance of a probe operator of the at least one probe operator for performing the join query.
7. The method of claim 6 , 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.
8. The method of claim 1 , wherein determining the actual size of the second relation is performed prior to performing the broadcast join.
9. The method of claim 1 , wherein determining the actual size of the second relation comprises:
performing the build operation on the second relation to partition the second relation; and
determining, based on the build operation, the actual size of the second relation.
10. The method of claim 1 , wherein the communication link in the synchronous state performs a local transfer of the first relation to a local instance of a probe operator of the at least one probe operator to perform the join query.
11. A system comprising:
a memory to store a plurality of relations; and
a processor, operatively coupled to the memory, the processor to:
receive a 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;
separate the equality join operation into a build operation and a probe operation;
determine an actual size of the second relation;
determine 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;
in response to determining to distribute at least one of the first and second relations using the broadcast join, set a 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, the at least one probe operator to join the first relation and the second relation based on the join condition.
12. The system of claim 11 , wherein to determine to distribute at least one of the first and second relations using the broadcast join is further based on a cost metric associated with the broadcast join of the first relation and the second relation.
13. The system of claim 12 , wherein the cost metric is a computational cost or a memory cost.
14. The system of claim 12 , wherein to determine to distribute at least one of the first and second relations using the broadcast join based on the cost metric, the processor is to:
determine that the broadcast join is a most efficient type of join operation based on the cost metric.
15. The system of claim 11 , wherein the processor is further to distribute the second relation using the broadcast join, wherein to use the broadcast join the processor is to assign at least part of the second relation to two or more processing nodes associated with the first relation.
16. The system of claim 15 , 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 of the at least one probe operator for performing the join query.
17. The system of claim 16 , 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.
18. The system of claim 11 , wherein the processor determines the actual size of the second relation prior to performing the broadcast join.
19. The system of claim 11 , wherein to determine the actual size of the second relation the processor is to:
perform the build operation on the second relation to partition the second relation; and
determine, based on the build operation, the actual size of the second relation.
20. The system of claim 11 , wherein the communication link in the synchronous state performs a local transfer of the first relation to a local instance of a probe operator of the at least one probe operator to perform the join query.
21. A non-transitory computer readable medium having instructions stored thereon that, when executed by a processor, cause the processor to:
receive a 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;
separate the equality join operation into a build operation and a probe operation;
determine, by the processor, an actual size of the second relation;
determine, by the processor, 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;
in response to determining to distribute at least one of the first and second relations using the broadcast join, set, by the processor, a 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, the at least one probe operator to join the first relation and the second relation based on the join condition.
22. The non-transitory computer readable medium of claim 21 , wherein to determine to distribute at least one of the first and second relations using the broadcast join is further based on a cost metric associated with the broadcast join of the first relation and the second relation.
23. The non-transitory computer readable medium of claim 22 , wherein the cost metric is a computational cost or a memory cost.
24. The non-transitory computer readable medium of claim 22 , wherein to determine to distribute at least one of the first and second relations using the broadcast join based on the cost metric the processor is to:
determine that the broadcast join is a most efficient type of join operation based on the cost metric.
25. The non-transitory computer readable medium of claim 21 , wherein the processor is further to distribute the second relation using the broadcast join, wherein to use the broadcast join the processor is to assign at least part of the second relation to two or more processing nodes associated with the first relation.
26. The non-transitory computer readable medium of claim 25 , 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 of the at least one probe operator for performing the join query.
27. The non-transitory computer readable medium of claim 26 , 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.
28. The non-transitory computer readable medium of claim 21 , wherein the processor determines the actual size of the second relation prior to performing the broadcast join.
29. The non-transitory computer readable medium of claim 21 , wherein to determine the actual size of the second relation the processor is to:
perform the build operation on the second relation to partition the second relation; and
determine, based on the build operation, the actual size of the second relation.
30. The non-transitory computer readable medium of claim 21 , wherein the communication link in the synchronous state performs a local transfer of the first relation to a local instance of a probe operator of the at least one probe operator to perform the join query.