System and Method for Join-Partitioning For Local Computability of Query Over Shared-Nothing Clusters

• US 20080189239A1
• Filed: 02/02/2007
• Published: 08/07/2008
• Est. Priority Date: 02/02/2007
• Status: Active Grant

• Current Assignee: Teradata US, Inc. (Teradata Corporation)
• Original Assignee: Aster Data Systems, Inc. (Teradata Corporation)
• Inventors: Bawa, Mayank, Candea, George, Argyros, Anastasios
• Granted Patent: US 8,156,107 B2
• Time in Patent Office: Days
• US Class Current: 707/2
• CPC Class Codes: G06F 16/22 Indexing; Data structures therefor; Storage structures
• CPC Class Codes: G06F 16/2471 Distributed queries
• CPC Class Codes: G06F 16/278 Data partitioning, e.g. horizontal or vertical partitioning
• CPC Class Codes: Y10S 707/966 Distributed
• CPC Class Codes: Y10S 707/971 Federated
• CPC Class Codes: Y10S 707/972 Partitioning

Abstract

The present invention addresses the problem of partitioning database relations across a plurality of hosts in a shared-nothing cluster while minimizing communication costs. A new partitioning approach for database relations—join-partitioning—is introduced that focuses on relations and their dependencies to a priori partition the plurality of relations to the plurality of hosts such that each host can locally compute a large class of queries over its partitioned data in isolation at query time, independent of other hosts in the cluster. Such an approach thus overcomes the primary bottleneck, i.e., the network, in deploying database systems over shared-nothing clusters by allowing them to seamlessly scale linearly to tens of thousands of hosts and manage tens or hundreds of terabytes of data. This description is not intended to be a complete description of, or limit the scope of, the invention. Other features, aspects, and objects of the invention can be obtained from a review of the specification, the figures, and the claims.

Claims

1. A system to support data partitioning for local computability of query, comprising:

a cluster, comprising a plurality of hosts;

a partitioning module operable to partition each of a plurality of relations of a dataset into a plurality of sub-relations and store them on the plurality of hosts; and

a query executing module operable to;

execute a query to the dataset over the plurality of partitioned sub-relations locally on each of the plurality of hosts without moving data among them or causing network bottlenecks in the cluster; and

generate result of the query by merging local query output from the plurality of hosts.

5. The system according to claim 1, wherein:

the query class is an equi-join SELECT-PROJECT-JOIN (SPJ) query.

9. The system according to claim 1, wherein:

the plurality of hosts in the cluster share no hard disk drive and/or computer memory among them.

10. The system according to claim 1, wherein:

each of the plurality of hosts is one of;

a server computer, a mainframe computer, a workstation, a laptop or desktop PC, a PDA, a Tablet PC, a Pocket PC, a cell phone, an electronic messaging device, a Java-enabled device, and a FPGA-based device.

11. The system according to claim 1, wherein:

the partitioning module is operable to preserve one or more foreign key constraints locally on each of the plurality of hosts.

12. The system according to claim 1, wherein:

the partitioning module is operable to enforce strongly typed database schema on the plurality of relations.

13. The system according to claim 1, wherein:

the partitioning module is operable to keep a plurality of partitions of the sub-relations balanced across the plurality of hosts via one or more of;

keeping track of the foreign keys for each of the plurality of sub-relations that are placed in each of the plurality of partitions using a bit-map per referenced relation;

keeping a score on the number of tuples of the sub-relations that have been placed in each of the plurality of partitions along with a count of the number of distinct dimension keys per relation in each of the plurality of partitions;

looking up the bit-maps to determine if there are one or more candidate partitions which have all the dimension keys that occur in a tuple;

inserting the tuple into the partition with the least score of the bit-maps if there are one or more candidate partitions or into the one with the least count if there is no candidate partition; and

updating the bit-maps to reflect the placement of the dimension keys in that partition as are the score and count.

14. The system according to claim 1, wherein:

the partitioning module is operable to model the plurality of relations via a snowflake schema.

15. The system according to claim 14, wherein:

the snowflake schema comprises a plurality of relations maintaining one or more of the following information;

a plurality of recurring events, a plurality of participants in the plurality of recurring events, summary of measures of the plurality of recurring events, and time of the plurality of recurring events.

16. The system according to claim 14, wherein:

the snowflake schema is a combination of one or more of;

one or more singleton relations, star schema, snowflaked-dimension schema, and one or more summary relations.

17. The system according to claim 14, wherein:

the partitioning module is operable to dynamically determine one or more partition candidates for each of the plurality of relations based on the snowflake schema.

2. A system to support data partitioning for local computability of query, comprising:

a plurality of databases operable to maintain data on a plurality of hosts in a cluster;

a partitioning module operable to partition each of a plurality of relations of a dataset into a plurality of sub-relations and store them on the plurality of databases; and

a query executing module operable to;

execute a query to the dataset over the plurality of partitioned sub-relations locally on each of the plurality of databases without moving data among them or causing network bottlenecks in the cluster; and

generate result of the query by merging local query output from the plurality of hosts.

3. The system according to claim 2, wherein:

each of the plurality of databases is one of;

a MySQL database, an IBM DB2 database, a PostgreSQL database, an Oracle database, and a Teradata database.

4. A system to support data partitioning for local computability of query, comprising:

a cluster, comprising a plurality of hosts; and

a partitioning module operable to;

partition each of a plurality of relations of a dataset into a plurality of sub-relations and store them on the plurality of hosts while preserving one or more foreign key constraints locally on each of the plurality of hosts; and

store the plurality of sub-relations locally on the plurality of hosts so that a class of queries to the dataset is computable locally over the plurality of hosts without moving data among them or causing network bottlenecks in the cluster.

6. A system to support data partitioning for local computability of query, comprising:

a cluster, comprising a plurality of hosts;

a partitioning module operable to partition each of a plurality of relations of a dataset into a plurality of sub-relations and store them on the plurality of hosts; and

a query executing module operable to;

identify at query time whether a query to the dataset is locally computable on each of the plurality of hosts independent of rest of the plurality of hosts;

execute the query over the plurality of partitioned sub-relations locally on each of the plurality of hosts without moving data among them or causing network bottlenecks in the cluster; and

generate result of the query by merging local query output from the plurality of hosts.

7. The system according to claim 6, wherein:

the query executing module is operable to identify local computability of the query at query time by;

constructing a characterization graph of the query whose vertices correspond to relation variables in the query for each of the plurality of databases; and

determine if there exists a vertex in the characterization graph such that every other vertex in the characteristic graph is reachable from it by following directed edges.

8. A system to support data partitioning for local computability of query, comprising:

a shared-nothing cluster compromising a plurality of hosts sharing no hard disk drive and/or computer memory among them;

a partitioning module operable to partition each of a plurality of relations of a dataset into a plurality of sub-relations and store them on the plurality of hosts while preserving one or more foreign key references locally on each of the plurality of hosts; and

a query executing module operable to;

identify at query time whether a query to the dataset is computable over the plurality of sub-relations locally on each of the plurality of hosts independent of rest of the plurality of hosts;

execute the query over the plurality of partitioned sub-relations stored on the plurality of hosts without moving data among them or causing network bottlenecks; and

generate result of the query by merging local query output from the plurality of hosts.

18. A method to support data partitioning for local computability of query, comprising:

partitioning each of a plurality of relations of a dataset into a plurality of sub-relations and store them on a plurality of hosts in a cluster;

executing a query to the dataset over the plurality of partitioned sub-relations locally on each of the plurality of hosts without moving data among them or causing network bottlenecks in the cluster; and

generating result of the query by merging local query output from the plurality of hosts.

23. The method according to claim 18, further comprising:

preserving one or more foreign key constraints locally on each of the plurality of hosts.

24. The method according to claim 18, further comprising:

enforcing strongly typed database schema on the plurality of relations.

25. The method according to claim 18, further comprising:

keeping a plurality of partitions of the sub-relations balanced across the plurality of hosts via one or more of;

keeping track of the foreign keys for each of the plurality of sub-relations that are placed in each of the plurality of partitions using a bit-map per referenced relation;

keeping a score on the number of tuples of the sub-relations that have been placed in each of the plurality of partitions along with a count of the number of distinct dimension keys per relation in each of the plurality of partitions;

looking up the bit-maps to determine if there are one or more candidate partitions which have all the dimension keys that occur in a tuple;

inserting the tuple into the partition with the least score of the bit-maps if there are one or more candidate partitions or into the one with the least count if there is no candidate partition; and

updating the bit-maps to reflect the placement of the dimension keys in that partition as are the score and count.

26. The method according to claim 18, further comprising:

modeling the plurality of relations via a snowflake schema.

27. The method according to claim 26, further comprising:

determining one or more partition candidates for each of the plurality of relations dynamically based on the snowflake schema.

19. A method to support data partitioning for local computability of query, comprising:

partitioning each of a plurality of relations of a dataset into a plurality of sub-relations and store them on the plurality of hosts while preserving one or more foreign key constraints locally on each of the plurality of hosts; and

storing the plurality of sub-relations locally on the plurality of hosts so that a class of queries to the dataset is computable locally over the plurality of hosts without moving data among them or causing network bottlenecks in the cluster.

20. A method to support data partitioning for local computability of query, comprising:

partitioning each of a plurality of relations of a dataset into a plurality of sub-relations and store them on a plurality of hosts in a cluster;

identifying at query time whether a query to the dataset is locally computable on each of the plurality of hosts independent of rest of the plurality of hosts;

executing the query over the plurality of partitioned sub-relations locally on each of the plurality of hosts without moving data among them or causing network bottlenecks in the cluster; and

generating result of the query by merging local query output from the plurality of hosts.

21. The method according to claim 20, further comprising:

identifying local computability of the query at query time by;

constructing a characterization graph of the query whose vertices correspond to relation variables in the query for each of the plurality of databases; and

determine if there exists a vertex in the characterization graph such that every other vertex in the characteristic graph is reachable from it by following directed edges.

22. A method to support data partitioning for local computability of query, comprising:

partitioning each of a plurality of relations of a dataset into a plurality of sub-relations and store them on a plurality of hosts in a shared-nothing cluster, while preserving one or more foreign key references locally on each of a plurality of hosts;

identifying at query time whether a query to the dataset is computable over the plurality of sub-relations locally on each of the plurality of hosts independent of rest of the plurality of hosts;

executing the query over the plurality of partitioned sub-relations stored on the plurality of hosts without moving data among them or causing network bottlenecks; and

generating result of the query by merging local query output from the plurality of hosts.

28. A machine readable medium having instructions stored thereon that when executed cause a system to:

partition each of a plurality of relations of a dataset into a plurality of sub-relations and store them on a plurality of hosts in a cluster;

execute a query to the dataset over the plurality of partitioned sub-relations locally on each of the plurality of hosts without moving data among them or causing network bottlenecks in the cluster; and

generate result of the query by merging local query output from the plurality of hosts.

29. A system to support data partitioning for local computability of query, comprising:

means for maintaining data on a plurality of hosts in a cluster;

means for partitioning each of a plurality of relations of a dataset into a plurality of sub-relations and store them on the plurality of databases;

means for executing a query to the dataset over the plurality of partitioned sub-relations locally on each of the plurality of databases without moving data among them or causing network bottlenecks in the cluster; and

means for generating result of the query by merging local query output from the plurality of hosts.