ch18Pushing instead of pulling

What you build: Pipelines that push batches from a scan into sinks, a sink that says when it has enough, and a tee that feeds one scan to two consumers, counting the bytes each way reads.

The question#

What changes when the scan drives the query, pushing rows up the plan, instead of the top asking for them?

Every operator you have built pulls. ch01’s projection asked its filter for a batch, the filter asked the scan, and nothing was read until the top of the plan asked. DuckDB runs its plans the other way round: the scan loops over the file and hands each batch to the operator above it, which hands its result on, until the batch reaches an operator that keeps it. The operators and the rows are the same. What changes is who holds the loop, and this chapter counts what follows from that.

Observe#

Two questions about the returned orders. The first needs every returned order:

-- ch18, Pushing instead of pulling: how many returned orders are there? Every
-- row counts.
SELECT count(*) AS returned
FROM 'fixtures/orders-sorted.parquet'
WHERE status = 'returned';

The second needs only one:

-- ch18, Pushing instead of pulling: is there a returned order at all? One row
-- answers it.
SELECT EXISTS (
    SELECT 1 FROM 'fixtures/orders-sorted.parquet' WHERE status = 'returned'
) AS any_returned;

DuckDB plans the second like this:

┌───────────────────────────┐
│         PROJECTION        │
│    ────────────────────   │
│          (#0 = 1)         │
│                           │
│          ~1 Rows          │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│    UNGROUPED_AGGREGATE    │
│    ────────────────────   │
│        Aggregates:        │
│        count_star()       │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         PROJECTION        │
│    ────────────────────   │
│             42            │
│                           │
│         ~4000 Rows        │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│      STREAMING_LIMIT      │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│       PARQUET_SCAN        │
│    ────────────────────   │
│         Function:         │
│        PARQUET_SCAN       │
│                           │
│          Filters:         │
│   status='returned' AND   │
│     status IS NOT NULL    │
│                           │
│         ~4000 Rows        │
└───────────────────────────┘

A third asks for the customers whose returns add up to more than the average returned order. It reads the returned orders twice, once for each customer’s total and once for the average:

-- ch18, Pushing instead of pulling: the customers whose returns add up to
-- more than the average returned order. The returned orders are read by two
-- consumers: the totals and the average.
WITH returned AS (
    SELECT customer_id, amount
    FROM 'fixtures/orders-sorted.parquet'
    WHERE status = 'returned'
)
SELECT customer_id, sum(amount) AS returned
FROM returned
GROUP BY customer_id
HAVING sum(amount) > (SELECT avg(amount) FROM returned)
ORDER BY customer_id;

Its twin asks DuckDB to read the returned orders once and keep them:

-- ch18, Pushing instead of pulling: the same customers, with the returned
-- orders read once and kept, for both consumers to read.
WITH returned AS MATERIALIZED (
    SELECT customer_id, amount
    FROM 'fixtures/orders-sorted.parquet'
    WHERE status = 'returned'
)
SELECT customer_id, sum(amount) AS returned
FROM returned
GROUP BY customer_id
HAVING sum(amount) > (SELECT avg(amount) FROM returned)
ORDER BY customer_id;

What DuckDB read for each, and how many scans its plan has:

1. The question that needs one row read a fraction of what the count read. The STREAMING_LIMIT above the scan keeps one row, and the scan stopped soon after. Something had to tell it to stop: in DuckDB, the scan is not asked for rows, it hands them up.
2. Two consumers of one input, two scans. Written plainly, the returned orders feed two parts of the plan, and each reads them from the file: nearly twice the bytes.
3. Kept, they are read once, and scanned twice. MATERIALIZED reads the file once into a result DuckDB holds in memory, and two CTE_SCANs read that result. It saves the second read by holding every returned order until both consumers are done.

Running a plan by having the operator at the bottom hand each batch up to the operator above it, and so on to the top, is the push model. An operator that keeps what it is handed, instead of handing it on, is a sink: an aggregate’s table, a join’s build side, the query’s result.

Predict, then measure#

Your engine pushes both queries. The scan tests the status, as in ch03, and pushes each row group’s returned orders on. For the first query, the sink wants one row; the panel runs it with a source that listens when the sink says it has enough, and with one that never listens. For the second, it runs two consumers of the returned orders pulled, each asking its own scan, and pushed, through one scan that hands every batch to both.

Predict the bytes each way reads. The panel shows the bytes of reading to the end, or of one scan, beside each.

The panel draws what your engine counted when the book was built. Its button runs the same pipelines again in your browser.

• A source that listens stops after the first row group with a returned order. It read the footer and one row group’s statuses. The one that never listens reads every row group, and its sink ignores every row after the first.
• Pulled, each consumer pays for its own scan. Two consumers, twice the bytes; the chart’s rising line adds a scan’s worth for each consumer more.
• Pushed, one scan serves every consumer. The flat line: the tee hands each batch to every consumer, and each takes in every returned order, from one read of the file.

Building it#

A step#

A pushed operator is a step: it is handed a batch, does its work, and pushes the result to the steps after it. What it returns is its answer to one question: does anything after it want more?

class Step:
    """One operator of a pushed pipeline. It is handed a batch, does its work on it, and pushes
    the result on to the steps after it. Its answer says whether anything after it wants more."""

    def __init__(self, name: str, detail: str, then: list[Step]) -> None:
        self.metrics = Metrics(name, detail)
        self.then = then
        # The counters read as a pulled plan's do: each step after this one takes its rows from it.
        for step in then:
            step.metrics.children.append(self.metrics)

    def push(self, batch: pa.RecordBatch) -> bool:
        raise NotImplementedError

    def finish(self) -> None:
        """The source has nothing more: pass the word on, so every sink can finish its work."""
        for step in self.then:
            step.finish()

    def take(self, batch: pa.RecordBatch) -> pa.RecordBatch:
        self.metrics.rows_in += batch.num_rows
        self.metrics.batches_in += 1
        return batch

    def emit(self, batch: pa.RecordBatch) -> pa.RecordBatch:
        self.metrics.rows_out += batch.num_rows
        self.metrics.batches_out += 1
        self.metrics.peak_memory_bytes = max(self.metrics.peak_memory_bytes, batch.get_total_buffer_size())
        return batch

Steps keep the same counters as pulled operators, and the same ledger holds: each step’s rows in are the rows handed to it.

The source’s loop#

In a pulled plan, the loop is at the top: the result asks for batches until there are none. In a pushed pipeline, the loop is at the bottom, in the source, and it is the only place a pipeline can stop:

def drive(source: Operator, step: Step, listen: bool = True) -> None:
    """Run a pipeline: the source's loop pushes each of its batches into ``step``. A source that
    listens stops as soon as nothing downstream wants more; one that does not reads to the end."""
    step.metrics.children.append(source.metrics)
    for batch in source.batches():
        wants_more = step.push(batch)
        if listen and not wants_more:
            break
    step.finish()

Make the source deaf, changing if listen and not wants_more: to if False:, and run the panel’s report on your edit: the first case reads to the end, as the second always did. In a pulled plan, stopping is free, since a scan nobody asks reads nothing more. In a pushed one, it is a message the source must be written to hear.

Steps and sinks#

A filter is a step that pushes on what passes. The sink that answers whether a row exists wants none once it has one:

class Filter(Step):
    """Keep the rows of each batch for which ``predicate`` is true, and push them on."""

    def __init__(self, description: str, predicate: BatchFunction, then: Step) -> None:
        super().__init__("Filter", description, [then])
        self.predicate = predicate

    def push(self, batch: pa.RecordBatch) -> bool:
        self.take(batch)
        return self.then[0].push(self.emit(batch.filter(self.predicate(batch))))

class Exists(Step):
    """A sink that answers whether any row reached it. The first row answers the question, so from
    then on it wants no more."""

    def __init__(self) -> None:
        super().__init__("Exists", "", [])
        self.found = False

    def push(self, batch: pa.RecordBatch) -> bool:
        self.take(batch)
        self.found = self.found or batch.num_rows > 0
        return not self.found

    def finish(self) -> None:
        self.emit(pa.record_batch({"exists": [self.found]}))

One scan, two consumers#

A tee hands every batch it is pushed to each step after it, and wants more while any of them does:

class Tee(Step):
    """One input pushed to several steps: every batch goes to each of them. It wants more while
    any of them does."""

    def __init__(self, then: list[Step]) -> None:
        super().__init__("Tee", f"{len(then)} consumers", then)

    def push(self, batch: pa.RecordBatch) -> bool:
        self.emit(self.take(batch))
        answers = [step.push(batch) for step in self.then]
        return any(answers)

The query’s two consumers are ch07’s hash aggregate, rewritten as a sink: the same loop, run on each batch as it is pushed, instead of on each batch it pulls. The first pipeline reads the file once and fills both:

def big_returners(
    root: Path, consumers: tuple[Consumer, ...] = (TOTALS, AVERAGE)
) -> tuple[list[Scan], list[Group]]:
    """queries/big_returners.sql pushed: one scan of the returned orders, and a tee that pushes
    every batch to each consumer's aggregate."""
    scan = Scan(root / ORDERS, ["customer_id", "amount"], filters=[RETURNED])
    groups = [Group(keys, aggregates, scan.schema()) for keys, aggregates in consumers]
    drive(scan, Tee(groups))
    return [scan], groups

Pipelines#

An aggregate cannot push anything on until its source has run dry, since any row might belong to any group. It ends the pipeline, and its groups are the source of the next one. An operator that must see all its input before it hands anything up is a pipeline breaker, and a pushed plan is cut into pipelines at every one: aggregates, sorts, and the build side of a hash join. The second pipeline starts from the totals the first one left in its sink:

def over_average(totals: pa.RecordBatch, average: float) -> pa.Table:
    """The second pipeline: its source is the totals the first one left in a sink, and it keeps the
    customers whose returns add up to more than the average returned order."""
    result = Collect()
    keep = Filter("returned > average", lambda b: pc.greater(b["returned"], average), result)
    drive(Source([totals], totals.schema), keep)
    return pa.Table.from_batches(result.batches, schema=totals.schema).sort_by("customer_id")

The book’s tests check that both queries give DuckDB’s answers, pushed and pulled, that every consumer of the tee takes in every row the scan hands up, and that pulled consumers read one scan’s bytes each.

Compare#

Your engine’s bytes, beside DuckDB’s:

• Both stop a scan for a sink that has enough. DuckDB’s STREAMING_LIMIT tells its scan it is finished, as your Exists tells your source. Each reads a little of the file beyond its footer; the two readers fetch it in different pieces.
• Your tee and DuckDB’s kept result read the same bytes. Both read the returned orders’ columns once. They differ in what they hold: your tee holds one batch at a time, and DuckDB’s materialised result holds every returned order until both consumers have read it.
• Pulled, both read the file twice. Your two scans read it twice in full; DuckDB’s second read is a little smaller than its first.
• DuckDB’s operators answer as your steps do. Each of its operators returns whether it needs more input or has finished, and each sink whether it wants more, and its pipeline’s loop stops on the answer.

What this cannot tell you#

• Which model is faster. Pulling and pushing do the same work on the same batches; the counters cannot tell them apart where both read the same bytes. The difference in time lies in the calls between operators, and the book does not time them: under Python’s interpreter a call costs far more than in compiled code, and the times would say more about Python than about the two models.
• What a tee costs when its consumers are uneven. Your tee pushes every batch to every consumer at once. If one consumer is slower, a real engine must hold batches for it, or slow the others down.
• How DuckDB decides when to keep a result. DuckDB 1.1.2 reads a WITH twice unless told otherwise. Other engines, and other versions, may decide for themselves, weighing a second read against the memory a kept result holds.

What this means for your design#

• Ask for what you need, and the engine can stop early. EXISTS, LIMIT without ORDER BY, and a join that needs one match are all ways of saying so. Problem 18.3 shows a limit that cannot stop.
• Know how your engine treats a WITH used twice. Some read its input again for every use; some keep the result in memory. Either can be the right choice; check which yours made, in its plan.
• Read a plan’s pipeline breakers. Aggregates, sorts and a join’s build side each end a pipeline, and each holds its input, or what it made of it, until the pipeline is done. The memory of Part III is held there.
• Expect push engines to parallelise by pipeline. DuckDB’s workers, as ch14 simulated them, each take a morsel and push it through a pipeline to its sink. A loop at the source is what lets a scheduler hand out the work.

Key takeaways#

Problems#

There are three problems. The first two are code with tests; the third is a slow query to diagnose, with no test. Write your answers to the first two in the workbench, and run the graders there. Your answers stay in this browser, and Reset to the stubs starts again.

18.1 Push a hash join. Write Build and Probe: the build side of ch08’s hash join as a sink, which ends its pipeline, and the probe side as a step in the next. The graders build the customers, push the orders through your probe, and compare the rows with DuckDB’s join; they check the probe refuses to start before the build side is ready, and passes on a sink’s answer when it has enough.

18.2 Gather the crumbs. A selective filter pushes on batches of a few rows each, and every batch costs every step above it a call. Write Coalesce: a step that gathers small batches until they hold a given number of rows, then pushes them on as one. The graders push random batches through it and after the returned orders’ filter, and check no row is lost or reordered, no batch goes on early or late, and a sink’s answer comes back through it.

18.3 Diagnose the limit that read everything. No test. A colleague asks for any ten orders, and then for the ten earliest:

-- ch18, Pushing instead of pulling (problem 18.3): the ten earliest orders.
SELECT order_id, order_date, amount
FROM 'fixtures/orders-sorted.parquet'
ORDER BY order_date, order_id
LIMIT 10;

The file is sorted by date, and the ten earliest orders are all in its first row group. Why did DuckDB read every row group for them, and only one for any ten? What would an engine need to know, and to do, to stop after the first row group? A good answer names the operator that breaks the pipeline, says why it cannot tell the scan to stop, and uses each row group’s range of dates to say when the scan could stop, as ch04 used the same statistics to skip.

Where to go next#

• Push-based execution. Thomas Neumann, Efficiently Compiling Efficient Query Plans for Modern Hardware, VLDB 2011: pipelines, pushed from the source to each breaker, as the unit an engine runs. ch19 takes up the rest of the paper.
• Morsels and pipelines. Viktor Leis, Peter Boncz, Alfons Kemper and Thomas Neumann, Morsel-Driven Parallelism, SIGMOD 2014: a scheduler handing pipelines’ morsels to workers.
• Push and pull compared. Amir Shaikhha, Mohammad Dashti and Christoph Koch, Push versus Pull-Based Loop Fusion in Query Engines, Journal of Functional Programming, 2018: the same plans run both ways, and where each stops early or not.
• DuckDB’s pipelines. DuckDB’s source, in src/parallel/pipeline_executor.cpp: the loop that pushes a source’s chunks through a pipeline’s operators into its sink, and stops when the sink says it is finished.

Built from commit dc21108.