Dynamic partition selection

• US 9,652,496 B1
• Filed: 06/25/2014
• Issued: 05/16/2017
• Est. Priority Date: 06/25/2014
• Status: Active Grant

• Current Assignee: Pivotal Software, Inc. (Broadcom, Inc.)
• Original Assignee: Pivotal Software, Inc. (Broadcom, Inc.)
• Inventors: Antova, Lyublena Rosenova, El-Helw, Amr, Soliman, Mohamed Y., Gu, Zhongxian, Petropoulos, Michail, Waas, Florian Michael
• Primary Examiner(s): Vital, Pierre
• Assistant Examiner(s): Antoine, Alicia
• Application Number:
  US14/314,863
• Time in Patent Office: 1,056 Days
• Field of Search: 707714
• CPC Class Codes: G06F 16/24537 of operators
• CPC Class Codes: G06F 16/24542 Plan optimisation
• CPC Class Codes: G06F 16/24544 Join order optimisation
• CPC Class Codes: G06F 16/24554 Unary operations; Data partitioning operations
• CPC Class Codes: G06F 16/2456 Join operations
• CPC Class Codes: G06F 16/27 Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor
• CPC Class Codes: G06F 16/284 Relational databases
• CPC Class Codes: G06F 16/9024 Graphs; Linked lists G06F16/9027 takes precedence
• CPC Class Codes: G06F 16/9027 Trees

Abstract

Methods, systems, and apparatus, including computer programs encoded on computer storage media, for dynamic partition selection. One of the methods includes receiving a representation of a query plan generated for a query, wherein the query plan includes a dynamic scan operator that represents a first computing node obtaining tuples of one or more partitions of a table from storage and transferring the tuples to a second computing node that executes a parent operator of the dynamic scan operator. A partition selector operator is generated corresponding to the dynamic scan operator. A location in the query plan is determined for the partition selector operator. A modified query plan is generated having the partition selector operator at the determined location.

Claims

1. A computer-implemented method comprising:

receiving a representation of a query plan generated for a query, the query plan comprising a first plurality of operators that, when executed by one or more computing nodes, cause the one or more computing nodes to compute a result for the query, wherein the query plan includes a dynamic scan operator that represents a first computing node obtaining tuples of one or more partitions of a table from storage and transferring the tuples to a second computing node that executes a parent operator of the dynamic scan operator;

generating a partition selector operator corresponding to the dynamic scan operator, wherein the partition selector operator represents a third computing node that executes the partition selector operator including determining one or more partition identifiers of partitions of the table and transferring the one or more partition identifiers to the dynamic scan operator of the first computing node;

determining a location in the query plan for the partition selector operator, including;

determining, for each operator of a subset of operators in the query plan, whether the dynamic scan operator occurs in a subtree of the query plan that is rooted at the respective operator of the subset of operators; and

determining, using results of the determinations of whether the dynamic scan operator occurs in a subtree of the query plan that is rooted at the respective operator of the subset of operators, a first operator in the query plan that is i) a parent operator of the partition selector operator and the dynamic scan operator or ii) a child operator of the partition selector operator;

determining, using the first operator, the location in the query plan for the partition selector operator; and

generating a modified query plan having the partition selector operator at the determined location, wherein the modified query plan includes a second plurality of operators that, when executed by one or more computing nodes, cause the one or more computing nodes to compute a result for the query.

2. The method of claim 1, wherein the query plan is represented as a graph, wherein each operator in the query plan is a node in the graph, and wherein each edge between a first graph node and a second graph node in the graph represents a first computing node, that executes a first operator represented by the first graph node, transferring output of the first operator to a second computing node that executes a second operator represented by the second graph node.

3. The method of claim 1, wherein determining a location in the query plan for the partition selector operator comprises:

determining that the dynamic scan operator does not occur in a subtree rooted at a particular operator; and

in response to determining that the dynamic scan operator does not occur in a subtree rooted at the particular operator, adding the partition selector operator as a parent operator of the particular operator in the query plan.

4. The method of claim 1, further comprising:

determining that the dynamic scan operator occurs in a subtree rooted at a particular operator; and

in response to determining that the dynamic scan operator occurs in a subtree rooted at the particular operator, pushing the partition selector operator to a child operator of the particular operator.

5. The method of claim 4, wherein pushing the partition selector operator to a child operator of the particular operator comprises recursively calling a partition location function for the child operator.

6. The method of claim 4, wherein the particular operator is a join operator that computes pairs of first tuples of a first table and second tuples of the table that have matching values.

7. The method of claim 6, further comprising:

determining that the dynamic scan operator is defined in an outer subtree of the join operator; and

in response to determining that the dynamic scan operator is defined in an outer subtree of the join operator, pushing the partition selector operator to an outer child operator of the join operator.

8. The method of claim 6, further comprising:

determining that the partition selector operator includes a predicate expression on a partitioning key; and

in response to determining that the partition selector operator includes a predicate expression on a partitioning key, pushing the partition selector operator to an outer child operator of the join operator.

9. The method of claim 6, further comprising:

determining that the partition selector operator does not include a predicate expression on a partitioning key; and

in response determining that the partition selector operator does not include a predicate expression on a partitioning key, pushing the partition selector operator to an inner child operator of the join operator.

10. The method of claim 4, wherein the particular operator is a select operator that requests, from the table, one or more tuples having respective values according to a first predicate expression in a query, and

wherein pushing the partition selector operator to a child operator of the particular operator comprises appending the first predicate expression to the partition selector operator,

wherein appending the first predicate expression to the partition selector operator causes a particular computing node to determine, from the first predicate expression and a partition selection function, one or more partitions of the table that may include tuples having respective values that satisfy the first predicate expression.

11. The method of claim 10, wherein the table is a multilevel partitioned table,

wherein the first predicate expression references a first partitioning key of the table, wherein the query includes a second predicate expression on a second partitioning key of the table, and wherein the particular computing node determines, by providing the first predicate expression and the second predicate expression as input to the partition selection function, one or more partitions of the table that may include tuples having respective values that satisfy the first predicate expression and the second predicate expression.

12. The method of claim 10, wherein appending the first predicate expression to the partition selector operator comprises:

receiving a second predicate expression of the partition selector operator; and

computing a combined predicate expression, the combined predicate expression comprising a conjunction of the first predicate expression with the second predicate expression, wherein the particular computing node determines, from the combined predicate expression and the partition selection function, one or more partitions of the table that may include tuples having respective values that satisfy the combined predicate expression.

13. A system comprising:

one or more first computers and one or more first storage devices storing instructions that are operable, when executed by the one or more first computers, to cause the one or more first computers to implement a select operator node that is operable to request, from a first table, one or more tuples having respective values according to a predicate expression in a query;

one or more second computers and one or more second storage devices storing instructions that are operable, when executed by the one or more second computers, to cause the one or more second computers to implement a partition selector node that is operable to determine, from the predicate expression in the query according to a partition selection function, one or more partitions of a table that may include tuples having respective values that satisfy the predicate expression and provides respective identifiers for the one or more partitions to a dynamic scanner node;

one or more third computers and one or more third storage devices storing instructions that are operable, when executed by the one or more third computers, to cause the one or more third computers to implement a dynamic scanner node that is operable to receive, from the partition selector node, respective identifiers of the one or more partitions, obtains tuples of the one or more partitions from storage, and provides the one or more obtained tuples to the select operator node; and

one or more fourth computers and one or more fourth storage devices storing instructions that are operable, when executed by the one or more fourth computers, to cause the one or more fourth computers to implement a master node that is operable to generate, using a representation of a query plan for the query, a modified query plan for the query that comprises a plurality of operators that, when executed by one or more computing nodes, cause the one or more computing nodes to compute a result for the query, wherein the modified query plan includes a select operator that represents the select operator node, a partition selector operator that represents the partition selector node, and a dynamic scan operator that represents the dynamic scanner node; and

wherein the master node is operable to generate the modified query plan by determining a location in the query plan for the partition selector operator by performing operations comprising;

determining, for each operator of a subset of operators in the query plan, whether the dynamic scan operator occurs in a subtree of the query plan that is rooted at the respective operator of the subset of operators;

determining, using results of the determinations of whether the dynamic scan operator occurs in a subtree of the query plan that is rooted at the respective operator of the subset of operators, a first operator in the query plan that is i) a parent operator of the partition selector operator and the dynamic scan operator or ii) a child operator of the partition selector operator; and

determining, using the first operator, the location in the query plan for the partition selector operator.

14. The system of claim 13, further comprising:

a sequence operator that causes the dynamic scanner node to begin obtaining tuples from the one or more partitions after the partition selector node has provided all the respective identifiers to the dynamic scanner node.

15. The system of claim 13, wherein the one or more partitions include less than all of the partitions of the table.

16. The system of claim 13, wherein:

the table is a multilevel partitioned table,

the predicate expression references a first partitioning key of the table,

the query includes a second predicate expression on a second partitioning key of the table, and

the partition selector node determines, by providing the predicate expression and the second predicate expression as input to the partition selection function, one or more partitions of the table that may include tuples having respective values that satisfy the predicate expression and the second predicate expression.

17. A system comprising:

one or more first computers and one or more first storage devices storing instructions that are operable, when executed by the one or more first computers, to cause the one or more first computers to implement a join operator node that is operable to compute, according to a predicate expression in a query, a join result of pairs of first tuples of a first table and second tuples of a second table that have matching values, including matching first values of a first attribute of the first table, and matching second values of a second attribute of the second table, wherein the second table is partitioned by the second attribute of the second table;

one or more second computers and one or more second storage devices storing instructions that are operable, when executed by the one or more second computers, to cause the one or more second computers to implement a table scanner node that is operable to obtain the first tuples of the first table from storage and provides the obtained first tuples to a partition selector node;

one or more third computers and one or more third storage devices storing instructions that are operable, when executed by the one or more third computers, to cause the one or more third computers to implement a partition selector node that is operable to determine, according to a partition selection function for the second table, one or more partitions of the second table that may include second tuples having respective second values that match first values of the first tuples for the first attribute, and provide respective identifiers for the one or more partitions of the second table to a dynamic scanner node;

one or more fourth computers and one or more fourth storage devices storing instructions that are operable, when executed by the one or more fourth computers, to cause the one or more fourth computers to implement a dynamic scanner node that is operable to receive, from the partition selector node, respective identifiers of the one or more partitions of the second table, obtain, using the respective identifiers of the one or more partitions of the second table, the second tuples of the one or more partitions from storage, and provide the obtained second tuples to the join operator node for use in computing the join result; and

one or more fifth computers and one or more fifth storage devices storing instructions that are operable, when executed by the one or more fifth computers, to cause the one or more fifth computers to implement a master node that is operable to generate, using a representation of a query plan for the query, a modified query plan for the query that comprises a plurality of operators that, when executed by one or more computing nodes, cause the one or more computing nodes to compute a result for the query, wherein the modified query plan includes a join operator that represents the join operator node, a table scanner operator that represents the table scanner node, a partition selector operator that represents the partition selector node, and a dynamic scan operator that represents the dynamic scanner node; and

wherein the master node is operable to generate the modified query plan by performing operations comprising;

determining that the dynamic scan operator is defined in a subtree on an outer side of the join operator; and

in response to determining that the dynamic scan operator is defined a subtree on the outer side of the join operator, pushing the partition selector operator to the subtree on the outer side of the join operator.

18. The system of claim 17, wherein the partition selector node provides the first tuples to the join operator node.

19. The system of claim 17, wherein the one or more partitions of the second table include less than all of the partitions of the second table.

20. The system of claim 17, wherein the partition selector node provides the first values of the first tuples to the partition selection function, and wherein the partition selection function returns respective identifiers for one or more partitions to the partition selector node.

21. The system of claim 17, wherein:

the join operator node computes the join result as part of a query plan for a query; and

a table scanner operator for the table scanner node is a child operator in the query plan of a partition selector operator for the partition selector node.

22. The system of claim 17, wherein:

the table scanner node is operable to obtain all of the first tuples from the first table; and

the dynamic scanner node is operable to obtain the second tuples of the one or more partitions of the second table without obtaining tuples from one or more additional partitions of the second table, the one or more partitions being different partitions in the second table than the one or more additional partitions.

23. A system comprising:

one or more computers and one or more storage devices storing instructions that are operable, when executed by the one or more computers, to perform operations comprising;

receiving a representation of a query plan generated for a query, the query plan comprising a first plurality of operators that, when executed by one or more computing nodes, cause the one or more computing nodes to compute a result for the query, wherein the query plan includes a dynamic scan operator that represents a first computing node obtaining tuples of one or more partitions of a table from storage and transferring the tuples to a second computing node that executes a parent operator of the dynamic scan operator;

generating a partition selector operator corresponding to the dynamic scan operator, wherein the partition selector operator represents a third computing node that executes the partition selector operator including determining one or more partition identifiers of partitions of the table and transferring the one or more partition identifiers to the dynamic scan operator of the first computing node;

determining a location in the query plan for the partition selector operator, including;

determining, for each operator of a subset of operators in the query plan, whether the dynamic scan operator occurs in a subtree of the query plan that is rooted at the respective operator of the subset of operators;

determining that the dynamic scan operator occurs in a subtree rooted at a particular operator of the subset of operators; and

in response to determining that the dynamic scan operator occurs in a subtree rooted at the particular operator of the subset of operators, pushing the partition selector operator to a child operator of the particular operator, wherein the child operator indicates the location for the partition selector operator in the query plan; and

generating a modified query plan having the partition selector operator at the determined location, wherein the modified query plan includes a second plurality of operators that, when executed by one or more computing nodes, cause the one or more computing nodes to compute a result for the query.