Parallel streaming of external data

• US 9,898,469 B1
• Filed: 08/25/2014
• Issued: 02/20/2018
• Est. Priority Date: 02/28/2014
• Status: Active Grant

• Current Assignee: Pivotal Software, Inc. (Broadcom, Inc.)
• Original Assignee: Pivotal Software, Inc. (Broadcom, Inc.)
• Inventors: Dorin, Dov Yaron, Goldshuv, Alon, Shacked, Alex, Lonergan, Luke
• Primary Examiner(s): To, Baoquoc N
• Assistant Examiner(s): Ahmed, Saba
• Application Number:
  US14/468,089
• Time in Patent Office: 1,275 Days
• Field of Search: 707693
• CPC Class Codes: G06F 16/116 Details of conversion of file system types or formats
• CPC Class Codes: G06F 16/13 File access structures, e.g. distributed indices arrangements of input from, or output to, record carriers G06F3/06
• CPC Class Codes: G06F 16/182 Distributed file systems
• CPC Class Codes: G06F 16/2456 Join operations
• CPC Class Codes: G06F 16/2471 Distributed queries
• CPC Class Codes: G06F 16/33 Querying
• CPC Class Codes: H04L 67/1097 for distributed storage of data in networks, e.g. transport arrangements for network file system [NFS], storage area networks [SAN] or network attached storage [NAS]

Abstract

Methods, systems, and apparatus, including computer programs encoded on computer storage media, for streaming external data in parallel from a second distributed system to a first distributed system. One of the methods includes receiving a query that requests a join of first rows of a first table in a first distributed system with second rows of an external table, the external table representing data in a second distributed system. Each of the segment nodes communicates with a respective extension service that obtains fragments from one or more data nodes of the second distributed system according to location information for the respective fragments, and provides to the segment node a stream of data corresponding to second rows of the external table. Each of the segment nodes computes joined rows between the first rows of the first table and the stream of data corresponding to second rows of the external table.

Claims

1. A computer-implemented method comprising:

receiving, by a first distributed system that comprises multiple segment nodes, a query that requests rows of an external table representing data stored in multiple data fragments on multiple respective data nodes in a second distributed system, wherein each of the data nodes of the second distributed system are distinct from the segment nodes of the first distributed system, wherein the first distributed system operates under control of a first master node and the second distributed system operates under control of a second master node that is distinct from the first master node, and wherein the second distributed system stores data in a second format that is different from a first format of data stored by the first distributed system;

initiating communication by a plurality of extension services between nodes of the first distributed system and nodes of the second distributed system, including a first extension service communicating with the master nodes of the first and second distributed system and a plurality of second extension services communicating between data nodes of the second distributed system and segment nodes of the first distributed system;

requesting, by the master node of the first distributed system from the first extension service, location information for a plurality of the data fragments in the second distributed system represented by the external table;

allocating each data fragment of the plurality of data fragments to a respective one of the plurality of second extension services and each of the second extension services to one of a plurality of segment nodes of the first distributed system, including providing each segment node with location information for respective fragments of the plurality of fragments allocated to the segment node;

receiving, by each of the plurality of segment nodes of the first distributed system, a stream of converted data in the first format, wherein each segment node communicates with a respective extension service to;

obtain fragments in the second format from one or more data nodes of the second distributed system according to the location information for the respective fragments,

convert the obtained fragments from the second format to the first format, and

provide to the segment node of the first distributed system a stream of converted data corresponding to the rows of the external table; and

computing a result for the received query using the stream of converted data corresponding to the rows of the external table.

2. The method of claim 1, wherein requesting, by the first master node of the first distributed system from the first extension service, location information for a plurality of the data fragments occurs after receiving the query.

3. The method of claim 1, wherein the extension service converts the obtained fragments to the first format at scan time.

4. The method of claim 1, wherein the location information includes a network location for a data node of the second distributed system for each fragment of the plurality of fragments.

5. The method of claim 1, further comprising:

receiving, from a user, a definition of the external table representing the multiple data fragments on multiple data nodes in the second distributed system.

6. The method of claim 1, wherein the second distributed system is a NoSQL database.

7. The method of claim 1, wherein the first format is a row-oriented format and wherein the second format is a column-oriented format.

8. The method of claim 1, wherein each segment node of the first distributed system computes a partial result for the query without storing the stream of converted data corresponding to the rows of the external table in nonvolatile memory of the first distributed system.

9. The method of claim 1, wherein each extension service converts the data from the second format to the first format after receiving the query and before computing a result for the received query.

10. 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 cause the one or more computers to perform operations comprising;

receiving, by a first distributed system that comprises multiple segment nodes, a query that requests rows of an external table representing data stored in multiple data fragments on multiple respective data nodes in a second distributed system, wherein each of the data nodes of the second distributed system are distinct from the segment nodes of the first distributed system, wherein the first distributed system operates under control of a first master node and the second distributed system operates under control of a second master node that is distinct from the first master node, and wherein the second distributed system stores data in a second format that is different from a first format of data stored by the first distributed system;

initiating communication by a plurality of extension services between nodes of the first distributed system and nodes of the second distributed system, including a first extension service communicating with the master nodes of the first and second distributed system and a plurality of second extension services communicating between data nodes of the second distributed system and segment nodes of the first distributed system;

requesting, by the master node of the first distributed system from the first extension service, location information for a plurality of the data fragments in the second distributed system represented by the external table;

allocating each data fragment of the plurality of data fragments to a respective one of the plurality of second extension services and each of the second extension services to one of a plurality of segment nodes of the first distributed system, including providing each segment node with location information for respective fragments of the plurality of fragments allocated to the segment node;

receiving, by each of the plurality of segment nodes of the first distributed system, a stream of converted data in the first format, wherein each segment node communicates with a respective extension service to;

obtain fragments in the second format from one or more data nodes of the second distributed system according to the location information for the respective fragments,

convert the obtained fragments from the second format to the first format, and

provide to the segment node of the first distributed system a stream of converted data corresponding to the rows of the external table; and

computing a result for the received query using the stream of converted data corresponding to the rows of the external table.

11. The system of claim 10, wherein requesting, by the first master node of the first distributed system from the first extension service, location information for a plurality of the data fragments occurs after receiving the query.

12. The system of claim 10, wherein the extension service converts the obtained fragments to the first format at scan time.

13. The system of claim 10, wherein the location information includes a network location for a data node of the second distributed system for each fragment of the plurality of fragments.

14. The system of claim 10, wherein the operations further comprise:

receiving, from a user, a definition of the external table representing the multiple data fragments on multiple data nodes in the second distributed system.

15. The system of claim 10, wherein the second distributed system is a NoSQL database.

16. The system of claim 10, wherein the first format is a row-oriented format and wherein the second format is a column-oriented format.

17. The system of claim 10, wherein each segment node of the first distributed system computes a partial result for the query without storing the stream of converted data corresponding to the rows of the external table in nonvolatile memory of the first distributed system.

18. The system of claim 10, wherein each extension service converts the data from the second format to the first format after receiving the query and before computing a result for the received query.

19. A computer program product, encoded on one or more non-transitory computer storage media, comprising instructions that when executed by one or more computers cause the one or more computers to perform operations comprising:

receiving, by a first distributed system that comprises multiple segment nodes, a query that requests rows of an external table representing data stored in multiple data fragments on multiple respective data nodes in a second distributed system, wherein each of the data nodes of the second distributed system are distinct from the segment nodes of the first distributed system, wherein the first distributed system operates under control of a first master node and the second distributed system operates under control of a second master node that is distinct from the first master node, and wherein the second distributed system stores data in a second format that is different from a first format of data stored by the first distributed system;

initiating communication by a plurality of extension services between nodes of the first distributed system and nodes of the second distributed system, including a first extension service communicating with the master nodes of the first and second distributed system and a plurality of second extension services communicating between data nodes of the second distributed system and segment nodes of the first distributed system;

requesting, by the master node of the first distributed system from the first extension service, location information for a plurality of the data fragments in the second distributed system represented by the external table;

allocating each data fragment of the plurality of data fragments to a respective one of the plurality of second extension services and each of the second extension services to one of a plurality of segment nodes of the first distributed system, including providing each segment node with location information for respective fragments of the plurality of fragments allocated to the segment node;

receiving, by each of the plurality of segment nodes of the first distributed system, a stream of converted data in the first format, wherein each segment node communicates with a respective extension service to;

obtain fragments in the second format from one or more data nodes of the second distributed system according to the location information for the respective fragments,

convert the obtained fragments from the second format to the first format, and

provide to the segment node of the first distributed system a stream of converted data corresponding to the rows of the external table; and

computing a result for the received query using the stream of converted data corresponding to the rows of the external table.

20. The computer program product of claim 19, wherein each extension service converts the data from the second format to the first format after receiving the query and before computing a result for the received query.