# US 2019/0171639 A1 · Method, medium, and system for joining data tables

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

- Number: US 2019/0171639 A1
- Document ID: US20190171639A1
- Status: Published application
- Kind: A1
- Application number: 16/084,529
- Filed: 2017-02-27
- Published: 2019-06-06
- Earliest claimed: 2016-03-14
- Assignees: Alibaba Group Holding Limited
- Applicants: Alibaba Group Holding Limited
- Inventors: Dong XU; Weiguang SUN; Jiehong LIAN; Longzhong WANG
- Web page: https://patentgrep.com/patent/US20190171639A1

## CPC

- G06F 16/2456: {Join operations}
- G06F 16/24544: {Join order optimisation}
- G06F 16/283: {Multi-dimensional databases or data warehouses, e.g. MOLAP or ROLAP}
- G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N)

## Abstract

Data tables that are located in a distributed data warehouse are joined with a target join-calculating algorithm that has been selected from a number of table joining algorithms which have been compared to each other based on the execution costs of each of the number of table joining algorithms.

## Claims

### Claim 1

A method of joining data tables, the method comprising:
determining a plurality of to-be-joined data tables in a distributed data warehouse that has a plurality of computing nodes and a plurality of storage nodes;
obtaining a plurality of table joining algorithms;
estimating a plurality of execution costs for the plurality of table joining algorithms such that each table joining algorithm has an estimated execution cost to join the plurality of to-be-joined data tables;
selecting a target algorithm from the plurality of table joining algorithms based on the estimated execution cost of each table joining algorithm; and
joining the plurality of to-be-joined data tables with the target algorithm.

### Claim 2 (depends on 1)

The method according to claim 1, wherein estimating the plurality of execution costs for the plurality of table joining algorithms includes estimating an execution cost for a table joining algorithm, wherein estimating the execution cost for the table joining algorithm includes:
determining a number of execution steps in the table joining algorithm;
estimating an execution cost for each execution step; and
determining the estimated execution cost for the table joining algorithm based on the execution cost for each execution step.

### Claim 3 (depends on 2)

The method according to claim 2, wherein each execution step has a number of operations, the number of operations include one or more key operations which have higher execution costs than other operations of the number of operations, and the execution cost of an execution step is based on the execution cost of the one or more key operations in the execution step.

### Claim 4 (depends on 3)

The method according to claim 3, wherein:
the execution steps of a Partitioned Sort Join (PSJ) table joining algorithm include:
a re-partition execution step having a plurality of key operations, the key operations in the re-partition execution step including a local read operation, a network read operation, a local sort operation, and a local write operation; and
a sort join execution step having a key operation, the key operation in the sort join execution step includes an output operation;
the execution steps of a Broadcasted Hash Join (BHJ) table joining algorithm include:
a broadcast execution step having a key operation, the key operation in the broadcast execution step includes a network read operation; and
a hash join execution step having a key operation, the key operation in the hash join execution step includes an output operation; and
the execution steps of a Blocked Hash Join (BKHJ) table joining algorithm include:
a broadcast distribution execution step having a plurality of key operations, the key operations in the broadcast distribution execution step include the local read operation, the network read operation, and the local write operation; and
a hash join execution step having a key operation, the key operation in the hash join execution step includes the output operation.

### Claim 5 (depends on 4)

The method according to claim 4, further comprising obtaining target parameters needed for each execution step, wherein obtaining target parameters includes:
for the re-partition execution step, obtaining parameters N, L, RC, RNC, and WC as target parameters needed for the re-partition execution step;
for the broadcast execution step, obtaining parameters Ni, Nk, D, L, and RNC as target parameters needed for the broadcast execution step;
for the broadcast distribution execution step, obtaining parameters N, L, RC, RNC, and WC as target parameters needed for the broadcast distribution execution step; and
for the sort join execution step or the hash join execution step, obtaining Nj and n as target parameters needed for the sort join execution step or the hash join execution step, where
N represents the total number of data records;
L represents the average length of each data record;
RC represents the unit cost for the local read operation;
RNC represents the unit cost for the network read operation;
WC represents the unit cost for the local write operation;
Nk represents a number of data records contained in a main data table in the to-be-joined data tables, where k is any value of 1 . . . n;
Ni represents a number of data records contained in the i-th auxiliary data table in the to-be-joined data tables, where i=1 . . . n and i*k;
D represents the size of the data block supported by each storage node based on a file system of the distributed data warehouse; and
Nj represents a number of data records contained in the j-th data table in the to-be-joined data tables, where j=1 . . . n; and
n represents a number of data tables in the to-be-joined data tables.

### Claim 6 (depends on 5)

The method according to claim 5, wherein estimating an execution cost includes:
for the re-partition execution step, estimating an execution cost for the local read operation as a data record number consumption of 0, a CPU consumption of 0, and an I/O consumption of N*L*RC (0, 0, N*L*RC), an execution cost for the network read operation as a data record number consumption of N, a CPU consumption of 0, and an I/O consumption of N*L*RNC (N, 0, N*L*RNC), an execution cost for the local sort operation as a data record number consumption of 0, a CPU consumption of N, and an I/O consumption of 0 (0, N, 0), and an execution cost for the local write operation as a data record number consumption of 0, a CPU consumption of 0, and an I/O consumption of N*L*WC (0, 0, N*L*WC) based on the parameters N, L, RC, RNC, and WC;
for the broadcast execution step, estimating the execution cost for the network read operation as a data record number consumption of ΣNi*M, a CPU consumption of 0, and an I/O consumption of ΣNi*M*L*RNC (ΣNi*M, 0, ΣNi*M*L*RNC) based on the parameters Ni, Nk, D, L, and RNC, where M=Nk/D;
for the broadcast distribution execution step, estimating the execution cost for the local read operation as a data record number consumption of 0, a CPU consumption of 0, and an I/O consumption of N*L*RC (0, 0, N*L*RC), the execution cost for the network read operation as a data record number consumption of N, a CPU consumption of 0, and an I/O consumption of N*L*RNC (N, 0, N*L*RNC), and the execution cost for the local write operation as a data record number consumption of 0, a CPU consumption of 0, and an I/O consumption of N*L*WC (0, 0, N*L*WC) based on the parameters N, L, RC, RNC, and WC; and
for the sort join execution step or the hash join execution step, estimating an execution cost for the output operation as a data record number consumption of J, a CPU consumption of 0, and an I/O consumption of 0 (J, 0, 0) based on the parameters Nj and n, where J=(ΠNj)1/n.

### Claim 7 (depends on 6)

The method according to claim 6, wherein estimating an execution cost further includes:
for the re-partition execution step, superposing the execution cost (0, 0, N*L*RC) of the local read operation, the execution cost (N, 0, N*L*RNC) of the network read operation, the execution cost (0, N, 0) of the local sort operation, and the execution cost (0, 0, N*L*WC) of the local write operation to obtain an execution cost (N, N, N*L*(RC+RNC+WC)) as an execution cost for the re-partition execution step;
for the broadcast execution step, using the execution cost (Σi*M, 0, ΣNi*M*L*RNC) for the network read operation as an execution cost for the broadcast execution step;
for the broadcast distribution execution step, superposing the execution cost (0, 0, N*L*RC) for the local read operation, the execution cost (N, 0, N*L*RNC) for the network read operation, and the execution cost (0, 0, N*L*WC) for the local write operation to obtain an execution cost (N, 0, N*L*(RC+RNC+WC)) as an execution cost for the broadcast distribution execution step; and
for the sort join execution step or the hash join execution step, using the execution cost (J, 0, 0) for the output operation as an execution cost for the sort join execution step or the hash join execution step.

### Claim 8 (depends on 7)

The method according to claim 7, wherein prior to, for the re-partition execution step, superposing the execution cost (0, 0, N*L*RC) of the local read operation, the execution cost (N, 0, N*L*RNC) of the network read operation, the execution cost (0, N, 0) of the local sort operation, and the execution cost (0, 0, N*L*WC) of the local write operation to obtain an execution cost (N, N, N*L*(RC+RNC+WC)) as an execution cost for the re-partition execution step:
determining whether a skewed distribution occurs with respect to data records contained in the to-be-joined data tables;
if a determination result is positive, then correcting the execution cost (N, 0, N*L*RNC) for the network read operation to (N, 0, P*N*L*p*RNC) and correcting the execution cost (0, 0, N*L*WC) for the local write operation to (0, 0, P*N*L*p*WC),
where p represents a distribution skewness; and
P represents a number of computing nodes for performing a join processing on the to-be-joined data tables.

### Claim 9 (depends on 8)

The method according to claim 8, wherein prior to, for the hash join execution step, using the execution cost (J, 0, 0) for the output operation as an execution cost for the hash join execution step:
determining whether a data table having a number of data records greater than the size D of the data block supported by each storage node exists in each auxiliary data table; and
if a determination result is positive, then correcting the execution cost (J, 0, 0) for the output operation to obtain a corrected execution cost (J, Nk*ΣNl, Nk*ΣNl*L*WC) as the execution cost for the hash join execution step,
where Nl represent the l-st data table having a number of data records greater than the size D of the data block supported by each storage node; and l=1 . . . n and l≠k.

### Claim 10

A non-transitory computer-readable storage medium having embedded therein program instructions, which when executed by a processor causes the processor to execute a method of joining data tables, the method comprising:
determining a plurality of to-be-joined data tables in a distributed data warehouse that has a plurality of computing nodes and a plurality of storage nodes;
obtaining a plurality of table joining algorithms;
estimating a plurality of execution costs for the plurality of table joining algorithms such that each table joining algorithm has an estimated execution cost to join the plurality of to-be-joined data tables;
selecting a target algorithm from the plurality of table joining algorithms based on the estimated execution cost of each table joining algorithm; and
joining the plurality of to-be-joined data tables with the target algorithm.

### Claim 11 (depends on 10)

The medium according to claim 10, wherein estimating the plurality of execution costs for the plurality of table joining algorithms includes estimating an execution cost for a table joining algorithm, wherein estimating the execution cost for the table joining algorithm includes:
determining a number of execution steps in the table joining algorithm;
estimating an execution cost for each execution step; and
determining the estimated execution cost for the table joining algorithm based on the execution cost for each execution step.

### Claim 12 (depends on 11)

The medium according to claim 11, wherein each execution step has a number of operations, the number of operations include one or more key operations which have higher execution costs than other operations of the number of operations, and the execution cost of an execution step is based on the execution cost of the one or more key operations in the execution step.

### Claim 13 (depends on 12)

The medium according to claim 12, wherein:
the execution steps of a Partitioned Sort Join (PSJ) table joining algorithm include:
a re-partition execution step having a plurality of key operations, the key operations in the re-partition execution step including a local read operation, a network read operation, a local sort operation, and a local write operation; and
a sort join execution step having a key operation, the key operation in the sort join execution step includes an output operation;
the execution steps of a Broadcasted Hash Join (BHJ) table joining algorithm include:
a broadcast execution step having a key operation, the key operation in the broadcast execution step includes a network read operation; and
a hash join execution step having a key operation, the key operation in the hash join execution step includes an output operation; and
the execution steps of a Blocked Hash Join (BKHJ) table joining algorithm include:
a broadcast distribution execution step having a plurality of key operations, the key operations in the broadcast distribution execution step include the local read operation, the network read operation, and the local write operation; and
a hash join execution step having a key operation, the key operation in the hash join execution step includes the output operation.

### Claim 14 (depends on 13)

The medium according to claim 13 wherein the method further comprises obtaining target parameters needed for each execution step, wherein obtaining target parameters includes:
for the re-partition execution step, obtaining parameters N, L, RC, RNC, and WC as target parameters needed for the re-partition execution step;
for the broadcast execution step, obtaining parameters Ni, Nk, D, L, and RNC as target parameters needed for the broadcast execution step;
for the broadcast distribution execution step, obtaining parameters N, L, RC, RNC, and WC as target parameters needed for the broadcast distribution execution step; and
for the sort join execution step or the hash join execution step, obtaining Nj and n as target parameters needed for the sort join execution step or the hash join execution step, where
N represents the total number of data records;
L represents the average length of each data record;
RC represents the unit cost for the local read operation;
RNC represents the unit cost for the network read operation;
WC represents the unit cost for the local write operation;
Nk represents a number of data records contained in a main data table in the to-be-joined data tables, where k is any value of 1 . . . n;
Ni represents a number of data records contained in the i-th auxiliary data table in the to-be-joined data tables, where i=1 . . . n and i≠k;
D represents the size of the data block supported by each storage node based on a file system of the distributed data warehouse; and
Nj represents a number of data records contained in the j-th data table in the to-be-joined data tables, where j=1 . . . n; and
n represents a number of data tables in the to-be-joined data tables.

### Claim 15 (depends on 14)

The medium according to claim 14, wherein estimating an execution cost includes:
for the re-partition execution step, estimating an execution cost for the local read operation as a data record number consumption of 0, a CPU consumption of 0, and an I/O consumption of N*L*RC (0, 0, N*L*RC), an execution cost for the network read operation as a data record number consumption of N, a CPU consumption of 0, and an I/O consumption of N*L*RNC (N, 0, N*L*RNC), an execution cost for the local sort operation as a data record number consumption of 0, a CPU consumption of N, and an I/O consumption of 0 (0, N, 0), and an execution cost for the local write operation as a data record number consumption of 0, a CPU consumption of 0, and an I/O consumption of N*L*WC (0, 0, N*L*WC) based on the parameters N, L, RC, RNC, and WC;
for the broadcast execution step, estimating the execution cost for the network read operation as a data record number consumption of ΣNi*M, a CPU consumption of 0, and an I/O consumption of Σi*M*L*RNC (Σi*M, 0, ΣNi*M*L*RNC) based on the parameters Ni, Nk, D, L, and RNC, where M=Nk/D;
for the broadcast distribution execution step, estimating the execution cost for the local read operation as a data record number consumption of 0, a CPU consumption of 0, and an I/O consumption of N*L*RC (0, 0, N*L*RC), the execution cost for the network read operation as a data record number consumption of N, a CPU consumption of 0, and an I/O consumption of N*L*RNC (N, 0, N*L*RNC), and the execution cost for the local write operation as a data record number consumption of 0, a CPU consumption of 0, and an I/O consumption of N*L*WC (0, 0, N*L*WC) based on the parameters N, L, RC, RNC, and WC; and
for the sort join execution step or the hash join execution step, estimating an execution cost for the output operation as a data record number consumption of J, a CPU consumption of 0, and an I/O consumption of 0 (J, 0, 0) based on the parameters Nj and n, where J=(ΠNj)1/n.

### Claim 16 (depends on 15)

The medium according to claim 15, wherein estimating an execution cost further includes:
for the re-partition execution step, superposing the execution cost (0, 0, N*L*RC) of the local read operation, the execution cost (N, 0, N*L*RNC) of the network read operation, the execution cost (0, N, 0) of the local sort operation, and the execution cost (0, 0, N*L*WC) of the local write operation to obtain an execution cost (N, N, N*L*(RC+RNC+WC)) as an execution cost for the re-partition execution step;
for the broadcast execution step, using the execution cost (ΣNi*M, 0, ΣNi*M*L*RNC) for the network read operation as an execution cost for the broadcast execution step;
for the broadcast distribution execution step, superposing the execution cost (0, 0, N*L*RC) for the local read operation, the execution cost (N, 0, N*L*RNC) for the network read operation, and the execution cost (0, 0, N*L*WC) for the local write operation to obtain an execution cost (N, 0, N*L*(RC+RNC+WC)) as an execution cost for the broadcast distribution execution step; and
for the sort join execution step or the hash join execution step, using the execution cost (J, 0, 0) for the output operation as an execution cost for the sort join execution step or the hash join execution step.

### Claim 17 (depends on 16)

The medium according to claim 16, wherein prior to, for the re-partition execution step, superposing the execution cost (0, 0, N*L*RC) of the local read operation, the execution cost (N, 0, N*L*RNC) of the network read operation, the execution cost (0, N, 0) of the local sort operation, and the execution cost (0, 0, N*L*WC) of the local write operation to obtain an execution cost (N, N, N*L*(RC+RNC+WC)) as an execution cost for the re-partition execution step:
determining whether a skewed distribution occurs with respect to data records contained in the to-be-joined data tables;
if a determination result is positive, then correcting the execution cost (N, 0, N*L*RNC) for the network read operation to (N, 0, P*N*L*p*RNC) and correcting the execution cost (0, 0, N*L*WC) for the local write operation to (0, 0, P*N*L*p*WC),
where p represents a distribution skewness; and
P represents a number of computing nodes for performing a join processing on the to-be-joined data tables.

### Claim 18 (depends on 17)

The medium according to claim 17, wherein prior to, for the hash join execution step, using the execution cost (J, 0, 0) for the output operation as an execution cost for the hash join execution step:
determining whether a data table having a number of data records greater than the size D of the data block supported by each storage node exists in each auxiliary data table; and
if a determination result is positive, then correcting the execution cost (J, 0, 0) for the output operation to obtain a corrected execution cost (J, Nk*ΣNl, Nk*ΣNl*L*WC) as the execution cost for the hash join execution step,
where Nl represent the l-st data table having a number of data records greater than the size D of the data block supported by each storage node; and l=1 . . . n and l≠k.

### Claim 19

A system for joining data tables, the system comprising:
a memory; and
a processor coupled to the memory, the processor to:
determine a plurality of to-be-joined data tables in a distributed data warehouse that has a plurality of computing nodes and a plurality of storage nodes;
obtain a plurality of table joining algorithms;
estimate a plurality of execution costs for the plurality of table joining algorithms such that each table joining algorithm has an estimated execution cost to join the plurality of to-be-joined data tables;
select a target algorithm from the plurality of table joining algorithms based on the estimated execution cost of each table joining algorithm; and
join the plurality of to-be-joined data tables with the target algorithm.

### Claim 20 (depends on 19)

The system according to claim 19, wherein estimating the plurality of execution costs for the plurality of table joining algorithms includes estimating an execution cost for a table joining algorithm, wherein estimating the execution cost for the table joining algorithm includes:
determining a number of execution steps in the table joining algorithm;
estimating an execution cost for each execution step; and
determining the estimated execution cost for the table joining algorithm based on the execution cost for each execution step.