Adaptive distribution for hash operation

1. A method for joining relations, the method comprising:

receiving a relational join query for a join operation comprising a predicate and a plurality of relations, wherein the plurality of relations comprises a first relation and a second relation;

calculating an actual size of the second relation by processing the second relation according to a build operation; and

determining whether to join the plurality of relations via a broadcasting join or a re-partitioning join based on the actual size of the second relation, an estimated size of the first relation, and a cost metric, wherein the join is performed via the broadcasting join in response to a multiple of the actual size of the second relation being less than a sum of the actual size of the second relation and the estimated size of the first relation, wherein the multiple corresponds to a number of times the second relation is to be broadcast.

2. The method of claim 1 , further comprising, in response to determining to join the plurality of relations via the broadcasting join, converting a link between a build operator and a probe operator into a synchronous link.

3. The method of claim 2 , wherein the synchronous link denotes a local one-to-one link between the build operator and the probe operator, and wherein the probe operator performs the broadcasting join.

4. The method of claim 3 , wherein converting the link comprises converting a communication link between the build operator and the probe operator from an initially inactive and adaptive state.

5. The method of claim 1 , further comprising, in response to determining to join the plurality of relations via the re-partitioning join, converting a link between a build operator and a probe operator into a partition link.

6. The method of claim 1 , wherein the first relation and the second relation are partitioned among a plurality of processors.

7. The method of claim 1 , further comprising partitioning the second relation and, during a re-partition join, sending a partition of the second relation to a local probe operator instance as a single pointer to a block of memory or a file on a disk.

8. The method of claim 7 , wherein partitioning the second relation comprises a partitioning operator comprising a hash function over at least one column of the second relation which appears in the predicate.

9. The method of claim 1 , wherein during a broadcasting join, a local partition of the first relation is forward to a local instance of a probe operator for performing the relational join query.

10. The method of claim 9 , wherein the broadcasting join further comprises broadcasting each partition of a plurality of partitions of the second relation to every instance of the probe operator in a distributed system.

11. The method of claim 1 , wherein during a re-partitioning join, a partitioning operation partitions the first relation across a plurality of processing nodes of a distributed system and the re-partitioning join is performed by a probe operator at each of the plurality of processing nodes.

12. The method of claim 11 , wherein a partitioning operator on the first relation is compatible with a partitioning function previously applied to the second relation such that each pair of tuples that satisfy the predicate is placed in the same partition and within the same instance of the probe operator in the distributed system.

13. The method of claim 1 , wherein the second relation is a base relation.

14. The method of claim 1 , wherein the second relation is an output of a sub-expression.

15. A system for joining relations, the system comprising non-transitory computer readable storage media storing instructions that, when executed by one or more processors, cause the one or more processors to:

receive a relational join query for a join operation comprising a predicate and a plurality of relations, wherein the plurality of relations comprises a first relation and a second relation;

calculate an actual size of the second relation by processing the second relation according to a build operation; and

determine whether to join the plurality of relations via a broadcasting join or a re-partitioning join based on the actual seize of the second relation, an estimated size of the first relation, and a cost metric, wherein the join is performed via the broadcasting join in response to a multiple of the actual size of the second relation being less than the sum of the actual size of the second relation and the estimated size of the first relation, wherein the multiple corresponds to a number of times the second relation is to be broadcast.

16. The system of claim 15 , wherein the instructions further cause the one or more processors to, in response to determining to join the plurality of relations via the broadcasting join, convert a link between a build operator and a probe operator into a synchronous link.

17. The system of claim 16 , wherein the synchronous link denotes a local one-to-one link between the build operator and the probe operator, and wherein the probe operator performs the broadcasting join.

18. The system of claim 17 , wherein the instructions cause the one or more processors to convert the link by converting a communication link between the build operator and the probe operator from an initially inactive and adaptive state.

19. The system of claim 15 , wherein the instructions further cause the one or more processors to, in response to determining to join the plurality of relations via the re-partitioning join, convert a link between a build operator and a probe operator into a partition link.

20. The system of claim 15 , wherein the first relation and the second relation are partitioned among a plurality of processors of a distributed system.

21. An apparatus for joining relations, wherein the apparatus is configured to:

receive a relational join query for a join operation comprising a predicate and a plurality of relations, wherein the plurality of relations comprises a first relation and a second relation;

calculate an actual size of the second relation by processing the second relation according to a build operation; and

determine whether to join the plurality of relations via a broadcasting join or a re-partitioning join based on the actual size of the second relation, an estimated size of the first relation, and a cost metric, wherein the join is performed via the broadcasting join in response to a multiple of the actual size of the second relation being less than the sum of the actual size of the second relation and the estimated size of the first relation, wherein the multiple corresponds to a number of times the second relation is to be broadcast.

22. The apparatus of claim 21 , wherein the relations are distributed over a plurality of processing nodes.

23. The apparatus of claim 21 , wherein the apparatus is configured to, in response to determining to join the plurality of relations via the broadcasting join, convert a link between a build operator and a probe operator into a synchronous link.

24. The apparatus of claim 23 , wherein the synchronous link denotes a local one-to-one link between the build operator and the probe operator, and wherein the probe operator performs broadcasting join.

25. The apparatus of claim 21 , wherein the apparatus is further configured to, in response to determining to join the plurality of relations via the re-partitioning join, convert a link between a build operator and a probe operator into a partition link.

26. The apparatus of claim 21 , wherein during a broadcasting join, a local partition of the first relation is forwarded to a local instance or a probe operator for performing the relational join query.

27. The apparatus of claim 21 , wherein during a re-partitioning join, a partitioning operator partitions the first relation across a plurality of processing nodes of a distributed system and the re-partitioning join is performed by a probe operator at each of the plurality of processing nodes.