Hierarchy of Servers for Query Processing of Column Chunks in a Distributed Column Chunk Data Store

• US 20110016127A1
• Filed: 09/20/2010
• Published: 01/20/2011
• Est. Priority Date: 12/19/2005
• Status: Active Grant

• Current Assignee: Google LLC (Alphabet Inc.)
• Original Assignee: Yahoo! Inc. (Apollo Global Management, Inc.)
• Inventors: Uppala, Radha Krishna
• Granted Patent: US 9,280,579 B2
• Time in Patent Office: Days
• US Class Current: 707/737
• CPC Class Codes: G06F 16/2365 Ensuring data consistency and integrity
• CPC Class Codes: G06F 16/245 Query processing
• CPC Class Codes: G06F 16/24532 of parallel queries
• 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/969 Horizontal partitioning
• CPC Class Codes: Y10S 707/973 Horizontal partitioning

Abstract

An improved system and method for query processing in a distributed column chunk data store is provided. A distributed column chunk data store may be provided by multiple storage servers operably coupled to a network. A storage server provided may include a database engine for partitioning a data table into the column chunks for distributing across multiple storage servers, a storage shared memory for storing the column chunks during processing of semantic operations performed on the column chunks, and a storage services manager for striping column chunks of a partitioned data table across multiple storage servers. Query processing may be performed by storage servers or query processing servers operably coupled by a network to storage servers in the column chunk data store. To do so, a hierarchy of servers may be dynamically determined to process execution steps of a query transformed for distributed processing.

Claims

1. A method comprising:

partitioning, by one or more computing devices, each column of a data table comprising one or more columns and one or more rows into two or more column chunks of a plurality of column chunks;

constructing, by the one or more computing devices, a plurality of parity column chunks, wherein each parity column chunk is computed using two or more column chunks of the plurality of column chunks;

individually distributing and storing, by the one or more computing devices, the plurality of column chunks and the plurality of parity column chunks across a plurality of storage servers.

2. The method of claim 1, wherein each parity column chunk of the plurality of parity column chunks is computed by performing a bitwise XOR on two or more column chunks of the plurality of column chunks.

3. The method of claim 1, wherein each parity column chunk of the plurality of parity column chunks is stored at one storage server of the plurality of storage servers where the two or more column chunks used to compute the parity column chunk are not stored.

4. The method of claim 1, further comprising:

receiving a query for one or more data of the data table;

transforming the query into a plurality of sub-queries;

distributing the plurality of sub-queries across a plurality of query servers;

receiving a plurality of sub-results from the plurality of query servers;

constructing a result comprising the one or more data by combining the plurality of sub-results;

transmitting the result in response to the query.

5. The method of claim 4, further comprising:

validating syntax and semantic content of the query; and

determining a plurality of execution steps of the query for distributed processing of the query by the plurality of query servers.

6. The method of claim 4, further comprising assigning one query server of the plurality of query servers to cache one or more column chunks of the plurality of column chunks that contain the one or more data.

7. The method of claim 4, further comprising:

processing, at each query server of the plurality of query servers, one or more sub-queries received from the first query server; and

transmitting, by each query server of the plurality of query servers, a sub-result comprising at least one of the one or more data to the first query server.

8. The method of claim 1, further comprising:

receiving a query for one or more data of the data table;

transforming the query into a plurality of sub-queries;

distributing the plurality of sub-queries across the plurality of storage servers;

receiving a plurality of sub-results from the plurality of storage servers respectively;

constructing a result comprising the one or more data by combining the plurality of sub-results;

transmitting the result in response to the query.

9. The method of claim 8, further comprising:

validating syntax and semantic content of the query; and

determining a plurality of execution steps of the query for distributed processing of the query by the plurality of storage servers.

10. The method of claim 8, further comprising:

processing, at each storage server of the plurality of storage servers, one or more sub-queries received from the query server; and

transmitting, by each storage server of the plurality of storage servers, a sub-result comprising at least one of the one or more data to the query server.

11. A method comprising:

partitioning, by a first server, each column of a data table comprising one or more columns and one or more rows into two or more column chunks of a plurality of column chunks;

constructing, by the first server, a plurality of parity column chunks, wherein each parity column chunk is computed using two or more column chunks of the plurality of column chunks;

individually distributing and storing, by the first server, the plurality of column chunks and the plurality of parity column chunks across a plurality of second servers;

receiving, at the first server, a request for one or more data of the data table;

transforming, by the first server, the query into a plurality of sub-queries;

distributing, by the first server, the plurality of sub-queries across a plurality of third servers;

receiving, at the first server, a plurality of sub-results from the plurality of third servers;

constructing, by the first server, a result comprising the one or more data by combining the plurality of sub-results;

transmitting, by the first server, the result in response to the query.

12. The method of claim 11, wherein each parity column chunk of the plurality of parity column chunks is computed by performing a bitwise XOR on two or more column chunks of the plurality of column chunks.

13. The method of claim 11, wherein each parity column chunk of the plurality of parity column chunks is stored at one second server of the plurality of second servers where the two or more column chunks used to compute the parity column chunk are not stored.

14. The method of claim 11, further comprising:

validating, by the first server, syntax and semantic content of the query; and

determining, by the first server, a plurality of execution steps of the query for distributed processing of the query by the plurality of third query servers.

15. The method of claim 11, further comprising assigning, by the first server, one third server of the plurality of third servers to cache one or more column chunks of the plurality of column chunks that contain the one or more data.

16. The method of claim 11, further comprising:

processing, by each third server of the plurality third servers, one or more sub-queries received from the first server;

distributing, by each third server of the plurality third servers, the one or more sub-queries received from the first server to one or more fourth servers;

transmitting, by each third server of the plurality third servers, one sub-result of the plurality of sub-results to the first server, wherein the one sub-result is determined based on processing performed by the one or more fourth servers;

17. The method of claim 11, further comprising:

processing, by each third server of the plurality third servers, one or more sub-queries received from the first server;

distributing, by each third server of the plurality third servers, the one or more sub-queries received from the first server to one or more second servers of the plurality of second servers;

transmitting, by each third server of the plurality third servers, one sub-result of the plurality of sub-results to the first server, wherein the one sub-result is determined based on processing performed by the one or more second servers;

18. One or more computer-readable non-transitory storage media embodying software operable when executed by one or more computer systems to:

partition each column of a data table comprising one or more columns and one or more rows into two or more column chunks of a plurality of column chunks;

construct a plurality of parity column chunks, wherein each parity column chunk is computed using two or more column chunks of the plurality of column chunks;

individually distribute and store the plurality of column chunks and the plurality of parity column chunks across a plurality of storage servers.

19. The media of claim 18, wherein the software is further operable when executed by one or more computer systems to:

receive a query for one or more data of the data table;

transform the query into a plurality of sub-queries;

distribute the plurality of sub-queries across a plurality of query servers;

receive a plurality of sub-results from the plurality of query servers;

construct a result comprising the one or more data by combining the plurality of sub-results;

transmit the result in response to the query.

20. The media of claim 18, wherein the software is further operable when executed by one or more computer systems to:

receive a query for one or more data of the data table;

transform the query into a plurality of sub-queries;

distribute the plurality of sub-queries across the plurality of storage servers;

receive a plurality of sub-results from the plurality of storage servers respectively;

construct a result comprising the one or more data by combining the plurality of sub-results;

transmit the result in response to the query.