Systems, methods, and devices for managing data skew in a join operation

1. A method for managing data skew, the method comprising:

detecting, by a local server comprising at least one hardware processor, data skew on a probe side of a join operation at a runtime of the join operation using a lightweight sketch data structure;

identifying, by the local server, a frequent probe-side join key on the probe side of the join operation during a probe phase of the join operation;

identifying, by the local server, a frequent build-side row having a frequent build-side join key corresponding to the identified frequent probe-side join key;

asynchronously distributing, by the local server in response to detecting the data skew on the probe side of the join operation, the identified frequent build-side row to a plurality of remote servers that frequently transmitted the frequent build-side join key;

asynchronously receiving, by each of the plurality of remote servers, the identified frequent build-side row; and

altering, by each of the plurality of remote servers, an input link to route a frequent probe-side row comprising the identified frequent probe-side join key to a local instance of the join operation.

2. The method of claim 1 , wherein each of the plurality of remote servers is configured to generate a separate hash table for the identified frequent build-side row.

3. The method of claim 1 , further comprising:

computing, by the local server, a hash value for the join operation;

selecting, by the local server, a rowset comprising a plurality of rows of the join operation; and

probing, by the local server, each of the plurality of rows of the rowset into a space saving algorithm using the hash value for the join operation.

4. The method of claim 3 , further comprising:

updating, by the local server, the space saving algorithm based on incoming data; and identifying, by the local server for each update to the space saving algorithm, a frequency indicating how frequently the identified frequent probe-side join key is probed as a side-effect of updating the space saving algorithm.

5. The method of claim 4 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers is also in response to the identified frequency exceeding a predetermined threshold.

6. The method of claim 3 , further comprising:

calculating, by the local server, a total number of rows of the join operation that have been probed into the space saving algorithm;

calculating, by the local server, a threshold per worker thread based on the total number of rows of the join operation that have been probed into the space saving algorithm; and

determining, by the local server based on the threshold per worker thread, whether the frequent build-side join key is frequent among all threads of at least one server among the local server and the plurality of remote servers.

7. The method of claim 1 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers occurs only after determining, by the local server to a threshold confidence level, that the identified frequent probe-side join key is frequent on the local server.

8. Non-transitory computer readable storage media storing instructions that, when executed by one or more hardware processors, cause the one or more hardware processors to perform operations comprising:

detecting, by a local server, data skew on a probe side of a join operation at a runtime of the join operation using a lightweight sketch data structure;

identifying, by the local server, a frequent probe-side join key on the probe side of the join operation during a probe phase of the join operation;

identifying, by the local server, a frequent build-side row having a frequent build-side join key corresponding to the identified frequent probe-side join key;

asynchronously distributing, by the local server in response to detecting the data skew on the probe side of the join operation, the identified frequent build-side row to a plurality of remote servers that frequently transmitted the frequent build-side join key;

asynchronously receiving, by each of the plurality of remote servers, the identified frequent build-side row; and

altering, by each of the plurality of remote servers, an input link to route a frequent probe-side row comprising the identified frequent probe-side join key to a local instance of the join operation.

9. The non-transitory computer readable storage media of claim 8 , the operations further comprising:

computing, by the local server, a hash value for the join operation,

selecting, by the local server, a rowset comprising a plurality of rows of the join operation; and

probing, by the local server, each of the plurality of rows of the rowset into a space saving algorithm using the hash value for the join operation.

10. The non-transitory computer readable storage media of claim 9 , the operations further comprising:

updating, by the local server, the space saving algorithm based on incoming data; and

identifying, by the local server for each update to the space saving algorithm, a frequency indicating how frequently the identified frequent probe-side join key is probed as a side-effect of updating the space saving algorithm.

11. The non-transitory computer readable storage media of claim 10 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers is also in response to the identified frequency exceeding a predetermined threshold.

12. The non-transitory computer readable storage media of claim 9 , the operations further comprising:

calculating, by the local server, a total number of rows of the join operation that have been probed into the space saving algorithm;

calculating, by the local server, a threshold per worker thread based on the total number of rows of the join operation that have been probed into the space saving algorithm; and

determining, by the local server based on the threshold per worker thread, whether the frequent build-side join key is frequent among all threads of at least one server among the local server and the plurality of remote servers.

13. The non-transitory computer readable storage media of claim 8 , wherein each of the plurality of remote servers is configured to generate a separate hash table for the identified frequent build-side row.

14. The non-transitory computer readable storage media of claim 8 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers occurs only after determining, by the local server to a threshold confidence level, that the identified frequent probe-side join key infrequent on the local server.

15. A system for managing data skew, the system comprising:

a local server comprising:

at least one local-server hardware processor; and

one or more local-server non-transitory computer readable storage media containing local-server instructions that, when executed by the at least one local-server hardware processor, cause the at least one local-server hardware processor to perform local-server operations comprising:

detecting data skew on a probe side of a join operation at a runtime of the join operation using a lightweight sketch data structure;

identifying a frequent probe-side join key on the probe side of the join operation during a probe phase of the join operation;

identifying a frequent build-side row having a frequent build-side join key corresponding to the frequent probe-side join key; and

asynchronously distributing, in response to detecting the data skew on the probe side of the join operation, the identified frequent build-side row to a plurality of remote servers that frequently transmitted the frequent build-side join key; and

the plurality of remote servers each comprising:

at least one remote-server hardware processor; and

one or more remote-server non-transitory computer readable storage media containing remote-server instructions that, when executed by the at least one remote-server hardware processor, cause the at least one remote-server hardware processor to perform remote-server operations comprising:

asynchronously receiving the identified frequent build-side row; and

altering an input link to route a frequent probe-side row comprising the identified frequent probe-side join key to a local instance of the join operation.

16. The system of claim 15 , the local-server operations further comprising:

computing a hash value for the join operation;

selecting a rowset comprising a plurality of rows of the join operation; and

probing each of the plurality of rows of the rowset into a space saving algorithm using the hash value for the join operation.

17. The system of claim 16 , the local-server operations further comprising:

updating the space saving algorithm based on incoming data; and

identifying, for each update to the space saving algorithm, a frequency indicating how frequently the identified frequent probe-side join key is probed as a side-effect of updating the space saving algorithm.

18. The system of claim 17 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers is also in response to the identified frequency exceeding a predetermined threshold.

19. The system of claim 16 , the local-server operations further comprising:

calculating a total number of rows of the join operation that have been probed into the space saving algorithm;

calculating a threshold per worker thread based on the total number of rows of the join operation that have been probed into the space saving algorithm; and

determining, based on the threshold per worker thread, whether the frequent build-side join key is frequent among all threads of at least one server among the local server and the plurality of remote servers.

20. The system of claim 15 , wherein each of the plurality of remote servers is configured to generate a separate hash table for the identified frequent build-side row.

21. The system of claim 15 , wherein asynchronously distributing the identified frequent build-side row to the plurality of remote servers occurs only after determining, by the local server to a threshold confidence level, that the identified frequent probe-side join key is frequent on the local server.