Adaptive distribution method for hash operations
A method, apparatus, and system including performing a partitioning function on a first table of a join query, determining a size of the first table based on the partitioning function, and distributing the first table to a set of nodes of a computing cluster based on the size of the first table.
1. A method comprising:
performing a partitioning function on a first table of a join query;
determining, by a processing device, a size of the first table based on the partitioning function;
determining, based at least in part on the size of the first table, to distribute the first table to a set of nodes of a computing cluster using a broadcast join; and
distributing, by the processing device, the first table to the set of nodes of the computing cluster.
2. The method of claim 1 , further comprising:
assigning at least part of the first table to two or more processing nodes associated with a second table.
3. The method of claim 2 , further comprising:
duplicating the first table across the two or more processing nodes associated with the second table.
4. The method of claim 3 , further comprising:
forwarding a local partition of the second table to a local instance of a probe operator for performing the join query.
5. The method of claim 1 , further comprising:
determining, based at least in part on the size of the first table, to distribute the first table using a partitioning join; and
assigning at least part of the first table to a node based on a key associated with a join condition.
6. The method of claim 5 , further comprising:
partitioning at least part of the first table across a set of nodes of the computing cluster, wherein the partitioning join is performed by each node in the set of nodes of the computing cluster.
7. The method of claim 5 , wherein each pair of tuples from the first and second tables that satisfy the join condition is assigned to the same node.
8. The method of claim 1 , further comprising:
forwarding one or more partitions of the first table to a build operator of a build operation.
9. The method of claim 8 , wherein the build operation determines the size the first table based on an amount of data processed by each build operator of the build operation.
10. The method of claim 1 , wherein determining an amount of data in the first table comprises determining an amount of data in a part of the first table that is a result of a sub-expression.
11. A system, comprising:
a memory to store a plurality of tables; and
a processor, operatively coupled with the memory, the processor configured to:
perform a partitioning function on a first table of a join query;
determine a size of the first table based on the partitioning function;
determine, based at least in part on the size of the first table, to distribute the first table to a set of nodes of a computing cluster using a broadcast join; and
distribute the first table to the set of nodes of the computing cluster.
12. The system of claim 11 , wherein the processor is further to:
assign at least part of the first table to two or more processing nodes associated with a second table.
13. The system of claim 12 , wherein the processor is further to:
duplicate the first table across the two or more processing nodes associated with the second table.
14. The system of claim 13 , wherein the processor is further to:
forward a local partition of the second table to a local instance of a probe operator for performing the join query.
15. The system of claim 11 , wherein the processor is further to:
determine, based at least in part on the size of the first table, to distribute the first table using a partitioning join; and
assign at least part of the first table to a node based on a key associated with a join condition.
16. The system of claim 15 , wherein the processor is further to:
partition at least part of the first table across a set of nodes of the computing cluster, and the partitioning join is performed by each node in the set of nodes of the computing cluster.
17. The system of claim 15 , wherein each pair of tuples from the first and second tables that satisfy the join condition is assigned to the same node.
18. The system of claim 11 , wherein the processor is further to:
forward one or more partitions of the first table to a build operator of a build operation.
19. The system of claim 18 , wherein the build operation determines the size of the first table based on an amount of data processed by each build operator of the build operation.
20. The system of claim 11 , wherein determining an amount of data in the first table comprises determining an amount of data in a part of the first table that is a result of a sub-expression.
21. A non-transitory computer readable medium having instructions stored thereon that, when executed by a processor, cause the processor to:
perform a partitioning function on a first table of a join query;
determine, by the processor, a size of the first table based on the partitioning function;
determine, based at least in part on the size of the first table, to distribute the first table to a set of nodes of a computing cluster using a broadcast join; and
distribute, by the processor, the first table to the set of nodes of the computing cluster.
22. The non-transitory computer readable medium of claim 21 , wherein the processor is further to:
assign at least part of the first table to two or more processing nodes associated with a second table.
23. The non-transitory computer readable medium of claim 22 , wherein the processor is further to:
duplicate the first table across the two or more processing nodes associated with the second table.
24. The non-transitory computer readable medium of claim 23 , wherein the processor is further to:
forward a local partition of the second table to a local instance of a probe operator for performing the join query.
25. The non-transitory computer readable medium of claim 21 , wherein the processor further to:
determine, based at least in part on the size of the first table, to distribute the first table using a partitioning join; and
assign at least part of the first table to a node based on a key associated with a join condition.
26. The non-transitory computer readable medium of claim 25 , wherein the processor is further to:
partition at least part of the first table across a set of nodes of the computing cluster, and the partitioning join is performed by each node in the set of nodes of the computing cluster.
27. The non-transitory computer readable medium of claim 25 , wherein each pair of tuples from the first and second tables that satisfy the join condition is assigned to the same node.
28. The non-transitory computer readable medium of claim 21 , wherein the processor is further to:
forward one or more partitions of the first table to a build operator of a build operation.
29. The non-transitory computer readable medium of claim 28 , wherein the build operation determines the size of the first table based on an amount of data processed by each build operator of the build operation.
30. The non-transitory computer readable medium of claim 21 , wherein determining an amount of data in the first table comprises determining an amount of data in a part of the first table that is a result of a sub-expression.