In modern large-scale distributed systems, analytics jobs submitted by various users often share similar work. Instead of optimizing jobs independently, multi-query optimization techniques can be employed to save a considerable amount of cluster resources. In this work, we introduce a novel method combining inmemory cache primitives and multi-query optimization, to improve the eficiency of data-intensive, scalable computing frameworks. By careful selection and exploitation of common (sub) expressions, while satisfying memory constraints, our method transforms a batch of queries into a new, more eficient one which avoids unnecessary recomputations. To find feasible and eficient execution plans, our method uses a cost-based optimization formulation akin to the multiple-choice knapsack problem. Experiments on a prototype implementation of our system show significant benefits of worksharing for TPC-DS workloads.
Modern technologies to analyze large amounts of data have
flourished in the past decade, with general-purpose cluster processing
frameworks such as MapReduce [
Large-scale analytics systems are deployed in shared
environments, whereby multiple users submit queries concurrently.
In this context, concurrent queries often perform similar work,
such as scanning and processing the same set of input data. The
research in [
Multi-query optimization (MQO) amounts to find similarities
among a set of queries and uses a variety of techniques to avoid
redundant work during query execution. For traditional database
systems, MQO trades some small optimization overheads for
increased query performance, using techniques such as sharing
sub-expressions [
In this paper, we study MQO in the context of distributed
computing engines such as Apache Spark [
We present the design of a MQO component that, given a set of concurrent queries, proceeds as follows. First, it analyzes query plans to find sharing opportunities, using logical plan fingerprinting and an eficient lookup procedure. Then it builds multiple sharing plans, using shared relational operators and scans, which subsume common work across the given query set. Sharing plans materialize their output relation in RAM. A cost-based optimization selects best sharing plans with dynamic programming, using cardinality estimation and a knapsack formulation of the problem, that considers a memory budget given to the MQO problem. The final step is a global query plan rewrite, including sharing plans which pipeline their output to modified consumer queries.
We validate our system with a prototype built for SparkSQL, using the standard TPC-DS benchmark for the experiments. Overall, our method achieves up to 80% reduction in query runtime, when compared to a setup with no worksharing. Our main contributions are as follows: • We propose a general approach to MQO for distributed computing frameworks. Our approach produces sharing plans, that are materialized in RAM, aiming at eliminating redundant work. • We cast the optimization problem of selecting the best sharing plans as a Multiple-choice Knapsack problem, and solve it eficiently through dynamic programming. • Our ideas materialize into a system prototype based on SparkSQL, which we evaluated using standard benchmarks, obtaining tangible improvements in aggregate query execution times. 2
MQO in RDBMSes. Multi-query optimization has been studied
extensively [
Materialized views can be used in conjunction with MQO to
reduce query response times [
Building upon MQO techniques in RDBMSes, Silva et al. [
More recently, [
We introduce a simple running example, that is rich enough to illustrate the gist of the MQO problem. Consider the following three concurrent queries: QUERY 1: SELECT name, dept_name, salary FROM employees, departments, salaries WHERE dep = dept_id
AND id = emp_id AND gender = 'F' AND location = 'us'
AND salary > 20000 ORDER BY salary DESC QUERY 2: SELECT name, dept_name, title,
to as title_expired_on FROM departments, employees, titles WHERE dep = dept_id
AND id = emp_id AND gender = 'F' AND location = 'us'
AND from >= 2010 QUERY 3: SELECT id, name, salary, from_date FROM employees, salaries WHERE id = emp_id
AND age > 30
AND SALARY > 30000
Figure 1 illustrates the optimized operator trees (logical plans) of the queries in the above example. The leaf nodes represent the base relations. Each intermediate node is a relational operator (Selection, Projection, Join, etc.). The arrows between nodes indicate data flow. Our MQO strategy uses such optimized logical plans to produce new plans – whose aim is to exploit sharing opportunities by caching distributed relations – which are translated into physical plans for execution.
First, we see that the three queries can share the scan of the employees, departments and salaries relations. Hence, a simple approach to work sharing would be to inject a cache operator in Query 1, which would steer the system to serve input relations from RAM instead of reading them from disk, when executing Query 2 and 3. A more refined approach could be to find common work (not only common I/O), in the form of similar subexpressions (SE) among the queries from the example, such as filtering and projecting records, and materialize intermediate results in RAM, so that to re-use such intermediate relations.
Figure 1 illustrates four examples of similar SEs, which are labelled as ψi , i = 1, 2, 3, 4 (we explain the meaning of this label in the next section). For example, consider the subexpression labelled as ψ2: all three queries share the same sub-tree structure, in the form Projectp (Filterf (employees)), but use diferent filtering predicates and projections. In principle, it is thus possible to save reading, parsing, filtering and projecting costs on the employees relation: by caching the intermediate output of a general form of subexpression, which subsumes the three similar sub-trees in each query. Such costs would be payed once, and the cached intermediate relation could serve three consumer queries. To this aim, we need to build a covering expression (CE) that combines the variants of the predicates appearing in the operators, e.g., considering ψ2 the corresponding CE could be:
Projectid, name, dep, age(Filtergender=F ∨ age>30(employees)) Similarly, the SEs labelled as ψ3 and ψ4 share the projection and filtering on department and salaries relations.
We anticipate that, in the context of our work, it is possible to rank some SEs according to the benefits they bring, in terms of reducing redundant work.
Sort [salary, DESC]
Project [name, dept_name, salary]
Join [id = emp_id]
Project [name, dept_name, title, to]
Join [id = emp_id]
Project [id, name, salary, from_date]
Join [id= emp_id] ψ2
Project [id, name, age]
For instance, the SE Projectp (Filterf (employees)) leads to additional savings when compared to the SE Filterf (employees), and caching the intermediate relation of the corresponding CE results in a smaller memory footprint because of its selectivity. More to the point, we now consider the SE labelled as ψ1 in Figure 1: Query 1 and 2 share a common sub-tree in their respective logical plans, that involves selections, projections and joins. In this case, selecting this SE as a candidate to build a CE between Query 1 and 2 contributes to decreased scanning, computation and communication costs. However, since caching a relation in RAM bears its own costs and must satisfy capacity constraints, materializing in RAM the output of the CE might reveal not beneficial after all: a join operator could potentially produce an intermediate relation too big to fit in RAM.
Overall, given an input query set, our problem amounts to explore a potentially very large search space, to identify SEs, to build the corresponding CEs – which we also call sharing plans, and to decide which CEs to include in the optimized output plan. Our MQO strategy aims at reducing the search space to build CEs by appropriately pruning SEs according to their rank. Furthermore, a cost-based selection of candidate CEs must ensure memory budget constraints to be met. 4
We consider a set of concurrent queries submitted by multiple users to be parsed, analyzed and individually optimized by a query optimizer. Our MQO method operates on a set of optimized logical plans corresponding to the set of input queries, that we call the input set. We approach the MQO problem with the following steps: §4.1: Similar subexpressions identification. The goal is to identify all common subexpressions in the input set. Two (or more) operators sharing similar subexpressions constitute a SE, which are candidates for building covering expressions (CEs). §4.2: Building sharing plans. The goal is to construct one or more groups of covering expressions CEs for a set of SEs. §4.3: Sharing plan selection. The goal is to select the best combination of CEs, using estimated costs and memory constraints. We model this step as a Multiple-Choice Knapsack problem, and use dynamic programming to solve it. §4.4: Query rewriting. The last step to achieve MQO is to rewrite the input query set such as to use selected sharing plans. 4.1
Finding similar subexpressions, given an input set of logical plans, has received considerable attention in the literature. What sets our approach apart from previous works lies behind the very nature of the resource we use to achieve work sharing: memory is limited, and the overall MQO process we present is seen as a constrained optimization problem, which strives to use caching with parsimony. Thus, we use a general rule of thumb that prefers a large number of CEs (built from the corresponding SEs) with small memory footprints instead of a small number of CEs with large memory requirements. This is in line with low-level systems considerations: data materialization in RAM is not cost-free, and current parallel processing frameworks are sometimes fragile, when it comes to memory management under pressure.
Armed with the above considerations, we first consider the
input to our task: we search SEs given a set of “locally optimized”
query plans, which are represented in a tree form. Such input
plans have been optimized by applying common rules such as
early filtering, predicate push-down, plan simplification and
collapsing [
Additional considerations are in order. Some operators produce output relations that are not easy to materialize in RAM: for example, binary operators such as join, generally produce large outputs that would deplete memory resources if cached. When searching for SEs, we recognize “cache unfriendly” operators and preempt them for being considered as valid candidates, either by selecting SEs that appear lower in the logical plan hierarchy (e.g., which could imply caching the input relations of a join), or by selecting SEs that subsume them (e.g., which could imply caching a relation resulting from filtering a join output). Currently, we treat the join, Cartesian product and Union as “cache unfriendly” operators. This means that our method does not produce SEs rooted at cache unfriendly operators; moreover, cache unfriendly operators can be shared inside a common SE only when they are syntactically equal.1 Next, we provide the necessary definitions that are then used to describe the identification of SEs.
Definition 4.1 (Sub-tree). Given a logical plan of a query Q, represented as a tree τ Q where leaf nodes are base relations and each intermediate node is a relational algebra operator, a subtree τsQ of τ Q is a continuous portion of the logical plan of Q containing an intermediate node of τ Q and all its descendant in τ Q . In other words, a sub-tree includes all the base relations and operators that are necessary to build its root.
If the context is clear, we denote a sub-trees simply as τ ,
without indicating from which query it has been derived. Given any
two sub-trees, we need to determine if they have the same
structure in terms of base relations and operators. To this aim, we
define a similarity function based on a modified Merkle Tree
(also known as hash tree)[
(u.label, u.attributes) otherwise.
Notice that this definition makes a distinction between loose and strict identifier. A loose identifier, such that used for projections and selections, allows the construction of a shared operator that subsumes the individual attributes with more general ones, which allows sharing computation among SEs. Instead, a strict identifier, such that used for all other operators (including joins and unions), imposes strict equality for two sub-graphs to be considered SEs. In principle, this restricts the applicability of a shared operator. However, given the above considerations about cache unfriendly operators, our approach still shares both I/O and computation.
Definition 4.2 (Fingerprint). is computed as
Given a sub-tree τ , its fingerprint h(ID(τroot)) τroot = leaf F (τ ) = h(ID(τroot)|F (τchild)) τroot = unary h(ID(τroot)|F (τl.child)|F (τr.child)) τroot = binary where h() is a robust cryptographic hash function, and the operation | indicates concatenation.
The fingerprint F (τ ) is computed recursively starting from the root of the sub-tree (τroot), down to the leaves (that is, input relations). If the root is a unary operator, we compute the ifngerprint of its child sub-tree ( τchild), conversely in case of a binary operator, we consider the left and right sub-trees (τl.child and τr.child). For the sake of clarity, we omit an additional sorting which ensures the isomorphic property for binary operators: for example, TableA join TableB and TableB join TableA are two isomorphic expressions, and have the same fingerprint.
We are now ready to define what a similar subexpression is.
1Our method can be easily extended for sharing similar join operators, for example
by applying the “equivalence classes” approach used in [
Definition 4.3 (Similar subexpression). A similar subexpression (SE) ω is a set of sub-trees that have the same fingerprint ψ , i.e. ω = {τi | F (τi ) = ψ }.
Algorithm 1 provides a pseudo-code of our procedure to find, given a set of input queries, the SEs according to Definition 4.3 that will be the input of the next phase, the search for covering expressions. The underlying idea is to avoid a brute-force search of fingerprints, which would produce a large number of SEs. Instead, by proceeding in a top-down manner when exploring logical plans, we produce fewer SEs candidates, by interrupting the lookup procedure as early and as “high” as possible.
The procedure uses a fingerprint table FT (line 2) to track SEs: this is a HashMap, where the key is a fingerprint ψ , and the value is a set of subtrees. Each logical plan from the input set of queries is examined in a depth-first manner. We first consider the whole query tree (line 4) and check if its root is a cache-friendly operator: in this case, we add the tree to the SEs identified by its ifngerprint. The method AddValueSet(ψ , τ ) retrieves the value (which is a set) from the HashMap FT given the key ψ (line 9), and adds the subtree τ to such a set – if the key does not exists, it adds it and create a value with a set containing the subtree τ . If the root is not a cache-friendly operator, or the logical plan contains a cache-unfriendly operator, then we need to explore the subtrees (line 13), i.e. we consider the root’s child (if the the operator at the root is unary) or children (otherwise).
At the end, we extract the set of SEs from the HashMap FT: we consider the SEs bigger than a threshold k in order to focus on SEs that ofer potential work sharing opportunities.
Going back to our running example, Algorithm 1 outputs a set of SEs as follows {ω1, ω2, ω3, ω4} – in Figure 1 the sub-trees corresponding to them are labelled ψ1, ψ2, ψ3 and ψ4, where ψi is the fingerprint of SE ωi . For instance, ω1 contains two sub-trees (one from Query 1, and one from Query 2), while ω2 contains three sub-trees, one from each query.
Given a list of candidate SEs, this phase aims at building covering subexpressions (CEs) corresponding to identified SEs, and generate a set of candidate groups of CEs for their final selection. Covering subexpressions. For each similar sub-query in the same SE ωi , the goal is to produce a new plan to “cover” all operations of each individual sub-query. Recall that all sub-trees τj within a SE ωi share the same sub-query plan fingerprint: that is, they operate on the same input relation(s) and apply the same relational operators to generate intermediate output relations. If the operator attributes are exactly the same across all τj , then the CE will be identical to any of the τj . In general, however, operators can have diferent attributes or predicates. In this case, the CE construction is slightly more involved.
First, we note that, by construction, the only shared operators we consider are projections and selections. Indeed, for cache unfriendly operators, the SE identification phase omits their fingerprint from the lookup procedure (see Algorithm 1, lines 8-9). Nevertheless, they could be included within a subtree, but they are in any case “surrounded” by cache-friendly operators (see for instance in Figure 1, the SE labeled as ψ1). As a consequence, a CE can be constructed in a top-down manner, by “OR-ing” the ifltering predicates and by “unioning” the projection columns of the corresponding operators in the SE. The CE thus produces and materializes all output records that are needed for its consumer queries. Figure 2 illustrates an example of CE for a simple SE of two sub-queries taken from the running example shown in Figure 1. In particular, we consider the SE labeled as ψ2.
The resulting CE contains the same operators as the subtrees τj ∈ ωi , but with modified predicates or attribute lists.
In general, we can build a CE, which we denote with Ω i , from a f () SE ωi , by applying a transformation function f (), [τ1, ...τm ] −−→ τi∗, which trasforms a collection of similar sub-trees to a single, covering sub-tree τi∗. Note that the resulting covering sub-tree has the fingerprint of the sub-trees in ωi .
Definition 4.4 (Covering subexpression). A Covering subexpression (CE) Ω i = f (ωi ) is a sub-tree τi∗ derived from the SE ωi by applying the transformation f (), with F (τi∗) = F (τj ) ∀τj ∈ ωi , such that all τj ∈ ωi can be derived from τi∗.
In summary, the query plan τi∗ that composes Ω i contains the same nodes as any subtree τj ∈ ωi , changing the predicates of the selections (OR of all the predicates in τj ) and projections (union of all the predicates in τj ).
Once the set of CEs, Ω = {Ω 1, Ω 2, . . . }, has been derived from the corresponding set of SEs, ω = {ω1, ω2, . . . }, we need to face the problem of CE selection. The main question we need to answer is: among the CEs contained in the set Ω , which ones should be cached? Each CE covers diferent portions of the query logical plans, therefore a CE may include another CE. Looking at the running example shown in Figure 1, we have that Ω 1 (derived from ω1, in the figure labeled with ψ1) contains Ω 3 (derived from ω3 and labeled in the figure with ψ3). If we decide to store Ω 1 in the cache, it becomes questionable to store Ω 3 as well.
The next step of our process is then to identify the potential combinations of mutually exclusive CEs that will be the input of the optimization problem: each combination will have a value and weight, where the value provides a measure of the work sharing opportunities, and the weight indicates the amount of space required to cache the CE in RAM. We start considering how to compute such values and weights, and we proceed with the algorithm to identify the potential combination of CEs.
estimation and cost modeling to reason about the benefit of using CEs. The objective is to estimate if a given CE, that could serve multiple consumer queries, yields lower costs than executing individually the original queries it subsumes.
The cardinality estimator component analyzes relational operators to estimate their output size. To do so, it first produces statistics about input relations. At relation level, it obtains the number of records and average record size. At column level, it collects the min and max values, approximates column cardinality and produces an equi-width histogram for each column.
The cost estimator component uses the results from cardinality estimation to approximate a (sub) query execution cost. We model the total execution cost of a (sub) query as a combination of CPU, disk and network I/O costs. Hence, given a sub-tree τj , we denote by CE (τj ) the execution cost of sub-tree τj . This component recursively analyzes, starting from the root of sub-tree τj , relational operators to determine their cost (and their selectivity), which is the multiplication between predefined constants (representative of the compute cluster running the parallel processing framework) and the estimated number of input and output records. Given a SE ω = {τ1, τ2, . . . , τm }, the total execution cost C(ωi ) related to the execution of all similar sub-trees τj ∈ ωi without the work-sharing optimization is given by
C(ωi ) =
CE (τj ). m Õ j=1 Instead, the cost of using the corresponding CE Ω i must account for both the execution cost of the common sub-tree τi∗, and materialization (CW ) and retrieving (CR ) costs associated to the cache operator we use in our approach, which accounts for write and read operations:
C(Ω i ) = CE (τi∗) + CW (|τi∗ |) + m · CR (|τi∗ |), where both CW (|τi∗ |) and CR (|τi∗ |) are functions of the cardinality |τi∗ | of the intermediate output relation obtained by executing τi∗. Eq. 2 indicates that retrieving costs are “payed” by each of the m consumer queries from the SE ωi that can use the CE Ω i .
Then, we can derive the value of a CE Ω i , denoted by v(Ω i ), as the diference between the cost of an unoptimized set of sub-trees (execution of ωi ) and the cost of the CE Ω i :
v(Ω i ) = C(ωi ) − C(Ω i ).
From Equations 1 and 2, we note that v(Ω i ) is an increasing function in m. Indeed, the more similar sub-queries a CE can serve, the higher its value.
Along with the value, we need to associate to a CE also a weight, since the memory is limited and we need to take into (1) (2) (3)
Input: Set Ω of CEs Output: Set of Knapsack items (potential CEs) 1: procedure GenerateKPitems(Ω = {Ω 1, Ω 2, . . . }) 2: Ω exp ← ∅ 3: while Ω not empty do 4: Ω i ← PopLargest(Ω ) 5: DescSet ← FindDescendant(Ω i , Ω ) 6: Groupi ← [Ω i ] ∪ Expand(DescSet) 7: Ω exp ← Ω exp ∪ {Groupi } 8: Remove(DescSet, Ω ) 9: end while 10: return Ω exp 11: end procedure account if a CE can fit in the cache. The weight, denoted by w(Ω i ) is the size required to cache in RAM the output of Ω i , i.e. w(Ω i ) = |τi∗ | =∆ |Ω i |.
Having defined the CE value and weight, we describe next the algorithm to identify the potential combination of CE.
problem of generating a combinatorial set of CEs, with their associated value and weight, to be given as an input to the multi-query optimization solver we have designed. Given the complexity of the optimization task, our goal is to produce a small set of valuable alternative options, which we call the candidate set of CEs. We present an algorithm to produce such a set, but first illustrate the challenges it addresses using the example shown in Fig. 1.
Let’s focus on CE Ω 1 (corresponding to the sub-trees labeled as ψ1). A naive enumeration of all possible choices of candidate CE to be cached leads to the following, mutually exclusive options: (i) Ω 1, (ii) Ω 2, (iii) Ω 3, (iv) both (Ω 2,Ω 3), (v) both (Ω 1,Ω 2), and (vi) both (Ω 1,Ω 3). Intuitively, however, it is easy to discern valuable from wasteful options. For example, the compound CE (Ω 1, Ω 2) could be a good choice, since Ω 2 can be cached to serve query 1 and 2 – and of course used to build Ω 1 – and for query 3. Conversely, caching the compound (Ω 1, Ω 3) brings less value, since it only benefits query 1 and query 2, but costs more than simply caching Ω 1, which also serves both query 1 and 2.
It is thus important to define how to compute the value and weight of compound CE. In this work we only consider compound CEs for which value and weight are additive in the values and weights of their components. This is achieved with compounds of disjoint CEs, i.e., those that have no common sub-trees.
For example, consider the two CEs Ω 1 and Ω 2, and the subtrees used to build them. The CE Ω 2 is included in Ω 1, but only some of the originating sub-trees of Ω 2 are included in the originating sub-trees of Ω 1 (in particular, the ones in query 1 and 2, but not in query 3). Given our definition of the value and the weight of CEs, the value and the weight of the compound (Ω 1, Ω 2) may not be equal to the sums of the values and of the weights of each individual CE, since part of the CE need to be reused to compute diferent sub-trees. Thus, we discard this option from the candidate set.
Algorithm 2 generates the candidate input for the optimization solver as a set of non-overlapping groups of CEs; then, the optimization algorithm selects a single candidate for each group in order to determine the best set of CEs to store in memory. Given the full set of Ω of CEs as input, we consider CE Ω i starting from the root of the logical plan and remove it from the set (line 4). We then look for its descendants from the input set Ω , i.e. all the CEs contained in Ω i (line 5). With a CE and its descendant, we build a list of options that contains (i) the CE itself and its individual descendants, and (ii) all the compounds of disjoint descendant CEs (line 6 and 7). We then remove the descendant from Ω and continue the search for other groups.
Considering our running example, we start from Ω = {Ω 1, Ω 2, Ω 3, Ω 4}. The “largest” CE is Ω 1, and its descendants are Ω 2 and Ω 3, therefore the list of mutually exclusive options for this group would be [Ω 1, Ω 2, Ω 3, (Ω 2, Ω 3)]. The output of Alg. 2 then is: {[Ω 1, Ω 2, Ω 3, (Ω 2, Ω 3)] , [Ω 4]} , (4) where the notation (·, ·) indicates a compound CE, and [·, ·] indicates a group of related CEs.
Note that a CE may be part of more than one larger CE: to keep the algorithm simple, we consider only the largest ancestor for each CE. To each option, we associate the value and the weight (in case of a compound, the sum of each component), that will be used by the optimization solver. 4.3
Next, we delve into our MQO problem formulation. In this work,
we model the process that selects which sharing plan to use as a
Multiple-choice Knapsack problem (MCKP) [
Our problem is thus to select which set of CEs (single, or compound) to include in the knapsack. The output of the previous phase (and in particular, the output of Algorithm 2) is a set containing m groups of mutually exclusive options, or items. Each group Gi , i = 1, 2, . . . , д, contains |Gi | items, which can be single CE or compounds of CEs. For instance, looking at our running example, the output shown in Eq. (4) contains д = 2 groups: the ifrst group has 4 items, the second group just one item. Given a group i, each item j has a value vi,j and a weight wi,j computed as described in Sect. 4.2.
The MCKP solver needs to choose at most one item from each group such that the total value is maximized, while the corresponding total weight must not exceed the capacity c. More formally, the problem can be cast as following:
Maximize subject to Õд |Gi |
Õ i=1 j=1 Õд |Gi |
Õ
i=1 j=1
|Gi |
Õ
j=1
vi,j xi,j
wi,j xi,j ≤ c
xi,j ≤ 1, ∀i = 1 . . . д
(5)
xi,j ∈ {0, 1}, ∀i = 1 . . . д, j = 1 . . . |Gi |
where the variable xi,j indicates if item j from group i has been
selected or not. The MCKP is a well-known NP-Hard problem:
in this work, we implement a dynamic programming technique
to solve it [
Note that alternative formulations exist, for which a provably optimal greedy algorithm can be constructed: for example, we could consider a fractional formulation of the knapsack problem. This approach, however, would be feasible only if the underlying query execution engine could support partial caching of a relation. As it turns out, the system we target in our work does support hierarchical storage levels for cached relations: what does not fit in RAM, is automatically stored on disk. Although this represents an interesting direction for future work (as it implies a linear time greedy heuristic can be used), in this paper we limit our attention to the 0/1 problem formulation. 4.4
The last step is to transform the original input queries to benefit from the selected combination of cache plans.
Recall that the output of a cache plan is materialized in RAM after its execution. Then, for each input query that is a consumer for a given cache plan, we build an extraction plan which manipulates the cached data to produce the output relation, as it would be obtained by the original input query. In other words, in the general case, we apply the original input query to the cached relation instead of using the original input relation. In the case of a CE subsuming identical SEs, the extraction plan is an identity: the original query simply replaces the sub-tree containing the CE by its cached intermediate relation. Instead, if shared operators are used – because of SEs having the same fingerprint but diferent attributes – we build an extraction plan that applies the original filter and projection predicates or attributes to “extract” relevant tuples from the cached relation produced from the CE.
Considering our running example, assume that the output of the MCKP solver is to store Ω 2 and Ω 3 in cache. Ω 3 derives from ω3, where the composing sub-trees (one from query 1, and one from query 2) are the same, therefore the extraction plan will be Ω 3 itself. Instead, ω2 (from which Ω 2 derives) contains sub-trees with diferent filtering and projection predicates: when Ω 2 is materialized in the cache, we need to apply the correct filtering (e.g., “gender = F”) and projection predicates to extract the actual result when considering the diferent queries. 5
We now present experimental results to evaluate the efectiveness
of our methodology, which we implement for the Apache Spark
and SparkSQL systems – the details of the implementation can
be found in our companion TR [
Results. We select a subset of all queries available in the TPC-DS benchmark, and focus on the 50 queries that can be successfully executed without failures or parsing errors. We present results for a setup in which we consider all the 50 queries and execute them in the order of their identifiers. Figure 3 shows the empirical Cumulative Distribution Function (CDF) of the runtime ratios between a system absorbing the workload with MQO enabled and disabled. Overall, we note that, for 60% of the queries, we obtain a 80% decrease of the runtime. In total, our approach reduces the runtime for 82% of the queries. On the other hand, 18% of the queries experience a larger runtime, which is explained by the overheads associated to caching. Overall, our optimizer has identified 60 SEs, and it has built 45 CEs. The cache used to store the output is approximately 26 GB (out of 120 GB available). The optimization process took less than 2 seconds, while the query runtime are in the order of tens of minutes (individually) and hours (all together).
Next, we consider an experimental setup in which we emulate the presence of a queuing component that triggers the execution of our worksharing optimization. In particular, since TPC-DS queries have no associated submission timestamp, we take a randomized approach (without replacement) to select which queries are submitted to the queuing component, and parametrize the latter with the number of queries – we call this parameter the window size – to accumulate before triggering our MQO mechanism. For a given window size, we repeat the experiment, i.e., we randomly select queries from the full TPC-DS workload, 20 times, and we build the corresponding empirical CDF of the runtime ratio, as defined above. We also measure the number of SEs identified within the window size, and show the corresponding empirical CDF. Given this setup, we consider all possible combinations of queries to assess the benefits of worksharing.
Figure 4 shows the boxplots of the runtime ratio (top) and number of similar subexpression identified (bottom) for diferent window sizes. The boxplots indicate the main percentiles (5%, 25%, 50%, 75%, 95%) of the empirical CDF, along with the average (red lines). The Figure shows a clear pattern: as the size of the window increases, there are more chances of finding a high number of SE, thus better sharing opportunities, which translates into reduced aggregate runtime. We observe a 20% decrease of the aggregate runtime (median) with a window size of only five queries, which ramps up to 45% when the window size is set to 20 queries.
Note that a queuing mechanism can introduce an additional
delay for the execution of a query, because the system needs
to accumulate a suficient number of queries in the window
before triggering their optimization and execution. Investigating
the trade-of between eficiency and delay, as well as studying
scheduling policies to steer system behavior is part of our future
research agenda. Due to space contraints, we refer to [
We presented a new approach to MQO that uses in-memory caching to improve the eficiency of computing frameworks such as Apache Spark. Our method takes a batch of input queries and ifnds common (sub)expressions, leading to the construction of covering expressions that subsume the individual work required by each query. To make the search problem tractable, we used several techniques: modified hash trees to quickly identify common sub-graphs, and an algorithm to enumerate (and prune) feasible common expressions. MQO was cast as a multiple-choice knapsack problem: each feasible common expression was associated with a value (representative of how much work could be shared among queries) and a weight (representative of the memory pressure imposed by caching the common data), and the goal was to ifll a knapsack representative of memory constraints.
To quantify the benefit of our approach, we implemented a prototype for Apache Spark SQL, and we used well-known workloads. Our results indicated that worksharing opportunities are frequent, and that our method brings substantial benefits in terms of reduced query runtime, with up to an 80% reduction for a large fraction of the submitted queries.