ch15Changing a table

What you build: A table read through its snapshots: lookups, position delete files applied in a scan, and a compaction planner.

The question#

What does it cost to find one row, or change one, in a table of Parquet files?

ch14 read a table that never changed. Real tables do. An order is cancelled, a refund corrects an amount, a customer asks to be forgotten, and late data arrives in small batches all day. A Parquet file cannot absorb any of that: its column chunks are encoded and compressed, and its footer records the offset of every one, so changing a value in the middle would move bytes the footer points at. Every change to a table of Parquet files is therefore a new file. This chapter measures what that costs, for the reads and the writes, and what the usual repair, compaction, costs in its turn. It starts with the operation Parquet was never built for: finding one row.

The experiment#

A table and its history#

fixtures/changes/ holds the orders from ch11, written by pyarrow in four files sorted by order_id, each with row groups, small pages and a page index. Then the table was changed, and every version was recorded as a snapshot: the list of data files, and of delete files, that make up the table at that moment. Table formats such as Apache Iceberg and Delta Lake keep exactly this list. Here it is one JSON file, _snapshots.json, beside the data:

copy-on-write deleted an order by writing its file again without it. The merge-on-read snapshots deleted orders by writing small position delete files instead, each naming one row of one data file. after-a-day adds new orders the way a stream of small batches does, one file per batch, and compacted is what a compaction job made of it.

Before the reader looks anything up, read the table’s metadata yourself: in Python with the json module, in Rust with the reader’s changes::read_snapshots, which parses the same file. The steps below run in the page in Python, and in a Codespace or at a desk in Rust. First, what the latest snapshot holds:

# What a snapshot says the table is: its data files and delete files, from its metadata.
import json

snapshots = json.load(open("fixtures/changes/_snapshots.json"))["snapshots"]
snapshot = next(s for s in snapshots if s["id"] == "after-a-day")

for f in snapshot["data_files"]:
    rows, size = f["record_count"], f["file_size"]
    print(f"{f['path']}: {rows} rows in {size} bytes, {size // rows} bytes a row")
data, deletes = snapshot["data_files"], snapshot["delete_files"]
print(len(data), "data files and", len(deletes), "delete files")

//! What a snapshot says the table is: its data files and delete files, from its metadata.

use parquet_lab::changes::read_snapshots;

fn main() {
    let text = std::fs::read_to_string("fixtures/changes/_snapshots.json")
        .expect("run this from the repository's root");
    let snapshots = read_snapshots(&text).expect("snapshots");
    let snapshot = snapshots.iter().find(|s| s.id == "after-a-day").unwrap();

    for f in &snapshot.data_files {
        let (path, rows, size) = (&f.path, f.record_count, f.file_size);
        println!(
            "{path}: {rows} rows in {size} bytes, {} bytes a row",
            size as i64 / rows
        );
    }
    let (data, deletes) = (snapshot.data_files.len(), snapshot.delete_files.len());
    println!("{data} data files and {deletes} delete files");
}

Compare the bytes a row costs in the original files with the bytes a row costs in the appended ones. Each small file carries its own footer, page headers and magic numbers for a handful of rows, so a row in a small file costs several times what it costs in a large one.

Next, the files a lookup of one order has to open. The snapshot records each data file’s range of order_id, as an Iceberg manifest does, so the reader can rule files out before opening them:

# Which files a lookup of one order must open: the data files whose order_id range
# holds it, and the delete files that name those.
import json

snapshots = json.load(open("fixtures/changes/_snapshots.json"))["snapshots"]
snapshot = next(s for s in snapshots if s["id"] == "after-a-day")
key = 300

data = [f["path"] for f in snapshot["data_files"]
        if f["order_id"][0] <= key <= f["order_id"][1]]
for path in data:
    print("order", key, "can only be in", path)
deletes = [d["path"] for d in snapshot["delete_files"] if d["data_file"] in data]
print("and these delete files may remove it:", ", ".join(deletes))

//! Which files a lookup of one order must open: the data files whose order_id range
//! holds it, and the delete files that name those.

use parquet_lab::changes::read_snapshots;

fn main() {
    let text = std::fs::read_to_string("fixtures/changes/_snapshots.json")
        .expect("run this from the repository's root");
    let snapshots = read_snapshots(&text).expect("snapshots");
    let snapshot = snapshots.iter().find(|s| s.id == "after-a-day").unwrap();
    let key = 300;

    let data: Vec<&str> = (snapshot.data_files.iter())
        .filter(|f| f.min_key <= key && key <= f.max_key)
        .map(|f| f.path.as_str())
        .collect();
    for path in &data {
        println!("order {key} can only be in {path}");
    }
    let deletes: Vec<&str> = (snapshot.delete_files.iter())
        .filter(|d| data.contains(&d.data_file.as_str()))
        .map(|d| d.path.as_str())
        .collect();
    println!(
        "and these delete files may remove it: {}",
        deletes.join(", ")
    );
}

One data file can hold order 300, because the files are sorted by order_id. But every delete file that names that data file might remove the order, so each of them must be read too.

Last, a delete file itself. It is an ordinary Parquet file with two columns, file_path and pos, as Iceberg defines them. The book’s reader, as you built it, reads it like any other file:

# A position delete file is a Parquet file of (file_path, pos): read it with the reader.
import sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import engine

deletes = open("fixtures/changes/deletes/delete-00.parquet", "rb").read()
for path, pos in engine.run(deletes, "SELECT file_path, pos FROM deletes").rows:
    data = open(f"fixtures/changes/{path}", "rb").read()
    order = engine.run(data, "SELECT order_id FROM orders").rows[pos][0]
    print(f"row {pos} of {path} is deleted: order {order}")

//! A position delete file is a Parquet file of (file_path, pos): read it with the reader.

use parquet_lab::engine::{run, Value};

fn main() {
    let deletes = std::fs::read("fixtures/changes/deletes/delete-00.parquet")
        .expect("run this from the repository's root");
    for row in run(&deletes, "SELECT file_path, pos FROM deletes")
        .unwrap()
        .rows
    {
        let [Value::Str(path), Value::Int(pos)] = &row[..] else {
            panic!("a delete file holds a path and a position");
        };
        let data = std::fs::read(format!("fixtures/changes/{path}")).unwrap();
        let orders = run(&data, "SELECT order_id FROM orders").unwrap().rows;
        let order = orders[*pos as usize][0].to_json().to_json();
        println!("row {pos} of {path} is deleted: order {order}");
    }
}

The delete file names a position, not an order: the row’s place in its data file, counting from zero. The data file still holds the order; the delete file says it is gone. Change the path to another delete file and run it again.

Find one order#

Now let the book’s reader look up order 300 in after-a-day, as a query engine would, through the simulated object store. The step prints every request, grouped by round trip: the requests in one round trip go out together, and each round trip waits for the one before it.

# The book's reader finds one order in a snapshot, and the store logs each request.
import sys
from itertools import groupby
from pathlib import Path
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab.changes import lookup
from parquet_lab.object_store import MemoryStore, NetworkModel

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

# Try order 250; then the snapshot "written"; then prefetch 65536.
snapshot, key, prefetch = "after-a-day", 300, 0
result = lookup(store, "", snapshot, key, prefetch, connections=4, model=NetworkModel())

for trip, (_, requests) in enumerate(groupby(result.requests, lambda r: r.phase), 1):
    print(f"round trip {trip}:")
    for r in requests:
        print(f"  {r.end_us // 1000:>3} ms: {r.method} {r.key} {r.range or '·'} {r.why}")
if result.found is None:
    print(f"no file holds order {key}")
else:
    path, pos = result.found
    gone = f", but {result.deleted_by} deletes it" if result.deleted_by else ""
    print(f"order {key} is row {pos} of {path}{gone}")
    for column, value in result.row:
        print(f"  {column}: {value}")
n, fetched, ms = len(result.requests), result.bytes_fetched, result.elapsed_us // 1000
print(f"{n} requests, {fetched} bytes, {ms} ms")

//! The book's reader finds one order in a snapshot, and the store logs each request.

use parquet_lab::changes::lookup;
use parquet_lab::object_store::{MemoryStore, NetworkModel};
use std::path::Path;

/// Every file under `dir`, under its path from `root`.
fn put_files(store: &mut MemoryStore, root: &Path, dir: &Path) {
    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() {
            put_files(store, root, &path);
        } else {
            let key = path.strip_prefix(root).unwrap().to_string_lossy();
            store.put(&key, std::fs::read(&path).expect("a file"));
        }
    }
}

fn main() {
    let mut store = MemoryStore::new(); // every file of the table, under its key
    let root = Path::new("fixtures/changes");
    put_files(&mut store, root, root);

    // Try order 250; then the snapshot "written"; then prefetch 65536.
    let (snapshot, key, prefetch) = ("after-a-day", 300, 0);
    let model = NetworkModel::default();
    let result = lookup(store, "", snapshot, key, prefetch, 4, model).expect("a lookup");

    let mut trips = 0;
    for (i, r) in result.requests.iter().enumerate() {
        if i == 0 || r.phase != result.requests[i - 1].phase {
            trips += 1;
            println!("round trip {trips}:");
        }
        let (ms, method, object) = (r.end_us / 1000, r.method.name(), &r.key);
        let range = r.range.as_deref().unwrap_or("·");
        println!("  {ms:>3} ms: {method} {object} {range} {}", r.why);
    }
    match (&result.found, &result.deleted_by) {
        (None, _) => println!("no file holds order {key}"),
        (Some((path, pos)), Some(by)) => {
            println!("order {key} is row {pos} of {path}, but {by} deletes it")
        }
        (Some((path, pos)), None) => println!("order {key} is row {pos} of {path}"),
    }
    for (column, value) in &result.row {
        println!("  {column}: {}", value.to_json());
    }
    let n = result.requests.len();
    let (fetched, ms) = (result.bytes_fetched, result.elapsed_us / 1000);
    println!("{n} requests, {fetched} bytes, {ms} ms");
}

Read the round trips from the top. The snapshots come first, since nothing else can be chosen without them. Then the delete files that name the one data file the order can be in, alongside that file’s trailer. Then its footer, which the trailer located; then the page indexes the footer located; then the pages the index chose in every column, to put the whole row back together. Each round trip needed a number the last one returned, so none could be sent sooner.

Change the order to 250. The reader finds it in its data file, decodes it, and only then finds that a delete file removes it: it read the data to learn that the answer is nothing. Then change the snapshot to written, which has no delete files, and set prefetch to 65536. The first read of the data file now asks for that much of its tail, which is the whole file: the chain shrinks to two round trips, and the bytes grow to the whole file, for one row.

The panel draws the same lookup’s requests over time, one lane per connection. Change the snapshot and the order. Whenever a file can hold the order, the chain has the same steps, and delete files widen only the second. Order 805 is past the end of written, so the snapshots alone answer; in after-a-day it is in a small append, and the whole chain is paid for a file of a few rows.

What one lookup costs#

The same lookups, and what a key-value store would pay for the same row:

A Parquet table has no index from a key to a row. Statistics narrow the search to a file, a row group and a page, and each narrowing is a round trip that needs the one before it. Then the pages are decoded, and to return the whole row the reader needs the matching page of every column, which is ch01’s cost of reconstructing a row, paid in requests.

This is the best case. The files are sorted by order_id, so one file and one page of each column hold the answer. On a table written in arrival order, every file’s range would span most keys, and a lookup would open every file. A key-value store answers the same question in one request of about one row’s bytes. That gap is the price of a layout built for scans, and no reader setting closes it.

Delete one order, two ways#

A table format can delete a row in two ways. Copy-on-write rewrites the row’s file without it. Merge-on-read writes a small delete file that names the row, and leaves the reader to take it out. Both first find the row, which is the lookup above:

Copy-on-write writes a whole file to change one row: its write amplification is the file’s size over the row’s. Nothing is left for readers to do afterwards, so every later scan costs what it did before. Merge-on-read writes far less, though still many times the row’s size, because the delete file is a Parquet file with its own footer. It moves the work to every reader: each later scan of that data file must fetch the delete file and apply it. That is read amplification, and it grows with every delete.

What deletes and small files do to every scan#

A scan of every row reads the snapshots, then every data file and every delete file at once, since none depends on another:

Each delete file is one more request for every scan. While the requests fit the connections, the extra requests overlap and cost little; once they do not, the scan takes another round of the store’s latency. The small appends of after-a-day are worse: each file is a request, and each brings its own footer, so the table carries many more bytes for a few more rows. And a position delete file removes a row only after the reader has decoded its page, so the deleted rows are decoded all the same.

Compaction#

Compaction repairs both. It reads the files that have deletes, and runs of small files, and writes them back as a few whole files with the deleted rows gone. The reader’s planner chooses the files for after-a-day:

# The book's reader plans a compaction: which files it would rewrite together.
import sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab.changes import plan_compaction, read_snapshots

snapshots = read_snapshots(open("fixtures/changes/_snapshots.json").read())
snapshot = next(s for s in snapshots if s.id == "after-a-day")

# Try target_rows=100; then small_rows=10; then the snapshot "merge-on-read-8".
plan = plan_compaction(snapshot, target_rows=200, small_rows=100)

for n, group in enumerate(plan, 1):
    print(f"group {n}: {group.rows_in} rows in, {group.rows_out} out, "
          f"{group.bytes_in} bytes to read")
    for path in group.data_files + group.delete_files:
        print(f"  {path}")
print(f"{len(plan)} files to write, from {sum(g.bytes_in for g in plan)} bytes read")

//! The book's reader plans a compaction: which files it would rewrite together.

use parquet_lab::changes::{plan_compaction, read_snapshots};

fn main() {
    let text = std::fs::read_to_string("fixtures/changes/_snapshots.json")
        .expect("run this from the repository's root");
    let snapshots = read_snapshots(&text).expect("snapshots");
    let snapshot = snapshots.iter().find(|s| s.id == "after-a-day").unwrap();

    // Try a target of 100 rows; then small files under 10 rows; then "merge-on-read-8".
    let (target_rows, small_rows) = (200, 100);
    let plan = plan_compaction(snapshot, target_rows, small_rows);

    for (n, group) in plan.iter().enumerate() {
        let (rows_in, rows_out, bytes) = (group.rows_in, group.rows_out, group.bytes_in);
        println!(
            "group {}: {rows_in} rows in, {rows_out} out, {bytes} bytes to read",
            n + 1
        );
        for path in group.data_files.iter().chain(&group.delete_files) {
            println!("  {path}");
        }
    }
    let read: u64 = plan.iter().map(|g| g.bytes_in).sum();
    println!("{} files to write, from {read} bytes read", plan.len());
}

Each data file with deletes is a group of its own, with its delete files, since two of them together would pass the target. The small appends go together into one group. Lower the target to 100 rows, and the appends split into two groups. Then set the small size to 10 rows instead: no append is small any more, and only the files with deletes are rewritten. On merge-on-read-8 the plan rewrites the files its delete files name and nothing else.

What carrying out the first plan costs, from the files pyarrow wrote for compacted:

The compaction is one large read and one large write, paid once. Every later scan is cheaper, and after enough scans the compaction has paid for itself. That arithmetic decides how often a table should be compacted: a table scanned all day repays it quickly, and one read once a week may never do so. The job also has to run somewhere, at some time, and it competes with the writers: if a writer deletes from data/part-1.parquet while compaction is rewriting it, one of the two commits must fail and retry.

Building it#

The reader’s changes module (python/parquet_lab/changes.py, or crates/parquet-lab/src/changes.rs) reads a table through its snapshots, which read_snapshots parses out of _snapshots.json. A snapshot is two lists, and the rows a data file still holds are its rows less the rows its delete files name:

class Snapshot:
    id: str
    summary: str
    data_files: list[DataFile]
    delete_files: list[DeleteFile]

    def deletes_for(self, data_file: str) -> list[DeleteFile]:
        """The delete files that name ``data_file``."""
        return [d for d in self.delete_files if d.data_file == data_file]

    def live_rows(self, f: DataFile) -> int:
        """The rows a data file still holds, by the snapshot's own counts: its records, less the
        positions its delete files name."""
        return f.record_count - sum(d.record_count for d in self.deletes_for(f.path))

pub struct Snapshot {
    pub id: String,
    pub summary: String,
    pub data_files: Vec<DataFile>,
    pub delete_files: Vec<DeleteFile>,
}

impl Snapshot {
    /// The delete files that name `data_file`.
    pub fn deletes_for(&self, data_file: &str) -> Vec<&DeleteFile> {
        self.delete_files
            .iter()
            .filter(|d| d.data_file == data_file)
            .collect()
    }

    /// The rows a data file still holds, by the snapshot's own counts: its records, less the
    /// positions its delete files name.
    pub fn live_rows(&self, f: &DataFile) -> i64 {
        f.record_count
            - self
                .deletes_for(&f.path)
                .iter()
                .map(|d| d.record_count)
                .sum::<i64>()
    }
}

Reading a delete file#

A position delete file is read with the reader’s query engine (ch12), since it is a Parquet file like any other:

def read_position_deletes(data: bytes) -> list[tuple[str, int]]:
    """The rows of a position delete file: which data file, and which position in it, counting
    from 0. Its columns are ``file_path`` and ``pos``, as Iceberg defines them."""
    out = []
    for row in engine.run(data, "SELECT file_path, pos FROM deletes").rows:
        match row:
            case [str(path), int(pos)] if not isinstance(pos, bool):
                out.append((path, pos))
            case _:
                raise ChangesError("a position delete file holds a path and a position")
    return out

pub fn read_position_deletes(bytes: &[u8]) -> Result<Vec<(String, i64)>, String> {
    let answer = engine::run(bytes, "SELECT file_path, pos FROM deletes")?;
    answer
        .rows
        .into_iter()
        .map(|row| match &row[..] {
            [Value::Str(path), Value::Int(pos)] => Ok((path.clone(), *pos)),
            _ => Err("a position delete file holds a path and a position".to_string()),
        })
        .collect()
}

Applying the deletes in a scan#

A scan fetches every delete file and every data file at once, since none depends on another. Then it decodes each data file’s rows in order, and skips each position a delete file names:

    deleted: set[tuple[str, int]] = set()
    for d in snapshot.delete_files:
        deleted.update(read_position_deletes(fetch(d.path, "read a delete file")))
    rows_decoded = live_rows = total = 0
    for f in snapshot.data_files:
        data = fetch(f.path, "read a data file")
        amounts = engine.run(data, "SELECT amount_cents FROM orders").rows
        for pos, row in enumerate(amounts):
            rows_decoded += 1
            if (f.path, pos) in deleted:
                continue  # a delete file names this row: it is no longer in the table
            live_rows += 1
            if isinstance(row[0], int):
                total += row[0]

    let mut deleted: Vec<(String, i64)> = Vec::new();
    for d in &snapshot.delete_files {
        let bytes = fetch(&d.path, "read a delete file")?;
        deleted.extend(read_position_deletes(&bytes)?);
    }
    let (mut rows_decoded, mut live_rows, mut sum) = (0, 0, 0);
    for f in &snapshot.data_files {
        let bytes = fetch(&f.path, "read a data file")?;
        let amounts = engine::run(&bytes, "SELECT amount_cents FROM orders")?.rows;
        for (pos, row) in amounts.iter().enumerate() {
            rows_decoded += 1;
            if deleted.contains(&(f.path.clone(), pos as i64)) {
                continue; // a delete file names this row: it is no longer in the table
            }
            live_rows += 1;
            if let Some(Value::Int(v)) = row.first() {
                sum += v;
            }
        }
    }

Finding one row#

A lookup reads the snapshots, then the delete files that name the files the key could be in, in the same round trip as the first read of the data file:

    deleted: list[tuple[str, int, str]] = []
    for d in deletes:
        why = "read a delete file that names a file the key may be in"
        got = s.get(f"{prefix}{d.path}", All(), why)
        deleted += [(path, pos, d.path) for path, pos in read_position_deletes(got.data)]

    let mut deleted: Vec<(String, i64, String)> = Vec::new();
    for d in &deletes {
        let got = s
            .get(
                &format!("{prefix}{}", d.path),
                GetRange::All,
                "read a delete file that names a file the key may be in",
            )
            .map_err(|e| e.to_string())?;
        for (path, pos) in read_position_deletes(&got.bytes)? {
            deleted.push((path, pos, d.path.clone()));
        }
    }

Then each candidate data file is read the way ch10 reads any file: its size known from the snapshot, statistics and the page index to find the page, and the matching pages of every column. scan_in is ch10’s scan, sharing the store with the requests before it, so the trace shows the whole chain. The row is found by its position, and a delete file naming that position removes it:

        strategy = Strategy(
            footer=FooterOptions(Known(f.file_size), prefetch),
            connections=connections,
            coalesce_gap=0,
            whole_chunks=False,
            mechanisms=Mechanisms(statistics=True, bloom=False, page_index=True),
        )
        r = scan_in(s, data_bytes, key_path, query, strategy)
        rows_decoded += r.rows_decoded
        if r.matches:
            pos = r.matches[0]
            found = (f.path, pos)
            deleted_by = next((by for p, n, by in deleted if p == f.path and n == pos), None)
            if deleted_by is None:
                row = list(zip(r.column_names, r.rows[0], strict=True))
            break

        let strategy = Strategy {
            footer: FooterOptions {
                size: SizeSource::Known(f.file_size),
                prefetch,
            },
            connections,
            coalesce_gap: Some(0),
            whole_chunks: false,
            mechanisms: Mechanisms {
                statistics: true,
                bloom: false,
                page_index: true,
            },
        };
        let r = scan::scan_in(&mut s, &bytes, &key_path, &query, strategy)?;
        rows_decoded += r.rows_decoded;
        if let Some(&pos) = r.matches.first() {
            found = Some((f.path.clone(), pos));
            deleted_by = deleted
                .iter()
                .find(|(p, n, _)| p == &f.path && *n as u64 == pos)
                .map(|(_, _, by)| by.clone());
            if deleted_by.is_none() {
                row = r
                    .column_names
                    .iter()
                    .cloned()
                    .zip(r.rows[0].iter().cloned())
                    .collect();
            }
            break;
        }

Which files a lookup opens, and how the compaction planner groups files, are the problems below; the reader’s own versions, which the steps ran, are in the same module.

Checking it#

The tests hold the reader to pyarrow: for every snapshot, the rows and the sum of amount_cents must be what pyarrow counts after applying the same deletes with its own compute functions. They look up orders across the whole range in five snapshots, and check that the planner rewrites exactly the files the compacted snapshot replaced:

python3 -m pytest python/tests -k "snapshot or lookup or compaction or position_delete"

cargo test -p parquet-lab --test fixtures -- snapshot lookup compaction position_delete

Ask a library#

pyarrow and the parquet crate read Parquet files, not tables. Neither knows that a delete file is anything but a file of paths and numbers, so applying one is yours to do, as the book’s reader does. Here both read data/part-1.parquet and every delete file, and take out the rows the delete files name:

# pyarrow reads Parquet files, not tables: the deletes are yours to apply.
from pathlib import Path
import pyarrow as pa, pyarrow.compute as pc, pyarrow.parquet as pq

path = "data/part-1.parquet"
rows = pq.read_table(f"fixtures/changes/{path}")
gone = []
for d in sorted(Path("fixtures/changes/deletes").glob("*.parquet")):
    named = pq.read_table(d).to_pylist()
    gone += [r["pos"] for r in named if r["file_path"] == path]

positions = pa.array(range(len(rows)))
live = rows.filter(pc.invert(pc.is_in(positions, value_set=pa.array(gone, pa.int64()))))
print(f"{path} holds {len(rows)} rows; {len(gone)} are deleted; {len(live)} are live")
print("their amounts sum to", pc.sum(live["amount_cents"]).as_py())

//! The `parquet` crate reads Parquet files, not tables: the deletes are yours to apply.

use parquet::file::reader::{FileReader, SerializedFileReader};
use parquet::record::{Row, RowAccessor};

fn rows(path: impl AsRef<std::path::Path>) -> Vec<Row> {
    let file = std::fs::File::open(path).expect("run this from the repository's root");
    let reader = SerializedFileReader::new(file).expect("a Parquet file");
    let rows = reader.get_row_iter(None).expect("rows");
    rows.map(|r| r.expect("a row")).collect()
}

fn main() {
    let path = "data/part-1.parquet";
    let data = rows(format!("fixtures/changes/{path}"));
    let mut gone = Vec::new();
    for d in std::fs::read_dir("fixtures/changes/deletes").unwrap() {
        for named in rows(d.unwrap().path()) {
            if named.get_string(0).unwrap() == path {
                gone.push(named.get_long(1).unwrap() as usize);
            }
        }
    }

    let live: Vec<&Row> = (data.iter().enumerate())
        .filter_map(|(i, r)| (!gone.contains(&i)).then_some(r))
        .collect();
    let (n, deleted) = (data.len(), gone.len());
    println!(
        "{path} holds {n} rows; {deleted} are deleted; {} are live",
        live.len()
    );
    let sum: i64 = live.iter().map(|r| r.get_long(5).unwrap()).sum();
    println!("their amounts sum to {sum}");
}

The live rows and their sum are the ones the book’s reader reports for that file. A table format’s own libraries apply delete files for you when they read a snapshot; the work they do is this. Change the path to data/part-2.parquet, and the other delete files apply instead.

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 deletes_with_a_library

What this cannot tell you#

How deep real metadata is. This table’s metadata is one file. Iceberg’s is a chain: a metadata file points to a manifest list, which points to manifests, which list the files. Each link is another round trip before the first data request, unless a reader caches it.

Other ways to delete. Iceberg also has equality deletes, which name values instead of positions (order_id = 250), and are cheaper to write and dearer to read. Newer formats use deletion vectors: a bitmap of deleted positions per data file, which is smaller and faster to apply than a Parquet file of positions. The costs move, and the shape stays: every form of merge-on-read is work that each reader repeats until compaction.

Concurrent writers. One writer changed this table. Two writers, or a writer and a compaction, commit through the table format’s optimistic concurrency: each writes its files, then tries to make its snapshot current, and one fails if the other got there first. A compaction that loses that race has done all its work for nothing.

What old snapshots cost. Every snapshot here stays readable, so the files of written are still stored after compacted replaced them. Real tables expire old snapshots and delete the files no snapshot needs, which ends time travel to those versions.

What a write costs. The store prices a write like a read of the same size. A real rewrite also decodes and re-encodes every row, and a table’s compaction usually runs on a cluster whose cost is the real bill.

Key takeaways#

Problems#

Three, in exercises/python/changing_a_table.py, or in Rust in exercises/src/changing_a_table.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.

15.1 The files a lookup opens. Given a snapshot and an order_id, return the data files whose range holds it and the delete files that name them. The test compares your answer with the reader’s for every order in every snapshot, and for made-up snapshots whose ranges overlap.

15.2 Plan a compaction. Choose the files to rewrite, those with deletes and those too small, and pack them into groups of at most a target number of rows, in order of order_id. The rules are in the stub. The test compares your plan with the reader’s across many targets and snapshots, and checks that for after-a-day it rewrites exactly the files compacted replaced.

15.3 Your own table. No test: the table is yours. Take a table you write to every day. Count its files, how many are small, and, if it uses merge-on-read, how many delete files it holds. Estimate how many times a day it is scanned, and what one compaction of it would read and write. A good answer says whether copy-on-write or merge-on-read suits the table’s mix of writes and reads, how often to compact it, and how many scans a compaction takes to pay for itself. If most of the table’s reads are lookups of single rows, it should say which store would serve those better.

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

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

cargo test -p exercises --test changing_a_table -- --ignored

Where to go next#

• The Apache Iceberg table specification defines snapshots, manifests, position and equality delete files, and deletion vectors.
• The Delta Lake protocol describes its deletion vectors and how a writer commits against others.
• O’Neil and others, The Log-Structured Merge-Tree, Acta Informatica, 1996, is where writing changes aside and merging them later began; compaction is its merge.
• pyarrow’s compute functions are what the library step used to take out deleted rows.
• ch16 maps what Apache Iceberg adds above Parquet files to the chapters that show it, and reads the field ids by which a table names its columns.

Built from commit c32f48e.