Self-described query execution in a massively parallel SQL execution engine
A query is executed in a massively parallel processing data storage system comprising a master node communicating with a cluster of multiple segments that access data in distributed storage by producing a self-described query plan at the master node that incorporates changeable metadata and information needed to execute the self-described query plan on the segments, and that incorporates references to obtain static metadata and information for functions and operators of the query plan from metadata stores on the segments. The distributed storage may be the Hadoop distributed file system, and the query plan may be a full function SQL query plan.
1. A method of query execution in a massively parallel processing (MPP) data storage system comprising a master node and a cluster of multiple distributed segments that access data in distributed storage, comprising:
generating a self-described query plan at the master node that is responsive to a query, wherein:
the self-described query plan comprises metadata or other information needed by one or more segments to execute the self-described query plan;
the metadata or other information comprises information relating to location of data in the distributed storage in connection with execution of the self-described query plan;
the master node maintains a transaction log that records information for changes to file system metadata pertaining a namespace,
the namespace is based at least in part on the cluster of multiple distributed segments; and
the metadata comprised in the self-described query plan is based at least in part on information recorded in the transaction log;
broadcasting the self-described query plan to the one or more segments for execution, wherein the self-described query plan comprises information respectively identifying the one or more segments to execute the self-described query plan; and
receiving a result associated with execution of the self-described query plan from the one or more segments.
2. The method of claim 1 , wherein the self-described query plan is executed such that the one or more segments do not access a central metadata store in connection with executing the self-described query plan.
3. The method of claim 1 , wherein the metadata or other information further comprises catalog information for functions and operators used in the self-described query plan for processing the data.
4. The method of claim 1 , wherein the metadata or other information is stored in a store at the master node.
5. The method of claim 1 , wherein in response to a determination that a part of the metadata or a part of the other information needed by the segments to execute the query plan is stored at the cluster of multiple distributed segments, the master node includes an identifier associated with the part of the metadata or the part of the other information that is stored at the cluster of multiple distributed segments and excludes the part of the metadata or the part of the other information that is stored at the cluster of multiple distributed segments from the query plan.
6. The method of claim 1 , wherein the segments comprise stateless servers, and the self-described query plan comprise all metadata or other information needed for execution of the self-described query plan such that all the metadata or the other information needed for execution is contained in the self-described query plan.
7. The method of claim 1 , further comprising storing the metadata and other information that is changeable in a central store at the master node, and further storing, at each of the one or more segments, static information comprising function and query operator information needed by the one or more segments for executing the self-described query plan.
8. The method of claim 1 , further comprising storing static information comprising type and built-in function and operator information in a read-only cache at each segment, and accessing the static information in the read-only cache of a segment by the self-described query plan as needed for execution of the self-described query plan on such segment.
9. The method of claim 8 , wherein the generating of the self-described query plan comprises incorporating in the self-described query plan a reference to the static information with which one of the distributed segments accesses the static information from the read-only cache as needed for execution of the self-described query plan.
10. The method of claim 1 , wherein the self-described query plan comprises a SQL query plan, and the method further comprises storing, at each of the one or more segments, static SQL function and operator information needed by the segments to execute the SQL query plan.
11. The method of claim 1 , wherein the cluster comprises distributed server nodes hosting the one or more segments, and the distributed storage comprises a distributed file system on the server nodes, and the self-described query plan is executed in parallel on the distributed server nodes.
12. The method of claim 1 , wherein the cluster comprises distributed relational database nodes.
13. The method of claim 1 , wherein the master node comprises a query planner configured to generate the self-described query plan, and the method further comprises slicing by the master node the self-described query plan into slices, informing each of the one or more segments to set up a query executor for each slice on each segment.
14. The method of claim 13 , wherein the self-described query plan slice and corresponding segment index information for each such segment to identify the metadata that each query executor needs to access to execute the self-described query plan is included in the self-described query plan.
15. The method of claim 1 , further comprising:
compressing the self-described query plan before the broadcasting of the self-described query plan.
16. The method of claim 1 , wherein the query plan comprises an identifier associated with information that is stored at the one or more segments.
17. A system, comprising:
one or more processors configured to:
generate a self-described query plan at a master node that is responsive to a query, wherein:
the self-described query plan comprises metadata or other information needed by one or more segments to execute the self-described query plan;
the metadata or other information comprises information relating to location of data in a distributed storage in connection with execution of the self-described query plan;
the master node maintains a transaction log that records information for changes to file system metadata pertaining a namespace,
the namespace is based at least in part on a cluster of multiple distributed segments; and
the metadata comprised in the self-described query plan is based at least in part on information recorded in the transaction log;
broadcast the self-described query plan to the one or more segments for execution, wherein the self-described query plan comprises information respectively identifying the one or more segments to execute the self-described query plan; and
receive a result associated with execution of the self-described query plan from the one or more segments; and
a memory coupled to the processor and configured to provide instructions to the one or more processors.
18. A non-transitory computer readable storage media for storing executable instructions for controlling an operation of one or more computers in a massively parallel processing (MPP) data storage system comprising a master node and a cluster of multiple distributed segments that access data in distributed storage to perform a method of query execution comprising:
generating a self-described query plan at the master node that is responsive to a query, wherein:
the self-described query plan comprises metadata or other information needed by one or more segments to execute the self-described query plan;
the metadata or other information comprises information relating to location of data in the distributed storage in connection with execution of the self-described query plan;
the master node maintains a transaction log that records information for changes to file system metadata pertaining a namespace,
the namespace is based at least in part on the cluster of multiple distributed segments; and
the metadata comprised in the self-described query plan is based at least in part on information recorded in the transaction log;
broadcast the self-described query plan to the one or more segments for execution, wherein the self-described query plan comprises information respectively identifying the one or more segments to execute the self-described query plan; and
receiving a result associated with execution of the self-described query plan from the one or more segments.
19. The method of claim 1 , wherein the information respectively identifying the one or more segments to execute the self-described query plan that is comprised in the self-described query plan corresponds to a slice number or segment index information pertaining to the one or more segments that are to execute the self-described query plan.
20. The method of claim 1 , wherein the self-described query plan is broadcast to all segments in the cluster of multiple distributed segments.