# US 10,055,472 B2 · Adaptive distribution method for hash operations

> Agents: this data is available as JSON from the patentgrep API and MCP server — see https://patentgrep.com/llms.txt

- Number: US 10,055,472 B2
- Document ID: US10055472B2
- Status: Granted
- Kind: B2
- Application number: 14/626,836
- Filed: 2015-02-19
- Published: 2015-08-20
- Granted: 2018-08-21
- Earliest claimed: 2014-02-19
- Assignees: Snowflake Computing, Inc.
- Applicants: Snowflake Computing Inc.
- Inventors: Benoit Dageville; Thierry Cruanes; Marcin Zukowski; Allison Waingold Lee; Philipp Thomas Unterbrunner
- Web page: https://patentgrep.com/patent/US10055472B2

## CPC

- G06F 16/273: {Asynchronous replication or reconciliation}
- A61F 5/566: {Intra-oral devices}
- G06F 9/4881: {Scheduling strategies for dispatcher, e.g. round robin, multi-level priority queues}
- G06F 9/5016: {the resource being the memory}
- G06F 9/5044: {considering hardware capabilities}
- G06F 9/5083: {Techniques for rebalancing the load in a distributed system}
- G06F 9/5088: {involving task migration}
- G06F 16/128: {Details of file system snapshots on the file-level, e.g. snapshot creation, administration, deletion (error detection or correction of the data by redundancy in operations or in hardware G06F11/14, G06F11/16)}
- G06F 16/148: {File search processing}
- G06F 16/1827: {Management specifically adapted to NAS (management of storage area networks [SAN] G06F3/067)}
- G06F 16/211: {Schema design and management}
- G06F 16/221: {Column-oriented storage; Management thereof}
- G06F 16/2365: {Ensuring data consistency and integrity}
- G06F 16/24532: {of parallel queries}
- G06F 16/24545: {Selectivity estimation or determination}
- G06F 16/24552: {Database cache management}
- G06F 16/2456: {Join operations}
- G06F 16/2471: {Distributed queries}
- G06F 16/254: {Extract, transform and load [ETL] procedures, e.g. ETL data flows in data warehouses}
- G06F 16/27: Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor
- G06F 16/283: {Multi-dimensional databases or data warehouses, e.g. MOLAP or ROLAP}
- G06F 16/951: Indexing; Web crawling techniques
- G06F 16/9535: Search customisation based on user profiles and personalisation
- G06F 16/9538: Presentation of query results
- H04L 67/1095: Replication or mirroring of data, e.g. scheduling or transport for data synchronisation between network nodes
- H04L 67/1097: for distributed storage of data in networks, e.g. transport arrangements for network file system [NFS], storage area networks [SAN] or network attached storage [NAS]
- H04L 67/568: Storing data temporarily at an intermediate stage, e.g. caching
- G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N)
- A61F: FILTERS IMPLANTABLE INTO BLOOD VESSELS; PROSTHESES; DEVICES PROVIDING PATENCY TO, OR PREVENTING COLLAPSING OF, TUBULAR STRUCTURES OF THE BODY, e.g. STENTS; ORTHOPAEDIC, NURSING OR CONTRACEPTIVE DEVICES; FOMENTATION; TREATMENT OR PROTECTION OF EYES OR EARS; BANDAGES, DRESSINGS OR ABSORBENT PADS; FIRST-AID KITS (dental prosthetics A61C)
- H04L: TRANSMISSION OF DIGITAL INFORMATION, e.g. TELEGRAPHIC COMMUNICATION (arrangements common to telegraphic and telephonic communication H04M)

## Abstract

A method, apparatus, and system for join operations of a plurality of relations that are distributed over a plurality of storage locations over a network of computing components.

## Claims

### Claim 1

A method for joining relations distributed over a plurality of processing nodes, the method comprising:
receiving a relational join query for a join operation comprising a predicate and a plurality of relations, wherein the desired join uses an equivalence operation, wherein the plurality of relations comprises a first relation and a second relation, and wherein the second relation is estimated to be smaller than the first relation;
processing the second relation according to a build operation, wherein in response to the build operation an actual size of the second relation is known after the second relation being fully consumed;
in response to the build operation, 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; and
performing the selected broadcasting join or re-partitioning join.

### Claim 2 (depends on 1)

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.

### Claim 3 (depends on 2)

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

### Claim 4 (depends on 1)

The method of claim 1, wherein during a re-partitioning join, a partitioning operation partitions the first relation across the 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 nodes.

### Claim 5 (depends on 4)

The method of claim 4, wherein the partitioning operation 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 operation in the distributed system.

### Claim 6 (depends on 5)

The method of claim 5, wherein the partitioning function previously applied to the second relation is performed during or before the build operation, and wherein the partitions of the second relation are not read back at the time of partitioning the first because the second relation has already been repartitioned.

### Claim 7 (depends on 1)

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 disk.

### Claim 8 (depends on 7)

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.

### Claim 9 (depends on 1)

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

### Claim 10 (depends on 1)

The method of claim 1, wherein the second relation is an output of a subexpression.

### Claim 11

A system for joining relations that are distributed over a plurality of processing nodes performing, the system comprising non-transitory storage media storing computer executable instructions which, when executed by one or more processors, cause the system to:
receive a relational join query for a join operation comprising a predicate and a plurality of relations, wherein the desired join uses an equivalence operation, wherein the plurality of relations comprises a first relation and a second relation, and wherein the second relation is estimated to be smaller than the first relation;
processing the second relation according to a build operation, wherein in response to the build operation an actual size of the second relation is known after the second relation being fully consumed;
in response to the build operation, 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; and joining of the first and second relations using the selected broadcasting join or repartitioning join.

### Claim 12 (depends on 11)

The system of claim 11, wherein during a broadcasting join, a local partition of the first relation is forwarded to a local instance of a probe operator for performing the relational join query and each partition of the second relation is broadcasted to every instance of the probe operation from the local partition.

### Claim 13 (depends on 12)

The system of claim 12, wherein a local partition of the first relation is forwarded to its local instance of the probe operation such that no network transfer occurs with the first relation.

### Claim 14 (depends on 11)

The system of claim 11 wherein during a re-partitioning join, a partitioning operation partitions the first relation across the 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.

### Claim 15 (depends on 14)

The system of claim 14, wherein the partitioning operation 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 operation in the distribution system.

### Claim 16 (depends on 11)

The system of claim 11, wherein the computer executable instructions which, when executed by one or more processors, further cause the system to partition the second relation and, during a re-partition join, send 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 disk.

### Claim 17

An apparatus for joining relations that are distributed over a plurality of processing nodes, 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 desired join uses an equivalence operation, wherein the plurality of relations comprises a first relation and a second relation, and wherein the second relation is estimated to be smaller than the first relation;
process the second relation according to a build operation, wherein in response to the build operation an actual size of the second relation is known after the second relation being fully consumed;
in response to the build operation, 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; and
joining of the first and second relations using the selected broadcasting join or repartitioning join.

### Claim 18 (depends on 17)

The apparatus of claim 17, 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.

### Claim 19 (depends on 17)

The apparatus of claim 17, wherein during a re-partitioning join, a partitioning operation partitions the first relation across the 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 nodes.

### Claim 20 (depends on 19)

The apparatus of claim 19, wherein the partitioning operation 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 operation in the distributed system.

### Claim 21

A system for joining relations that are distributed over a plurality of processing nodes, the system comprising:
means for receiving a relational join query for a join operation comprising a predicate and a plurality of relations, wherein the desired join uses an equivalence operation, wherein the plurality of relations comprises a first relation and a second relation, and wherein the second relation is estimated to be smaller than the first relation;
means for processing the second relation according to a build operation, wherein in response to the build operation an actual size of the second relation is known after being fully consumed; means for determining, in response to the build operation, whether to join the plurality of relations via a broadcasting join or a re-partitioning join based primarily on the actual size of the second relation, an estimated size of the first relation, and a cost metric; and
means for joining of the first and second relations using the selected broadcasting join or re-partitioning join.