Build-side skew handling for hash-partitioning hash joins in distributed database query execution
Provided herein are systems, methods, and computer-storage media for managing data skew in hash join operations. A skew manager partitions build-side row data into multiple sets corresponding to hash-join-build (HJB) instances based on hash values. The skew manager detects skew in a build-side row set associated with a first HJB instance by analyzing the number of rows. Upon detecting skew, the skew manager redirects data rows to at least a second HJB instance. The method involves configuring skew caches, generating histograms, and detecting frequent hash values to identify skew. It also includes communicating skew notifications, broadcasting probe-side row data, and adjusting partitioning of probe-side data. The disclosed techniques further include buffering build-side row sets in streams and performing join operations based on these streams, enhancing efficiency in distributed computing environments.
1 . A system comprising:
at least one hardware processor of a first server of a plurality of servers; and
at least one memory storing instructions that cause the at least one hardware processor to perform operations comprising:
partitioning, at the first server, build-side row data into a plurality of build-side row sets based on a hash value, the plurality of build-side row sets corresponding to a plurality of hash-join-build (HJB) instances;
detecting, at the first server, a skew in a build-side row set of the plurality of build-side row sets based on a number of build-side rows in the build-side row set and an average number of build-side rows per HJB instance of the plurality of HJB instances, the build-side row set assigned to a first HJB instance of the plurality of HJB instances; and
redirecting, according to a distribution scheme, from the first server, one or more data rows of the build-side row set having the hash value to at least a second server of the plurality of servers, the second server hosting at least a second HJB instance of the plurality of HJB instances, the redirecting being based on the skew.
2 . The system of claim 1 , the operations comprising:
configuring a first set of skew caches at a data consumer side comprising a plurality of join processes, the plurality of join processes corresponding to the plurality of HJB instances, the first set of skew caches storing a first plurality of hash values associated with the plurality of HJB instances.
3 . The system of claim 2 , the operations comprising:
configuring a second set of skew caches at a build-side hash link producer side, the build-side hash link producer side associated with receiving the build-side row data and generating the plurality of build-side row sets, the second set of skew caches storing a second plurality of hash values associated with the partitioning of the build-side row data.
4 . The system of claim 3 , the operations comprising:
generating a first histogram based on a frequency distribution of the first plurality of hash values; and
generating a second histogram based on a frequency distribution of the second plurality of hash values.
5 . The system of claim 4 , the operations comprising:
detecting at least one frequent hash value of the first plurality of hash values and the second plurality of hash values based on the first histogram and the second histogram; and
detecting the skew based on the at least one frequent hash value.
6 . The system of claim 1 , the operations comprising:
communicating a notification of the detected skew in the build-side row set to a probe-side hash link producer.
7 . The system of claim 6 , the operations comprising:
broadcasting probe-side row data to a plurality of hash-join-probe (HJP) instances corresponding to the plurality of HJB instances, the broadcasting based on the notification.
8 . The system of claim 7 , the operations comprising:
adjusting partitioning of the probe-side row data at each HJP instance of the plurality of HJP instances to match distribution of build-side row data at a corresponding HJB instance of the plurality of HJB instances after the redirecting.
9 . The system of claim 1 , the operations comprising:
buffering a second build-side row set associated with the second HJB instance at a first stream, the first stream configured at the second HJB instance.
10 . The system of claim 9 , the operations comprising:
buffering the one or more data rows of the build-side row set at a second stream configured at the second HJB instance; and
performing a join operation associated with the second HJB instance, the join operation based on the first stream and the second stream.
11 . A method comprising:
partitioning, by at least one hardware processor of a first server of a plurality of servers, build-side row data into a plurality of build-side row sets based on a hash value, the plurality of build-side row sets corresponding to a plurality of hash-join-build (HJB) instances;
detecting, at the first server, a skew in a build-side row set of the plurality of build-side row sets based on a number of build-side rows in the build-side row set and an average number of build-side rows per HJB instance of the plurality of HJB instances, the build-side row set assigned to a first HJB instance of the plurality of HJB instances; and
redirecting, according to a distribution scheme, from the first server, one or more data rows of the build-side row set having the hash value to at least a second server of the plurality of servers, the second server hosting at least a second HJB instance of the plurality of HJB instances, the redirecting being based on the skew.
12 . The method of claim 11 , further comprising:
configuring a first set of skew caches at a data consumer side comprising a plurality of join processes, the plurality of join processes corresponding to the plurality of HJB instances, the first set of skew caches storing a first plurality of hash values associated with the plurality of HJB instances.
13 . The method of claim 12 , further comprising:
configuring a second set of skew caches at a build-side hash link producer side, the build-side hash link producer side associated with receiving the build-side row data and generating the plurality of build-side row sets, the second set of skew caches storing a second plurality of hash values associated with the partitioning of the build-side row data.
14 . The method of claim 13 , further comprising:
generating a first histogram based on a frequency distribution of the first plurality of hash values; and
generating a second histogram based on a frequency distribution of the second plurality of hash values.
15 . The method of claim 14 , further comprising:
detecting at least one frequent hash value of the first plurality of hash values and the second plurality of hash values based on the first histogram and the second histogram; and
detecting the skew based on the at least one frequent hash value.
16 . The method of claim 11 , further comprising:
communicating a notification of the detected skew in the build-side row set to a probe-side hash link producer.
17 . The method of claim 16 , further comprising:
broadcasting probe-side row data to a plurality of hash-join-probe (HJP) instances corresponding to the plurality of HJB instances, the broadcasting based on the notification.
18 . The method of claim 17 , further comprising:
adjusting partitioning of the probe-side row data at each HJP instance of the plurality of HJP instances to match distribution of build-side row data at a corresponding HJB instance of the plurality of HJB instances after the redirecting.
19 . The method of claim 11 , further comprising:
buffering a second build-side row set associated with the second HJB instance at a first stream, the first stream configured at the second HJB instance.
20 . The method of claim 19 , further comprising:
buffering the one or more data rows of the build-side row set at a second stream configured at the second HJB instance; and
performing a join operation associated with the second HJB instance, the join operation based on the first stream and the second stream.
21 . A computer-storage medium comprising instructions that, when executed by one or more processors of a machine, configure the machine to perform operations comprising:
partitioning at a first server of a plurality of servers, build-side row data into a plurality of build-side row sets based on a hash value, the plurality of build-side row sets corresponding to a plurality of hash-join-build (HJB) instances;
detecting, at the first server, a skew in a build-side row set of the plurality of build-side row sets based on a number of build-side rows in the build-side row set and an average number of build-side rows per HJB instance of the plurality of HJB instances, the build-side row set assigned to a first HJB instance of the plurality of HJB instances; and
redirecting, according to a distribution scheme, from the first server, one or more data rows of the build-side row set having the hash value to at least a second server of the plurality of servers, the second server hosting at least a second HJB instance of the plurality of HJB instances, the redirecting being based on the skew.
22 . The computer-storage medium of claim 21 , the operations comprising:
configuring a first set of skew caches at a data consumer side comprising a plurality of join processes, the plurality of join processes corresponding to the plurality of HJB instances, the first set of skew caches storing a first plurality of hash values associated with the plurality of HJB instances.
23 . The computer-storage medium of claim 22 , the operations comprising:
configuring a second set of skew caches at a build-side hash link producer side, the build-side hash link producer side associated with receiving the build-side row data and generating the plurality of build-side row sets, the second set of skew caches storing a second plurality of hash values associated with the partitioning of the build-side row data.
24 . The computer-storage medium of claim 23 , the operations comprising:
generating a first histogram based on a frequency distribution of the first plurality of hash values; and
generating a second histogram based on a frequency distribution of the second plurality of hash values.
25 . The computer-storage medium of claim 24 , the operations comprising:
detecting at least one frequent hash value of the first plurality of hash values and the second plurality of hash values based on the first histogram and the second histogram; and
detecting the skew based on the at least one frequent hash value.
26 . The computer-storage medium of claim 21 , the operations comprising:
communicating a notification of the detected skew in the build-side row set to a probe-side hash link producer.
27 . The computer-storage medium of claim 26 , the operations comprising:
broadcasting probe-side row data to a plurality of hash-join-probe (HJP) instances corresponding to the plurality of HJB instances, the broadcasting based on the notification.
28 . The computer-storage medium of claim 27 , the operations comprising:
adjusting partitioning of the probe-side row data at each HJP instance of the plurality of HJP instances to match distribution of build-side row data at a corresponding HJB instance of the plurality of HJB instances after the redirecting.
29 . The computer-storage medium of claim 21 , the operations comprising:
buffering a second build-side row set associated with the second HJB instance at a first stream, the first stream configured at the second HJB instance.
30 . The computer-storage medium of claim 29 , the operations comprising:
buffering the one or more data rows of the build-side row set at a second stream configured at the second HJB instance; and
performing a join operation associated with the second HJB instance, the join operation based on the first stream and the second stream.