# US 12,001,428 B2 · Build-side skew handling for hash-partitioning hash joins > Agents: this data is available as JSON from the patentgrep API and MCP server — see https://patentgrep.com/llms.txt - Number: US 12,001,428 B2 - Document ID: US12001428B2 - Status: Granted - Kind: B2 - Application number: 18/047,872 - Filed: 2022-10-19 - Published: 2024-04-25 - Granted: 2024-06-04 - Earliest claimed: 2022-10-19 - Assignees: Snowflake Inc. - Applicants: Snowflake Inc. - Inventors: Xinzhu Cai; Florian Andreas Funke - Web page: https://patentgrep.com/patent/US12001428B2 ## CPC - G06F 16/24537: {of operators} - G06F 16/2255: {Hash tables} - G06F 16/2456: {Join operations} - G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N) ## Abstract Provided herein are systems and methods for handling build-side skew. For example, a method includes computing a plurality of hash values for a join operation. The join operation uses a corresponding plurality of row sets. The plurality of hash values are sampled to detect a frequent hash value. A build-side row set is partitioned using the frequent hash value to generate a partitioned build-side row set. The build-side row set is selected from the plurality of row sets. The partitioned build-side row set is distributed to a plurality of hash-join-build (HJB) instances executing at a corresponding plurality of servers. ## Claims ### Claim 1 A system comprising: at least one hardware processor of a first server; and at least one memory storing instructions that cause the at least one hardware processor to perform operations comprising: generating at the first server, a plurality of build-side row sets using build-side row data received from a second server; computing, at the first server, a plurality of hash values for a join operation using the build-side row data, the join operation using the plurality of build-side row sets as build-side input and a plurality of probe-side row sets of probe-side row data originating from a third server as probe-side input, data from the probe-side input being probed to detect a match with data from build-side input during the join operation; sampling the plurality of hash values associated with the plurality of build-side row sets to detect a frequent hash value, the frequent hash value appearing at least a threshold number of times in the plurality of hash values for the build-side row data; partitioning a build-side row set of the plurality of build-side row sets to generate a partitioned build-side row set, the build-side row set using the frequent hash value, and the partitioned build-side row set including a plurality of hash partitions of the build-side row data, a number of rows in the build-side row data being smaller than a number of rows in the probe-side row data after the partitioning of the build-side row set; distributing the plurality of hash partitions to a plurality of hash-join-build (HJB) instances executing at a corresponding plurality of additional servers, each server of the plurality of additional servers executing one of the plurality of HJB instances using a hash partition of the plurality of hash partitions; and outputting a result of the join operation based on completion of execution of the plurality of HJB instances at the plurality of additional servers. ### Claim 2 (depends on 1) The system of claim 1, the operations comprising: determining a size of a build partition available at each of the plurality of HJB instances. ### Claim 3 (depends on 2) The system of claim 2, the operations comprising: distributing the plurality of hash partitions to the plurality of HJB instances according to the determined size of the build partition available at each of the plurality of HJB instances. ### Claim 4 (depends on 1) The system of claim 1, the operations comprising: distributing the plurality of hash partitions to the plurality of HJB instances according to a round-robin scheme. ### Claim 5 (depends on 1) The system of claim 1, the operations comprising: sampling a subset of the plurality of hash values; determining a sample partitioned row set using each hash value of the subset, to generate a plurality of sampled partitioned row sets; and determining a size for each sampled partitioned row set of the plurality of sampled partitioned row sets. ### Claim 6 (depends on 5) The system of claim 5, the operations comprising: detecting a skew in a selected partitioned row set of the plurality of sampled partitioned row sets, when the size of the selected partitioned row set is greater than a threshold row set size; and selecting the hash value corresponding to the selected partitioned row set as the frequent hash value. ### Claim 7 (depends on 1) The system of claim 1, the operations comprising: encoding the frequent hash value for transmission to a third server, the third server configured as at least one hash link producer, the at least one hash link producer coupled to a plurality of hash-join-probe (HJP) instances executing at the plurality of additional servers. ### Claim 8 (depends on 7) The system of claim 7, the operations comprising: causing partitioning of a probe-side row set by the at least one hash link producer based on the transmission of the frequent hash value, to generate a partitioned probe-side row set, the partitioning of the probe-side row set using the frequent hash value. ### Claim 9 (depends on 8) The system of claim 8, the operations comprising: causing distributing the partitioned probe-side row set to the plurality of HJP instances executing at the plurality of additional servers. ### Claim 10 (depends on 1) The system of claim 1, the operations comprising: determining a set of frequent hash values based on a plurality of frequent hash values received from a hash link producer configured at one or more of the plurality of additional servers; and selecting the frequent hash value from the plurality of frequent hash values, the frequent hash value being common between the plurality of hash values and the plurality of frequent hash values. ### Claim 11 A method comprising: generating, by at least one hardware processor of a first server, a plurality of build-side row sets using build-side row data received from a second server; computing, by the at least one hardware processor, a plurality of hash values for a join operation using the build-side row data, the join operation using the plurality of build-side row sets as build-side input and a plurality of probe-side row sets of probe-side row data originating from a third server as probe-side input, data from the probe-side input being probed to detect a match with data from build-side input during the join operation; sampling the plurality of hash values associated with the plurality of build-side row sets to detect a frequent hash value, the frequent hash value appearing at least a threshold number of times in the plurality of hash values for the build-side row data; partitioning a build-side row set of the plurality of build-side row sets to generate a partitioned build-side row set, the build-side row set using the frequent hash value, and the partitioned build-side row set including a plurality of hash partitions of the build-side row data, a number of rows in the build-side row data being smaller than a number of rows in the probe-side row data after the partitioning of the build-side row set; distributing the plurality of hash partitions to a plurality of hash-join-build (HJB) instances executing at a corresponding plurality of additional servers, each server of the plurality of additional servers executing one of the plurality of HJB instances using a hash partition of the plurality of hash partitions; and outputting a result of the join operation based on completion of execution of the plurality of HJB instances at the plurality of additional servers. ### Claim 12 (depends on 11) The method of claim 11, further comprising: determining a size of a build partition available at each of the plurality of HJB instances. ### Claim 13 (depends on 12) The method of claim 12, further comprising: distributing the plurality of hash partitions to the plurality of HJB instances according to the determined size of the build partition available at each of the plurality of HJB instances. ### Claim 14 (depends on 11) The method of claim 11, further comprising: distributing the plurality of hash partitions to the plurality of HJB instances according to a round-robin scheme. ### Claim 15 (depends on 11) The method of claim 11, further comprising: sampling a subset of the plurality of hash values; determining a sample partitioned row set using each hash value of the subset, to generate a plurality of sampled partitioned row sets; and determining a size for each sampled partitioned row set of the plurality of sampled partitioned row sets. ### Claim 16 (depends on 15) The method of claim 15, further comprising: detecting a skew in a selected partitioned row set of the plurality of sampled partitioned row sets, when the size of the selected partitioned row set is greater than a threshold row set size; and selecting the hash value corresponding to the selected partitioned row set as the frequent hash value. ### Claim 17 (depends on 11) The method of claim 11, further comprising: encoding the frequent hash value for transmission to a third server, the third server configured as at least one hash link producer, the at least one hash link producer coupled to a plurality of hash-join-probe (HJP) instances executing at the plurality of additional servers. ### Claim 18 (depends on 17) The method of claim 17, further comprising: causing partitioning of a probe-side row set by the at least one hash link producer based on the transmission of the frequent hash value, to generate a partitioned probe-side row set, the partitioning of the probe-side row set using the frequent hash value. ### Claim 19 (depends on 18) The method of claim 18, further comprising: causing distributing the partitioned probe-side row set to the plurality of HJP instances executing at the plurality of additional servers. ### Claim 20 (depends on 11) The method of claim 11, further comprising: determining a set of frequent hash values based on a plurality of frequent hash values received from a hash link producer configured at one or more of the plurality of additional servers; and selecting the frequent hash value from the plurality of frequent hash values, the frequent hash value being common between the plurality of hash values and the plurality of frequent hash values. ### Claim 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: generating, by at least one hardware processor of a first server, a plurality of build-side row sets using build-side row data received from a second server; computing, by the at least one hardware processor, a plurality of hash values for a join operation using the build-side row data, the join operation using the plurality of build-side row sets as build-side input and a plurality of probe-side row sets of probe-side row data originating from a third server as probe-side input, data from the probe-side input being probed to detect a match with data from build-side input during the join operation; sampling the plurality of hash values associated with the plurality of build-side row sets to detect a frequent hash value, the frequent hash value appearing at least a threshold number of times in the plurality of hash values for the build-side row data; partitioning a build-side row set of the plurality of build-side row sets to generate a partitioned build-side row set, the build-side row set using the frequent hash value, and the partitioned build-side row set including a plurality of hash partitions of the build-side row data, a number of rows in the build-side row data being smaller than a number of rows in the probe-side row data after the partitioning of the build-side row set; distributing the plurality of hash partitions to a plurality of hash-join-build (HJB) instances executing at a corresponding plurality of additional servers, each server of the plurality of additional servers executing one of the plurality of HJB instances using a hash partition of the plurality of hash partitions; and outputting a result of the join operation based on completion of execution of the plurality of HJB instances at the plurality of additional servers. ### Claim 22 (depends on 21) The computer-storage medium of claim 21, the operations comprising: determining a size of a build partition available at each of the plurality of HJB instances. ### Claim 23 (depends on 22) The computer-storage medium of claim 22, the operations comprising: distributing the plurality of hash partitions to the plurality of HJB instances according to the determined size of the build partition available at each of the plurality of HJB instances. ### Claim 24 (depends on 21) The computer-storage medium of claim 21, the operations comprising: distributing the plurality of hash partitions to the plurality of HJB instances according to a round-robin scheme. ### Claim 25 (depends on 21) The computer-storage medium of claim 21, the operations comprising: sampling a subset of the plurality of hash values; determining a sample partitioned row set using each hash value of the subset, to generate a plurality of sampled partitioned row sets; and determining a size for each sampled partitioned row set of the plurality of sampled partitioned row sets. ### Claim 26 (depends on 25) The computer-storage medium of claim 25, the operations comprising: detecting a skew in a selected partitioned row set of the plurality of sampled partitioned row sets, when the size of the selected partitioned row set is greater than a threshold row set size; and selecting the hash value corresponding to the selected partitioned row set as the frequent hash value. ### Claim 27 (depends on 21) The computer-storage medium of claim 21, the operations comprising: encoding the frequent hash value for transmission to a third server, the third server configured as at least one hash link producer, the at least one hash link producer coupled to a plurality of hash-join-probe (HJP) instances executing at the plurality of additional servers. ### Claim 28 (depends on 27) The computer-storage medium of claim 27, the operations comprising: causing partitioning of a probe-side row set by the at least one hash link producer based on the transmission of the frequent hash value, to generate a partitioned probe-side row set, the partitioning of the probe-side row set using the frequent hash value. ### Claim 29 (depends on 28) The computer-storage medium of claim 28, the operations comprising: causing distributing the partitioned probe-side row set to the plurality of HJP instances executing at the plurality of additional servers. ### Claim 30 (depends on 21) The computer-storage medium of claim 21, the operations comprising: determining a set of frequent hash values based on a plurality of frequent hash values received from a hash link producer configured at one or more of the plurality of additional servers; and selecting the frequent hash value from the plurality of frequent hash values, the frequent hash value being common between the plurality of hash values and the plurality of frequent hash values.