CMSC724: Query Evaluation Techniques
Instructor: Amol Deshpande
This chapter follows “Query Evaluation Techniques for Large Databases” (Goetz Graefe, ACM Computing Surveys 25(2), 1993), beginning with its execution-engine architecture and then sorting and hashing.
The organizing question is: once the optimizer has chosen a plan, how does the engine carry it out? We will distinguish what an operator computes from how it computes it, then see how operators cooperate without storing every intermediate result. These ideas provide the foundation for parallelism, vectorization, and compilation in the next three chapters.
1. Why This Paper
I highly recommend reading this survey. More than three decades after publication, it remains a useful blueprint for query execution engines. The hardware has changed, but its central abstractions—operators, iterators, and exchanges—still help us understand newer systems.
A personal note: this was the first academic paper I ever read. My advisor in my senior year of my bachelor’s handed it to me and said to read it, and then we would talk about research. That was in 1996 or 1997.
The survey emphasizes disk I/O because that was often the dominant cost in those days. For an in-memory workload, CPU overhead and memory access become much more visible. Keep that shift in mind: an interface that was inexpensive compared to a disk read, may become costly when repeated millions of times over memory-resident data.
2. The Query Processing Pipeline
Consider a query that joins Orders with Customers, keeps orders above $100, and sums their amounts by customer region. The system must first determine what the query means, then choose an efficient way to compute it.
- Parse: turn SQL text into a structured representation; reject invalid syntax.
- Validate and bind: resolve table and column names, check types, and verify that the referenced objects exist.
- Resolve views: incorporate the definitions of referenced views into the query representation.
- Optimize: choose join order, access paths (such as a scan or index lookup), and physical algorithms using estimated costs.
- Compile, when applicable: translate expressions or parts of the physical plan into executable code.
- Execute: run the operators and produce the result.
These stages may overlap in an implementation. The useful distinction is that planning normally reasons about metadata and statistics; execution processes the query’s input tuples. Statistics include row counts and value distributions, which help estimate intermediate result sizes. Some systems also sample data during planning or revise decisions during execution.
A plan can be prepared once and reused for repeated executions. Reuse saves planning or compilation time, although changing parameters or data distributions can make an old plan less suitable.
Interpretation versus compilation. An interpreter executes a general operator implementation and evaluates the plan’s expressions at runtime. Compilation specializes code for the chosen plan—for example, fixing attribute types and the predicate in the generated code. A system can mix the two, compiling expressions while interpreting the surrounding plan. Chapter 4 examines the benefits and costs of specialization.
Updates
An UPDATE can contain joins and predicates, so finding the target rows may require the same planning and execution machinery as a read query.
Updates introduce an additional correctness issue: changing data must not cause the scan to encounter the same row again. For example, increasing a salary while scanning a salary index could move the row ahead of the scan and update it twice. This is the Halloween problem. One solution is to collect target row identifiers before applying changes; other plans can avoid the problem through their access path or execution rules. Separating discovery from modification is useful, but fully deferred updates are not universal.
3. Logical versus Physical Algebra
A logical operator specifies the result to compute. A physical operator specifies an algorithm that computes it. A tuple is a row; an operator consumes streams of tuples and produces another stream.
For our example, the logical plan says “filter orders, join customers, group by region.” A physical plan might use a sequential scan, a hash join, and hash aggregation. Another equivalent plan might use an index scan and a merge join.
Intersection illustrates the same distinction: logically, retain values present in both inputs. Physically, sort the inputs and merge them, or build a hash table on one and look up values from the other. The optimizer chooses among implementations with different costs and requirements.
A physical plan still leaves implementation details to the engine: choosing a sort operator does not specify every buffer allocation or comparison routine. Also, equivalence must respect SQL semantics, including duplicates and NULLs; not every set-algebra rewrite applies unchanged to SQL.
The mapping is not one-to-one, in either direction
Several logical operators may share one physical implementation. A scan can apply a selection (row filtering) and projection (choosing or computing columns) as it reads. This avoids passing unwanted rows or columns to another operator.
One logical operator may require several physical operators. A join implemented by sort-merge may require two sorts followed by a merge join. If the inputs are already ordered suitably, the sorts can disappear.
An implementation can be asymmetric even when the logical operation is symmetric. An inner equijoin can exchange its inputs logically, but a hash join has distinct build and probe inputs. It first stores build tuples in a hash table, then looks up matching tuples for each probe tuple. The build side usually needs more memory.
Related operations can share code—for example, inner joins and semi-joins may use the same lookup machinery—but differ in what they emit. A semi-join returns a qualifying row from one side once, rather than every matching pair.
Parallel execution adds further physical decisions: where tuples live, how they move, and how work is divided. We return to those decisions in Chapter 2.
4. How Do Operators Pass Tuples to Each Other?
An algorithm for each operator is only part of an execution engine. We also need a rule for when each operator runs and where intermediate tuples go.
Materialization versus pipelining
Materialization stores an intermediate result in memory or on disk before its consumer processes it. This simplifies execution and permits reuse, but costs space and writes followed by reads. For example, materializing a filtered Orders relation means storing every surviving order before beginning the join.
Pipelining lets the consumer process tuples as the producer generates them. A surviving order can immediately probe the customer hash table, and its joined result can update an aggregate. This reduces intermediate storage and can shorten the time to the first result.
Some operations impose a blocking boundary: they must consume input before producing the relevant output. Sorting is the clearest example. A hash join blocks on its build side but can stream its probe side once the hash table is ready. Ordinary hash aggregation consumes the input before emitting final group totals; an aggregation over input sorted by group can finish one group at a time.
Who schedules the pipeline?
One design gives each operator a thread and passes tuples through queues. It allows operators to run concurrently, but introduces queue management, synchronization, and scheduling overhead. A bounded queue also needs backpressure: a fast producer must wait when its consumer cannot keep up.
Properly blocked threads need not consume CPU while waiting. The issue is that coordinating a thread per operator can cost more than the small amount of useful work per tuple. Queue-based pipelines are useful at selected parallel boundaries, but need not be used at every operator boundary.
The iterator model, also called the Volcano model, provides a simpler way to run a pipeline on one thread: when a consumer needs input, it calls its producer. Ordinary function calls determine execution order, with no queue between adjacent operators.
Materialization and recovery are separate choices. Storing an intermediate result does not automatically make it a recoverable checkpoint. A system must also track completed work and retain the result across failures. Some engines restart a failed query; others retry tasks using saved intermediate data.
5. The Iterator Model: open, next, close
Each operator implements three methods:
open()initializes its state and opens its inputs.next()returns the next output tuple, or an end-of-stream indication.close()releases resources and closes its inputs.
A scan’s next() reads the next row. A selection’s next() repeatedly calls its child until a row passes the predicate. A hash join builds its hash table during initialization (or its first call), then probes it as rows arrive from the other input. A merge join advances through sorted inputs, aligning equal keys.
Following one tuple through a plan
For selection(scan(Orders)), the caller asks the selection for a tuple. The selection asks the scan; if the amount is too small, it asks again. Once it finds a qualifying order, it returns it. The caller sees a stream of qualifying orders without knowing how they were obtained.
selection.next():
loop:
t = input.next()
if t is END: return END
if predicate(t): return tComposability is the advantage: a join can consume a scan, a selection, or another join through the same interface. The inputs must expose the expected schema and satisfy any required properties, but their internal algorithms need not be known to the consumer.
The state problem
There is a wrinkle that makes implementation trickier than it looks. Suppose the probe tuple r1 matches two build tuples, producing r1s1 and r1s2. You are only allowed to return one tuple per call. So you must remember, across calls, that you still owe the caller r1s2.
If you are writing this in Python, yield makes the operators nearly trivial, because the language maintains that state for you. Without it, you must track your position in the hash bucket by hand. You can always materialize all matches into a list and return from it, but there are more efficient implementations.
Preconditions
Merge join is the standard example of an operator with a precondition: its inputs must be sorted. That is not its job to arrange — it is the query planner’s responsibility to insert an explicit sort if needed. Several operators have preconditions like this.
Also note that the merge-join description in the table assumes no duplicates; handling duplicates makes the code considerably more involved.
Where it falls down
The cost becomes visible when data is in memory. Returning one tuple may require several nested next() calls, type checks, and expression evaluations. Across millions of tuples, even small overheads add up. Indirect calls can also prevent the compiler from optimizing across operator boundaries.
The iterator model remains a useful abstraction; its tuple-at-a-time interface is the performance concern. Vectorization amortizes calls over batches. Compilation can fuse adjacent operators into a loop and remove their runtime calls. Chapters 3 and 4 develop these two responses.
Pull and push describe control flow. In pull execution, a consumer requests input; in push execution, a producer delivers output to its consumer. Either can pipeline data, and either can operate on tuples or batches. Parallel execution often runs the same pipeline over different data partitions, rather than assigning a separate core to each operator.
6. The Exchange Operator
An exchange operator transports tuples between execution contexts, such as threads, processes, or machines, while exposing the same stream interface to surrounding operators.
Suppose one machine scans orders and another consumes them. The consumer calls next() on its exchange, which obtains buffered tuples from the remote producer. The producer runs its local pipeline and sends batches through the exchange. Communication can be asynchronous; it need not involve one network request per tuple.
Depending on the plan, an exchange can gather several streams, broadcast one stream, or repartition tuples by a key. It hides the mechanics of communication from the join or aggregate above it, although the optimizer must still choose the right distribution. Buffering and backpressure keep producers and consumers coordinated.
This is the architectural payoff of a common interface: parallelism can be introduced at selected boundaries without rewriting every relational operator.
7. Plan Shapes
The terms describe the structure of a binary join tree:
- Left-deep: each join’s right child is a base relation, as in
((A ⋈ B) ⋈ C) ⋈ D. - Right-deep: each join’s left child is a base relation, as in
A ⋈ (B ⋈ (C ⋈ D)). - Bushy: both children may be joins, as in
(A ⋈ B) ⋈ (C ⋈ D).
Draw the tree rather than relying only on the name: conventions in the literature sometimes differ. More importantly, tree shape does not by itself determine the hash-build side. That is an additional physical choice.
For the left-deep expression above, building on the right means building tables for B, C, and D; the intermediate stream from A can then probe them successively. Building on the left instead stores intermediate join results. Which is preferable depends on their sizes and the memory budget—base relations are not always smaller.
A bushy plan permits two subjoins to be computed independently, but their results may be expensive to store. A plan can also be a directed acyclic graph (DAG) when one intermediate result feeds multiple consumers. Such sharing requires coordination or materialization beyond a simple tree of iterators.
8. Sorting
How Volcano implements sort as an iterator:
opendoes most of the work. If the input fits in memory, it reads everything and quick-sorts. If not, it does an external merge sort except for the final merge.nexteither hands back already-sorted tuples, or performs that final merge incrementally.
The paper notes it is probably better to do all the work in next, with special-case code for the first call.
External merge sort
The classic picture: memory holds three tuples, you need to sort twelve. Group the input into runs of three, sort each in memory, and write them back as sorted runs. Then merge two at a time — two input slots plus one output slot is exactly the three you have — producing runs of six, then twelve.
The key property is that merging needs only the head of each run in memory, however large the runs are.
In an external sort, buffers hold pages, not individual tuples. With B buffer pages, an initial run can hold roughly B pages, and a merge can combine up to B − 1 runs while reserving one output buffer. This number of inputs is the merge fan-in. Each merge pass multiplies the run size by roughly the fan-in; the number of passes depends on input size and available memory. Streaming the final merge into the next operator avoids writing and rereading the final sorted file.
Replacement selection
Replacement selection can generate initial runs longer than memory capacity. Keep a min-heap of up to H tuples: a heap supports retrieving and removing the smallest active value efficiently.
Emit the smallest value and read a replacement. If the new value is at least as large as the last value emitted, it can remain in the current run; otherwise, freeze it for the next run. When no active values remain, finish the run and reactivate the frozen values.
For example, after emitting 4, an incoming 7 is eligible for the current run, while an incoming 2 must wait. This rule maintains sorted order without keeping the entire run in memory.
Under the usual random-input assumptions, runs average about 2H tuples; already sorted input can produce one run. Longer runs can reduce merging, but maintaining the heap adds CPU work and can have worse locality than sorting fixed-size chunks. The benefit depends on whether fewer I/O passes compensate for that work.
9. Hashing
Hashing organizes rows into buckets using their keys. For an equality join, build a hash table on one input and probe it with keys from the other, which can arrive in any order. Equality checks distinguish actual matches from hash collisions.
If the build input does not fit in memory, partition both inputs using the same hash of the join key and write partitions to disk. Matching rows reach the same partition pair. Read one pair at a time, build a hash table on its build partition, and probe with its counterpart. This is Grace hash join: partitioning lets a large join run within a smaller memory budget.
Two refinements:
Recursive partitioning. If a build partition still exceeds its memory budget, partition the corresponding pair again using another hash function. Limited memory, concurrent queries, or skew can make this necessary. A heavily repeated key cannot be split just by changing its hash function; it needs special treatment.
Hybrid hash join. Keep one partition in memory during the initial partitioning pass rather than writing it out and reading it back. It can be done reactively, and it works very well when the smaller input is just barely larger than memory.
A dictionary is a useful mental model, but a SQL join must preserve duplicates. If two build rows and three probe rows share a key, the join emits six pairs. A hash-table entry therefore needs to retain all matching build rows, and probing must check actual key equality because different keys can hash to the same bucket.
10. Sorting versus Hashing
The survey emphasizes a duality: many operations bring equal keys together either by sorting them into adjacent positions or by hashing them into the same bucket. Group-by, duplicate elimination, and equality joins all have implementations based on these alternatives.
For an equality join, hashing often avoids sorting and permits streaming after the build phase. Sorting is attractive when inputs are already ordered or when later operators can reuse that order. Both can spill to disk, and both can suffer from uneven data distributions.
The figure traces a sequence of research comparisons. Read such results as measurements of particular implementations, workloads, and machines. Core count, memory bandwidth, key width, duplicates, and the cost of producing output can all change the outcome.
The optimizer should consider the whole plan: a locally more expensive sort-merge join may pay for itself if its output order removes a later sort. Conversely, sorting only to feed an order-insensitive consumer may be wasted work.
A factor-of-two execution improvement can have substantial operational value even when it is not an order-of-magnitude research result. But separate the source of the gain: a better algorithm, a better implementation, or a better match to the benchmark’s hardware and data.
11. Sorting Networks and Vectorization
A sorting network is a fixed arrangement of compare-exchange operations. Each compares two designated positions and places the smaller value first. The positions and execution order do not depend on the input, which makes small networks suitable for SIMD implementations using comparisons and shuffles.
A complete four-element sorting network compares (1,2) and (3,4), then (1,3) and (2,4), then (2,3). Starting from [4,1,3,2], those stages give [1,4,2,3], [1,3,2,4], and finally [1,2,3,4]. The figure shows the related bitonic construction; a bitonic merge assumes its input already has the required bitonic order.
A bitonic sorting network uses \(O(n\log^2 n)\) comparisons, rather than the \(O(n\log n)\) of efficient general comparison sorts. Its regular structure can still be worthwhile for small blocks; larger sorts can use it as a building block. Fixed compare-exchange operations can be implemented without data-dependent branches, though the generated code determines whether that happens.
The theme connecting these sections is intermediate data movement: iterator pipelines avoid unnecessary materialization; external sorting and hashing manage unavoidable materialization; the next chapters ask how to reduce communication and CPU overhead as well.