# US 12,118,009 B2 · Supporting query languages through distributed execution of query engines > Agents: this data is available as JSON from the patentgrep API and MCP server — see https://patentgrep.com/llms.txt - Number: US 12,118,009 B2 - Document ID: US12118009B2 - Status: Granted - Kind: B2 - Application number: 16/657,916 - Filed: 2019-10-18 - Published: 2020-02-13 - Granted: 2024-10-15 - Earliest claimed: 2017-07-31 - Assignees: Splunk Inc. - Applicants: Splunk Inc. - Inventors: Arindam Bhattacharjee; Sourav Pal; Timothy Tully - Web page: https://patentgrep.com/patent/US12118009B2 ## CPC - G06F 16/2452: Query translation - G06F 16/2471: {Distributed queries} - G06F 16/24526: {Internal representations for queries} - G06F: ELECTRIC DIGITAL DATA PROCESSING (computer systems based on specific computational models G06N) ## Abstract Systems and methods are described for distributed processing a query in a first query language utilizing a query execution engine intended for single-device execution. While distributed processing provides numerous benefits over single-device processing, distributed query execution engines can be significantly more difficult to develop that single-device engines. Embodiments of this disclosure enable the use of a single-device engine to support distributed processing, by dividing a query into multiple stages, each of which can be executed by multiple, concurrent executions of a single-device engine. Between stages, data can be shuffled between executions of the engine, such that individual executions of the engine are provided with a complete set of records needed to implement an individual stage. Because single-device engines can be significantly less difficult to develop, use of the techniques described herein can enable a distributed system to rapidly support multiple query languages. ## Claims ### Claim 1 A computer-implemented method, comprising: receiving a query in a first query language to be applied to a set of data records; parsing the query to identify multiple query stages; generating, for each query stage of the multiple query stages, a sub-query, in the first query language, wherein each sub-query is configured to cause each of multiple worker nodes, to implement the query stage with respect to a subset of the set of data records obtained at the worker node, each sub-query representing a distinct executable query in the first query language that corresponds to a distinct query stage of the multiple query stages; based on a determination that a first query stage of the multiple query stages corresponds to a first native operation and a determination the multiple worker nodes are configured to execute the first native operation, generating one or more instructions to execute the first native operation; based on a determination that a second query stage of the multiple query stages does not correspond to any native operation, determining not to generate one or more instructions to execute a native operation; generating instructions for shuffling records between the multiple worker nodes at a point in time between at least two of the multiple query stages; and communicating the instructions for shuffling records, the one or more instructions to execute the first native operation, and the sub-query corresponding to the second query stage of the multiple query stages, to the multiple worker nodes for concurrent implementation, wherein each worker node includes a distinct executor for processing sub-queries in the first query languages. ### Claim 2 (depends on 1) The computer-implemented method of claim 1, wherein the query is a Structured Query Language query. ### Claim 3 (depends on 1) The computer-implemented method of claim 1 further comprising generating instructions for generating the set of data records from a set of unstructured data and distributing the instructions to the multiple worker nodes. ### Claim 4 (depends on 1) The computer-implemented method of claim 1, wherein at a non-initial stage of the multiple query stages, the subset of the records obtained at each worker node includes intermediary records generated during a prior stage of the multiple query stages. ### Claim 5 (depends on 1) The computer-implemented method of claim 1, wherein the multiple query stages include at least one map operation stage and at least one reduce operation stage. ### Claim 6 (depends on 1) The computer-implemented method of claim 1, wherein the multiple query stages include a map operation stage, and wherein implementation of the map operation stage implements a filter with respect to the set of data records. ### Claim 7 (depends on 1) The computer-implemented method of claim 1, wherein the multiple query stages include a reduce operation stage, and implementation of the reduce operation stage combines at least two records into a single record. ### Claim 8 (depends on 1) The computer-implemented method of claim 1, wherein the multiple query stages include a reduce operation stage, and implementation of the reduce operation stage determines an order of at least two records. ### Claim 9 (depends on 1) The computer-implemented method of claim 1, wherein the multiple query stages include a reduce operation stage, and wherein the sub-query for a query stage prior to the reduce operation stage comprises a first sub-query to implement an operation of the prior query stage and a second sub-query to implement a pre-shuffle reduce operation supporting the reduce operation stage. ### Claim 10 (depends on 1) The computer-implemented method of claim 1, wherein the sub-query for at least one query stage corresponds to multiple sub-queries that collectively implement the at least one query stage. ### Claim 11 (depends on 1) The computer-implemented method of claim 1, wherein the sub-query for at least one query stage corresponds to multiple sub-queries that collectively implement the at least one query stage, the multiple sub-queries comprising a compaction query and a transformation query. ### Claim 12 (depends on 1) The computer-implemented method of claim 1 further comprising distributing the distinct executor to each of the multiple worker nodes. ### Claim 13 (depends on 1) The computer-implemented method of claim 1, wherein parsing the query to identify multiple query stages comprises implementing a parser to parse the query and implementing a query planner to identify the multiple query stages. ### Claim 14 (depends on 1) The computer-implemented method of claim 1, wherein the instructions further comprise instructions for shuffling records between the multiple worker nodes between each of the multiple query stages. ### Claim 15 (depends on 1) The computer-implemented method of claim 1 further comprising communicating to the multiple worker nodes instructions for returning partial search results generated based on implementation of the multiple query stages to an aggregator configured to aggregate the partial search results into complete search results for the query. ### Claim 16 (depends on 1) The computer-implemented method of claim 1 wherein the sub-queries of each of the multiple query stages in combination are logically equivalent to the query. ### Claim 17 (depends on 1) The computer-implemented method of claim 1, wherein the instructions for shuffling records between the multiple worker nodes include identification of a shuffle key, and wherein the shuffle key indicates a field of the records to utilize in redistributing the records between the multiple worker nodes. ### Claim 18 (depends on 1) The computer-implemented method of claim 1, wherein the instructions for shuffling records between the multiple worker nodes include identification of a shuffle key, wherein the shuffle key indicates a field of the records to utilize in redistributing the records between the multiple worker nodes, and wherein the records are redistributed between the multiple worker nodes based on a hash of a value of the shuffle key. ### Claim 19 (depends on 1) The computer-implemented method of claim 1, wherein the instructions for shuffling records between the multiple worker nodes include identification of a shuffle key, wherein the shuffle key indicates a field of the records to utilize in redistributing the records between the multiple worker nodes, and wherein the records are redistributed between the multiple worker nodes based on a range of values of the shuffle key. ### Claim 20 (depends on 1) The computer-implemented method of claim 1 further comprising: implementing a first query stage of the at least two query stages at each of the multiple worker nodes; shuffling records between the multiple worker nodes in accordance with the instructions; and implementing a second query stage of the at least two query stages at each of the multiple worker nodes. ### Claim 21 (depends on 1) The computer-implemented method of claim 1 further comprising: implementing each query stage of the multiple query stages at each of the multiple worker nodes, wherein implementing each query stage comprises shuffling records between the multiple worker nodes in accordance with the instructions; aggregating partial results generated at each of the multiple worker nodes as a result of implementing each query stage into a set of complete results; and returning the set of complete results in response to the query. ### Claim 22 (depends on 1) The computer-implemented method of claim 1, wherein the multiple query stages represent a first portion of the query, and wherein the method further comprises: implementing each query stage of the multiple query stages at each of the multiple worker nodes, wherein implementing each query stage comprises shuffling records between the multiple worker nodes in accordance with the instructions; aggregating partial results generated at each of the multiple worker nodes as a result of implementing each query stage into a set of results; joining the set of results with a second set of results generated based on a second portion of the query to result in a set of complete results; and returning the set of complete results in response to the query. ### Claim 23 (depends on 1) The computer-implemented method of claim 1 further comprising implementing each query stage of the multiple query stages at each of the multiple worker nodes, wherein at one or more worker nodes, at least one query stage is implemented multiple times concurrently on multiple processor cores. ### Claim 24 (depends on 1) The computer-implemented method of claim 1, wherein each execution of the distinct executor is a single-threaded process. ### Claim 25 (depends on 1) The computer-implemented method of claim 1, wherein the distinct executor represents code configured for execution on a single device. ### Claim 26 A system comprising: a data store including computer-executable instructions; and a processor in communication with the data store and configured to execute the computer-executable instructions to: receive a query in a first query language to be applied to a set of data records; parse the query to identify multiple query stages; generate, for each query stage of the multiple query stages, a sub-query, in the first query language, wherein each sub-query is configured to cause each of multiple worker nodes, to implement the query stage with respect to a subset of the set of data records obtained at the worker node, each sub-query representing a distinct executable query in the first query language that corresponds to a distinct query stage of the multiple query stages; based on a determination that a first query stage of the multiple query stages corresponds to a first native operation and a determination the multiple worker nodes are configured to execute the first native operation, generate one or more instructions to execute the first native operation; based on a determination that a second query stage of the multiple query stages does not correspond to any native operation, determine not to generate one or more instructions to execute a native operation; generate instructions for shuffling records between the multiple worker nodes at a point in time between at least two of the multiple query stages; and communicate the instructions for shuffling records, the one or more instructions to execute the first native operation, and the sub-query corresponding to the second query stage of the multiple query stages, to the multiple worker nodes for concurrent implementation, wherein each worker node includes a distinct executor for processing sub-queries in the first query language. ### Claim 27 (depends on 26) The system of claim 26, wherein the processor is further configured to execute the computer-executable instructions to: obtain, from each worker node, partial search results generated based on implementation of each of the multiple query stages; aggregate the partial search results from each worker node to result in complete search results; and return the complete search results in response to the query. ### Claim 28 Non-transitory computer-readable media comprising computer-executable instructions that, when executed by a computing system, cause the computing system to: receive a query in a first query language to be applied to a set of data records; parse the query to identify multiple query stages; generate, for each query stage of the multiple query stages, a sub-query, in the first query language, wherein each sub-query is configured to cause each of multiple worker nodes, to implement the query stage with respect to a subset of the set of data records obtained at the worker node, each sub-query representing a distinct executable query in the first query language that corresponds to a distinct query stage of the multiple query stages; based on a determination that a first query stage of the multiple query stages corresponds to a first native operation and a determination the multiple worker nodes are configured to execute the first native operation, generate one or more instructions to execute the first native operation; based on a determination that a second query stage of the multiple query stages does not correspond to any native operation, determine not to generate one or more instructions to execute a native operation; generate instructions for shuffling records between the multiple worker nodes at a point in time between at least two of the multiple query stages; and communicate the instructions for shuffling records, the one or more instructions to execute the first native operation, and the sub-query corresponding to the second query stage of the multiple query stages, to the multiple worker nodes for concurrent implementation, wherein each worker node includes a distinct executor for processing sub-queries in the first query language. ### Claim 29 (depends on 28) The non-transitory computer-readable media of claim 28, wherein the instructions further cause the computing system to: obtain, from each worker node, partial search results generated based on implementation of each of the multiple query stages; aggregate the partial search results from each worker node to result in complete search results; and return the complete search results in response to the query. ### Claim 30 (depends on 28) The non-transitory computer-readable media of claim 28, wherein at a non-initial stage of the multiple query stages, the subset of the records obtained at each worker node includes intermediary records generated during a prior stage of the multiple query stages.