Systems, methods, and devices for managing data skew in a join operation
Systems, methods, and devices, for managing data skew during a join operation are disclosed. A method includes computing a hash value for a join operation and detecting data skew on a probe side of the join operation at a runtime of the join operation using a lightweight sketch data structure. The method includes identifying a frequent probe-side join key on the probe side of the join operation during a probe phase of the join operation. The method includes identifying a frequent build-side row having a build-side join key corresponding with the frequent probe-side join key. The method includes asynchronously distributing the frequent build-side row to one or more remote servers.
1. A method comprising:
transmitting, from a first server to a second server, a frequent build-side join key of a frequent build-side row of a join operation being conducted at least in part by the second server;
asynchronously receiving the frequent build-side row at the first server from the second server, the frequent build-side row being asynchronously distributed by the second server to the first server in response to (i) the first server having transmitted the frequent build-side join key to the second server at a frequency that exceeds a predetermined threshold and (ii) the second server having detected data skew on a probe side of the join operation at a runtime of the join operation;
generating, at the first server for the received frequent build-side row, a local hash table for processing a local instance of a portion of the join operation; and
altering, at the first server, a local input link to route a frequent probe-side row to the local instance of the portion of the join operation, the frequent probe-side row comprising a frequent probe-side join key corresponding to the frequent build-side join key, the second server having identified the frequent probe-side join key as being frequent to a threshold confidence level.
2. The method of claim 1 , the frequent build-side row having been identified by the second server at least in part by identifying the frequent build-side row as having the frequent build-side join key corresponding to the frequent probe-side join key.
3. The method of claim 1 , further comprising:
asynchronously receiving a second frequent build-side row at the first server from the second server, the second frequent build-side row corresponding to the join operation being conducted at least in part by the second server, the second frequent build-side row comprising a second frequent build-side join key;
generating, at the first server for the received second frequent build-side row, a second local hash table for processing the local instance of the portion of the join operation; and
altering, at the first server, a local input link to route, to the local instance of the portion of the join operation, a second frequent probe-side row comprising a second frequent probe-side join key corresponding to the second frequent build-side join key.
4. The method of claim 1 , also carried out in parallel by each server in a set of one or more additional remote servers to which the second server also asynchronously distributes the frequent build-side row.
5. The method of claim 1 , wherein altering the local input link to route the frequent probe-side row to the local instance of the portion of the join operation comprises changing the local input link from routing the frequent probe-side row to a specific remote server to instead routing the frequent probe-side row to the local instance of the portion of the join operation.
6. The method of claim 1 , wherein:
the first server comprises a first execution platform for processing the local instance of the portion of the join operation;
the second server comprises a second execution platform for processing a portion of the join operation; and
asynchronously receiving the frequent build-side row at the first server from the second server comprises asynchronously receiving the frequent build-side row at the first execution platform from the second execution platform.
7. A first server comprising:
at least one hardware processor; and
one or more non-transitory computer-readable storage media containing instructions that, when executed by the at least one hardware processor, cause the at least one hardware processor to perform operations comprising:
transmitting, from a first server to a second server, a frequent build-side join key of a frequent build-side row of a join operation being conducted at least in part by the second server;
asynchronously receiving the frequent build-side row at the first server from the second server, the frequent build-side row being asynchronously distributed by the second server to the first server in response to (i) the first server having transmitted the frequent build-side join key to the second server at a frequency that exceeds a predetermined threshold and (ii) the second server having detected data skew on a probe side of the join operation at a runtime of the join operation;
generating, at the first server for the received frequent build-side row, a local hash table for processing a local instance of a portion of the join operation; and
altering, at the first server, a local input link to route a frequent probe-side row to the local instance of the portion of the join operation, the frequent probe-side row comprising a frequent probe-side join key corresponding to the frequent build-side join key, the second server having identified the frequent probe-side join key as being frequent to a threshold confidence level.
8. The first server of claim 7 , the frequent build-side row having been identified by the second server at least in part by identifying the frequent build-side row as having the frequent build-side join key corresponding to the frequent probe-side join key.
9. The first server of claim 7 , the operations further comprising:
asynchronously receiving a second frequent build-side row at the first server from the second server, the second frequent build-side row corresponding to the join operation being conducted at least in part by the second server, the second frequent build-side row comprising a second frequent build-side join key;
generating, at the first server for the received second frequent build-side row, a second local hash table for processing the local instance of the portion of the join operation; and
altering, at the first server, a local input link to route, to the local instance of the portion of the join operation, a second frequent probe-side row comprising a second frequent probe-side join key corresponding to the second frequent build-side join key.
10. The first server of claim 7 , wherein altering the local input link to route the frequent probe-side row to the local instance of the portion of the join operation comprises changing the local input link from routing the frequent probe-side row to a specific remote server to instead routing the frequent probe-side row to the local instance of the portion of the join operation.
11. The first server of claim 7 , wherein:
the first server further comprises a first execution platform for processing the local instance of the portion of the join operation;
the second server comprises a second execution platform for processing a portion of the join operation; and
asynchronously receiving the frequent build-side row at the first server from the second server comprises asynchronously receiving the frequent build-side row at the first execution platform from the second execution platform.
12. One or more non-transitory computer-readable storage media containing instructions that, when executed by at least one hardware processor of a first server, cause the at least one hardware processor to perform operations comprising:
transmitting, from a first server to a second server, a frequent build-side join key of a frequent build-side row of a join operation being conducted at least in part by the second server;
asynchronously receiving the frequent build-side row at the first server from the second server, the frequent build-side row being asynchronously distributed by the second server to the first server in response to (i) the first server having transmitted the frequent build-side join key to the second server at a frequency that exceeds a predetermined threshold and (ii) the second server having detected data skew on a probe side of the join operation at a runtime of the join operation;
generating, at the first server for the received frequent build-side row, a local hash table for processing a local instance of a portion of the join operation; and
altering, at the first server, a local input link to route a frequent probe-side row to the local instance of the portion of the join operation, the frequent probe-side row comprising a frequent probe-side join key corresponding to the frequent build-side join key, the second server having identified the frequent probe-side join key as being frequent to a threshold confidence level.
13. The one or more non-transitory computer-readable storage media of claim 12 , the frequent build-side row having been identified by the second server at least in part by identifying the frequent build-side row as having the frequent build-side join key corresponding to the frequent probe-side join key.
14. The one or more non-transitory computer-readable storage media of claim 12 , wherein altering the local input link to route the frequent probe-side row to the local instance of the portion of the join operation comprises changing the local input link from routing the frequent probe-side row to a specific remote server to instead routing the frequent probe-side row to the local instance of the portion of the join operation.