Hierarchy of servers for query processing of column chunks in a distributed column chunk data store
View Patent ↗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.
1. A method implemented on at least one machine, each of which has at least one processor, storage, and a communication platform, comprising:
accessing, at a first query processing server, a query for data of a table comprising a plurality of columns and a plurality of rows, wherein:
the table has been partitioned into a plurality of column chunks by independently partitioning each column of the table into two or more column chunks based on a predetermined partitioning policy, the predetermined partitioning policy comprising information related to the number of column chunks into which the each column of the table should be partitioned;
a plurality of parity column chunks are created, each parity column chunk being created based on two or more column chunks of the plurality of column chunks; and
the plurality of column chunks and the plurality of parity column chunks have been distributed and stored across a plurality of storage servers;
transforming, by the first query processing server, the query into a plurality of sub-queries;
distributing, by the first query processing server, the plurality of sub-queries to a plurality of second query processing servers;
receiving, at the first query processing server, the data from the plurality of second query processing servers; and
transmitting, by the first query processing server, the data in response to the query.
2. The method of claim 1 , further comprising validating, by the first query processing server, syntax and semantic content of the query.
3. The method of claim 1 , further comprising:
determining, at the plurality of second query processing servers, one or more execution steps for each sub-query;
distributing, by the plurality of second query processing servers, the one or more execution steps determined for each sub-query to the plurality of third query processing servers;
receiving, at the plurality of second query processing servers, one or more sub-results for each sub-query from the plurality of third query processing servers;
combining, at the plurality of second query processing servers, the one or more sub-results for each sub-query to construct the data; and
transmitting, by the plurality of second query processing servers, the data to the first query processing server.
4. The method of claim 3 , wherein when distributing the one or more execution steps determined for each sub-query to the plurality of third query processing servers, for each sub-query, if one or more of the plurality of third query processing servers cache one or more of the plurality of column chunks referenced by the sub-query, then distributing the one or more execution steps determined for the sub-query to the one or more of the plurality of third query processing servers.
5. The method of claim 3 , further comprising:
executing, at the plurality of third query processing servers, the one or more execution steps for each sub-query to retrieve the one or more sub-results for each sub-query from the plurality of storage servers; and
transmitting, by the plurality of third query processing servers, the one or more sub-results for each sub-query to the plurality of second query processing servers.
6. The method of claim 5 , wherein the one or more execution steps for each sub-query are executed in parallel at the plurality of third query processing servers.
7. The method of claim 1 , further comprising:
determining, at the plurality of second query processing servers, one or more execution steps for each sub-query;
distributing, by the plurality of second query processing servers, the one or more execution steps determined for each sub-query to the plurality of storage servers;
receiving, at the plurality of second query processing servers, one or more sub-results for each sub-query from the plurality of storage servers;
combining, at the plurality of second query processing servers, the one or more sub-results for each sub-query to construct the data; and
transmitting, by the plurality of second query processing servers, the data to the first query processing server.
8. The method of claim 7 , further comprising:
executing, at the plurality of storage servers, the one or more execution steps for each sub-query to retrieve the one or more sub-results for each sub-query; and
transmitting, by the plurality of storage servers, the one or more sub-results for each sub-query to the plurality of second query processing servers.
9. The method of claim 1 , further comprising:
computing the plurality of parity column chunks using the plurality of column chunks, wherein each parity column chunk is computed by performing a bitwise XOR on the two or more column chunks; and
distributing and storing the plurality of parity column chunks across the plurality of storage servers, wherein each parity column chunk is stored at a storage server where the two or more column chunks used to compute the parity column chunk are not stored.
10. The method of claim 1 , wherein one or more of the plurality of column chunks stored in the plurality of storage servers have been compressed.
11. A method implemented on at least one machine, each of which has at least one processor, storage, and a communication platform, comprising:
accessing, at a first query processing server, a query for data of a table comprising a plurality of columns and a plurality of rows, wherein:
the table has been partitioned into a plurality of column chunks by independently partitioning each column of the table into two or more column chunks based on a predetermined partitioning policy, the predetermined partitioning policy comprising information related to the number of column chunks into which the each column of the table should be partitioned;
a plurality of parity column chunks are created, each parity column chunk being created based on two or more column chunks of the plurality of column chunks;
the plurality of column chunks and the plurality of parity column chunks have been distributed and stored across a plurality of storage servers; and
one or more of the plurality of column chunks have been cached at the first query processing server;
transforming, by the first query processing server, the query into a plurality of sub-queries, wherein each sub-query references a portion of the data;
separating, by the first query processing server, the plurality of sub-queries into one or more first sub-queries and one or more second sub-queries, wherein each first sub-query references one or more of the plurality of column chunks that have been cached at the first query processing server, and each second sub-query references one or more of the plurality of column chunks that have not been cached at the first query processing server;
distributing, by the first query processing server, the one or more second sub-queries to a plurality of second query processing servers;
processing, by the first query processing server, the one or more first sub-queries using the one or more of the plurality of column chunks cached at the first query processing server to determine a sub-result for each first sub-query;
transmitting, by the first query processing server, the sub-result determined for each first sub-query to the plurality of second query processing servers;
receiving, at the first query processing server, the data from the plurality of second query processing servers; and
transmitting, by the first query processing server, the data in response to the query.
12. The method of claim 11 , further comprising validating, by the first query processing server, syntax and semantic content of the query.
13. The method of claim 11 , further comprising:
determining, at the plurality of second query processing servers, one or more execution steps for each second sub-query;
distributing, by the plurality of second query processing servers, the one or more execution steps determined for each second sub-query to a plurality of third query processing servers;
receiving, at the plurality of second query processing servers, one or more sub-results for each second sub-query from the plurality of third query processing servers;
combining, at the plurality of second query processing servers, the sub-result for each first sub-query and the one or more sub-results for each second sub-query to construct the data; and
transmitting, by the plurality of second query processing servers, the data to the first query processing server.
14. The method of claim 13 , further comprising:
executing, at the plurality of third query processing servers, the one or more execution steps for each second sub-query to retrieve the one or more sub-results for each second sub-query from the plurality of storage servers; and
transmitting, by the plurality of third query processing servers, the one or more sub-results for each second sub-query to the plurality of second query processing servers.
15. The method of claim 14 , wherein the one or more execution steps for each second sub-query are executed in parallel at the plurality of third query processing servers.
16. The method of claim 11 , further comprising:
determining, at the plurality of second query processing servers, one or more execution steps for each second sub-query;
distributing, by the plurality of second query processing servers, the one or more execution steps determined for each second sub-query to the plurality of storage servers;
receiving, at the plurality of second query processing servers, one or more sub-results for each second sub-query from the plurality of storage servers;
combining, at the plurality of second query processing servers, the sub-result for each first sub-query and the one or more sub-results for each second sub-query to construct the data; and
transmitting, by the plurality of second query processing servers, the data to the first query processing server.
17. The method of claim 11 , further comprising:
computing a plurality of parity column chunks using the plurality of column chunks, wherein each parity column chunk is computed by performing a bitwise XOR on the two or more column chunks; and
distributing and storing the plurality of parity column chunks across the plurality of storage servers, wherein each parity column chunk is stored at a storage server where the two or more column chunks used to compute the parity column chunk are not stored.
18. A computer-readable non-transitory storage medium storing computer-readable instructions that, when executed by one or more processors, cause a first query processing server to:
access a query for data of a table comprising a plurality of columns and a plurality of rows, wherein:
the table has been partitioned into a plurality of column chunks by independently partitioning each column of the table into two or more column chunks based on a predetermined partitioning policy, the predetermined partitioning policy comprising information related to the number of column chunks into which the each column of the table should be partitioned;
a plurality of parity column chunks are created, each parity column chunk being created based on two or more column chunks of the plurality of column chunks; and
the plurality of column chunks and the plurality of parity column chunks have been distributed and stored across a plurality of storage servers;
transform the query into a plurality of sub-queries; distribute the plurality of sub-queries to a plurality of second query processing servers;
receive the data from the plurality of second query processing servers; and
transmit the data in response to the query.
19. The media of claim 18 , wherein the computer-readable instructions further causes the first query processing server to:
compute the plurality of parity column chunks using the plurality of column chunks, wherein each parity column chunk is computed by performing a bitwise XOR on the two or more column chunks; and
distribute and storing the plurality of parity column chunks across the plurality of storage servers, wherein each parity column chunk is stored at a storage server where the two or more column chunks used to compute the parity column chunk are not stored.