Reparallelization for workload skewing database operations
A method for parallelization may include identifying, within a sequence of operations comprising a query pipeline for executing a query, a workload skewing operation whose input dataset and output dataset exhibit an above-threshold difference in size. A reparallelization point may be inserted in the query pipeline subsequent to the workload skewing operation. The reparallelization point including a scheduling operation configured to determine a task size for performing one or more operations following the workload skewing operation in the query pipeline. The query may be executed by performing the sequence of operations comprising the query pipeline. Related systems and computer program products are also provided.
1 . A system, comprising:
at least one data processor; and
at least one memory storing instructions which, when executed by the at least one data processor, cause operations comprising:
determining whether an operation within a query pipeline is a workload skewing operation, wherein the query pipeline is an architecture in which a query is divided into a sequence of operations for execution of the query, wherein a size of an output dataset of the workload skewing operation is more than a threshold larger or more than a threshold smaller than an input dataset of the workload skewing operation, and wherein the query pipeline comprises one or more first operations following a first scheduling operation that is performed to determine a first task size for the one or more first operations based on a first quantity of time required to perform the one or more first operations on a first portion of the output dataset;
in response to determining that an operation is a workload skewing operation, inserting, in the query pipeline, a reparallelization point subsequent to the workload skewing operation, wherein the reparallelization point includes a second scheduling operation that is performed to copy the output dataset of the workload skewing operation into one or more buffers and determine a second task size for one or more second operations following the second scheduling operation in the query pipeline based on a second quantity of time required to perform the one or more second operations on a second portion of the output dataset smaller than the first portion of the output dataset, wherein the second scheduling operation is further configured to schedule, based at least on the second task size, one or more corresponding tasks for performance by a corresponding quantity of threads once the one or more buffers are full;
repeating, at each of multiple subsequent points of the query pipeline, the determining and the inserting to insert one or more additional reparallelization points in the query pipeline in response to determining that one or more additional operations is a workload skewing operation; and
executing the query by at least performing the sequence of operations comprising the query pipeline.
2 . The system of claim 1 , wherein data associated with each task is pushed to the one or more second operations following the workload skewing operation.
3 . The system of claim 2 , wherein the second scheduling operation is configured to avoid parallelization based at least on a first time required to gather the output dataset of the workload skewing operation into one or more buffers exceeding the second quantity of time required to perform the one or more second operations following the workload skewing operation.
4 . The system of claim 2 , wherein the second scheduling operation is configured to avoid parallelization based at least on there being less than a threshold quantity of tasks of the second task size.
5 . The system of claim 2 , wherein the one or more buffers are configured to accommodate data for multiple tasks, and wherein the second scheduling operation is configured to schedule the multiple tasks in parallel.
6 . The system of claim 1 , wherein the workload skewing operation comprises a selective join operation, an expanding join operation, or a selective table scan operation.
7 . The system of claim 1 , wherein the workload skewing operation is identified based on a compile time estimation or a runtime estimation of a first size of the input dataset and a second size of the output dataset.
8 . A computer-implemented method, comprising:
determining whether an operation within a query pipeline is a query is a workload skewing operation, wherein the query pipeline is an architecture in which a query is divided into a sequence of operations for execution of the query, wherein a size of an output dataset of the workload skewing operation is more than a threshold larger or more than a threshold smaller than an input dataset of the workload skewing operation, and wherein the query pipeline comprises one or more first operations following a first scheduling operation that is performed to determine a first task size for the one or more first operations based on a first quantity of time required to perform the one or more first operations on a first portion of the output dataset;
in response to determining that an operation is a workload skewing operation, inserting, in the query pipeline, a reparallelization point subsequent to the workload skewing operation, wherein the reparallelization point includes a second scheduling operation that is performed to copy the output dataset of the workload skewing operation into one or more buffers and determine a second task size for one or more second operations following the second scheduling operation in the query pipeline based on a second quantity of time required to perform the one or more second operations on a second portion of the output dataset smaller than the first portion of the output dataset, wherein the second scheduling operation is further configured to schedule, based at least on the second task size, one or more corresponding tasks for performance by a corresponding quantity of threads once the one or more buffers are full;
repeating, at each of multiple subsequent points of the query pipeline, the determining and the inserting to insert one or more additional reparallelization points in the query pipeline in response to determining that one or more additional operations is a workload skewing operation; and
executing the query by at least performing the sequence of operations comprising the query pipeline.
9 . The method of claim 8 , wherein data associated with each task is pushed to the one or more second operations following the workload skewing operation.
10 . The method of claim 9 , wherein the second scheduling operation is configured to avoid parallelization based at least on a first time required to gather the output dataset of the workload skewing operation into one or more buffers exceeding the second quantity of time required to perform the one or more second operations following the workload skewing operation.
11 . The method of claim 9 , wherein the second scheduling operation is configured to avoid parallelization based at least on there being less than a threshold quantity of tasks of the second task size.
12 . The method of claim 9 , wherein the one or more buffers are configured to accommodate data for multiple tasks, and wherein the second scheduling operation is configured to schedule the multiple tasks in parallel.
13 . The method of claim 8 , wherein the workload skewing operation is identified based on a compile time estimation or a runtime estimation of a first size of the input dataset and a second size of the output dataset.
14 . A non-transitory computer readable medium storing instructions, which when executed by at least one data processor, result in operations comprising:
determining whether an operation within a query pipeline is a workload skewing operation, wherein the query pipeline is an architecture in which a query is divided into a sequence of operations for execution of the query, wherein a size of an output dataset of the workload skewing operation is more than a threshold larger or more than a threshold smaller than an input dataset of the workload skewing operation, and wherein the query pipeline comprises one or more first operations following a first scheduling operation that is performed to determine a first task size for the one or more first operations based on a first quantity of time required to perform the one or more first operations on a first portion of the output dataset;
in response to determining that an operation is a workload skewing operation, inserting, in the query pipeline, a reparallelization point subsequent to the workload skewing operation, wherein the reparallelization point includes a second scheduling operation that is performed to copy the output dataset of the workload skewing operation into one or more buffers and determine a second task size for one or more second operations following the second scheduling operation in the query pipeline based on a second quantity of time required to perform the one or more second operations on a second portion of the output dataset smaller than the first portion of the output dataset, wherein the second scheduling operation is further configured to schedule, based at least on the second task size, one or more corresponding tasks for performance by a corresponding quantity of threads once the one or more buffers are full;
repeating, at each of multiple subsequent points of the query pipeline, the determining and the inserting to insert one or more additional reparallelization points in the query pipeline in response to determining that one or more additional operations is a workload skewing operation; and
executing the query by at least performing the sequence of operations comprising the query pipeline.