# US 2014/0195558 A1 · System and method for distributed database query engines > Agents: this data is available as JSON from the patentgrep API and MCP server — see https://patentgrep.com/llms.txt - Number: US 2014/0195558 A1 - Document ID: US20140195558A1 - Status: Published application - Kind: A1 - Application number: 13/735,820 - Filed: 2013-01-07 - Published: 2014-07-10 - Earliest claimed: 2013-01-07 - Assignees: Facebook, Inc. - Inventors: Raghotham Murthy; Rajat Goel - Web page: https://patentgrep.com/patent/US20140195558A1 ## CPC - G06F 16/2471: {Distributed queries} - G06F 16/2358: {Change logging, detection, and notification (replication G06F16/27)} - G06F 16/24539: {using cached or materialised query results} - G06F 16/24552: {Database cache management} - G06F 16/951: Indexing; Web crawling techniques - G06F 16/9535: Search customisation based on user profiles and personalisation - G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N) ## Abstract Techniques for a system capable of performing low-latency database query processing are disclosed herein. The system includes a gateway server and a plurality of worker nodes. The gateway server is configured to divide a database query, for a database containing data stored in a distributed storage cluster having a plurality of data nodes, into a plurality of partial queries and construct a query result based on a plurality of intermediate results. Each worker node of the plurality of worker nodes is configured to process a respective partial query of the plurality of partial queries by scanning data related to the respective partial query that stored on at least one data node of the distributed storage cluster and generate an intermediate result of the plurality of intermediate results that is stored in a memory of that worker node. ## Claims ### Claim 1 A system comprising: a gateway server configured to generate a plurality of partial queries from a database query for a database containing data stored in a distributed storage cluster that has a plurality of data nodes, and to construct a query result based on a plurality of intermediate results; and a plurality of worker nodes, wherein each worker node of the plurality of worker nodes is configured to process a respective partial query of the plurality of partial queries by scanning data related to the respective partial query and stored on at least one data node of the distributed storage cluster, and wherein each worker node of the plurality of worker nodes is further configured to generate an intermediate result of the plurality of intermediate results that is stored in a memory of that worker node. ### Claim 2 (depends on 1) The system of claim 1, wherein each worker node of the plurality of worker nodes is further configured to process the respective partial query of the plurality of partial queries by scanning a portion of the data related to the respective partial query that is stored on the at least one data node of the distributed storage cluster and to generate an approximate intermediate result that is stored in the memory of that worker node. ### Claim 3 (depends on 2) The system of claim 2, wherein the gateway server is further configured to construct an approximate query result based on at least one approximate intermediate result. ### Claim 4 (depends on 1) The system of claim 1, wherein the gateway server is further configured to construct an approximate query result based on a portion of the plurality of intermediate results. ### Claim 5 (depends on 1) The system of claim 1, wherein the gateway server is further configured to identify a straggling worker node, further divide a partial query that is assigned to the straggling worker node into a plurality of subordinate partial queries, and assign the plurality of subordinate partial queries to some of the plurality of worker nodes, wherein the straggling worker node is a worker node that either fails to report a rate of progress to the gateway server or reports the rate of progress below a predetermined value after a predetermined time period to the gateway server. ### Claim 6 (depends on 1) The system of claim 1, wherein each worker node of the plurality of the worker nodes is a service running a respective data node within the distributed storage cluster. ### Claim 7 (depends on 1) The system of claim 1, further comprising: a metadata cache configured to cache table level metadata of the database and file level metadata of the distributed storage cluster. ### Claim 8 (depends on 7) The system of claim 7, wherein the metadata cache is configured to retain cached metadata from a previous database query for the database query. ### Claim 9 (depends on 1) The system of claim 1, wherein each worker node of the plurality of the worker nodes periodically sends heartbeat messages to the gateway server to report status of a partial query processing by that worker node. ### Claim 10 (depends on 1) The system of claim 1, wherein the gateway server is further configured to receive an instruction from a client device to return an approximate query result or terminate a processing of the database query. ### Claim 11 (depends on 1) The system of claim 1, wherein the gateway server is further configured to instruct the worker nodes to immediately return approximate intermediate results, and to return an approximate query result based on the approximate intermediate results to a client device. ### Claim 12 (depends on 1) The system of claim 1, wherein the database query includes a request for an approximate query result. ### Claim 13 (depends on 1) The system of claim 1, wherein the query result is accompanied by an indication of a portion of related data stored in the data nodes that has been scanned for the query result. ### Claim 14 (depends on 1) They system of claim 1, wherein the database is a Hive data warehouse system and the distributed storage cluster is a Hadoop cluster. ### Claim 15 A method comprising: receiving from a client device a database query for a database containing data stored in a distributed storage cluster that has a plurality of data nodes; dividing the database query into a plurality of partial queries; sending each of the partial queries to a respective worker node of a plurality of worker nodes, wherein each worker node is a service running on a data node of the distributed storage cluster; retrieving a plurality of intermediate results for the partial queries from the worker nodes, wherein each intermediate result is processed by a respective worker node of the worker nodes by scanning related data stored in a data node on which the perspective worker node runs; and generating a query result based on the plurality of intermediate results. ### Claim 16 (depends on 15) The method of claim 15, further comprising: returning the query result along with a portion indicator to the client device, wherein the portion indicator indicates the portion of related data stored in the data nodes that has been scanned for the query result. ### Claim 17 (depends on 15) The method of claim 15, further comprising: instructing the worker nodes to immediately return approximate query results; and wherein the step of retrieving comprises: retrieving a plurality of approximate intermediate results for the partial queries from the worker nodes, wherein each approximate intermediate result is processed by a respective worker node of the worker nodes by scanning a portion of related data stored in a data node on which the perspective worker node runs. ### Claim 18 (depends on 15) The method of claim 15, further comprising: for each partial query, retrieving metadata regarding which data node stores data related to the partial query; and the step of sending comprises: sending each of the partial queries to a respective worker node of a plurality of worker nodes based on the metadata. ### Claim 19 A method comprising: receiving a database query from a client device, for a database containing data stored in a distributed storage cluster having a plurality of data nodes; dividing the database query into a plurality of partial queries; sending each of the partial queries to a respective worker node of a plurality of worker nodes, wherein each worker node is a service running on a data node of the distributed storage cluster; identifying a straggling worker node, dividing a partial query that is assigned to the straggling worker node into a plurality of subordinate partial queries, and assigning the plurality of subordinate partial queries to some of the plurality of worker nodes; retrieving a plurality of intermediate results for the partial queries from the worker nodes, wherein each intermediate result is processed by a respective worker node of the worker nodes by scanning related data stored in a data node on which the perspective worker node runs; and generating a query result based on the plurality of intermediate results. ### Claim 20 (depends on 19) The method of claim 19, wherein the step of identifying comprises: identifying a straggling worker node by monitoring heartbeat messages that the worker nodes periodically send, wherein the straggling worker node is identified when heartbeat messages from the straggling worker node are not received for a predetermined time period, or when a heartbeat message from the straggling worker node is received and the heartbeat message including a number representing a status of a partial query processing by the straggling worker node that is below a threshold value.