Apparatus and method for operating a distributed database with foreign tables

• US 10,579,634 B2
• Filed: 08/30/2012
• Issued: 03/03/2020
• Est. Priority Date: 08/30/2012
• Status: Active Grant

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Citus Data Bilgi Islemleri Ticaret A.S.
• Inventors: Erdogan, Ozgun Ali, Pathak, Sumedh Suhas, Moshayedi, Hadi
• Primary Examiner(s): Hu, Jensen
• Application Number:
  US13/600,126
• Publication Number:
  US 20140067792A1
• Time in Patent Office: 2,742 Days
• Field of Search: 707705
• CPC Class Codes: G06F 16/2471 Distributed queries
• CPC Class Codes: G06F 16/256 in federated or virtual databases
• 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/278 Data partitioning, e.g. horizontal or vertical partitioning

Abstract

A system includes a coordinator node and worker nodes in communication with the coordinator node. Each worker node stores data blocks. Each data block has data in a semi-structured format and each data block has an associated foreign table declaration specifying conversion of the data in the semi-structured format into a tabular format interpretable by a query language. A query processor executed by the coordinator node produces a distributed query plan in response to a query language query. The distributed query plan includes sub-queries. The sub-queries are executed by selected worker nodes of the worker nodes. The selected worker nodes use foreign table declarations to convert data in semi-structured formats into tabular formats of a distributed database to provide tabular data in response to the query language query.

Claims

1. A system, comprising:

a coordinator node;

a plurality of worker nodes in communication with the coordinator node, wherein each worker node stores a plurality of data blocks, wherein each data block has data in a same semi-structured format, each data block is a partition of a same distributed database, each data block has a same foreign table declaration, and the foreign table declaration includes command(s) for converting the data in the semi-structured format into converted data in a tabular format; and

a query processor executed by the coordinator node to produce a distributed query plan in response to a query in a structured query language,

wherein the distributed query plan is partitioned into a plurality of sub-queries, each of which corresponds to a particular data block;

wherein for each sub-query corresponding to the particular data block;

a worker node is selected, the selected worker node contains the corresponding particular data block;

the corresponding sub-query is executed by the selected corresponding worker node of the plurality of worker nodes;

the selected worker node uses a local version of the same foreign table declaration to convert data of the particular data block in the same semi-structured format into converted data in the tabular format; and

the selected worker node executes the sub-query on the converted data to generate a sub-query result;

wherein each sub-query result is merged to produce a query result.

2. The system of claim 1 wherein the coordinator node mergers the tabular data from the selected worker nodes to provide a query result.

3. The system of claim 1 wherein the query processor directs each sub query to a worker node that contains the related data block.

4. The system of claim 1 wherein the coordinator node maintains foreign table metadata, wherein the foreign table metadata characterizes the tabular format.

5. The system of claim 4 wherein the query processor accesses the foreign table metadata to develop the distributed query plan.

6. The system of claim 1 wherein the coordinator node maintains metadata on the data blocks, wherein the metadata includes the location of the data blocks across the plurality of worker nodes.

7. The system of claim 6 wherein the query processor accesses the metadata on the data blocks to develop the distributed query plan.

8. The system of claim 1 wherein the query processor evaluates alternate network input/output costs to develop the distributed query plan.

9. The system of claim 1 wherein the query processor evaluates statistics on the data blocks to develop the distributed query plan.

10. The system of claim 1 wherein the query processor issues commands to copy one or more data blocks between the plurality of worker nodes to execute the distributed query plan.

11. The system of claim 1 wherein the query processor repartitions the data in the data blocks to execute the distributed query plan.

12. The system of claim 1 wherein a data block is replicated across the plurality of worker nodes.

13. The system of claim 1 wherein a data block is stored as a text file.

14. The system of claim 1 wherein a data block is stored as a binary file.

15. The system of claim 1 wherein the selected worker nodes evaluate statistics on the data blocks to develop a sub-query plan.

16. The system of claim 1 wherein the selected worker nodes evaluate data block indexes to develop a sub-query plan.

17. The system of claim 1 wherein the query language sub-query is a Structured Query Language (SQL) query.

18. A computer implemented method, comprising:

storing a plurality of data blocks on each of a plurality of networked worker nodes, wherein the data blocks have a same semi-structured format and each data block is a partition of a same distributed database;

associating each data block with the same foreign table declaration specifying conversion of the same semi-structured format into converted data in a tabular format;

producing a distributed query plan in response to a query in a structured query language, wherein the distributed query plan is partitioned into a plurality of sub-queries, each of which corresponds to a particular data block;

wherein for each of the sub-queries corresponding to the particular data block;

selecting a worker node from the plurality of networked worker nodes, the selected worker node containing the corresponding particular data block;

executing the corresponding sub-query at the selected worker node, wherein executing includes using instances of the foreign table declaration to convert data of the particular data block in the semi-structured format into converted data in the tabular format and; and

the selected worker node executes the sub-query on the converted data to generate a sub-query result; and

merging each sub-query result to produce a query result.

19. The computer implemented method of claim 18 further comprising directing each sub-query to a worker node that contains the related data block.

20. The computer implemented method of claim 18 further comprising generating foreign table metadata, wherein the foreign table metadata characterizes the tabular formats.

21. The computer implemented method of claim 20 further comprising accessing the foreign table metadata to develop the distributed query plan.

22. The computer implemented method of claim 18 further comprising generating metadata on the data blocks, wherein the metadata includes the location of the data blocks across the networked worker nodes.

23. The computer implemented method of claim 22 further comprising accessing the metadata on the data blocks to develop the distributed query plan.

24. The computer implemented method of claim 18 further comprising evaluating alternate network input/output costs to develop the distributed query plan.

25. The computer implemented method of claim 18 further comprising evaluating statistics on the data blocks to develop the distributed query plan.

26. The computer implemented method of claim 18 further comprising issuing commands to copy one or more data blocks to execute the distributed query plan.

27. The computer implemented method of claim 18 further comprising repartitioning the data in data blocks to execute the distributed query plan.

28. The computer implemented method of claim 18 wherein a data block is replicated across the networked worker nodes.

29. The computer implemented method of claim 18 further comprising storing a data block as a text file.

30. The computer implemented method of claim 18 further comprising storing a data block as a binary file.

31. The computer implemented method of claim 18 further comprising evaluating statistics on the data blocks to develop a sub-query plan.

32. The computer implemented method of claim 18 further comprising evaluating data block indexes to develop a sub-query plan.

33. The computer implemented method of claim 18 wherein the query language sub-query is a Structured Query Language (SQL) query.