ch14Lakehouse and beyond

What you build: Many files, partitions, a table log and file pruning over the same object store.

The question#

How does one Parquet file become one part of a larger analytical system?

Every chapter so far read one file. A table that grows every day is not one file: it is many, written at different times by different writers, in one directory of an object store. Before a reader can use anything from ch09 or ch10, it has to know which files there are, and it would rather not open the ones that cannot hold an answer. This chapter queries a table of nine files, and compares three ways of finding them.

The experiment#

A table of nine files#

fixtures/table/ holds the orders from ch11, written by pyarrow’s dataset writer. The writer split the rows by country into one directory per value, and wrote at most a hundred rows to a file.

Each step below is a few lines of code. In Python, run them in the page, change them with Edit, and run them again. In Rust, open them in a Codespace, or run one at a desk with cargo run -p walkthroughs --bin and its name.

List the table. Walk the directories and print each Parquet file with the directory it sits in and its size:

# The table's files, found by walking its directories: each one's partition and size.
from pathlib import Path

table = Path("fixtures/table")
for path in sorted(table.rglob("*.parquet")):
    key = path.relative_to(table).as_posix()  # country=UK/part-0.parquet
    directory = key.split("/")[0]  # a name, an equals sign and a value
    name, _, value = directory.partition("=")
    print(f"{key}: {name} is {value}, {path.stat().st_size} bytes")

//! The table's files, found by walking its directories: each one's partition and size.

use std::fs;

/// The names in a directory, in order.
fn names(dir: &str) -> Vec<String> {
    let entries = fs::read_dir(dir).expect("run this from the repository's root");
    let mut names: Vec<String> = entries
        .map(|e| {
            e.expect("an entry")
                .file_name()
                .to_string_lossy()
                .into_owned()
        })
        .collect();
    names.sort();
    names
}

fn main() {
    let table = "fixtures/table";
    for directory in names(table) {
        // A name, an equals sign and a value. The log's directory, _delta_log, has none.
        let Some((name, value)) = directory.split_once('=') else {
            continue;
        };
        for file in names(&format!("{table}/{directory}")) {
            let key = format!("{directory}/{file}"); // country=UK/part-0.parquet
            let size = fs::metadata(format!("{table}/{key}"))
                .expect("a file")
                .len();
            println!("{key}: {name} is {value}, {size} bytes");
        }
    }
}

Every file sits in a directory named for one country, and every directory name has the same shape. Most countries have one file. UK has three, since it has more orders than one file holds, and US has a second file of only a few orders. Keep only the paths whose value is UK: that filter is the whole of what the next section calls ruling out by path.

Partitions are in the paths#

The directory country=UK is a Hive-style partition: a name, an equals sign and a value. The files under it do not hold a country column at all. The dataset writer removed it, since every row in the directory has the same value, and a reader restores it from the path. Ask the book’s reader for country from one file alone:

PYTHONPATH=python python3 -m parquet_lab query "fixtures/table/country=UK/part-0.parquet" \
    "SELECT country FROM orders"

cargo run -p pqlab -- query "fixtures/table/country=UK/part-0.parquet" "SELECT country FROM orders"

The reader answers with an error: the file has no column country. Ask for order_id instead, and it answers.

A condition on a partition column is answered before any file is opened. A LIST request returns every key under the table’s prefix, and country = 'UK' keeps the keys whose path says country=UK. That is the cheapest skipping in the book: it costs nothing beyond the listing, and it never reads a byte of a file it rules out.

A condition on any other column learns nothing from a path. Listing can only find files; it cannot say what is in them.

A log that lists the files#

A table format keeps, beside the data files, a record of which files make up the table. Delta Lake’s is a transaction log: a directory of numbered JSON files, each a list of actions. An add action puts a file in the table, with its partition values, its size, and statistics about its columns. A remove action takes one out. The table is whatever the actions add up to.

fixtures/table/_delta_log/ holds one such file, in the shape Delta Lake uses, and the first step skipped it: its directory has no equals sign.

Read the log. Parse each line, and for each add, print the file and the statistics the log records for order_id:

# The table's log, read by hand: one action a line, and the statistics each add records.
import json

log = open("fixtures/table/_delta_log/00000000000000000000.json")
for n, line in enumerate(log, 1):
    action = json.loads(line)
    kind = next(iter(action))  # protocol, metaData, add or remove
    if kind != "add":
        print(f"line {n}: {kind}")
        continue
    add = action["add"]
    stats = json.loads(add["stats"])  # JSON inside a JSON string: parse it a second time
    low, high = stats["minValues"]["order_id"], stats["maxValues"]["order_id"]
    rows = stats["numRecords"]
    print(f"line {n}: add {add['path']}, {rows} rows, order_id {low} to {high}")

//! The table's log, read by hand: one action a line, and the statistics each add records.

use parquet_lab::json::Json;

fn main() {
    let log = std::fs::read_to_string("fixtures/table/_delta_log/00000000000000000000.json")
        .expect("run this from the repository's root");
    for (n, line) in (1..).zip(log.lines()) {
        let action = Json::parse(line).expect("one JSON object a line");
        let Json::Obj(fields) = &action else { continue };
        let kind = &fields[0].0; // protocol, metaData, add or remove
        if kind != "add" {
            println!("line {n}: {kind}");
            continue;
        }
        let add = &fields[0].1;
        // The statistics are JSON inside a JSON string: parse them a second time.
        let text = add.get("stats").and_then(Json::as_str).expect("statistics");
        let stats = Json::parse(text).expect("JSON");
        let order_id = |k: &str| stats.get(k).and_then(|m| m.get("order_id")).unwrap();
        let (low, high) = (
            order_id("minValues").to_json(),
            order_id("maxValues").to_json(),
        );
        let path = add.get("path").and_then(Json::as_str).unwrap();
        let rows = stats.get("numRecords").unwrap().to_json();
        println!("line {n}: add {path}, {rows} rows, order_id {low} to {high}");
    }
}

The first two lines say which version of the protocol the log follows and what the table’s schema is. Every line after them adds one file, and the files are the ones the first step found. The statistics are the same kind ch08 read from footers, a minimum and a maximum for each column, copied into the log when the file was written. A reader of the log knows each file’s range of order_id without opening it.

Change order_id to status, or print stats["nullCount"]: the log records every column.

The UK files show what the writer’s order gives the reader. The dataset writer wrote rows in the order it received them, by order_id, and started a new file every hundred rows, so the three UK files hold three separate ranges. The other countries have one file each, spanning the whole range:

Querying the table#

Query the table. The book’s reader does both kinds of ruling out, then fetches the files that survive through the simulated object store from ch02 and queries them together. Put the table’s files in a store, and ask it for the UK orders below 200, with the files found from the log:

# The book's reader queries the table: what it decided about each file, and every request.
import sys
from pathlib import Path
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab.object_store import MemoryStore, NetworkModel
from parquet_lab.table import Discovery, query

store = MemoryStore()  # every file of the table, under its key in a simulated object store
for path in Path("fixtures/table").rglob("*"):
    if path.is_file():
        store.put(path.relative_to("fixtures").as_posix(), path.read_bytes())

sql = "SELECT count(*) FROM orders WHERE country = 'UK' AND order_id < 200"
# Try Discovery.LIST_AND_PRUNE, then Discovery.LIST; then 1 or 8 connections.
result = query(store, "table/", sql, Discovery.LOG, 4, NetworkModel())

for f in result.files:
    print(f"{f.key}: {f.why}")
for r in result.requests:
    when = f"{r.start_us // 1000:>3}-{r.end_us // 1000:>3} ms"
    print(f"{when} on {r.connection + 1}: {r.method} {r.key}, {r.bytes_returned} bytes")
print(*result.answer.columns)
for row in result.answer.rows:
    print(*row)

//! The book's reader queries the table: what it decided about each file, and every request.

use parquet_lab::object_store::{MemoryStore, NetworkModel};
use parquet_lab::table::{query, Discovery};

/// Every file under `dir`, with its path.
fn files(dir: &std::path::Path, out: &mut Vec<std::path::PathBuf>) {
    for entry in std::fs::read_dir(dir).expect("run this from the repository's root") {
        let path = entry.expect("an entry").path();
        if path.is_dir() {
            files(&path, out)
        } else {
            out.push(path)
        }
    }
}

fn main() {
    let mut paths = Vec::new();
    files("fixtures/table".as_ref(), &mut paths);
    let mut store = MemoryStore::new(); // every file of the table, under its key
    for path in paths {
        let key = path.to_string_lossy().replacen("fixtures/", "", 1); // table/...
        store.put(&key, std::fs::read(&path).expect("a file"));
    }

    let sql = "SELECT count(*) FROM orders WHERE country = 'UK' AND order_id < 200";
    // Try Discovery::ListAndPrune, then Discovery::List; then 1 or 8 connections.
    let model = NetworkModel::default();
    let result = query(store, "table/", sql, Discovery::Log, 4, model).expect("an answer");

    for f in &result.files {
        println!("{}: {}", f.key, f.why);
    }
    for r in &result.requests {
        let when = format!("{:>3}-{:>3} ms", r.start_us / 1000, r.end_us / 1000);
        let (on, method, key) = (r.connection + 1, r.method.name(), &r.key);
        println!("{when} on {on}: {method} {key}, {} bytes", r.bytes_returned);
    }
    println!("{}", result.answer.columns.join(" "));
    for row in &result.answer.rows {
        let cells: Vec<String> = row.iter().map(|v| v.to_json().to_json()).collect();
        println!("{}", cells.join(" "));
    }
}

Every file has its reason. The paths rule out the files outside country=UK. The log’s statistics rule out two of the three UK files, whose ranges of order_id start too high. The reader fetches the log, then one file, and the count comes from that file alone.

Change Discovery.LOG to Discovery.LIST_AND_PRUNE (Discovery::ListAndPrune in Rust). The reader now lists the keys instead of reading the log, and rules out by path alone, so all three UK files are read, and the count is the same. With Discovery.LIST it reads every file. Then change the connections from 4 to 1 and to 8, and watch the times: whole files are small, so the time is almost all waiting for first bytes, and connections decide it.

Three ways to find the files#

The tables below run five queries with the files found each of the three ways. The first reads the whole table. With no condition, every way reads every file, and the log costs more bytes than listing, since it is fetched in full, for the same number of requests:

A condition on the partition column is answered by the path. Listing and pruning by path reads exactly the files the log reads:

A condition on any other column needs the log. Listing reads every file for a range of order_id. The log rules out the files whose ranges miss it: two of the UK files and the small US file.

Statistics rule out files for any column they cover. For status = 'refunded' the log skips the small US file, whose few rows are all paid. That skip is luck, a property of one file’s contents, not of the table’s design.

The two combine. For the step’s query, country = 'UK' AND order_id < 200, the path rules out six files and the log’s statistics rule out two more:

Time follows requests. Each file is fetched whole in one request, as ch10 found best for small files, and with four connections the table’s files arrive in rounds of four. Fewer files means fewer rounds.

What the log adds beyond skipping#

A listing tells a reader what is in a directory now. A log tells it what is in the table, which is not the same:

• A half-written file is not in the table. A writer adds a file to the log only after the file is complete, so a reader never sees a file that is still being written.
• A replaced file stays readable. A job that rewrites a partition adds the new files and removes the old ones in one log entry. A reader of the earlier entry keeps reading the old files, which are not deleted until no reader needs them.
• The whole table has one history. Each entry is a version of the table, and a reader can ask for any version the log still holds.

These are why table formats exist, and the files do not change for any of them. Each file is a Parquet file exactly as the earlier chapters read it.

Building it#

The steps read the paths and the log by hand, and then asked the reader for a plan. The reader’s table module writes the same reading once: partition values from a path, the files a log adds and removes, a rule for each condition, and the reader’s query engine, which ch12 ran on one file, over the files that survive.

Partition values from a path#

Every directory of the form name=value is a partition value, and values are percent-encoded where a path could not hold them:

def partition_values(path: str) -> list[tuple[str, str]]:
    """The partition values in a Hive-style path: every ``name=value`` directory, in order.
    Values are percent-encoded when they hold characters a path cannot, such as ``/`` or ``=``."""
    dirs = path.split("/")[:-1]
    return [(unescape(k), unescape(v)) for k, _, v in (d.partition("=") for d in dirs if "=" in d)]

pub fn partition_values(path: &str) -> Vec<(String, String)> {
    let dirs = path.split('/').collect::<Vec<_>>();
    dirs[..dirs.len().saturating_sub(1)]
        .iter()
        .filter_map(|d| d.split_once('='))
        .map(|(k, v)| (unescape(k), unescape(v)))
        .collect()
}

Reading the log#

Each line is one action. A remove takes out a file an earlier add put in, and an add carries the file’s path, partition values and statistics. The statistics are JSON inside a JSON string, so they are parsed twice, as the second step did:

        remove = action.get("remove")
        if isinstance(remove, dict):
            path = remove.get("path") if isinstance(remove.get("path"), str) else ""
            files = [f for f in files if f.key != path]
        add = action.get("add")
        if not isinstance(add, dict):
            continue
        key = add.get("path")
        if not isinstance(key, str):
            raise TableError(f"log line {n}: an add with no path")
        values = add.get("partitionValues")
        partition = (
            [(k, v if isinstance(v, str) else "") for k, v in values.items()]
            if isinstance(values, dict)
            else []
        )
        # The statistics are themselves JSON, stored as a string.
        stats = None
        raw = add.get("stats")
        if isinstance(raw, str):
            try:
                stats = _file_stats(json.loads(raw))
            except ValueError:
                stats = None

        if let Some(remove) = action.get("remove") {
            let path = remove
                .get("path")
                .and_then(Json::as_str)
                .unwrap_or_default();
            files.retain(|f| f.key != path);
        }
        let Some(add) = action.get("add") else {
            continue;
        };
        let key = add
            .get("path")
            .and_then(Json::as_str)
            .ok_or(format!("log line {}: an add with no path", n + 1))?;
        let partition = match add.get("partitionValues") {
            Some(Json::Obj(kv)) => kv
                .iter()
                .map(|(k, v)| (k.clone(), v.as_str().unwrap_or_default().to_string()))
                .collect(),
            _ => Vec::new(),
        };
        // The statistics are themselves JSON, stored as a string.
        let stats = add
            .get("stats")
            .and_then(Json::as_str)
            .and_then(|s| Json::parse(s).ok())
            .map(|s| file_stats(&s));

The file’s size comes from the add too. A line of the schema, metaData, gives the table’s columns, which the reader needs when every file is ruled out and there is nothing to read them from.

Ruling out a file#

A condition is tried against the file’s partition values first. A partition value is text, and compared as text:

def rules_out(file: TableFile, c: engine.Condition, use_stats: bool) -> str | None:
    """Why a condition rules a file out, if it does: from its partition values, then from the
    statistics a log records for it."""
    v = literal(c.value)
    p = next((value for k, value in file.partition if k == c.column), None)
    if p is not None:
        if c.op is Op.IS_NULL:
            ok = False
        elif c.op is Op.IS_NOT_NULL:
            ok = True
        else:
            ok = c.op.matches((p.encode() > c.value.encode()) - (p.encode() < c.value.encode()))
        return None if ok else f"its partition is {c.column}={p}"

fn rules_out(file: &TableFile, c: &engine::Condition, use_stats: bool) -> Option<String> {
    use std::cmp::Ordering::*;
    let v = literal(&c.value);
    if let Some((_, p)) = file.partition.iter().find(|(k, _)| k == &c.column) {
        let ok = match c.op {
            Op::IsNull => false,
            Op::IsNotNull => true,
            op => op.matches(Some(p.as_bytes().cmp(c.value.as_bytes()))),
        };
        return (!ok).then(|| format!("its partition is {}={p}", c.column));
    }

A column that is not a partition is tried against the statistics the log records for the file, with the comparison ch09 made against a footer’s statistics. There is one more rule: a JSON number and a JSON string prove nothing about each other, so a comparison across kinds rules nothing out.

    if not _comparable(v, low) or not _comparable(v, high):
        return None
    lo, hi = engine.order(v, low), engine.order(v, high)
    out = {
        Op.EQ: lo == -1 or hi == 1,
        Op.NOT_EQ: lo == 0 and hi == 0,
        Op.LT: lo != 1,
        Op.LT_EQ: lo == -1,
        Op.GT: hi != -1,
        Op.GT_EQ: hi == 1,
    }[c.op]
    return f"the log says {c.column} runs from {json_text(low)} to {json_text(high)}" if out else None

    if !comparable(&v, &min) || !comparable(&v, &max) {
        return None;
    }
    let (lo, hi) = (v.order(&min), v.order(&max));
    let out = match c.op {
        Op::Eq => lo == Less || hi == Greater,
        Op::NotEq => lo == Equal && hi == Equal,
        Op::Lt => lo != Greater,
        Op::LtEq => lo == Less,
        Op::Gt => hi != Less,
        Op::GtEq => hi == Greater,
        Op::IsNull | Op::IsNotNull => false,
    };
    let show = |x: &Value| x.to_json().to_json();
    out.then(|| {
        format!(
            "the log says {} runs from {} to {}",
            c.column,
            show(&min),
            show(&max)
        )
    })
}

Querying the files together#

The reader’s query engine, from ch12, reads the files that survive as one table. A column named in a partition, rather than in the files’ schema, takes its value from the path, the same value for every row of the file:

                    v = next((v for k, v in source.partition if k == col), None)
                    cells = [(None, v)] * rg.num_rows

                    Col::Partition(p) => {
                        let v = source
                            .partition
                            .iter()
                            .find(|(k, _)| k == p)
                            .map(|(_, v)| Value::Str(v.clone()))
                            .unwrap_or(Value::Null);
                        vec![(None, v); rg.num_rows as usize]
                    }
                };

Checking it#

The table’s files were written by pyarrow, and pyarrow answered each query over the whole table when the fixtures were generated. The tests hold the reader to those answers with every way of finding the files, and check that the log lists exactly the files a listing finds:

python3 -m pytest python/tests -k table

cargo test -p parquet-lab --test fixtures table

Ask a library#

At work a library finds the files. pyarrow’s dataset module reads a directory as one table: with Hive partitioning it reads country from the paths, and it skips _delta_log, as it skips every name that starts with an underscore. It does not read Delta logs. It rules out files by path, and then row groups by the statistics in each file’s footer, so it opens every file the path keeps. The parquet crate reads files, not tables: you walk the directories, split each name on =, and read each footer yourself:

# The same table to pyarrow: a dataset of the directory, with its partition read from the paths.
import pyarrow.dataset as ds

dataset = ds.dataset("fixtures/table", format="parquet", partitioning="hive")
print(dataset.schema.field("country"))  # from the paths: no file holds it
uk = ds.field("country") == "UK"
condition = uk & (ds.field("order_id") < 200)  # try 300

for fragment in dataset.get_fragments(filter=uk):  # the paths rule out the others
    s = fragment.row_groups[0].statistics["order_id"]  # from the file's footer
    kept = [rg.id for rg in fragment.subset(condition, schema=dataset.schema).row_groups]
    print(f"{fragment.path}: order_id {s['min']} to {s['max']}, row groups kept {kept}")
print("rows:", dataset.count_rows(filter=condition))

//! The same table to the `parquet` crate, which reads files, not tables: walk the directories,
//! take the partition from each path and the range of order_id from each file's footer.

use std::fs::{self, File};
use std::path::PathBuf;

use parquet::file::metadata::ParquetMetaDataReader;
use parquet::file::statistics::Statistics;

/// The paths in a directory, in order.
fn paths(dir: &str) -> Vec<PathBuf> {
    let entries = fs::read_dir(dir).expect("run this from the repository's root");
    let mut paths: Vec<PathBuf> = entries.map(|e| e.expect("an entry").path()).collect();
    paths.sort();
    paths
}

fn main() {
    for dir in paths("fixtures/table") {
        let name = dir.file_name().unwrap().to_string_lossy().into_owned();
        let Some((_, country)) = name.split_once('=') else {
            continue; // _delta_log, which the crate knows nothing about
        };
        for path in paths(&dir.to_string_lossy()) {
            let file = File::open(&path).expect("a file");
            let md = ParquetMetaDataReader::new()
                .parse_and_finish(&file)
                .expect("a Parquet file");
            let rows = md.file_metadata().num_rows();
            // One row group a file; order_id is its first column.
            let Some(Statistics::Int64(s)) = md.row_group(0).column(0).statistics() else {
                continue;
            };
            let (low, high) = (s.min_opt().unwrap(), s.max_opt().unwrap());
            // Could it hold country = 'UK' AND order_id < 200? Try 300.
            let kept = country == "UK" && *low < 200;
            let key = path.strip_prefix("fixtures/table").unwrap().display();
            println!("{key}: {rows} rows, order_id {low} to {high}, kept {kept}");
        }
    }
}

pyarrow’s schema has a country column that no file holds. Its filter on the path keeps the three UK files, and the condition on order_id keeps a row group only in the first, from its footer: the plan the log made, paid for with a read of each footer rather than one read of the log. The count is the reader’s. The crate prints every file’s range of order_id from its footer, and the ranges are the ones the log recorded. Change the limit to 300 in either, and the second UK file is kept too.

In Python, the first run loads pyarrow into the page, a much larger download than the other steps. In Rust the step needs the parquet crate, in walkthroughs/libraries, so it runs in a Codespace or at a desk:

cargo run -q --manifest-path walkthroughs/libraries/Cargo.toml --bin table_with_a_library

What this cannot tell you#

How a real log grows. This reader reads one log file. A real Delta log has one file per version, and a checkpoint, a Parquet file of its own, that summarises many versions so a reader need not replay them all. Apache Iceberg keeps the same information differently: a metadata file points to manifest lists, which point to manifests, which list the data files with their statistics. Both answer the same question this chapter asked, and ch16 describes Iceberg’s tree.

Typed partitions. Partition values in a path are text. This reader compares them as text, which is right for country and wrong for a numeric partition such as year compared with a number of a different length. A table format records each partition column’s type in its schema.

Concurrent writers. The log’s value is that two writers cannot both commit the same version. That needs the object store to refuse the second write, and this simulated store has one reader and no writers.

What a real table costs. Nine small files make the differences between the three ways easy to see. A table with millions of files makes LIST itself slow, since a LIST returns keys a page at a time, and makes the log the only practical way to plan a query at all.

Key takeaways#

Problems#

Three, in exercises/python/lakehouse_and_beyond.py, or in Rust in exercises/src/lakehouse_and_beyond.rs. The first two have tests. The third has none.

The editor below holds the Python stubs and keeps your edits in this browser. Run the tests runs the graders in the page, the same tests that run at a desk.

14.1 Partition values. Read the partition values in a Hive-style path, percent-encoding included. The test compares your answer with the reader’s for every file in the table and a set of awkward paths.

14.2 Plan from the log. Given a log and a range of order_id, return the files that could hold a row in it. The test compares your files with the reader’s decisions for many ranges, checks against the files themselves that no file with a matching row is dropped, and runs a log that removes a file and adds one with no statistics.

14.3 Your own table. No test: the data is yours. Pick a table you query, and the three conditions its queries use most. Decide which column to partition by, and in what order to write the rows within each partition. For each condition, say whether the path, the log’s statistics, or neither would rule files out. A good answer names one condition neither helps, and says what it would cost to change the layout so that one did.

For the Rust stubs, or to work at a desk, run the chapter’s graders with these commands. Add -k problem_14_1 in Python, or problem_14_1 before the -- in Rust, to run one problem:

python3 -m pytest exercises/python/tests/test_lakehouse_and_beyond.py --problems

cargo test -p exercises --test lakehouse_and_beyond -- --ignored

Where to go next#

• The Delta Lake protocol specifies the log, its actions and its checkpoints.
• The Apache Iceberg table specification describes manifests and how they carry statistics per file.
• Armbrust and others, Delta Lake: High-Performance ACID Table Storage over Cloud Object Stores, VLDB 2020, sets out why a log over object storage was needed.
• pyarrow’s dataset documentation covers partitioned writing and reading, the way the fixture was written.
• ch15 changes this kind of table: lookups, deletes and compaction.
• The appendices collect the book’s terms.

Built from commit c32f48e.