Scalable query processing
Embodiments of the present disclosure may provide a dynamic query execution model. This query execution model may provide acceleration by scaling out parallel parts of a query (also referred to as a fragment) to additional computing resources, for example computing resources leased from a pool of computing resources. Execution of the parts of the query may be coordinated by a parent query coordinator, where the query originated, and a fragment query coordinator.
1. A method comprising:
receiving, by one or more processors, a query directed at a data set;
generating a query plan to execute the query, the query plan including a plurality of operators connected by links;
identifying a portion of the query plan that is eligible for fragment processing based on the type of link connecting at least two operators of the plurality of operators;
executing, by a parent query coordinator using one or more computing resources of a first set of computing resources assigned to the parent query coordinator, the identified portion of the query plan on a first batch of files of the data set to generate a first batch result;
transmitting instructions to a fragment query coordinator for the fragment query coordinator to execute the identified portion of the query on a second batch of files of the data set to generate a second batch result using a second set of computing resources assigned to the fragment query coordinator;
combining the first batch result and the second batch result to generate combined results; and
executing remaining portion of the query plan on the combined results to generate a response to the query.
2. The method of claim 1 , wherein each link connects a first and second operator of the plurality of operators and the type of link indicates whether the first operator is executable by a computing resource without communicating with another computing resource.
3. The method of claim 1 , further comprising:
loading files of the data set into a first shared file queue as a continuous scanset;
grouping a first set of files as the first batch and providing the first batch to the parent query coordinator; and
grouping a second set of files as the second batch and providing the second batch to the fragment query coordinator.
4. The method of claim 3 , further comprising:
providing additional batches serially until all files in the continuous scanset have been provided.
5. The method of claim 1 , wherein identifying the portion of the query plan that is eligible for fragment processing is further based on a set of criteria.
6. The method of claim 5 , wherein the set of criteria includes whether an output of execution of the identified portion is less than an input of the identified portion.
7. The method of claim 5 , wherein the set of criteria includes whether the identified portion is executable by a computing resource without communicating with another computing resource.
8. A system comprising:
one or more processors of a machine; and
a memory storing instructions that, when executed by the one or more processors, cause the machine to perform operations comprising:
receiving a query directed at a data set;
generating a query plan to execute the query, the query plan include including a plurality of operators connected by links;
identifying a portion of the query plan that is eligible for fragment processing based on the type of link connecting at least two operators of the plurality of operators;
executing, by a parent query coordinator using one or more computing resources of a first set of computing resources assigned to the parent query coordinator, the identified portion of the query plan on a first batch of files of the data set to generate a first batch result;
transmitting instructions to a fragment query coordinator for the fragment query coordinator to execute the identified portion of the query on a second batch of files of the data set to generate a second batch result using a second set of computing resources assigned to the fragment query coordinator;
combining the first batch result and the second batch result to generate combined results; and
executing remaining portion of the query plan on the combined results to generate a response to the query.
9. The system of claim 8 , wherein each link connects a first and second operator of the plurality of operators and the type of link indicates whether the first operator is executable by a computing resource without communicating with another computing resource.
10. The system of claim 8 , the operations further comprising:
loading files of the data set into a first shared file queue as a continuous scanset;
grouping a first set of files as the first batch and providing the first batch to the parent query coordinator; and
grouping a second set of files as the second batch and providing the second batch to the fragment query coordinator.
11. The system of claim 10 , the operations further comprising:
providing additional batches serially until all files in the continuous scanset have been provided.
12. The system of claim 8 , wherein identifying the portion of the query plan that is eligible for fragment processing is further based on a set of criteria.
13. The system of claim 12 , wherein the set of criteria includes whether an output of execution of the identified portion is less than an input of the identified portion.
14. The system of claim 12 , wherein the set of criteria includes whether the identified portion is executable by a computing resource without communicating with another computing resource.
15. A non-transitory machine-storage medium embodying instructions that, when executed by a machine, cause the machine to perform operations comprising:
receiving a query directed at a data set;
generating a query plan to execute the query, the query plan include including a plurality of operators connected by links;
identifying a portion of the query plan that is eligible for fragment processing based on the type of link connecting at least two operators of the plurality of operators;
executing, by a parent query coordinator using one or more computing resources of a first set of computing resources assigned to the parent query coordinator, the identified portion of the query plan on a first batch of files of the data set to generate a first batch result;
transmitting instructions to a fragment query coordinator for the fragment query coordinator to execute the identified portion of the query on a second batch of files of the data set to generate a second batch result using a second set of computing resources assigned to the fragment query coordinator;
combining the first batch result and the second batch result to generate combined results; and
executing remaining portion of the query plan on the combined results to generate a response to the query.
16. The non-transitory machine-storage medium of claim 15 , wherein each link connects a first and second operator of the plurality of operators and the type of link indicates whether the first operator is executable by a computing resource without communicating with another computing resource.
17. The non-transitory machine-storage medium of claim 15 , further comprising:
loading files of the data set into a first shared file queue as a continuous scanset;
grouping a first set of files as the first batch and providing the first batch to the parent query coordinator; and
grouping a second set of files as the second batch and providing the second batch to the fragment query coordinator.
18. The non-transitory machine-storage medium of claim 17 , further comprising:
providing additional batches serially until all files in the continuous scanset have been provided.
19. The non-transitory machine-storage medium of claim 15 , wherein identifying the portion of the query plan that is eligible for fragment processing is further based on a set of criteria.
20. The non-transitory machine-storage medium of claim 19 , wherein the set of criteria includes whether an output of execution of the identified portion is less than an input of the identified portion.
21. The non-transitory machine-storage medium of claim 19 , wherein the set of criteria includes whether the identified portion is executable by a computing resource without communicating with another computing resource.