Data Intensive Systems: Lecture 7

Physical Query Plans

A SQL query is not executed directly. The DBMS first parses and rewrites it into a logical plan, whose nodes are relational algebra operators. The optimizer then chooses a physical plan, which keeps the same meaning but fixes how the work will be done.

Physical query plan. A query plan annotated with execution choices: the processing model, the implementation algorithm for each operator, and the access path used to read each base relation.

• Processing model. The scheduling rule for moving tuples between operators.
• Operator implementation. The concrete algorithm for an algebra operator, such as nested loops join or hash join.
• Access path. The method used to reach base data, such as a file scan, an index lookup, or a multi-index RID lookup.

The logical tree says what result is needed. The physical plan says how much data is moved, when operators run, and which pages are read. For SELECT R.id, S.cdate FROM R JOIN S ON R.id = S.id WHERE S.value > 100, one logical tree places the predicate on S before the join.

Processing Models

After the DBMS has an operator tree, every edge in the tree must carry tuples from a child operator to its parent. A processing model defines how that exchange happens: one tuple at a time, or one full intermediate result at a time.

The choice can dominate runtime. If a join produces one million intermediate rows and a later WHERE condition keeps only ten, a materialized plan may store the million rows before filtering them. A pipelined plan can test rows as they appear and avoid storing most of them.

Iterator model. A pipelined, demand-driven model where each operator exposes Open(), Next(), and Close(). Next() returns one tuple per call, or an end marker when the operator has no more output.

• Single-tuple output. One successful Next() call returns one row, not a full relation.
• EOF marker. When an operator is exhausted, Next() signals that no more tuples are available.
• Demand-driven control. An operator does work only when its parent asks for another tuple.
• Operator state. Each operator keeps enough state to resume where the previous Next() call stopped.

In the usual Volcano form, parents drive the work by calling Next() on their children. The lifecycle is fixed:

1. Open() initializes the operator, opens its children, and allocates local state such as scan cursors, predicate state, or hash tables.
2. Next() produces one output tuple if possible. It may call Next() on one or more children, update local state, and either return a tuple or keep searching.
3. Next() eventually returns EOF. EOF propagates upward when every child path needed by the operator is exhausted.
4. Close() releases temporary state and closes children.

Project.Next():
  while (t := child.Next()) != EOF:
    return project(t)
  return EOF
 
HashJoin.Next():
  if not built:
    while (r := left.Next()) != EOF:
      buildHashTable(r.id, r)
    built = True
 
  while True:
    if pending is not empty:
      return pending.pop()
    s = right.Next()
    if s == EOF:
      return EOF
    pending = [join(r, s) for r in probe(s.id)]
 
Select.Next():
  while (s := child.Next()) != EOF:
    if s.value > 100:
      return s
  return EOF

The important cost property is not the order of these operators, but the size of the intermediate state. Iterator execution does not store a full result at each edge. It stores only enough local state to resume: scan positions, a predicate cursor, pending join matches, or a hash table for a blocking build phase.

Once the hash table has been built, the probe side can stream. A qualifying tuple can move upward immediately instead of first being written into a full intermediate relation. This saves memory and avoids spill I/O when the intermediate result would be large, but it adds per-tuple function-call overhead because every output row travels through repeated Next() calls.

Some operators are blocking. A sort must usually see all input before it can return the first sorted tuple. A hash join must build its hash table before probing it. Blocking operators break a pipeline into stages, even inside an iterator engine.

Materialization model. A batch model where each operator consumes all input, stores its full result, and passes that result to its parent.

Instead of Next(), think of each operator exposing Output(): it returns a complete collection of result tuples. The parent does not receive a tuple until the child has finished its full output.

Its main properties are:

• Batch output. Each operator materializes its entire result before passing it upward.
• Hints. The DBMS can push down hints such as LIMIT or needed-column lists to reduce unnecessary materialization.
• Flexible output. An operator may materialize whole tuples or only the columns its parent needs.

Project.Output():
  out = []
  for t in child.Output():
    out.add(project(t))
  return out
 
HashJoin.Output():
  out = []
  for r in left.Output():
    buildHashTable(r.id, r)
  for s in right.Output():
    for r in probe(s.id):
      out.add(join(r, s))
  return out
 
Select.Output():
  out = []
  for s in child.Output():
    if s.value > 100:
      out.add(s)
  return out

Materialization moves the cost to operator boundaries. Each Output() call returns a complete relation, so a parent starts only after the child has finished and stored its result. This can be simple and efficient when intermediates are small, because the parent reads compact arrays or temporary files rather than repeatedly calling into child operators.

The weakness appears when an intermediate is large. A large filtered input, join result, or projected result must be held in memory or written to disk before the next operator can reduce it further. In analytical plans, this extra write-read cycle can dominate the cost even when the logical tree is good.

Any pipeline edge can be materialized without changing the result, only the cost. The iterator stores positions; materialization stores results.

Plan processing direction. The direction in which control moves through the plan tree. It is related to the processing model, but it names a different choice: who starts the work.

Pull direction. Top-to-bottom control. The root starts execution by requesting tuples from its children. The iterator model is the classic pull model.

1. The client asks the root for output.
2. The root calls a child function, usually Next().
3. The request travels down until a leaf scan can produce data.
4. Tuples return upward through function calls.

Pull execution gives precise output control. It is natural for LIMIT, cursors, and interactive clients because the root can stop asking when enough rows have been produced. Its cost is per-tuple call overhead, often through virtual function calls and branch-heavy operator code.

Push direction. Bottom-to-top control. Leaves start producing data and push tuples into parent operators.

1. A leaf scan produces a tuple or batch.
2. The tuple is pushed into the next operator.
3. Each operator processes the tuple and pushes resulting tuples upward.
4. Work continues until sources are exhausted or a parent stops accepting input.

Push execution can keep data in CPU caches and registers more effectively, especially in compiled or vectorized pipelines. It is also more flexible for dynamic re-optimization because runtime information appears as data flows upward. Its challenge is control: if a child produces many tuples, the parent must absorb them, buffer them, or stop the producer; blocking operators such as sort and hash build phases still require special handling.

The purpose of these choices is not to change the query answer. They change when work happens, how much intermediate data exists, and where the CPU overhead appears. Iterator/pull execution favors streaming and output control. Materialization favors simple batches. Push execution favors tight pipelines but needs more careful flow control.

Selections

Logically, selection $\sigma$ keeps the rows satisfying a predicate; physically, the question is how to reach those rows. For SELECT * FROM Reserves R WHERE R.rname < 'C%', the DBMS must decide how to get to the matching rows in Reserves, not only how to test the predicate after reading them. This is where access paths first become visible.

Selectivity. The fraction of tuples expected to satisfy a predicate. If relation $R$ has $M$ pages and selectivity $s$, the result size is roughly $sM$ pages when matching tuples are clustered.

Selectivity is also called the reduction factor. It is estimated from statistics, and it drives the physical choice: a predicate that returns 0.1% of a relation should usually use an index, while a predicate that returns most pages may be cheaper as a scan.

Sequential scan. Read the entire relation page by page. The cost is $M$ page reads. This is the fallback access path: it is always available, and it is often reasonable when many tuples qualify.

Index scan. Use an existing index to find qualifying data entries, then retrieve the corresponding records. The cost depends on selectivity, clustering, and the index height.

Multi-index scan. Use more than one index, combine the RID sets, and then fetch the records. This helps for complex WHERE clauses where several indexed predicates narrow the result together.

For a simple selection, the common cases are:

• No index, unsorted file. Scan the whole relation. Cost: $M$ page reads.
• No index, sorted file. Binary-search to the first matching page, then scan matching pages. Cost: search cost plus selected pages.
• Index on the selection attribute. Use the index to find qualifying data entries, then retrieve records.

With Reserves at 1000 pages, the unsorted scan costs 1000 I/Os. If the file is sorted on rname, a range such as rname < 'C%' costs roughly a binary search plus the matching pages, for example about 10 I/Os plus selectivity * 1000 pages.

For an index scan, clustering decides the data-read cost: a clustered index reads about as many pages as hold matches, while an unclustered index may read close to one page per matching tuple, since each RID can lead somewhere else.

Example 1. Clustered and unclustered index cost.

Suppose Reserves has $100{,}000$ tuples in $1000$ pages, and 10% of tuples satisfy rname < 'C%'. The qualifying tuples occupy about $100$ pages if the index is clustered, so the data-read cost is a little over $100$ I/Os. With an unclustered index, the $10{,}000$ qualifying tuples may live on unrelated pages, so the worst case is close to $10{,}000$ data-page reads.

RID sorting for an unclustered index. An optimization that turns many scattered record fetches into fewer page fetches by grouping RIDs by page id.

The process is:

1. Search the index. Collect all RIDs of tuples satisfying the predicate.
2. Sort RIDs by page id. Group RIDs that refer to the same data page.
3. Fetch pages in order. Read each data page once and retrieve all matching tuples on that page.

Without RID sorting, six matching tuples might cause repeated reads such as P1, P3, P1, P5, P3, P1, which reads P1 three times and P3 twice. With RID sorting, the DBMS reads P1, P3, and P5 once each.

General selections combine predicates. The optimizer wants a form where index results can be combined cleanly. AND corresponds to intersection of RID sets. OR corresponds to union of RID sets.

Selection with AND. A predicate where every condition must hold. AND usually helps indexes because each condition reduces the candidate set.

For day < '8/9/94' AND rname = 'Paul', two standard methods are:

The first method uses one access path and CPU filtering. The second lets both indexes narrow independently.

Selection with OR. A predicate where a tuple may satisfy any branch. OR is harder because using only one branch's index misses tuples that satisfy another branch.

For (day < '8/9/94' AND rname = 'Paul') OR bid = 5 OR sid = 3, the top-level connective is OR. The DBMS cannot simply use the day index, because tuples with bid = 5 or sid = 3 may fail the day predicate and still belong in the answer. The fix is to rewrite the condition so AND is on top.

Conjunctive normal form (CNF). A predicate written as an AND of clauses: C1 AND C2 AND ... AND Ck. Each clause Ci is an OR of smaller terms, such as ai1 OR ai2 OR ....

Within one CNF clause, OR means union of matching RID sets. Between clauses, AND means intersection.

Example 2. Rewriting an OR-heavy predicate.

Let $A$ be day < 8/9/94, $B$ be rname = 'Paul', $C$ be bid = 5, and $D$ be sid = 3. The predicate (A AND B) OR C OR D has OR at the top, so using just one index can miss tuples that match a different branch. Distributing OR over AND gives (A OR C OR D) AND (B OR C OR D).

The DBMS executes the CNF form by RID-set operations:

1. First conjunct. Lookup day < 8/9/94, bid = 5, and sid = 3; union their RID sets to get RIDs1.
2. Second conjunct. Lookup rname = 'Paul', bid = 5, and sid = 3; union their RID sets to get RIDs2.
3. Final result. Intersect RIDs1 and RIDs2, fetch those records, and apply any remaining predicates not covered by indexes.

Index choice depends on the index key.

$B^+$-tree index matching. A $B^+$-tree on $\langle a,b,c\rangle$ helps only when the predicate starts from the left side of the key.

• Matches prefix a. a = 5, or a range such as a > 10.
• Matches prefix a,b. a = 5 AND b = 3.
• Matches full key. a = 5 AND b = 3 AND c = 7.
• Does not match. b = 3, because it skips a.
• Partial use only. a = 5 AND c = 7 can use a = 5, but missing b prevents using c for navigation.

This is the leftmost-prefix rule at work: equalities on early columns, then at most one range predicate on the next column.

Hash index matching. A hash index on $\langle a,b\rangle$ requires exact equality on every key column.

• Matches. a = 5 AND b = 3.
• Does not match. a = 5, because b is missing.
• Does not match. b = 3, because a is missing.
• Does not match. a = 5 AND b > 3, because a range cannot identify one hash bucket.
• Does not match. a < 10 AND b = 3, because a range on a cannot identify one hash bucket.

The rule is: a hash index needs exact equality on every key column. No ranges, and no partial key.

Projection

Selections remove rows; projection $\pi$ removes columns. Without duplicate elimination, projection is just a scan that emits fewer attributes. SELECT DISTINCT is the hard part, because two tuples that differed only in removed columns become equal after projection.

For SELECT DISTINCT R.sid, R.bid FROM Reserves R, the DBMS must remove duplicate (sid, bid) pairs. Sorting solves this by making equal projected tuples adjacent.

Let $M$ be the number of pages in the original relation, $T$ the number of pages after dropping unneeded columns, and $P$ the number of external-sort passes over the projected file. A straightforward implementation has four I/O terms: M, T, 2PT, and T. These correspond to reading the original relation, writing the projected file, sorting that projected file, and reading it once more to remove adjacent duplicates.

Example 3. Projection by ordinary sorting.

If Reserves has $M=1000$ pages, the projected columns occupy $T=250$ pages, and sorting takes $P=2$ passes, the four terms are 1000, 250, 1000, and 250 I/Os. The total is 2500 I/Os, ignoring the final write of the query result.

The sort can be specialized for projection. During pass 0, the DBMS reads the original relation, drops unwanted columns, sorts each in-memory run on the projected attributes, and writes only the projected runs. During merge, duplicates are removed as soon as equal projected tuples become adjacent. With enough buffers for a two-pass sort, the cost becomes

For the same numbers, this is $1500$ I/Os.

Hash-based projection. A duplicate-elimination algorithm that partitions projected tuples by a hash function, then removes duplicates inside each partition.

Hashing solves the same grouping problem without producing sorted output. The first phase reads the relation, drops unwanted fields, and applies a hash function $h_1$ to the remaining tuple. Each tuple is written to one of $B-1$ partitions, where $B$ is the number of buffer pages. Identical projected tuples always go to the same partition, so duplicates never need to be compared across partitions.

The second phase reads one partition at a time and uses an in-memory hash table with a second hash function $h_2$ to discard duplicates. If a partition does not fit in memory, the same method is applied recursively to that partition.

Under uniform partitioning, each partition has about $T/(B-1)$ pages. To process a partition in memory, the DBMS needs

which is commonly summarized as $B > \sqrt{T}$. If the partitions fit, hash projection also costs $M+2T$: read the original relation, write projected partitions, and read the partitions for duplicate elimination.

Sort-based projection is the standard choice when the output order is useful for a later operator, or when skew makes hash partitions unreliable. Hash projection is attractive when enough memory is available and no sorted result is needed. If all projected attributes are already covered by an index, an index-only scan can avoid reading the base records.

Joins

Selections and projections are unary; a join $\bowtie$ reads two inputs, so the algorithm choice matters far more. A multiway join is usually executed as a sequence of pairwise joins, $R \bowtie S \bowtie T$ as $(R \bowtie S) \bowtie T$. Executing the definition literally, a cross product followed by a selection, is hopeless at scale because the intermediate product is enormous; physical join algorithms exist to avoid forming pairs that cannot match.

Assume relation $R$ has $M$ pages and $p_R$ tuples per page. Relation $S$ has $N$ pages and $p_S$ tuples per page. This cost model counts reads and ignores CPU cost and the cost of writing the output.

Simple nested loops join. For each tuple in the outer relation, scan the entire inner relation.

If $R$ is outer and $S$ is inner, the I/O cost is

The outer relation is scanned once. Then, for every outer tuple, the inner relation is scanned again.

Example 4. Tuple-oriented nested loops.

Let Reserves be $R$ with $M=1000$ pages and $p_R=100$, and let Sailors be $S$ with $N=500$ pages and $p_S=80$. With Reserves outer, the cost is 50,001,000 I/Os, about 139 hours at 10 ms per I/O. With Sailors outer, the cost is 40,000,500 I/Os, about 111 hours. The inner/outer choice matters even for the simplest algorithm.

Page-oriented nested loops join. For each page of the outer relation, scan every page of the inner relation and compare the tuples inside the two pages.

The cost is

For the same Reserves outer and Sailors inner example, this gives 501,000 I/Os, about 1.4 hours at 10 ms per I/O. This is far better than tuple-oriented nested loops because the inner relation is scanned once per outer page, not once per outer tuple.

Block nested loops join. A nested-loops join that reads many outer pages into memory at once, then scans the whole inner relation for that block.

If the outer block holds $b$ pages, the cost is

The buffer pool usually reserves one page for the inner scan and one page for output, so $b$ is roughly the remaining number of available pages.

Every nested-loops improvement is the same move: amortize one scan of the inner relation over more of the outer.

Example 5. Using buffers.

With Reserves outer, Sailors inner, and block size $b=100$, the outer relation forms 10 blocks. The DBMS reads Reserves once and scans Sailors once per block, for 6000 I/Os, about one minute at 10 ms per I/O.

One-pass hash join. A join that builds an in-memory hash table on one relation, then scans the other relation and probes the hash table for matches.

If the build relation fits in memory, the cost is

The build side should normally be the smaller input, because it must fit in memory together with the hash table overhead. For the running numbers, a one-pass hash join costs $1500$ I/Os, about 15 seconds at 10 ms per I/O.

One-pass sort-merge join. A join that sorts both inputs in memory and then merges them on the join key.

If both inputs fit in memory, the read cost is also $M+N$. This is the same one-pass cost as hash join, but the memory assumption is stronger: both relations must be available for in-memory sorting. In disk-oriented settings, sort-merge join is often a two-pass or multi-pass algorithm rather than a true one-pass algorithm.

The running example gives the scale.

All these plans compute the same relational join, but their I/O behavior differs by orders of magnitude.