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, 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;
re-partitioning the first relation to a plurality of build operator instances;
determining, by a processing device, whether to replicate the first relation to each of a plurality of probe operator instances or to re-partition the second relation between the plurality of probe operator instances, wherein the determining is based at least in part on an actual size of the first relation and an estimated size of the second relation;
based on the determining, distributing the first relation or the second relation over a plurality of communication links of a data communication network to the processing nodes of the cluster associated with the probe operation; and
performing, at the processing nodes of the cluster associated with the probe operation, the relational join query using at least one of a hash join, a sort-merge join, or a nested-loop join to generate a third relation.
2. The method of claim 1 , wherein the determining is further based on a cost metric.
3. The method of claim 1 , wherein the actual size of the first relation is determined during execution of the join operation.
4. The method of claim 1 , 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 1 , wherein the first relation comprises tuples, the method further comprising:
forwarding the tuples to a plurality of build operator instances in accordance with a partition move; 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 replicate the first relation to each of the plurality of probe operator instances:
setting links between the plurality of build operator instances and the plurality of probe operator instances to broadcast links;
setting links between the second relation and the plurality of probe operator instances 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 operator instances.
7. The method of claim 1 , further comprising, upon determining to re-partition the second relation between the plurality of probe operator instances:
setting links between the plurality of build operator instances and the plurality of probe operator instances to synchronous links;
setting links between the second relation and the plurality of probe operator instances to partition links; and
sending the second relation through the partition links to the plurality of probe operator instances.
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;
re-partition the first relation to a plurality of build operator instances;
determine, based at least in part on an actual size of the first relation and an estimated size of the second relation, whether to replicate the first relation to each of a plurality of probe operator instances or to re-partition the second relation between the plurality of probe operator instances;
based on the determination, distribute the first relation or the second relation over a plurality of communication links of a data communication network to the processing nodes of the cluster associated with the probe operation; and
perform, at the processing nodes of the cluster associated with the probe operation, the relational join query via at least one of a hash join, a sort-merge join, or a nested-loop join to generate a third relation.
9. The system of claim 8 , wherein the determination is further based on a cost metric.
10. The system of claim 8 , wherein the actual size of the first relation is determined during execution of the join operation.
11. The system of claim 8 , 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 8 , wherein the first relation comprises tuples, and the processing device is further to:
forward the tuples to the plurality of build operator instances in accordance with a partition move; 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 replicate the first relation to each of the plurality of probe operator instances, the processing device is further to:
set links between the plurality of build operator instances and the plurality of probe operator instances to broadcast links;
set links between the second relation and the plurality of probe operator instances 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 operator instances.
14. The system of claim 8 , wherein if the processing device determines to re-partition the second relation between the plurality of probe operator instances, the processing device is further to:
set links between the plurality of build operator instances and the plurality of probe operator instances to synchronous links;
set links between the second relation and the plurality of probe operator instances to partition links; and
send the second relation through the partition links to the plurality of probe operator instances.
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;
re-partition the first relation to a plurality of build operator instances;
determine, based at least in part on an actual size of the first relation and an estimated size of the second relation, whether to replicate the first relation to each of a plurality of probe operator instances or to re-partition the second relation between the plurality of probe operator instances;
based on the determination, distribute the first relation or the second relation over a plurality of communication links of a data communication network to the processing nodes of the cluster associated with the probe operation; and
perform, at the processing nodes of the cluster associated with the probe operation, the relational join query via at least one of a hash join, a sort-merge join, or a nested-loop join to generate a third relation.
16. The non-transitory computer readable medium of claim 15 , wherein the determination is further based on a cost metric.
17. The non-transitory computer readable medium of claim 15 , 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 15 , wherein the first relation comprises tuples, and the processing device is further to:
forward the tuples to the plurality of build operator instances in accordance with a partition move; 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 replicate the first relation to each of the plurality of probe operator instances, the processing device is further to:
set links between the plurality of build operator instances and the plurality of probe operator instances to broadcast links;
set links between the second relation and the plurality of probe operator instances 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 operator instances.
20. The non-transitory computer readable medium of claim 15 , wherein if the processing device determines to re-partition the second relation between the plurality of probe operator instances, the processing device is further to:
set links between the plurality of build operator instances and the plurality of probe operator instances to synchronous links;
set links between the second relation and the plurality of probe operator instances to partition links; and
send the second relation through the partition links to the plurality of probe operator instances.