MEMORY-AWARE JOINS BASED IN A DATABASE CLUSTER

• US 20160026667A1
• Filed: 07/22/2015
• Published: 01/28/2016
• Est. Priority Date: 07/22/2014
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Mukherjee, Niloy, Zait, Mohamed, Loaiza, Juan, Marwah, Vineet, Lahiri, Tirthankar, Kulkarni, Kartik, Yan, Jiaqi
• Granted Patent: US 10,002,148 B2
• Time in Patent Office: Days
• US Class Current: 1/1
• CPC Class Codes: G06F 16/2282 Tablespace storage structures; Management thereof
• CPC Class Codes: G06F 16/2456 Join operations
• CPC Class Codes: G06F 16/278 Data partitioning, e.g. horizontal or vertical partitioning
• CPC Class Codes: G06F 3/0604 Improving or facilitating administration, e.g. storage management
• CPC Class Codes: G06F 3/0644 Management of space entities, e.g. partitions, extents, pools
• CPC Class Codes: G06F 3/0679 Non-volatile semiconductor memory device, e.g. flash memory, one time programmable memory [OTP]

Abstract

Techniques are described herein for distributing data from one or more partitioned tables across the volatile memories of a cluster. In memory copies of data from partitioned tables are grouped based on the data falling within the same partition criteria. These groups are used for assigning data from corresponding partitions to the same node when distributing data from partitioned tables across the volatile memories of a multi-node cluster. When a query requires a join between rows of partitioned tables, the work for the join query is divided into work granules that correspond to partition-wise join operations. Those partition-wise join operations are assigned to nodes by a query coordinator based on the partition-to-node mapping located in the node of the query coordinator.

Claims

1. A method comprising:

prior to receiving a join query that requires work to be performed on a first set of data that belongs to a first partitioned object that resides on-disk, performing the steps of;

pre-loading into volatile memory data from the first partitioned object;

wherein the first partitioned object includes a first plurality of partitions;

wherein pre-loading the first partitioned object includes;

mapping the first plurality of partitions to a plurality of partition groups;

wherein each of the plurality of partition groups corresponds to corresponding partition criteria; and

assigning each partition group of the plurality of partition groups to a corresponding host node of a plurality of host nodes;

pre-loading each given partition of the first plurality of partitions into volatile memory of a host node that corresponds to the partition group to which the given partition is mapped;

in response to receiving the join query, distributing work required by the join query to the plurality of host nodes based on which partition groups have been assigned to each of the plurality of host nodes.

2. The method of claim 1 further comprising, prior to receiving the join query, performing the steps of:

pre-loading into volatile memory data from a second partitioned object;

wherein the second partitioned object includes a second plurality of partitions;

wherein pre-loading the second partitioned object includes;

mapping the second plurality of partitions to the plurality of partition groups; and

pre-loading each partition of the second plurality of partitions into volatile memory of the host node that corresponds to the partition group to which the given partition is mapped;

wherein the join query requires a join between the first partitioned object and the second partitioned object.

3. The method of claim 2, wherein:

the first partitioned object is composite partitioned according to a plurality of partitioning schemes;

the second partitioned object is partitioned according to at least one partitioning scheme; and

the method further comprises;

determining a common partitioning scheme among;

the plurality of partitioning schemes, and

the at least one partitioning scheme; and

establishing the plurality of partition groups based on the common partitioning scheme.

4. The method of claim 2, wherein:

the method further comprises creating a partition-to-node mapping that reflects how the first plurality of partitions are mapped to the plurality of partition groups; and

the step of mapping the second plurality of partitions to the plurality of partition groups comprises;

scanning a partition-to-node mapping; and

mapping the second plurality of partitions to the plurality of partition groups based on the partition-to-node mapping;

5. The method of claim 1, wherein:

the first object is composite partitioned according to a plurality of partitioning schemes; and

the method further comprises;

identifying one or more hash partitioning schemes of the plurality of partitioning schemes;

after identifying the one or more hash partitioning schemes, identifying, from the one or more hash partitioning schemes, a highest cardinality partitioning scheme; and

establishing the plurality of partition groups based on the highest cardinality partitioning scheme.

6. The method of claim 1, wherein:

the first object is composite partitioned according to a plurality of partitioning schemes; and

the method further comprises;

determining that none of the plurality of partitioning schemes is a hash partitioning scheme;

after determining that none of the plurality of partitioning schemes is a hash partitioning scheme, identifying, from the plurality of partitioning schemes, a highest cardinality partitioning scheme; and

establishing the plurality of partition groups based on the highest cardinality partitioning scheme.

7. The method of claim 2, wherein:

the first partitioned object is composite partitioned according to a first plurality of partitioning schemes;

the second partitioned object is composite partitioned according to a second plurality of partitioning schemes;

the method further comprises;

determining a plurality of common partitioning schemes among;

the first plurality of partitioning schemes, and

the second plurality of partitioning schemes; and

identifying one or more hash partitioning schemes of the plurality of common partitioning schemes;

after identifying the one or more hash partitioning schemes, identifying, from the one or more hash partitioning schemes, a highest cardinality partitioning scheme;

establishing the plurality of partition groups based on the highest cardinality partitioning scheme.

8. The method of claim 2, wherein:

the first partitioned object is composite partitioned according to a first plurality of partitioning schemes;

the second partitioned object is composite partitioned according to a second plurality of partitioning schemes;

the method further comprises;

determining a plurality of common partitioning schemes among;

the first plurality of partitioning schemes, and

the second plurality of partitioning schemes; and

determining that none of the plurality of common partitioning schemes is a hash partitioning scheme;

after determining that none of the plurality of partitioning schemes is a hash partitioning scheme, identifying, from the plurality of partitioning schemes, a highest cardinality partitioning scheme; and

establishing the plurality of partition groups based on the highest cardinality partitioning scheme.

9. The method of claim 1, wherein:

each partition group of the plurality of partitioned groups is assigned to two or more corresponding host nodes of the plurality of host nodes; and

pre-loading each given partition of the first plurality of partitions into volatile memory includes pre-loading each given partition of the plurality of partitions into volatile memory of each of the plurality of host nodes that correspond to the partition group to which the given partition is mapped.

10. The method of claim 2, wherein:

the first partitioned object is partitioned based on a first column;

the step of mapping the first plurality of partitions to the plurality of partition groups is based on values in the first column;

the second partitioned object is partitioned based on a second column;

the step of mapping the second plurality of partitions to the plurality of partition groups is based on values in the second column;

the join query joins the first partitioned object with the second partitioned object based on the first column of the first partitioned object and the second column of the second partitioned object;

each host node performs a join between;

one or more partitions of the first partitioned object that have been pre-loaded into volatile memory of the host node; and

one or more partitions of the second partitioned object that have been pre-loaded into volatile memory of the host node.

11. The method of claim 2, further comprising:

the first partitioned object is partitioned based on a first column;

the step of mapping the first plurality of partitions to the plurality of partition groups is based on values in the first column;

the second partitioned object is partitioned based on a second column;

the step of mapping the second plurality of partitions to the plurality of partition groups is based on values in the second column;

the join query joins the first partitioned object with the second partitioned object based on columns other than the first column and the second column;

each host node sends to a single node;

partitions of the first partitioned object that have been pre-loaded in volatile memory of the host node; and

partitions of the second partitioned object that have been pre-loaded in volatile memory of the host node;

the single node performing work required by the join query, where performing work required by the join query includes performing a partition-wise join between the first partitioned object and the second partitioned object based on partitions received from the plurality of host nodes.

12. One or more non-transitory computer-readable media storing instructions which, when executed by one or more hardware processors, cause performance of a method comprising:

prior to receiving a join query that requires work to be performed on a first set of data that belongs to a first partitioned object that resides on-disk, performing the steps of;

pre-loading into volatile memory data from the first partitioned object;

wherein the first partitioned object includes a first plurality of partitions;

wherein pre-loading the first partitioned object includes;

mapping the first plurality of partitions to a plurality of partition groups;

wherein each of the plurality of partition groups corresponds to corresponding partition criteria; and

assigning each partition group of the plurality of partition groups to a corresponding host node of a plurality of host nodes;

pre-loading each given partition of the first plurality of partitions into volatile memory of a host node that corresponds to the partition group to which the given partition is mapped;

in response to receiving the join query, distributing work required by the join query to the plurality of host nodes based on which partition groups have been assigned to the plurality of host nodes.

13. The one or more non-transitory computer-readable media of claim 12, further comprising, prior to receiving the join query, performing the steps of:

pre-loading into volatile memory data from a second partitioned object;

wherein the second partitioned object includes a second plurality of partitions;

wherein pre-loading the second partitioned object includes;

mapping the second plurality of partitions to the plurality of partition groups; and

pre-loading each partition of the second plurality of partitions into volatile memory of the host node that corresponds to the partition group to which the given partition is mapped;

wherein the join query requires a join between the first partitioned object and the second partitioned object.

14. The one or more non-transitory computer-readable media of claim 13, wherein:

the first partitioned object is composite partitioned according to a plurality of partitioning schemes;

the second partitioned object is partitioned according to at least one partitioning scheme; and

the method further comprises;

determining a common partitioning scheme among;

the plurality of partitioning schemes, and

the at least one partitioning scheme; and

establishing the plurality of partition groups based on the common partitioning scheme.

15. The one or more non-transitory computer-readable media of claim 13, wherein:

the method further comprises creating a partition-to-node mapping that reflects how the first plurality of partitions are mapped to the plurality of partition groups; and

the step of mapping the second plurality of partitions to the plurality of partition groups comprises;

scanning a partition-to-node mapping; and

mapping the second plurality of partitions to the plurality of partition groups based on the partition-to-node mapping;

16. The one or more non-transitory computer-readable media of claim 12, wherein:

the first object is composite partitioned according to a plurality of partitioning schemes; and

the method further comprises;

identifying one or more hash partitioning schemes of the plurality of partitioning schemes;

after identifying the one or more hash partitioning schemes, identifying, from the one or more hash partitioning schemes, a highest cardinality partitioning scheme;

establishing the plurality of partition groups based on the highest cardinality partitioning scheme.

17. The one or more non-transitory computer-readable media of claim 12, wherein:

the first object is composite partitioned according to a plurality of partitioning schemes; and

the method further comprises;

determining that none of the plurality of partitioning schemes is a hash partitioning scheme;

after determining that none of the plurality of partitioning schemes is a hash partitioning scheme, identifying, from the plurality of partitioning schemes, a highest cardinality partitioning scheme; and

establishing the plurality of partition groups based on the highest cardinality partitioning scheme.

18. The one or more non-transitory computer-readable media of claim 13, wherein:

the first partitioned object is composite partitioned according to a first plurality of partitioning schemes;

the second partitioned object is composite partitioned according to a second plurality of partitioning schemes;

the method further comprises;

determining a plurality of common partitioning schemes among;

the first plurality of partitioning schemes, and

the second plurality of partitioning schemes; and

identifying one or more hash partitioning schemes of the plurality of common partitioning schemes;

after identifying the one or more hash partitioning schemes, identifying, from the one or more hash partitioning schemes, a highest cardinality partitioning scheme;

establishing the plurality of partition groups based on the highest cardinality partitioning scheme.

19. The one or more non-transitory computer-readable media of claim 13, wherein:

the first partitioned object is composite partitioned according to a first plurality of partitioning schemes;

the second partitioned object is composite partitioned according to a second plurality of partitioning schemes;

the method further comprises;

determining a plurality of common partitioning schemes among;

the first plurality of partitioning schemes, and

the second plurality of partitioning schemes; and

determining that none of the plurality of common partitioning schemes is a hash partitioning scheme;

after determining that none of the plurality of partitioning schemes is a hash partitioning scheme, identifying, from the plurality of partitioning schemes, a highest cardinality partitioning scheme; and

establishing the plurality of partition groups based on the highest cardinality partitioning scheme.

20. The one or more non-transitory computer-readable media of claim 12, wherein:

each partition group of the plurality of partitioned groups is assigned to two or more corresponding host nodes of the plurality of host nodes; and

pre-loading each given partition of the first plurality of partitions into volatile memory includes pre-loading each given partition of the plurality of partitions into volatile memory of each of the plurality of host nodes that correspond to the partition group to which the given partition is mapped.

21. The one or more non-transitory computer-readable media of claim 13, wherein:

the first partitioned object is partitioned based on a first column;

the step of mapping the first plurality of partitions to the plurality of partition groups is based on values in the first column;

the second partitioned object is partitioned based on a second column;

the step of mapping the second plurality of partitions to the plurality of partition groups is based on values in the second column;

the join query joins the first partitioned object with the second partitioned object based on the first column of the first partitioned object and the second column of the second partitioned object;

each host node performs a join between;

one or more partitions of the first partitioned object that have been pre-loaded into volatile memory of the host node; and

one or more partitions of the second partitioned object that have been pre-loaded into volatile memory of the host node.

22. The one or more non-transitory computer-readable media of claim 13, further comprising:

the first partitioned object is partitioned based on a first column;

the step of mapping the first plurality of partitions to the plurality of partition groups is based on values in the first column;

the second partitioned object is partitioned based on a second column;

the step of mapping the second plurality of partitions to the plurality of partition groups is based on values in the second column;

the join query joins the first partitioned object with the second partitioned object based on columns other than the first column and the second column;

each host node sends to a single node;

partitions of the first partitioned object that have been pre-loaded in volatile memory of the host node; and

partitions of the second partitioned object that have been pre-loaded in volatile memory of the host node;

the single node performing work required by the join query, where performing work required by the join query includes performing a partition-wise join between the first partitioned object and the second partitioned object based on partitions received from the plurality of host nodes.