Distributed database systems gain relevance both in industry and academia. However, existing research on query optimization for relational database systems focuses largely on systems running on a single machine. Work on distributed systems neglects available workload information in database systems. In this work, we present optimization strategies to fully leverage the potential of distributed systems running on modern cloud architectures with fast networks. We focus on the optimal assignment of tasks to compute nodes and the joint optimization of join ordering and distribution layout of data. Furthermore, we introduce distributed plans and simulation-based evaluations using a new cost model for computation time.
Considering the very high bandwidth available in modern cloud systems, distributed processing becomes more and more attractive to not only handle large data sizes but also to improve processing performance. Still, good physical plans are key for eficient execution. We argue that distributed execution engines require changes to existing query optimizers for optimal performance. Existing join ordering algorithms yield suboptimal results because they fail to model the cost of data transfers. Furthermore, they need to spread computational load while avoiding waiting on data when assigning tasks to machines. We plan to contribute the following components to investigate new optimization opportunities:
distributed execution.
Methods like hash distributed joins require repartitioning of their input data. Our strategy chooses favorable distributions and introduces necessary data shufling. 2. A new operator-based computation time estimation method that allows us to compare the cost of transferring data over the network against local processing time. 3. An optimization method for the task assignment problem which determines on which node each part of a distributed query should be executed to minimize query response time.
VLDB 2023 PhD Workshop, co-located with the 49th International
1 2
3 Pipeline Break Pipelining Required Data
Shufle Stage (c) Distributed query plan base relations.
across various cluster setups using simulationbased evaluation. 5. A new join ordering algorithm that can take effects of distributed data partitioning and execution into account to jointly optimize distribution layout. use of (2) to estimate computational load and can be evaluated using (4). (5) also has a direct impact on query performance and can be compared against existing algorithms using (1).
Res B
1 2 7 1 9 8 4 2 6 5 3
3 1
B 2 1
B
2 (a) Initial query plan (b) New pipeline boundaries time estimations to track execution times accu- (3) will directly improve query response time. It makes rately. We can verify the quality of assignments
number of publications on distributed optimization is
rather limited. Microsoft extended the search space of
SQL Server’s query optimizer with data distribution
information for cost-based search [
There is also a lot of related work in the area of big
data that covers similar optimization aspects, such as
task scheduling [
• Scattered: Tuples are partitioned without a key.
tency, cloud cost, and energy consumption. We choose to optimize for latency, as we expect that optimizing for lower execution and transfer times will improve results in all metrics.
4.1. Distributed Plan Generation
First, we define the characteristics of distributed systems Data Units can be base relations from a database,
for which we want to optimize. We focus our work on intermediate results, or the final result of a query. They
OLAP systems, but the concepts are also applicable to contain the available attributes and the estimated number
transactional workloads. The system of concern has dis- of tuples in the data unit. Each data unit is annotated with
aggregated compute from storage to allow flexible scaling a partition layout determining the type of partitioning
of compute nodes similar to Snowflake [
We generalize relations, intermediate results, and final one data unit as input and returns one data unit as output. The only diference between input and output data unit 4000 is their respective partition layout. cyenu23000000
We initially create distributed query plans from phys- rFeq1000 ical plans created for single machines, as depicted in 0 1.00 1.25 1.50 Relati1v.e7C5ost of Ass2ig.0n0ment 2.25 2.50 2.75 Figure 1a. Our method takes a query plan and partition layout information for all base relations and distributes Figure 2: Cost distribution of 100 thousand sampled task the plan in several passes. assignments for TPC-H Q21 on 16 machines.
First, the operators in the plan are combined to pipelines. Next, we determine the best partition layout at each operator and split pipelines where necessary, as The task assignment optimizer focuses on the choice of in Figure 1b. Finally, we explicitly name the data units which node should execute which tasks. Each node can at the ends of each pipeline. The output of a pipeline execute any task, if the necessary data is transferred acmay have a diferent partition layout than the required cordingly. However, this will have significant impact input layout of its scanning pipeline. In this case, we on performance. We want to evenly spread the comcreate two data units and link them with a shufle stage, putational load among nodes and minimize time spent as shown for 5 and 6 in Figure 1c. waiting on data transfers. Each assignment also has an
Distributing single-node plans like this will not yield efect on subsequent execution, as it determines on which
optimal results as the original plan does not incorporate node the resulting partition will reside.
any information about the distributed system. Ultimately, As depicted in Figure 2, good task assignments can
the optimizer should consider distribution in all phases. improve performance over 2x. The number of possible
Most stages of this method can be reused for such an end- assignments assign = tansokdess grows exponentially in the
to-end optimizer, and we can create distributed plans to number of nodes, which renders exhaustive
enumeraconduct experiments early. Our work on this rule-based tion of assignments infeasible. Using our computation
plan generation is mostly done. time estimates and estimated time spent for data
transfers, we plan to build a heuristic optimizer for the task
4.2. Computation Time Estimation assignment problem that is able to generate good
assignments in short time. We will consider sampling-based
Traditional single-node query optimizers rely on rela- and greedy methods to find good initial plans in short
tively simple cost models because cardinality estimation time and refining methods like iterative improvement
errors outweigh the efect of more detailed models [
by min. At least one of them will finish and therefore make a new partition available at some node. Finally, it ifnds all pending operations that can start since now new partitions are available. By accumulating all values of min, we can compute the overall runtime of the query.
Our implementation of this simulator is ready for use.
We use it to evaluate randomly sampled task assignments
and give their execution time distribution in Figure 2. As
our implementation is fast, it can easily simulate tens of
thousands of executions per second and is hence suitable
for direct integration in the optimization loop.
Not only the new problem of task assignment has
optimization potential. Join ordering algorithms for
singlenode execution yield deficient plans for distributed
execution [
Distributed query processing opens potential for optimizations at many diferent stages of physical plan generation. This work proposes approaches to use that potential in several ways. We describe a way to lift current physical query plans for distributed execution. Then, we create a simulation-based evaluation method for these plans. We highlight the importance of the task assignment problem and sketch several methods to find good assignments. Finally, we present our vision for new enumeration-based join ordering algorithms that jointly optimize the distribution of data with the join ordering.