ch05Where work happens

What you build: A table scan that reads a table's metadata and opens only the files that might match, and a scan handed to storage that tests the rows itself.

The question#

When a table is many files, what decides which of them are opened, and where are the rows tested?

In ch03 and ch04, your scan skipped row groups and pages inside one file. Before it could skip anything, it had to open the file and read its footer, because the statistics it skipped by live there. A table in a data lake is rarely one file. It is hundreds or thousands, and a scan that must open each one to learn what it holds pays for every footer before it reads a row. There is a second cost the earlier chapters took for granted: every byte your scan tested had already crossed from the storage to the engine. This chapter moves both decisions. It decides which files to open before opening any, and then asks what changes when the storage tests the rows itself.

Observe#

The table is orders-by-month: the sorted orders of ch03, cut into one file per month. Each file sits in a directory named for its month, such as month=2024-03. The query is chapter 3’s fortnight, asked of every file in the table through a glob:

-- ch05: chapter 3's fortnight of orders, from a table of monthly files,
-- found by a glob
SELECT order_id, customer_id, amount
FROM read_parquet('fixtures/orders-by-month/*/*.parquet')
WHERE order_date >= DATE '2024-03-01'
  AND order_date < DATE '2024-03-15';

DuckDB’s plan is one scan, with the columns and the dates pushed into it, as it was for one file:

┌───────────────────────────┐
│       READ_PARQUET        │
│    ────────────────────   │
│         Function:         │
│        READ_PARQUET       │
│                           │
│        Projections:       │
│          order_id         │
│        customer_id        │
│           amount          │
│                           │
│          Filters:         │
│ order_date>='2024-03-01': │
│ :DATE AND order_date<'2024│
│     -03-15'::DATE AND     │
│   order_date IS NOT NULL  │
│                           │
│         ~4065 Rows        │
└───────────────────────────┘

The plan does not say how many files the scan will open. Now the same fortnight, asked of the directories’ names instead of the dates alone:

-- ch05: the same fortnight, asked of the month in each file's directory
SELECT order_id, customer_id, amount
FROM read_parquet(
  'fixtures/orders-by-month/*/*.parquet', hive_partitioning = true
)
WHERE month = '2024-03'
  AND order_date < DATE '2024-03-15';

┌───────────────────────────┐
│       READ_PARQUET        │
│    ────────────────────   │
│         Function:         │
│        READ_PARQUET       │
│                           │
│        Projections:       │
│          order_id         │
│        customer_id        │
│           amount          │
│                           │
│          Filters:         │
│  order_date<'2024-03-15': │
│  :DATE AND order_date IS  │
│          NOT NULL         │
│                           │
│       File Filters:       │
│    (month = '2024-03')    │
│                           │
│    Scanning Files: 1/12   │
│                           │
│         ~338 Rows         │
└───────────────────────────┘

1. The second plan filters files, not rows. Naming each directory for the value of a column, month=2024-03, is Hive partitioning, after the Hadoop warehouse that made it common. The month column is in no file. DuckDB reads it from each file’s path, so the test month = '2024-03' rules a file out from its name alone, and the plan says how many files are left to scan.
2. The first plan cannot do that. Its dates are inside the files. To learn a file’s dates, DuckDB must open it and read its footer, where the row group statistics are.
3. The query had to change to use the layout. The month and the dates say the same thing twice, and a query that asks only for the dates gains nothing from the directories.

A table format does better than directory names. It keeps, beside the data files, a table metadata file: a list of the table’s data files, each with the smallest and largest value of every column in it. Apache Iceberg keeps this in its manifests, and Delta Lake in its transaction log. A reader of the table metadata can rule out a file by any column, from one read, without opening the file.

Predict, then measure#

Your engine gains a table scan. By its metadata, it reads the table metadata first and opens only the files whose bounds might hold a match. By its files, it lists the table’s directory and opens every file, as DuckDB’s glob does, and skips a file’s row groups by the statistics in its footer.

Predict how many files each scan opens. The table holds the year’s orders in date order, a month to a file, and the query asks for a fortnight.

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

• By its metadata, the scan opens one file. The bounds of every other month rule it out, and the scan never fetches a byte of it.
• By its files, the scan opens every file. Each file’s footer rules out its one row group, so the scan decodes the same rows; but to read a footer, the scan must first ask where it is, then fetch it. Every file costs requests before the scan knows it holds nothing.

Building it#

A file’s bounds, before it is opened#

The table metadata stores each file’s bounds as a Parquet file’s statistics store them: the smallest and largest value, encoded as the column’s values are, with the comparator that orders them. So a comparison can hand them to the Parquet book’s reader, which decides for a file as it decides for a row group:

    def against_bounds(self, comparator: stats.Comparator, lower: bytes, upper: bytes) -> prune.Decision:
        """Whether a whole file can be skipped, from the bounds a table's metadata keeps for it (ch05).

        The metadata stores each column's bounds as the file's statistics store them, PLAIN-encoded,
        with the comparator that orders them, so the Parquet book's reader can decide as it does for
        a row group, before the file is opened.
        """
        value = prune.literal(comparator, _literal(self.value))
        predicate = prune.Predicate(0, prune.Op(self.op), value, comparator)
        return prune.against_bounds(predicate, (lower, upper), None, 1)

The table scan#

The table scan reads the table metadata, or lists the directory, then opens the files it could not rule out, one after another. Each file it opens is read by chapter 4’s scan, with the same filters, so everything the earlier chapters built still applies inside each file. All the files share one store, so the table’s counters are every request the scan made:

class TableScan(Operator):
    """Read a table of many Parquet files, one batch per row group of each file it opens (ch05).

    With ``metadata``, the scan first reads the table's metadata: a list of the table's files,
    each with the bounds of every column, as Iceberg keeps them in its manifests. A file whose
    bounds rule out every row is never opened. Without it, the scan lists the table's files and
    opens every one, and only a file's own footer can rule its rows out. Either way, each file it
    opens is read by a :class:`Scan` with the same filters, sharing one store, so the table's
    counters are every request the scan made.
    """

    #: The table's metadata, in the table's directory.
    METADATA = "metadata.json"

    def __init__(
        self, table: str | Path, columns: list[str], filters: list[Comparison] = (), metadata: bool = True
    ) -> None:
        detail = f"{Path(table).name}/: {', '.join(columns)}"
        if filters:
            detail += f"; {' AND '.join(map(str, filters))}"
        detail += "; by its metadata" if metadata else "; by its files"
        super().__init__("TableScan", detail, [])
        self.table = Path(table)
        self.columns = columns
        self.filters = list(filters)
        self.metadata = metadata
        self.skipped: dict[str, str] = {}
        """Each file the scan never opened, with the reason."""
        self.opened: list[str] = []
        """Each file the scan opened, in the order it opened them."""

    def batches(self) -> Iterator[pa.RecordBatch]:
        objects = MemoryStore()
        for path in sorted(self.table.rglob("*")):
            if path.is_file():
                objects.put(path.relative_to(self.table).as_posix(), path.read_bytes())
        store = TracingStore(objects, NetworkModel())
        if self.metadata:
            listing = json.loads(store.get(self.METADATA, All(), "the table's metadata").data)["files"]
        else:
            found = store.list("", "list the table's files")
            listing = [{"path": key} for key, _ in found if key.endswith(".parquet")]
        self.skipped, self.opened = {}, []
        self.count_requests(store)

        for entry in listing:
            why = self.ruled_out(entry) if self.metadata else None
            if why is not None:
                self.skipped[entry["path"]] = why
                continue
            self.opened.append(entry["path"])
            scan = Scan(self.table / entry["path"], self.columns, self.filters, store=(store, entry["path"]))
            for batch in scan.batches():
                self.count_requests(store)
                yield self.emit(batch)
            # The file's scan counted the rows it decoded and held; the table's are the sum.
            self.metrics.rows_in += scan.metrics.rows_in
            self.metrics.batches_in += scan.metrics.batches_in
            self.metrics.peak_memory_bytes = max(
                self.metrics.peak_memory_bytes, scan.metrics.peak_memory_bytes
            )
        self.count_requests(store)

    def ruled_out(self, entry: dict) -> str | None:
        """Why no row of a file can pass the filters, from its bounds in the table's metadata, or
        None if some row might."""
        for f in self.filters:
            bounds = entry["bounds"].get(f.column)
            if bounds is None:
                continue
            comparator = stats.Comparator[bounds["comparator"]]
            decision = f.against_bounds(
                comparator, bytes.fromhex(bounds["lower"]), bytes.fromhex(bounds["upper"])
            )
            if decision.skip:
                return f"{f}: {decision.why}"
        return None

Make ruled_out rule out nothing, and run the panel’s report: the scan by the metadata now opens every file, and pays for the metadata as well.

The plan#

def early_march_table(root: Path, metadata: bool = True) -> Operator:
    """queries/early_march_table.sql: the fortnight from the table of monthly files (ch05). With
    ``metadata``, the scan reads the table's metadata first; without it, it opens every file."""
    columns = ["order_id", "customer_id", "amount"]
    return TableScan(root / "fixtures" / "orders-by-month", columns, filters=EARLY_MARCH, metadata=metadata)

The book’s tests run the table scan with several predicates, both ways, and compare its rows with DuckDB’s. They read every file the metadata ruled out, row by row, to make sure it held no match, and check that the metadata lists every file with its true row count and size.

Storage that tests the rows#

Every scan so far fetched bytes from the storage and tested the rows in the engine. An object store answers ranges of bytes and nothing more, so an engine reading from one has no other choice. Some storage can do more. Given a file, the columns and the filters, it scans the file where the file lives and sends back only the rows that pass. Vast DataBase, for example, takes a query’s columns and predicates in its client’s request, evaluates the predicates next to its disks, and returns the matching rows as Arrow batches.

Your engine can simulate such storage with the scan it already has, run on the storage’s side. The storage sends the rows as an Arrow IPC stream, the format Arrow uses to move batches between processes, and the engine’s scan makes one request and reads what comes back:

class ComputingStore:
    """A store of Parquet files that can run a scan itself, on the storage's side."""

    def __init__(self, root: str | Path) -> None:
        self.root = Path(root)

    def select(self, key: str, columns: list[str], filters: list[Comparison]) -> tuple[bytes, Metrics]:
        """Scan ``key`` where it lives, with the page index where the file has one, and return the
        rows that pass as an Arrow IPC stream, with the storage's own scan's counters."""
        scan = Scan(self.root / key, columns, filters, page_index=True)
        table = scan.run()
        sink = pa.BufferOutputStream()
        with pa.ipc.new_stream(sink, table.schema) as writer:
            writer.write_table(table)
        return sink.getvalue().to_pybytes(), scan.metrics

class StorageScan(Operator):
    """Hand the whole scan to the storage: one request carries the file, the columns and the
    filters, and the response carries only the rows that pass.

    ``bytes_read`` counts the response, which is every byte that crossed the network. The storage
    decoded more than it sent; what it read of its disks is :attr:`storage`, its own scan's
    counters, reported beside this operator's and never added to them.
    """

    def __init__(
        self, store: ComputingStore, key: str, columns: list[str], filters: list[Comparison] = ()
    ) -> None:
        detail = f"{key}: {', '.join(columns)}"
        if filters:
            detail += f"; {' AND '.join(map(str, filters))}"
        super().__init__("StorageScan", detail + "; in the storage", [])
        self.store = store
        self.key = key
        self.columns = columns
        self.filters = list(filters)
        self.storage: Metrics | None = None
        """The storage's own scan's counters, once the scan has run."""

    def batches(self) -> Iterator[pa.RecordBatch]:
        payload, self.storage = self.store.select(self.key, self.columns, self.filters)
        self.metrics.requests = 1
        self.metrics.bytes_read = len(payload)
        for batch in pa.ipc.open_stream(payload):
            self.take(batch)
            yield self.emit(batch)

    def schema(self) -> pa.Schema:
        return Scan(self.store.root / self.key, self.columns).schema()

The engine counts what crossed the network; the storage keeps its own scan’s counters, which say what it read from its disks. The two are reported side by side and never added. Here is chapter 3’s fortnight, from the sorted and the shuffled file, tested in your engine and in the storage:

• The storage sends the result, not the data. The rows that pass are a small share of what the scan read, so far fewer bytes crossed the network, in one request.
• The storage still reads what the scan read. Moving the test changed where the bytes were read, not how many. On the shuffled file, the storage read every row group, as your engine did in ch03.
• The order of the rows stops mattering to the network. Both files send about the same bytes, because both hold the same matching rows. It still matters to the storage’s disks.

Compare#

Your table scans beside DuckDB’s, on the same table:

• All four hand up the same rows. Which files a scan opens changes what it reads, never what it returns.
• By its files, your scan reads exactly what DuckDB’s glob reads. Both open every file, read each footer, and read the row group of March alone.
• The table metadata saves requests more than bytes. One file opened instead of every file cut the requests most. The metadata is not free: it holds the bounds of every column of every file, and on a table this small it is a large share of what the scan read. On a table of thousands of files, it is one read against thousands of footers.
• By the month in the path, DuckDB reads the fewest bytes. A path costs nothing to read. It opened one file besides March’s; from the bytes, it read that file’s footer, which it seems to use for the table’s schema. Its plan does not say.

What this cannot tell you#

• How long listing takes. The simulated store lists a directory in one request. A real object store returns its keys a page at a time, a request per page, and a table of many directories lists slowly. Table metadata avoids listing altogether.
• What a real table format stores. Iceberg’s metadata is a tree: a metadata file names a snapshot, whose manifest list names manifests, which list the data files with their bounds. The book’s table metadata is one file with the same bounds, and none of the snapshots, schema history or deletes that make a table format a format.
• What the storage’s work costs. The storage that tests rows decoded as many rows as your engine would have. The counters say where the bytes moved, not how busy the storage was, or whether it shares its processors with other queries.
• Any version of DuckDB but the pinned one. The comparison shows what DuckDB 1.1.2 opens and reads for this table.

What this means for your design#

• Lay a table out by what its queries filter on. Files cut by month let a query by date skip most of them. A query by customer gains nothing from the same files, as problem 5.3 shows.
• Keep table metadata with column bounds. A reader of it rules files out by any column, from one read, without opening them. It saves most where the files are many.
• Beware many small files. Each file a scan opens costs requests for its footer before it yields a row, and the bounds of a small file are often as wide as a large one’s.
• Filter in the storage when the network is the constraint. Storage that tests rows sends the result instead of the data, which helps most when the predicate is selective. It does not help the storage’s own disks: a badly laid out file still costs a full read, only somewhere else.

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.

5.1 Which files? Write files_to_open: given the table metadata and some comparisons, the files that might hold a row passing all of them. Decode the bounds yourself, from the bytes the metadata stores. The graders check, for many predicates on every column, that you open every file holding a match, and that you open no file whose bounds rule it out.

5.2 Which directories? Write months_to_read: given the paths of a table in Hive’s layout and a range of dates, the files whose month holds a day in the range, decided from the paths alone. The graders try many ranges, including ones that end on the first of a month and ones that hold no day at all.

5.3 Diagnose the slow query. No test. A colleague asks the table for one customer’s orders, customer_id = 17, and finds it no faster than reading the whole table. Your table scan, by its metadata:

Why did the table metadata rule out no file? What would the bounds have to look like for it to rule some out, and what change to the way the table is written would make them look like that? What would that change cost the chapter’s query by date? A good answer reads the bounds in the table, names a layout that serves the customer query, and says what it does to the query by month.

Where to go next#

• The table format’s own account. The Apache Iceberg specification describes manifests, the column bounds they keep, and the partitioning Iceberg derives from a column’s values without putting them in a path.
• Hive partitioning in DuckDB. DuckDB’s Hive partitioning documentation says how it reads a column from a path and filters files by it.
• Arrow’s stream format. The Arrow columnar format specification describes the IPC stream the storage sends: a schema, then batches, in the same layout as in memory, so nothing is converted on arrival.
• Where the idea comes from. Michael Armbrust and others, Delta Lake: High-Performance ACID Table Storage over Cloud Object Stores, VLDB 2020, on why a table needs metadata beyond its files. Xiangyao Yu and others, PushdownDB: Accelerating a DBMS Using S3 Computation, ICDE 2020, on filtering in the storage and what it saves.
• A footer, byte by byte. Parquet, byte by byte decodes the footer every file-by-file scan reads, in its chapter on how readers read.

Built from commit dc21108.