U.S. patent application number 15/450389 was filed with the patent office on 2017-06-22 for self-described query execution in a massively parallel sql execution engine.
The applicant listed for this patent is EMC IP Holding Company LLC. Invention is credited to Lei Chang, Lirong Jian, Luke Lonergan, Lili Ma, Tao Ma, Zhanwei Zhanwei.
Application Number | 20170177665 15/450389 |
Document ID | / |
Family ID | 51389310 |
Filed Date | 2017-06-22 |
United States Patent
Application |
20170177665 |
Kind Code |
A1 |
Chang; Lei ; et al. |
June 22, 2017 |
SELF-DESCRIBED QUERY EXECUTION IN A MASSIVELY PARALLEL SQL
EXECUTION ENGINE
Abstract
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.
Inventors: |
Chang; Lei; (Beijing,
CN) ; Zhanwei; Zhanwei; (Beijing, CN) ; Ma;
Tao; (Beijing, CN) ; Lonergan; Luke; (San
Carlos, CA) ; Jian; Lirong; (Beijing, CN) ;
Ma; Lili; (Beijing, CN) |
|
Applicant: |
Name |
City |
State |
Country |
Type |
EMC IP Holding Company LLC |
Hopkinton |
MA |
US |
|
|
Family ID: |
51389310 |
Appl. No.: |
15/450389 |
Filed: |
March 6, 2017 |
Related U.S. Patent Documents
|
|
|
|
|
|
Application
Number |
Filing Date |
Patent Number |
|
|
13853060 |
Mar 29, 2013 |
9626411 |
|
|
15450389 |
|
|
|
|
61769043 |
Feb 25, 2013 |
|
|
|
Current U.S.
Class: |
1/1 |
Current CPC
Class: |
G06F 16/182 20190101;
H05K 999/99 20130101; H04L 65/60 20130101; G06F 16/11 20190101;
G06F 16/24524 20190101; G06F 16/1858 20190101; G06F 16/148
20190101; G06F 16/43 20190101; G06F 16/2471 20190101; G06F 16/27
20190101; H04L 67/1097 20130101; G06F 16/10 20190101; G06F 16/113
20190101; G06F 16/217 20190101; G06F 16/245 20190101; G06F 16/2455
20190101; G06F 16/24532 20190101; G06F 16/24542 20190101; G06F
16/907 20190101; G06F 16/2453 20190101 |
International
Class: |
G06F 17/30 20060101
G06F017/30 |
Claims
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 and other information
needed by one or more segments to execute the self-described query
plan, wherein the metadata and other information comprise
information relating to locations of data in the distributed
storage in connection with execution of the self-described query
plan; communicating the self-described query plan to the one or
more segments for execution; and receiving a result associated with
execution of the self-described query plan from the one or more
segments, 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 is query
plan.
2. The method of claim 1, wherein the metadata and other
information further comprises catalog information for functions and
operators used in the self-described query plan for processing the
data.
3. The method of claim 1, wherein the metadata and other
information are stored in a store at the master node.
4. The method of claim 1, wherein in the event 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.
5. The method of claim 1, wherein the segments comprise stateless
servers, and wherein the self-described query plan comprise all
metadata and information needed for execution of the self-described
query plan such that all the metadata and information needed for
execution is contained in the self-described query plan.
6. 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.
7. 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.
8. The method of claim 7, 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.
9. 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.
10. 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 wherein the self-described query plan is executed
in parallel on the distributed server nodes.
11. The method of claim 1, wherein the cluster comprises
distributed relational database nodes.
12. 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.
13. The method of claim 10, wherein the self-described query plan
slice and 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.
14. The method of claim 1, further comprising: compressing the
self-described query plan before the broadcasting of the
self-described query plan.
15. The method of claim 1, wherein the query plan comprises an
identifier associated with information that is stored at the
cluster of multiple distributed segments.
16. A system, comprising: one or more processors configured to:
generate a self-described query plan at the master node that is
responsive to a query, wherein the self-described query plan
comprises metadata and other information needed by one or more
segments to execute the self-described query plan, wherein the
metadata and other information comprise information relating to
locations of data in the distributed storage in connection with
execution of the self-described query plan; communicate the
self-described query plan to the one or more segments for
execution; and receive a result associated with execution of the
self-described query plan from the one or more segments, 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; and a memory coupled to
the processor and configured to provide instructions to the one or
more processors.
17. Computer readable storage media for storing executable
instructions for controlling the 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 and other
information needed by one or more segments to execute the
self-described query plan, wherein the metadata and other
information comprise information relating to locations of data in
the distributed storage in connection with execution of the
self-described query plan; communicating the self-described query
plan to the one or more segments for execution; and receiving a
result associated with execution of the self-described query plan
from the one or more segments, 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.
Description
CROSS REFERENCE TO OTHER APPLICATIONS
[0001] This application is a continuation of co-pending U.S. patent
application Ser. No. 13/853,060, entitled SELF-DESCRIBED QUERY
EXECUTION IN A MASSIVELY PARALLEL SQL EXECUTION ENGINE filed Mar.
29, 2013 which is incorporated herein by reference for all
purposes, which claims priority to U.S. Provisional Application No.
61/769,043, entitled INTEGRATION OF MASSIVELY PARALLEL PROCESSING
WITH A DATA INTENSIVE SOFTWARE FRAMEWORK filed Feb. 25, 2013 which
is incorporated herein by reference for all purposes.
BACKGROUND OF THE INVENTION
[0002] This invention relates generally to massively parallel
processing (MPP) data storage systems and methods for big data
applications, and more particularly to new and improved MPP system
architectures comprising large clusters of commodity servers, and
associated query execution models for accessing data in such
systems.
[0003] Most successful companies use data to their advantage. The
data are no longer easily quantifiable facts, such as point of sale
transaction data. Rather, companies retain, explore, analyze, and
manipulate all the available information in their purview.
Ultimately, they may analyze the data to search for evidence of
facts, and insights that lead to new business opportunities or
which leverage their existing strengths. This is the business value
behind what is often referred to as "Big Data".
[0004] Big data is "big" because it comprises massive quantities,
frequently hundreds of terabytes or more, of both structured and
unstructured data. Among the problems associated with such big data
is the difficulty of quickly and efficiently analyzing the data to
obtain relevant information. Conventional relational databases
store structured data and have the advantage of being compatible
with the structured query language (SQL), a widely used powerful
and expressive data analysis language. Increasingly, however, much
of big data is unstructured or multi-structured data for which
conventional relational database architectures are unsuited, and
for which SQL is unavailable. This has prompted interest in other
types of data processing platforms.
[0005] The Apache Software Foundation's open source Hadoop
distributed file system (HDFS) has rapidly emerged as one of the
preferred solution for big data analytics applications that grapple
with vast repositories of unstructured or multi-structured data. It
is flexible, scalable, inexpensive, fault-tolerant, and is well
suited for textual pattern matching and batch processing, which has
prompted its rapid rate of adoption by big data. HDFS is a simple
but extremely powerful distributed file system that can be
implemented on a large cluster of commodity servers with thousands
of nodes storing hundreds of petabytes of data, which makes it
attractive for storing big data. However, Hadoop is a non-SQL
compliant, and, as such, does not have available to it the richness
of expression and analytic capabilities of SQL systems. SQL based
platforms are better suited to near real-time numerical analysis
and interactive data processing, whereas HDFS is better suited to
batch processing of large unstructured or multi-structured data
sets.
[0006] A problem with such distinctly different data processing
platforms is how to combine the advantages of the two platforms by
making data resident in one data store available to the platform
with the best processing model. The attractiveness of Hadoop in
being able to handle large volumes of multi-structured data on
commodity servers has led to its integration with MapReduce, a
parallel programming framework that integrates with HDFS and allows
users to express data analysis algorithms in terms of a limited
number of functions and operators, and the development of SQL-like
query engines, e.g., Hive, which compile a limited SQL dialect to
interface with MapReduce. While this addresses some of the
expressiveness shortcomings by affording some query functionality,
it is slow and lacks the richness and analytical power of true SQL
systems.
[0007] One reason for the slowness of HDFS with MapReduce is the
necessity for access to metadata information needed for executing
queries. In a distributed file system architecture such as HDFS the
data is distributed evenly across the multiple nodes. If the
metadata required for queries is also distributed among many
individual metadata stores on the multiple distributed nodes, it is
quite difficult and time-consuming to maintain consistency in the
metadata. An alternative approach is to use a single central
metadata store that can be centrally maintained. Although a single
metadata store can be used to address the metadata consistency
problem, it has been impractical in MPP database systems. A single
central metadata store is subject to large numbers of concurrent
accesses from multiple nodes running parallel queries, such as is
the case with HDFS, and this approach does not scale well. The
system slows rapidly as the number of concurrent accesses to the
central store increases. Thus, while HDFS has many advantages for
big data applications, it also has serious performance
disadvantages. A similar problem exists in using a central metadata
store in conventional MPP relational databases that requires large
numbers of concurrent access. What is needed is a different
execution model and approach for executing queries in such
distributed big data stores.
[0008] It is desirable to provide systems and methods that afford
execution models and approaches for massively parallel query
processing in distributed file systems that address the foregoing
and other problems of MPP distributed data storage systems and
methods, and it is to these ends that the present invention is
directed.
BRIEF DESCRIPTION OF THE DRAWINGS
[0009] Various embodiments of the invention are disclosed in the
following detailed description and the accompanying drawings.
[0010] FIG. 1 is a diagrammatic view of the architecture of an MPP
database system embodying the invention;
[0011] FIG. 2 illustrates in more detail the architecture of a
preferred embodiment of an MPP database in accordance with the
invention;
[0012] FIG. 3 is a diagrammatic view illustrating an embodiment of
a primary master of the system of FIGS. 1; and
[0013] FIG. 4 is a diagrammatic view illustrating an example of a
slicing operation of the invention.
DETAILED DESCRIPTION
[0014] The invention can be implemented in numerous ways, including
as a process; an apparatus; a system; a composition of matter; a
computer program product embodied on a computer readable storage
medium; and/or a processor, such as a processor configured to
execute instructions stored on and/or provided by a memory coupled
to the processor. In this specification, these implementations, or
any other form that the invention may take, may be referred to as
techniques. In general, the order of the steps of disclosed
processes may be altered within the scope of the invention. Unless
stated otherwise, a component such as a processor or a memory
described as being configured to perform a task may be implemented
as a general component that is temporarily configured to perform
the task at a given time or a specific component that is
manufactured to perform the task. As used herein, the term
`processor` refers to one or more devices, circuits, and/or
processing cores configured to process data, such as computer
program instructions.
[0015] A detailed description of one or more embodiments of the
invention is provided below along with accompanying figures that
illustrate the principles of the invention. The invention is
described in connection with such embodiments, but the invention is
not limited to any embodiment. The scope of the invention is
limited only by the claims and the invention encompasses numerous
alternatives, modifications and equivalents. Numerous specific
details are set forth in the following description in order to
provide a thorough understanding of the invention. These details
are provided for the purpose of example and the invention may be
practiced according to the claims without some or all of these
specific details. For the purpose of clarity, technical material
that is known in the technical fields related to the invention has
not been described in detail so that the invention is not
unnecessarily obscured.
[0016] This invention is particularly well adapted for use with a
new MPP database system of the assignee of this invention
comprising the native integration of a Hadoop HDFS and a massively
parallel SQL distributed database with a massively parallel SQL
query engine that affords true (full) SQL processing for Hadoop,
and will be described in that context. It will be appreciated,
however, that this is illustrative of only one utility of the
invention and that the invention has applicability to other types
of systems and methods.
[0017] FIG. 1 is a diagrammatic view that illustrates the
architecture of an MPP database system 100 in which the invention
may be employed. The system 100 may comprise a large cluster of a
plurality of segment hosts 110 which may be commodity servers. The
servers may each comprise a central processing unit (CPU), memory,
storage and input/output (I/O) which may all be conventional and
are not shown in the figure. The segment hosts may be interfaced
via a network interconnect 114 to a primary master host 118 and a
standby master host 120, and the hosts are also interfaced to a
distributed storage layer 124 that stores user data. The
distributed storage layer may comprise a distributed file system,
such as the Hadoop distributed file system (HDFS), running on the
segment hosts, or may comprise a cluster of distributed relational
database nodes. Each segment host 110 may include a plurality of
segments 130 which may be computing units of a distributed database
engine formed by the cluster. The segments may be programs
executing on the segment hosts servers. The programs comprise
executable instructions embodied in computer readable media such as
physical memory or other computer program product for controlling
the servers to perform operations in accordance with the invention
as described herein.
[0018] The primary master 118, as will be described in more detail
below, may be responsible for accepting queries from a client 134,
planning queries, dispatching query plans to the segments for
execution on the stored data in the distributed storage layer, and
collecting the query results from the segments. The standby master
120 may be a warm backup for the primary master that takes over if
the primary master fails. The primary master and the standby master
may also be servers comprising conventional CPU, storage, memory
and I/O components that execute instructions embodied in memory or
other physical non-transitory computer readable storage media to
perform the operations in accordance with the invention described
herein. In addition to interfacing the segment hosts to the primary
master and standby master, the network interconnect 114 also
communicates tuples between execution processes on the
segments.
[0019] As will be described in more detail below, when the primary
master receives a query, it parses, optimizes and plans the query
using a query planner and optimizer, which in a preferred
embodiment are a SQL query planner and optimizer, and dispatches a
query plan to the segments for execution. In accordance with the
invention, after the query planning phase and prior to dispatch,
the primary master converts the query plan to a self-described
query plan that may comprise multiple slices. The self-described
query plan is a self-contained plan that includes all of the
information and metadata needed by each segment for execution of
the plan. The self-described query plan includes, for instance, the
locations of the files that store the tables accessed in the plan
and the catalog information for the functions, operators and other
objects in the query plan. In one embodiment, the information may
also include the functions needed for processing the data. This
information may be stored in a central metadata store 140 in the
local file system or storage of the primary master from which it
may be retrieved and inserted into the query plan to form the
self-described query plan.
[0020] There are a number of advantages to the invention. Since the
self-described query plan is self-contained, it may contain all of
the information need for its execution. This obviates the need for
any access by a segment to a central metadata store in order to
execute the plan, thereby avoiding large volumes of network traffic
and correspondingly slower response times, and avoids the necessity
of the segment hosts storing the necessary metadata locally.
Moreover, metadata is typically small and conveniently stored in
one central location. Therefore, metadata consistency can be easily
maintained. Furthermore, since the metadata may be stored in a
local file system on the primary master node the insertion of the
metadata into the self-described query plan 136 is fast. Following
generation the self-described query plan may be broadcast to the
segments 130 for execution, as indicated in the figure. In
accordance with the invention, the segments may be stateless, i.e.,
they act as slave workers and have no need to maintain any state
information for persistent files, functions and so on. This
advantageously permits the segments to be relatively simple and
fast.
[0021] Several optimizations are possible. One optimization is that
the self-described query plan may be compressed prior to dispatch
to decrease the network costs for broadcasting the plan. For a deep
sliced query on partitioned tables, the query plan size may be
quite large, for example, more than 100 MB. It is preferable to
decrease the size of the query plan that must be broadcast by
compressing the plan. In a preferred embodiment, a local read-only
cache may be maintained on each segment (as will be described) to
store static read-only information that seldom if ever changes,
such as type information, built-in function information, query
operators, etc. This information may include the functions and
operators needed to execute the self-described query plan, e.g.,
Boolean functions such as SUM, and various query operators so that
the plan itself need only contain a reference to the functions and
operators, and identifiers for their argument or object values. The
read-only cache may be initialized during a system bootstrap phase.
Then, when a self-described query plan is constructed at the
master, changeable metadata and information maintained in the
master may be incorporated into the plan. Since it is known that
any static read-only information may be obtained from the local
caches on the segments, it is unnecessary to place this information
into the plan that is broadcast.
[0022] FIG. 2 illustrates in more detail the architecture of an MPP
database 200 that integrates a Hadoop HDFS distributed file system
and a parallel distributed relational database. Instead of writing
to discs as in a traditional database, the data is written to the
HDFS distributed file system. The HDFS file system may be a
traditional hierarchical file organization that comprises
directories and allows users and applications to store files inside
of these directories. A plurality of segment hosts 210 are
interfaced via a network interconnect 214 to a master host 218. The
segment hosts may comprise commodity servers as described above for
FIG. 1, and each segment host may comprise a plurality of segments
220. Each segment reads and writes data to a DataNode 226 where the
data is stored in a plurality of data blocks 224 in its local file
system. Each data block may store 64 MB of data. The DataNodes 226
are responsible for serving read and write requests from file
system clients, and may also perform functions such as block
creation, deletion and replication. The system may further include
a NameNode 230 which serves as a master for the HDFS. The NameNode
stores and maintains the HDFS namespace, and may maintain a
transaction log that persistently records changes that occur to the
file system metadata. It also may maintain an image of the entire
file system namespace and a block map in its memory. The DataNodes
226 and the NameNode 230 may comprise software programs that run on
the servers. As described in connection with FIG. 1, the master
host 218 may accept queries from clients, optimize, plan and
dispatch self-directed query plans to the segments for execution,
and receive the results of the queries from the segments.
[0023] FIG. 3 illustrates in more detail the construction and
processing of a self-described query plan in accordance with the
invention. As previously described, and as shown in the figure, a
primary master 300 may include a SQL query parser and optimizer 310
for parsing and optimizing an input query to produce a SQL query
plan, and may include a query dispatcher 314 for broadcasting the
resulting query plan 316 to a plurality of segments 320 in a large
cluster of segment hosts 324. Each segment includes a plurality of
query executors 330 comprising processes that run on the segments
to execute the query plan that is dispatched to the segments.
Preferably, each segment also includes a read-only cache 334 that
stores static read-only information such as type information,
built-in function information, etc., as described above. After the
parser and optimizer 310 generates a query plan 316, the query plan
is converted into multiple slices 340, where each slice represents
a unit of work that can be performed separately. For each slice,
there is a query executor 330 in each segment. Once the query plan
has been produced and sliced, the number of query executors needed
to execute the query plan is known. The query dispatcher 314 may
then inform the segments to set up the number of query executors
330 required for execution. Next, the metadata information 344
required to execute the query plan, such as file location, catalog
information, etc., may be inserted into the query plan as
previously described to convert the query plan into a
self-described query plan. The query dispatcher 314 may then
broadcast the self-described query plan to all query executors in
all segments.
[0024] When each query executor 330 receives the query plan, it can
determine what command in the query plan that query executor should
process. Because the query executors are set up by the query
dispatcher after the self-described query plan has been generated,
the query dispatcher knows the slice number and segment index
information and may include this information in the query plan that
is broadcast. Upon receiving the query plan, a query executor can
look up the metadata information that the query executor needs to
process the command. This metadata may be found either from the
self-described query plan itself or from the read-only data in the
read-only cache 334 in the segment. The self-described query plan
may include an identifier or other indicator that designates the
read-only information stored in the read-only cache that the query
executor needs to execute the command. Once the query executor
executes the command, it may follow the information in the command
and send the results to a next-indicated query executor or return
the results to the query dispatcher in the primary master. As will
be appreciated, broadcasting the self-described query plan to each
segment is a new execution model and process that enables full SQL
functionality to be extended to distributed file systems such as
RDFS.
[0025] FIG. 4 illustrates an example of slicing and processing a
self-described query plan in accordance with the invention. In the
example illustrated in the figure, the query plan is sliced into
three slices, and each slice is to be executed on a query executor
in two segments. The first slice (Slice 1) comprises process steps
410 and 412. The second slice (Slice 2) comprises steps 420-428.
The third slice (Slice 3) comprises steps 430 and 432.
[0026] After the master generates the query plan shown in FIG. 4,
the query dispatcher will communicate with the two segments in the
cluster and inform the segments to establish three query executor
processes.
[0027] As shown in the figure, the query plan execution begins for
Slice 1 at 410 with a sequential scan on a table containing sales
data by query executors on two segments. Each query executor will
only perform the sequential scan command on its own table data. For
example, the query executor for Slice 1 on segment 1 will only
perform the sequential scan on the table sales data for segment 1.
Information on the portion of the table the query executor should
process is obtained from the metadata information embedded in the
self-described query plan. After the query executor performs the
sequential scan for Slice 1 at 410, it will perform a redistribute
motion at 412 to distribute the data to the query executors on the
two segments for Slice 2. Similarly, for Slice 2 the query
executors will perform a sequential scan of a table of customer
data at 420, hash the results at 422, perform a hash join operation
at 424 with results of the redistribute motion operation at 412,
execute a hash aggregate command at 426, and a perform a
redistribute motion operation at 428. Finally, for Slice 3, the
query executors will execute a hash aggregate command at 430 and a
gather motion command at 422 to gather the results.
[0028] In each process step of FIG. 4, the query plan will either
directly include the metadata or other information needed by the
query executors on a segment to execute the process step, or an
identifier to the required metadata or information in the read-only
cache on the segment.
[0029] As will be appreciated, the invention affords a new
self-described query execution model and process that has
significant advantages in increasing the speed and functionality of
MPP data stores by decreasing network traffic and affording better
control of metadata consistency. The process of the invention
allows full advantage to be taken of Hadoop HDFS and other
distributed file systems for multi-structured big data by extending
to such systems a massively parallel execution SQL engine and the
functionality.
[0030] While the foregoing has been with respect to preferred
embodiments of the invention, it will be appreciated that changes
to these embodiments may be made without departing from the
principles and spirit of the invention, the scope of which is
defined in the appended claims.
[0031] Although the foregoing embodiments have been described in
some detail for purposes of clarity of understanding, the invention
is not limited to the details provided. There are many alternative
ways of implementing the invention. The disclosed embodiments are
illustrative and not restrictive.
* * * * *