--- title: "Open Table Formats" book: open-table-formats edition: 2026.11.1 verified: 2026-11-06 edition_code: 7TCH-8M62 canonical: https://rollingedition.com/open-table-formats license: CC BY 4.0 attribution: "Open Table Formats, Edition 2026.11.1, examples passed Nov 6, 2026. By Rolling Edition, https://rollingedition.com/open-table-formats. Text CC BY 4.0 (https://creativecommons.org/licenses/by/4.0/). Code MIT, copyright Rolling Edition and contributors. Written with AI. Verified by people." --- Edition 2026.11.1. Examples passed Nov 6, 2026. 39 of 39 examples pass. Written with AI. Verified by people. To cite a section, name Edition 2026.11.1 and link to the section's permanent link: https://rollingedition.com/open-table-formats/s/{section_id} # Build a first table, end to end By the end of this chapter you can commit batches of rows to a table, read the table back in DuckDB, and check the totals. This book is for data and analytics engineers who write SQL and Python and have to decide how their lakehouse tables are stored, changed, cleaned up, shared and moved between formats. The book builds one result: a pageviews table that you store, change, read as of an earlier snapshot, clean up, register in a catalog and move between formats. This chapter builds a rough version of it, one table written in three table formats (Apache Iceberg, Delta Lake and DuckLake) and read back into DuckDB, so you can compare the totals. Each later chapter makes one step of the result solid: - Chapter 2 reads the table as it was at an earlier snapshot. - Chapter 3 adds and renames columns without rewriting files. - Chapter 4 changes the partitioning without rewriting old files. - Chapter 5 corrects rows with deletes, updates and merges. - Chapter 6 compacts the small files that hourly appends leave. - Chapter 7 expires snapshots and removes files that no snapshot reaches. - Chapter 8 registers tables in a catalog and queries them from DuckDB. - Chapter 9 moves a table between formats and checks the copy. - Chapter 10 chooses a format from what your tools support. You know SQL, Python and Parquet, and you know what a partition is. This chapter does not explain them. The book follows one case throughout: an hourly pageviews table, which we call the pageviews table. Each row says how many times one public Wikipedia article was viewed in one hour, in one language edition. The rows cover six hours of 3 June 2024, from 00:00 to 05:59 UTC. They come from the [Wikimedia pageviews dumps](https://dumps.wikimedia.org/other/pageviews/2024/2024-06/), which the Wikimedia project releases under CC0. The book keeps the 500 most-viewed titles in each hour, across the English, German, French and Japanese editions, after special pages are removed. The day is fixed and in the past, so the dumps for it will not change, and the data in the repository will not go stale. [The data folder's README](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/data/README.md) gives the source, the licence and the exact cut. The data is messy on purpose. Titles carry underscores and non-Latin scripts, and the counts are skewed: the top title in hour 00 has far more views than the title at position 500. Every row in the cut has a `response_bytes` of 0, as the [data folder's README](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/data/README.md) records. The book keeps the column as the cut gives it, so it is 0 throughout. These values stay in the data, because a table format has to store them as they are. ## What a table format adds to files {#what-a-table-format-adds} A folder of Parquet files can hold a table, but the folder does not say which files belong to the table. Anything that reads the folder sees every file in it, including a file a writer left half-finished. A table format fixes this with metadata: a record of the table's schema and of the data files that make up the table at each point in its history. A change to the table is a commit, which writes new metadata in a single step. That step is the one change that moves the table from one state to the next. Each table format has its own metadata and its own words for these ideas. Iceberg calls the state after a commit a snapshot. Delta Lake numbers its versions. DuckLake records its snapshots in a [catalog file](#create-catalog-and-table). This chapter writes all three, so you can see the same rows in each. Where the metadata sits differs too. An Iceberg table keeps a metadata file that names its current snapshot. In format version 2, which pyiceberg writes, each snapshot points to a manifest list, the manifest list names manifests, and each manifest names data files along with their statistics. The [Iceberg spec](https://iceberg.apache.org/spec/) defines each of those layers. A Delta Lake table keeps an ordered log of commit files in a `_delta_log` folder beside its data, as the [Delta Lake protocol](https://raw.githubusercontent.com/delta-io/delta/v4.4.1/PROTOCOL.md) describes. A DuckLake table keeps its metadata in the catalog file that you attach, and writes its data files into a folder beside that file. The example below loads six hours into Iceberg, Delta Lake and DuckLake, one hour at a time, then reads each table back and compares the totals. Example `rough-lakehouse` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Load six hours into Iceberg, Delta Lake and DuckLake, then read each table back.""" from lib.lakehouse import ( format_report, load_three_formats, new_warehouse, print_rows, safe_error, sql_rows, scan_arrow, ) TOP_THREE = """ SELECT sum(views) AS views, page_title FROM pageviews GROUP BY page_title ORDER BY views DESC, page_title LIMIT 3 """ def main(): try: with new_warehouse() as wh: iceberg, delta_path, lake = load_three_formats(wh, range(6)) report = format_report(wh, iceberg, delta_path, lake) for name, (rows, views, files) in report.items(): print(f"{name} rows: {rows}") print(f"{name} views: {views}") print(f"{name} data files: {files}") totals = {(rows, views) for rows, views, _ in report.values()} print(f"totals match: {len(totals) == 1}") print("top three page titles over six hours (views, page_title):") print_rows(sql_rows(wh.con, TOP_THREE, keep_order=True, pageviews=scan_arrow(iceberg)), keep_order=True) except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text iceberg rows: 2999 iceberg views: 1807299 iceberg data files: 6 delta rows: 2999 delta views: 1807299 delta data files: 6 ducklake rows: 2999 ducklake views: 1807299 ducklake data files: 6 totals match: True top three page titles over six hours (views, page_title): 399885 Main_Page 102089 メインページ 54796 Wikipedia:Hauptseite ``` Each format reports the same row count and the same view total. Each format also wrote one data file per hour: - Iceberg: 2999 rows, 1807299 views, 6 data files. - Delta Lake: 2999 rows, 1807299 views, 6 data files. - DuckLake: 2999 rows, 1807299 views, 6 data files. The last check line, `totals match: True`, says the three totals are equal. The example also prints the three most-viewed titles over the six hours, read from the Iceberg table alone: `Main_Page`, `メインページ` and `Wikipedia:Hauptseite`, the main pages of the English, Japanese and German editions. The example prints no snapshot IDs, version numbers or file names, since they change from run to run. [The section on appending each hour](#append-each-hour) explains what it prints instead. DuckLake needs one extra step on your own machine. Run `INSTALL ducklake` and then `LOAD ducklake` in DuckDB. The book's runs instead load a copy of DuckDB's official, signed `ducklake` extension from a file in the repository, so they run without a network connection. DuckLake also has an inlining default that matters for the file counts: by default, it keeps inserts of 10 rows or fewer in its catalog rather than in a Parquet file. The [DuckLake data inlining page](https://ducklake.select/docs/stable/duckdb/advanced_features/data_inlining) states the default: "Data inlining is enabled by default with a row limit of 10." The example sets the limit to 0, so every insert writes a file, and the file counts mean the same thing in every chapter. ### One load loop, three commit styles {#one-load-loop} The helper `load_three_formats()` in `examples/lib/lakehouse.py` holds the loop that [rough-lakehouse](#what-a-table-format-adds) calls. The loop writes each hour to all three tables, and each line that writes is one commit in one format: ```python for hour in hours: rows = read_cut_hour(hour) iceberg.append(rows) write_deltalake(delta_path, rows, mode="append") lake.register("hour", rows) lake.execute(f"INSERT INTO {DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE} SELECT * FROM hour") lake.unregister("hour") ``` `read_cut_hour` returns one hour of the cut as an Arrow table, with the same five columns in every format. The Iceberg append creates a snapshot. The Delta append creates a new version, and it creates the table on the first call, when the path does not yet hold one. The DuckLake insert writes a Parquet file and records a snapshot in the catalog file. The three calls differ, and the table each one writes is the same six hours of rows. The loop itself checks nothing. It runs the six loads in order and stops if one of them raises an error. The checks come afterwards, in the example, where each format's totals are read back and compared. ### A commit names the table's files {#files-metadata-and-commit} The [Apache Iceberg table spec](https://iceberg.apache.org/spec/) defines a snapshot in one sentence: "A snapshot represents the state of a table at some time and is used to access the complete set of data files in the table." A snapshot therefore does two things. It names the data files that the table holds at that point, and it records the operation that made it. The spec lists the kinds of operation. For an append, the operation is `append`, which the spec describes as "Only data files were added and no files were removed." The pageviews table makes this concrete. Each hour you load becomes a data file, and each load is one commit. After the first commit the table holds hour 00's file and nothing else. After the second commit it holds two files, and so on. A data file is never edited. A commit adds files, and the new snapshot names the files the table holds after the commit. The table's count of data files comes from its snapshot, not from its folder. [rough-lakehouse](#what-a-table-format-adds) prints the number of data files each format reports for the table. That number is what the table says it holds, and a listing of the folder can show a different number. **Working rule:** Ask the table what it holds, through its current snapshot, and not by listing its folder. The folder holds every file a writer put there, and the snapshot says which of those files the table reads. ## Why each format is taught through a Python library {#why-python-libraries} The [rough-lakehouse](#what-a-table-format-adds) example wrote each format through a different library, and this section says why. The book writes Apache Iceberg and Delta Lake with the Python libraries that a reader will use to run them: pyiceberg, at version 0.12.0, and deltalake, at version 1.6.6. DuckDB, at version 2.0.0, reads what those libraries write. It reads an Iceberg table as an Arrow table from a scan, and it reads a Delta table as an Arrow table from a query that deltalake runs. DuckLake is the exception. DuckDB runs DuckLake itself, through its `ducklake` extension, and attaches the table's catalog file. The book does not use DuckDB's `iceberg` or `delta` extensions, and no example runs them. | Format | Written by | Read by DuckDB through | | --- | --- | --- | | Apache Iceberg | pyiceberg, 0.12.0 | Arrow table from a scan | | Delta Lake | deltalake, 1.6.6 | Arrow table from deltalake's `QueryBuilder` | | DuckLake | DuckDB's `ducklake` extension, on DuckDB 2.0.0 | `ATTACH` to a catalog file | The choice has a cost. A reader who runs Iceberg through Spark or Trino meets more of the format there, with features this book does not cover. The book names those places and does not teach those engines. The gain is fewer moving parts: each format's own library writes it, and DuckDB reads the output. Each library's version is pinned, and the chapters name the pinned library version, not the version of the format's specification. The pins live in `examples/requirements.txt`, which lists every package with one exact version per line. A reader who installs from that file gets the libraries this chapter ran against. Use the Python version that `examples/.python-version` names, because pip does not select one. **Working rule:** Write and read each table with the library whose version you pin, and say which version your reads use. Reach for an engine's own extension only when you have measured what the library cannot do for you. ## Land three hours in a table {#land-three-hours} The [rough-lakehouse](#what-a-table-format-adds) example showed the whole loop, so this section slows it down. Landing a batch takes two steps: create the table, then commit each batch to it. The pageviews table lands with hours 00, 01 and 02 first, so each hour is one commit. That example loads hours 03 to 05 the same way, after the first three. The examples in the rest of this section keep to the Iceberg table, because it has the most to show. ### Create the table in a catalog {#create-catalog-and-table} An Iceberg table needs a catalog, which is the registry that maps a table name to the table's current metadata. A Delta Lake table is found by its path, with no catalog, and a DuckLake table lives in its catalog file. The book uses pyiceberg's SQL catalog on SQLite for the Iceberg table. The catalog is one file, and the table's data files go in a warehouse folder beside it. The catalog holds pointers and metadata, not rows. Inside the catalog file, a table name such as `wiki.pageviews_hourly` maps to the location of the table's current metadata file. A commit writes a new metadata file and moves that mapping, so anyone who asks the catalog for the table gets the metadata that the most recent commit wrote. The helper `new_warehouse()` in `examples/lib/lakehouse.py` makes a temporary catalog and warehouse for each run, and removes them when the run ends. Because SQLite is a file and needs no server, the whole catalog runs on a laptop with the network off. The table has five columns, and its schema says what each one holds: ```python Schema( NestedField(1, "hour_utc", TimestampType(), required=False), NestedField(2, "domain_code", StringType(), required=False), NestedField(3, "page_title", StringType(), required=False), NestedField(4, "views", LongType(), required=False), NestedField(5, "response_bytes", LongType(), required=False), ) ``` The `hour_utc` column holds a timestamp without a time zone, in UTC, and the two counts are 64-bit integers. The helper `create_iceberg_table()` creates the table in the catalog, unpartitioned. Chapter 4 adds a partition spec to it. **Working rule:** Write the schema out in code, with the type each column really has, before the first commit. The types are then decided on purpose, and not by whichever batch happens to arrive first. ### Append each hour as a commit {#append-each-hour} Each append is one commit. The example below creates the table, appends hours 00, 01 and 02 in order, and then lists the snapshots in commit order with their operation and the rows each commit added. Example `append-hours` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Create the Iceberg table, append hours 00 to 02 as three commits, then repeat hour 00.""" from lib.lakehouse import ( create_iceberg_table, iceberg_file_count, new_warehouse, read_cut_hour, safe_error, scan_arrow, snapshot_labels, ) def main(): try: with new_warehouse() as wh: table = create_iceberg_table(wh) for hour in range(3): table.append(read_cut_hour(hour)) snapshots = table.snapshots() labels = snapshot_labels([snap.snapshot_id for snap in snapshots]) for snap in snapshots: label = labels[snap.snapshot_id] print(f"{label} operation: {snap.summary.operation.value}") print(f"{label} rows added: {snap.summary['added-records']}") print(f"rows in table: {scan_arrow(table).num_rows}") print(f"data files: {iceberg_file_count(table)}") table.append(read_cut_hour(0)) repeat = table.current_snapshot() print(f"repeat of hour 00 operation: {repeat.summary.operation.value}") print(f"repeat of hour 00 rows added: {repeat.summary['added-records']}") print(f"rows in table after the repeat: {scan_arrow(table).num_rows}") except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text snapshot 1 operation: append snapshot 1 rows added: 499 snapshot 2 operation: append snapshot 2 rows added: 500 snapshot 3 operation: append snapshot 3 rows added: 500 rows in table: 1499 data files: 3 repeat of hour 00 operation: append repeat of hour 00 rows added: 499 rows in table after the repeat: 1998 ``` The first hour has 499 rows. The second and third have 500 each, so hour 00 is one row short. The data's privacy check removed one row from hour 00, as the data README explains. The three commits add 1499 rows in total, across 3 data files, one for each commit. Each snapshot's operation is `append`, the append operation the spec defines. Each commit also writes a summary, and pyiceberg returns the snapshots in commit order, which is how the example reads the operation and the rows added. The summary is what the table records about its own commits, so it is the first place to look when a count is wrong. The example prints `snapshot 1`, `snapshot 2` and `snapshot 3` in place of the snapshot IDs. An Iceberg snapshot ID is a long integer that the spec requires to be unique, and pyiceberg chooses it when the commit is made, so it changes from run to run. The example maps each ID to its position in commit order, so the printed output is the same on every run. **Working rule:** Make one commit per batch you can name, and compare the rows each commit added with the count you expect from its source before you make the next commit. A wrong count is simpler to trace while its commit is the most recent one, and harder once more commits sit on top of it. ## Query the table with DuckDB {#query-with-duckdb} With the table in place, you want to ask it questions. DuckDB runs SQL, but an Iceberg table is not a DuckDB table. The book reads the table's current snapshot as an Arrow table, which pyiceberg returns from a scan, and registers that Arrow table with DuckDB under the name `pageviews`. DuckDB then runs ordinary SQL over it, with no Iceberg extension. The scan is the table's own reading of the snapshot. It lists the data files that the snapshot names, reads them, and hands over only the rows they hold. DuckDB sees exactly those rows and nothing that the snapshot does not name. The example below appends the three hours and asks two questions: the five most-viewed titles in hour 00, and the total views across the three hours. It checks the total against the sum that the scan itself gives. Example `scan-into-duckdb` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Read the current Iceberg snapshot into DuckDB and check the totals.""" from lib.lakehouse import ( create_iceberg_table, new_warehouse, print_rows, read_cut_hour, safe_error, scan_arrow, sql_rows, ) def main(): try: with new_warehouse() as wh: table = create_iceberg_table(wh) for hour in range(3): table.append(read_cut_hour(hour)) arrow = scan_arrow(table) top = sql_rows( wh.con, """ SELECT views, page_title FROM pageviews WHERE hour_utc = TIMESTAMP '2024-06-03 00:00:00' ORDER BY views DESC, page_title LIMIT 5 """, keep_order=True, pageviews=arrow, ) print("top five pages for hour 00 (views, page_title):") print_rows(top, keep_order=True) duckdb_total = sql_rows(wh.con, "SELECT sum(views) FROM pageviews", pageviews=arrow)[0][0] pyiceberg_total = sum(arrow["views"].to_pylist()) print(f"total views in DuckDB: {duckdb_total}") print(f"total views in the scan: {pyiceberg_total}") print(f"totals match: {duckdb_total == pyiceberg_total}") except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text top five pages for hour 00 (views, page_title): 61107 Main_Page 13791 メインページ 9492 Wikipedia:Hauptseite 7511 Wikipédia:Accueil_principal 3781 笹生優花 total views in DuckDB: 899451 total views in the scan: 899451 totals match: True ``` The list holds the five titles with the most views in hour 00, and the most-viewed title comes first. The query orders the rows by views, largest first, with the title as the tie-breaker, and the example prints them in that order. Each line shows the views first, then the title, separated by a tab. The total is 899451 views in DuckDB and 899451 views in the scan, and `totals match: True`. A total that matches the scan's sum shows that DuckDB received every row the scan returned. It does not check the rows themselves, and it does not check the top-five query. A check on a total costs one line, so it is worth running after every change to a query. Checking the rows against their source is the job of the later chapters, which compare each change against the data they came from. **Working rule:** Read an Iceberg table into DuckDB as the Arrow table from a scan, and compare a total with the scan's own sum, to confirm that DuckDB received every row. Then check each query's result against a source you can compare it with. ## What goes wrong in production {#first-table-production-failures} The first chapter's path is short, and it hides four problems that you will meet in production. **A repeated load adds the same rows again.** An append adds the data files it is given and removes none, as the spec's definition of the operation says. Nothing in the table keeps a second copy of hour 00 out. A commit that carries a second copy of hour 00 looks like any other commit. The [append-hours](#append-each-hour) example shows it: the repeated commit's operation is `append`, it adds 499 rows, the same count as the first copy, and the table then holds 1998 rows, with no error. The only record that the load was meant to run once is the count you wrote down before it ran. Chapter 5 shows how to correct rows that were loaded wrongly. **The folder holds more than the table.** A count of the files in the warehouse folder answers a different question from the count of the table's files. The folder holds every file a writer put there, and a snapshot defines what the table holds. A tool that counts Parquet files in the folder reports a number that the table never promised. Chapters 6 and 7 deal with files that the table no longer reaches. **Snapshot IDs, versions and snapshot numbers are different kinds of number.** Iceberg IDs are 64-bit values. Delta Lake versions count from 0 in commit order. DuckLake numbers its snapshots as well, and it records a snapshot each time it creates a schema or a table, as well as each time it inserts rows. A script that stores one format's number and reads it as another format's version reads the wrong state. **A small DuckLake insert writes no file.** By default, DuckLake keeps inserts of 10 rows or fewer in its catalog, so a file count of zero after a small insert is the expected result, not a failure. The examples in this book set the row limit to 0. A reader who copies an insert into their own catalog should know the default first. **Working rule:** Write down the row count each commit should add, next to the load that makes it, and compare that count with the commit's summary after every load. The table does not check for a repeated load. A count you wrote down in advance is the check that does. ## Decisions this chapter made {#first-table-decisions} - **Three examples in this chapter.** `rough-lakehouse` does the whole result in three formats, so each later chapter changes one part of a result the reader has already run. The other two, `append-hours` and `scan-into-duckdb`, build the Iceberg table one commit at a time and read it in DuckDB. - **Hours 00 to 02 first, then 03 to 05.** The Iceberg examples land three hours and the rough example lands six. The totals in this chapter cover all six hours. - **Unpartitioned tables.** Chapter 1 creates tables without a partition spec. Chapter 4 changes the partitioning. The pyiceberg-core package is pinned beside pyiceberg, because the day and hour transforms in chapter 4 need it. - **Inlining off in the examples.** DuckLake examples set `DATA_INLINING_ROW_LIMIT` to 0, so each insert writes a Parquet file and the file counts mean the same thing wherever the book prints them. - **Delta tables are read through deltalake's `QueryBuilder` only.** The reading helper in `examples/lib/lakehouse.py` takes that path, and each later chapter that reads a Delta table uses the same helper. - **Real data, not a stand-in.** The examples read the cut of the six public dumps that [the data README](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/data/README.md) describes. The book ships no generated data for this chapter. - **SQLite for the Iceberg catalog.** A SQLite file needs no server, so the Iceberg examples run on a laptop with the network off. The examples also turn off DuckDB's automatic extension download, so a run that needs the network fails instead of fetching. - **Labels stand in for IDs.** The examples print `snapshot 1`, `snapshot 2` and so on in place of snapshot IDs, so the printed output does not change from run to run. # Read a table as it was at an earlier snapshot By the end of this chapter you can read a table as it was after an earlier commit, in each of the three formats. At the end you can list a table's commits, read one of them by the number its format uses, and compare the numbers across formats without mixing them up. This chapter follows the book's running case, the hourly pageviews table `wiki.pageviews_hourly`. A dashboard total for the table changed after the third hour landed. The owner asks two questions: what the total was before that commit, and whether the table changed or only the query did. Answering them needs the table's history, and the ability to read an old state of the table. The table has five columns: `hour_utc`, `domain_code`, `page_title`, `views` and `response_bytes`. The dashboard sums `views`. The examples in this chapter build their own copy of the table the way [chapter 1](#land-three-hours) does: they create it and append hours 00, 01 and 02 as three commits. Hour 00 has 499 rows, and each of the other two hours has 500. ## Every commit is a snapshot {#every-commit-is-a-snapshot} [Chapter 1](#append-each-hour) lands the three hours as three commits. Each commit leaves a snapshot behind, and that is what you read later. A commit changes the table's metadata in one atomic step, so a reader sees the table either before the commit or after it, never in between. The [Iceberg spec](https://iceberg.apache.org/spec/) describes each change as "an atomic swap" of metadata files. In Apache Iceberg, each commit creates a snapshot, which the [Iceberg table spec](https://iceberg.apache.org/spec/) defines as "the state of a table at some time," used "to access the complete set of data files in the table." A snapshot has a unique ID, a creation time, and a summary. The summary has a required field, `operation`, which is one of `append`, `replace`, `overwrite` or `delete`. An `append` added data files and removed none. A later commit adds a snapshot and leaves the earlier ones alone, which is what lets the dashboard owner ask about a state that is no longer the newest. The example below lists the three snapshots in commit order, with each one's operation and rows added. Example `snapshot-history` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """List each commit to one table: Iceberg snapshots and Delta Lake versions.""" from deltalake import DeltaTable, write_deltalake from lib.lakehouse import ICEBERG_ID, PAGEVIEWS_SCHEMA, new_warehouse, read_cut_hour, snapshot_labels def main(): with new_warehouse() as wh: table = wh.catalog.create_table(ICEBERG_ID, schema=PAGEVIEWS_SCHEMA) delta_path = str(wh.root / "delta" / "pageviews_hourly") for hour in range(3): batch = read_cut_hour(hour) table.append(batch) write_deltalake(delta_path, batch, mode="append") snapshots = sorted(table.snapshots(), key=lambda snapshot: snapshot.sequence_number) labels = snapshot_labels([snapshot.snapshot_id for snapshot in snapshots]) for snapshot in snapshots: label = labels[snapshot.snapshot_id] operation = snapshot.summary.operation.value print(f"{label} operation: {operation}") print(f"{label} rows added: {snapshot.summary['added-records']}") history = sorted(DeltaTable(delta_path).history(), key=lambda entry: entry["version"]) for entry in history: rows = entry["operationMetrics"]["num_added_rows"] print(f"version {entry['version']} operation: {entry['operation']}") print(f"version {entry['version']} rows added: {rows}") if __name__ == "__main__": main() ``` Output: ```text snapshot 1 operation: append snapshot 1 rows added: 499 snapshot 2 operation: append snapshot 2 rows added: 500 snapshot 3 operation: append snapshot 3 rows added: 500 version 0 operation: WRITE version 0 rows added: 499 version 1 operation: WRITE version 1 rows added: 500 version 2 operation: WRITE version 2 rows added: 500 ``` The first commit added 499 rows, which is hour 00. Hour 00 has one row fewer than the others because the data cut dropped one row from it, as the [data README](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/data/README.md) explains. The second and third each added 500 rows. All three commits used `append`, so none of them rewrote or removed a file. The dashboard owner's question starts here: the table has three states to choose from, and the history names them. pyiceberg gives each snapshot a long integer ID. We do not print those IDs. The example maps them, in commit order, to the labels "snapshot 1", "snapshot 2" and "snapshot 3" with `snapshot_labels`, and prints only the labels. The same three hours went to a Delta Lake table through `write_deltalake`, one write per hour. Delta Lake calls each commit a version, and its history lists the operation and the rows written for each one. Delta numbers its versions from 0, so the first write is version 0. The book prints Delta's own numbers, so the example sorts the history by version and prints "version 0", "version 1" and "version 2". Each Delta write is a `WRITE` operation that adds the same rows as the matching Iceberg snapshot: 499 rows for version 0, then 500 and 500. The history answers the first half of the owner's question. It says which commit changed the table and how many rows that commit added. An `append` only adds rows, so it cannot change a total for hours that are already in the table. If the total for an hour moved, the history must name an `overwrite`, a `delete` or a `replace` as the cause, and you then read the snapshots on either side of it. The history does not say what a query did, and that part of the question needs the reads in the next section. **Working rule:** Before you read or change a table, list its history in the format you use, and use that list as the only source of commit numbers. Sort the list by commit order, label Iceberg snapshots with `snapshot_labels` before you print them, and print Delta versions as Delta numbers them. Read the operation with each commit. An `append` only adds files, so it cannot have removed rows, while an `overwrite` or `delete` can. ### What a snapshot points to {#what-a-snapshot-points-to} A snapshot holds no rows. It points to the files that hold them. In format versions 2 and 3, a snapshot records a manifest list, and that is where its files are found. The spec's definition says the snapshot is the entry point to the complete set of data files in the table, so a reader starts from the snapshot and follows it to the files. That makes a snapshot a set of pointers, and the consequence matters for every section that follows. A snapshot reads only while the files it names still exist. Time travel is therefore not a backup. Time travel reads files that the table still keeps, and the table keeps them while some snapshot names them. [Chapter 7](#snapshots-keep-files-alive) removes the old snapshots and the files only they named, which ends the time travel to those states. The commit's operation shows what sharing is possible. The spec defines `append` as the operation where "only data files were added and no files were removed." So snapshot 3 names the files of hour 00 again, because the third commit did not rewrite them. Several snapshots can name the same file, and a commit costs only the files it wrote plus the pointers to them. The row counts in [Snapshots in pyiceberg](#snapshots-in-pyiceberg) show the pointers at work. Snapshot 2 reads back 999 rows, which is the 499 rows of hour 00 plus the 500 of hour 01, and nothing from hour 02. Snapshot 3 reads all three hours. Each snapshot reads the rows that its commits had made visible by then, and no more. The gap between the two counts is the rows that the third commit added, 500 of them, so the history and the reads agree. **Working rule:** Treat an old snapshot as readable only while its files exist. Before you remove files, decide which snapshots you keep, and expire the others first. A query over an old snapshot that fails after a cleanup is a missing file. The fix is to find that file in a copy of the table, not to read a different snapshot in its place. ## Read an earlier snapshot {#read-an-earlier-snapshot} Reading an earlier snapshot means naming it when you scan the table. Each format has its own way to name one. Iceberg takes a snapshot ID. DuckLake takes a snapshot number in `AT (VERSION => n)`. Delta Lake takes a version number. The three numbers do not count the same things. The owner's first question is what the total was before the third hour landed. The answer is the state after the second commit, snapshot 2. ### Snapshots in pyiceberg {#snapshots-in-pyiceberg} In pyiceberg 0.12.0, `scan(snapshot_id=...)` reads the table as of that snapshot, and a scan with no snapshot ID reads the current snapshot. The example takes the IDs in commit order, reads snapshots 2 and 3 by ID, and sums the views in each, so the case's totals come from the table's own history. It also reads the current table and compares it with snapshot 3. Example `read-as-of-snapshot` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Read an Iceberg table as of an earlier snapshot, by snapshot ID.""" from lib.lakehouse import ( ICEBERG_ID, PAGEVIEWS_SCHEMA, new_warehouse, read_cut_hour, safe_error, scan_arrow, snapshot_labels, sql_rows, ) def main(): with new_warehouse() as wh: table = wh.catalog.create_table(ICEBERG_ID, schema=PAGEVIEWS_SCHEMA) for hour in range(3): table.append(read_cut_hour(hour)) snapshots = sorted(table.snapshots(), key=lambda snapshot: snapshot.sequence_number) ids = [snapshot.snapshot_id for snapshot in snapshots] labels = snapshot_labels(ids) for snapshot_id in (ids[1], ids[2]): arrow = scan_arrow(table, snapshot_id=snapshot_id) rows, views = sql_rows(wh.con, "SELECT count(*), sum(views) FROM t", t=arrow)[0] print(f"{labels[snapshot_id]} rows: {rows}") print(f"{labels[snapshot_id]} views: {views}") current = scan_arrow(table) last = scan_arrow(table, snapshot_id=ids[2]) print(f"current rows: {current.num_rows}") print(f"current equals snapshot 3: {current.equals(last)}") try: scan_arrow(table, snapshot_id=2) except Exception as exc: print(f"read by label: {safe_error(exc, wh)}") if __name__ == "__main__": main() ``` Output: ```text snapshot 2 rows: 999 snapshot 2 views: 580963 snapshot 3 rows: 1499 snapshot 3 views: 899451 current rows: 1499 current equals snapshot 3: True read by label: ValueError: Snapshot not found: 2 ``` Snapshot 2 has 999 rows and 580963 views in total. That total is the answer to the owner's first question: what the table held before the third commit. Snapshot 3 has 1499 rows and 899451 views. The current table gives 1499 rows, and the example prints `True` for the comparison. Reading the table with no ID gives the same rows as its newest snapshot, and the example compares the rows themselves, not only their counts. The views column answers the owner's second question. Snapshot 3 has more views than snapshot 2, and the difference is the views that hour 02 added. The total changed because the third commit added rows, not because the dashboard's query changed. Two reads of the same table show that without any change to the query. A reader who has the label "snapshot 2" in mind might pass the number 2 as the ID. The call fails, and that failure is the safe outcome: `ValueError: Snapshot not found: 2` The error names the number it was given, which is not a snapshot ID, and nothing in the table changed. The reader then knows to look the ID up. The labels exist for the prose, and the IDs for the scan. The example gets each ID from `table.snapshots()`, and it maps each ID to its label with `snapshot_labels`, which takes the IDs in commit order. Nothing in the example computes an ID from a count, and it prints only labels, never IDs. **Working rule:** Get snapshot IDs from `table.snapshots()` and pass the ID you looked up, never a label's number. Map the IDs to labels in commit order before you print anything. When the question is the state after a commit, find that commit in the history first, and read the snapshot it created. ### Read a commit by its number in each format {#versions-in-ducklake-and-delta} Delta Lake reads an earlier version with `DeltaTable(path, version=n)`. The version is the commit's position in the table's history, counted from 0, so version 0 is the first write. The helper `read_delta` takes the path and an optional version and returns the rows as a sorted Arrow table. In this book, every Delta read goes through deltalake's `QueryBuilder`, which the helper calls for you. The [decisions section](#snapshot-decisions) gives the reason. DuckLake keeps a snapshot for each change to the catalog, and the change includes creating a schema and creating a table. DuckLake numbers its snapshots from 0 as well, so the first insert in a new catalog is not number 1. The [DuckLake time travel page](https://ducklake.select/docs/stable/duckdb/usage/time_travel) shows the clause for reading one, `SELECT * FROM tbl AT (VERSION => 3);`, and says the snapshot to query can be specified by timestamp or "explicitly using a snapshot identifier". The book uses the identifier. You read the number from `ducklake_snapshots`, not from your own count of inserts. The three-version example writes the same three hours to a Delta table and a DuckLake table, one commit per hour in each format. It reads each commit back by its number and prints both row counts. It also counts the DuckLake snapshots before the first insert, and shows one failure. Example `three-version-numbers` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Commit the same three hours to Delta Lake and DuckLake, then read each commit back by its version number.""" from deltalake import DeltaTable, write_deltalake from lib.lakehouse import (DUCKLAKE_ALIAS, NAMESPACE, TABLE, attach_ducklake, connect_ducklake, new_warehouse, read_cut_hour, read_delta, safe_error, snapshot_labels, sql_rows) LAKE_TABLE = f"{DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE}" CATALOG_SNAPSHOTS = f"SELECT snapshot_id FROM ducklake_snapshots('{DUCKLAKE_ALIAS}')" TIME_COLUMN = f"SELECT snapshot_time FROM ducklake_snapshots('{DUCKLAKE_ALIAS}')" def main(): with new_warehouse() as wh: delta_path = str(wh.root / "delta" / TABLE) connect_ducklake(wh.con) attach_ducklake(wh.con, DUCKLAKE_ALIAS, wh.root) wh.con.execute(f"CREATE SCHEMA {DUCKLAKE_ALIAS}.{NAMESPACE}") sql_rows(wh.con, f"CREATE TABLE {LAKE_TABLE} AS SELECT * FROM empty", empty=read_cut_hour(0).slice(0, 0)) for hour in range(3): batch = read_cut_hour(hour) write_deltalake(delta_path, batch, mode="append") sql_rows(wh.con, f"INSERT INTO {LAKE_TABLE} SELECT * FROM batch", batch=batch) inserts = sorted(row[0] for row in sql_rows(wh.con, CATALOG_SNAPSHOTS + " WHERE map_contains(changes, 'tables_inserted_into')")) before = sql_rows(wh.con, CATALOG_SNAPSHOTS + f" WHERE snapshot_id < {inserts[0]}") print(f"snapshots before the first insert: {len(before)}") lake_labels = snapshot_labels(sorted(row[0] for row in sql_rows(wh.con, CATALOG_SNAPSHOTS))) delta_versions = list(range(DeltaTable(delta_path).version() + 1)) for hour, (delta_version, snapshot) in enumerate(zip(delta_versions, inserts, strict=True)): lake = sql_rows(wh.con, f"SELECT count(*) FROM {LAKE_TABLE} AT (VERSION => {snapshot})")[0][0] rows = len(read_delta(delta_path, version=delta_version)) print(f"after hour {hour:02d}, Delta version {delta_version} rows: {rows}") print(f"after hour {hour:02d}, DuckLake {lake_labels[snapshot]} rows: {lake}") try: wh.con.execute(TIME_COLUMN).fetchall() except Exception as exc: print(f"select time: {safe_error(exc, wh)}") if __name__ == "__main__": main() ``` Output: ```text snapshots before the first insert: 3 after hour 00, Delta version 0 rows: 499 after hour 00, DuckLake snapshot 4 rows: 499 after hour 01, Delta version 1 rows: 999 after hour 01, DuckLake snapshot 5 rows: 999 after hour 02, Delta version 2 rows: 1499 after hour 02, DuckLake snapshot 6 rows: 1499 select time: InvalidInputException: Invalid Input Error: Required module 'pytz' failed to import, due to the following Python exception: ModuleNotFoundError: No module named 'pytz' ``` Before the first insert, the DuckLake catalog holds 3 snapshots: the default schema that DuckLake creates on attach, the `wiki` schema and the table. This book labels DuckLake's snapshots from 1, setup included, so the first insert is "snapshot 4", which is DuckLake's own number 3 plus one. A reader who counts inserts from 1 gets the wrong label. After hour 00, Delta version 0 has 499 rows and DuckLake snapshot 4 has 499. After hour 01, Delta version 1 has 999 rows and DuckLake snapshot 5 has 999. After hour 02, which is the full table, Delta version 2 has 1499 rows and DuckLake snapshot 6 has 1499. The hour is what makes the rows comparable, because the formats do not share a numbering. The table below sets the three numbers side by side. | Format | Number that names a commit | Which changes it counts | How you read it | | --- | --- | --- | --- | | Iceberg | Snapshot ID, a long integer | Each commit to the table | `scan(snapshot_id=...)` | | Delta Lake | Version, counted from 0 | Each commit to the table | `DeltaTable(path, version=n)` | | DuckLake | Snapshot number, counted from 0 (the book's label is one higher) | Every change to the catalog, including schema and table creation | `AT (VERSION => n)` | **Working rule:** Compare formats by the hour each commit loaded, never by raw number. Print Iceberg and DuckLake snapshots with their labels, print Delta versions as Delta numbers them, and bind the counts to them. Read Delta only through `read_delta`, and read DuckLake only with numbers that come from `ducklake_snapshots`. If a table in one format gets commits that the other does not, the labels stop lining up, and the history of each format becomes the only reliable record. ## What goes wrong in production {#snapshot-production-failures} The failures in this chapter are the ones a reader is likely to meet first. The first two appear in this chapter's examples. The third and fourth are argued from the documentation and the book's rules, and the fifth and sixth lean on [chapter 1](#what-a-table-format-adds) and [chapter 7](#snapshots-keep-files-alive). The first is the number that is not an ID. A failed scan is a safe failure, because the error says what was wrong. Worse failures come from reading a number that is valid in one format and means something else in another, or from a query that runs and returns something unexpected. The [read-as-of-snapshot example](#snapshots-in-pyiceberg) shows the first kind: the ID guess fails loudly, and the error text names the value that was passed. The second failure comes from a query that fetches a time column. The three-version example includes a select of `snapshot_time` from `ducklake_snapshots`, and the call fails: `InvalidInputException: Invalid Input Error: Required module 'pytz' failed to import, due to the following Python exception:` The message says the pytz module is missing. The pinned environment does not install it. DuckDB's Python client needs pytz to turn a `TIMESTAMPTZ` column such as `snapshot_time` into Python values. The example avoids the problem by selecting only `snapshot_id` and filtering on `changes`, so it never fetches `snapshot_time`. To read the time anyway, cast it with `snapshot_time::VARCHAR`. Leave out the column or cast it; we would rather not add a package the book does not pin. The third failure is a timestamp read. DuckLake accepts a timestamp in `AT (TIMESTAMP => ...)`, and Iceberg snapshots have a creation time. A timestamp picks a state by the clock, and the clock is the time when each commit ran. In production that is a good way to ask for "the table as it was at nine in the morning". An example would have to take a clock time from the run, which changes on every run, so the book uses IDs and version numbers instead. Choose the timestamp form when you need a point in wall-clock time and you have a record of when each commit ran. Choose the number form when you need the same answer every time. The fourth failure is a number that drifts. Two formats can hold the same table, and the same label can then name different states. The book's example commits the same hours in the same order, so the labels agree. A production table that receives a repair in one format and not the other will not agree. Even with identical histories, the raw numbers do not line up across formats, because DuckLake's snapshot numbers also count schema and table creation. A dashboard that reads "version 4" from both formats will show different totals, so compare commits by their labels. The check is to compare the row counts for the label in both formats before you trust either number. The three-version example does that. The fifth failure is about where DuckLake puts small writes. The [data inlining page](https://ducklake.select/docs/stable/duckdb/advanced_features/data_inlining) says "Data inlining is enabled by default with a row limit of 10." A small insert can therefore stay in the catalog and create no Parquet file, so a file count taken after it reads low. The example's batches have 500 rows, which is over that limit, so under the documented default each of its inserts is written as a file. A count of data files is only meaningful once you know the row limit, and [chapter 1](#what-a-table-format-adds) names the default limit. The sixth failure is the one this chapter warns about most. An old snapshot shows the table as it was written, and only while the files it names still exist. Nothing fails at the cleanup itself. The error appears when a reader asks for a snapshot whose files are gone. [Chapter 7](#snapshots-keep-files-alive) shows which files a cleanup removes. **Working rule:** Read the error text before you change the query. It names the value, the module or the version that failed, and that is the fastest route to the fix. Compare each count in one format with the same label in the other format before you trust either one. Keep the snapshot your question needs readable until the question is answered, and expire only the snapshots that no question still needs. Record the label and the format's name with every number you report, so that a reader of the report can reproduce it. ## Decisions this chapter made {#snapshot-decisions} - **Labels, not IDs.** The examples print labels such as "snapshot 2" for Iceberg and DuckLake. The Iceberg IDs and the DuckLake snapshot numbers never reach the output. Delta prints its version numbers, which count from 0 and are the numbers a reader passes to `DeltaTable`. A reader can rerun an example and compare its text line by line, and a diff shows a change in the data or the library, never a change in a random ID. The labels are mapped in commit order, so they keep their meaning when a library returns its history in another order. - **IDs and version numbers, not timestamps.** A timestamp read depends on the clock and on when each commit ran. We read by ID or number, so that a reader gets the same rows from the same commit whenever the example runs. A production reader who needs wall-clock time should use the timestamp form and keep a record of when each commit ran. - **Delta reads through `read_delta` only.** The helper reads through deltalake's `QueryBuilder`. At these pins, a process that reads Delta into Arrow with `to_pyarrow_table()` aborted at exit in some of our test runs. We fix the read path rather than work around the abort, so no example calls `os._exit` or hides the exit status. - **DuckLake through the helpers only.** The examples load DuckLake with `connect_ducklake` and attach it with `attach_ducklake`. We never set `allow_unsigned_extensions`. The book's runs load DuckDB's official, signed DuckLake extension from the vendored file in the repository, after the helper checks both hashes against `SOURCE.toml`, so an offline run needs no download. Only on a reader's own machine, when that vendored file is absent, does `connect_ducklake` run `INSTALL ducklake`. - **Narrow queries on the catalog.** The examples read `ducklake_snapshots` by named column and leave out `snapshot_time`, because that column needs a module the pins do not install. We would rather name the columns we need than add a dependency that the pins do not record. - **Labels counted from one for Iceberg and DuckLake, Delta's own numbers for Delta.** Iceberg and DuckLake snapshots get "snapshot N", counted from 1 over every snapshot the format records. For DuckLake that includes the three setup snapshots, so DuckLake's own number is N minus 1. Delta prints its real version, counted from 0. Each label says only which commit it is in its own table. It never says that two formats hold the same state, and the example checks that by comparing row counts. - **Failures printed, not raised.** Two examples end with a call that fails on purpose, and the example catches the error and prints it. The failure is shown because a reader will meet it, and the error text is printed in full so the reader can match it against their own output. - **Read the history before the state.** The examples list each format's commits, and only then read a commit by its number. A chapter reader gets the operation and the row count for each commit before any scan, so a count that looks wrong can be checked against the commit that produced it. # Add and rename columns without rewriting files By the end of this chapter you can add a column to a table that already holds data, rename a column in Iceberg without rewriting its data files, and say why a rename in Delta Lake is a rewrite of the whole table. The versions this book pins are pyiceberg 0.12.0 and deltalake 1.6.6. The running case is `wiki.pageviews_hourly`, the table of hourly view counts for six hours of Wikimedia pageviews that [the first chapter](#first-table) loads. Two requests reach it after the first three hours have been written. The first asks for a lineage column: each row should record which hourly load wrote it, so that a bad load can be traced back to its rows. The table needs a column for that, and the rows it already holds were written before the column existed. The second is a naming fix: the dashboards call the title column `article_title`, while the table calls it `page_title`, the name the dump uses. Both are schema changes on a table that already has data. For each one, the question is what happens to the files that were written before the change. A schema change is a metadata change in principle. The data files stay where they are, and the format records the added or renamed columns in its metadata. Whether the old files still read correctly depends on how the format finds a column inside a file. A format that finds columns by name has trouble when a name changes. A format that finds them by something that does not change does not. This chapter shows which of the two each format is, what each change costs, and which change the pinned Delta Lake library cannot make. ## Columns are tracked by ID {#columns-tracked-by-id} A data file holds values for some columns, and the table's schema says which column each value belongs to. The simple way to match them is by name. That works until a name changes. The schema then names a column the old file does not have under that name, and the reader has to guess what the old values belong to. A reader that matches by name and finds no column under the old name has two bad options: fail the query, or fill the column with null without saying so. The second is the quieter failure, and a dashboard can show it for weeks before anyone looks. We count the quiet null as the worse failure, which is a judgement, not a measured result. Matching by identity avoids both, because the column keeps its identity whatever its name is. Iceberg gives every column an integer field ID when the column is created, and it matches columns by that ID. The name is a label that can change. The Iceberg specification says so directly: renaming a field "must change the name, but not the field ID" [1]. When a reader opens an old file, it matches each column by field ID. If the file holds no value for a field, the spec's projection rule first checks two cases that do not apply here, an identity-partition value and a name mapping. It then returns the field's `initial-default` when one is set, and otherwise returns null: "Return `null` in all other cases" [1]. The spec defines `initial-default` for a column added with a default. This chapter does not set one, so an added column reads as null in every old row. You can see the IDs yourself. The schema of an Iceberg table lists each field with its ID, and [the example in the rename section](#rename-column-without-rewrite) prints the ID of one column before and after a rename. The ID is the value to check when you want to know whether a rename kept the column's identity. Delta Lake starts from a different place. Its table schema names each column, and a reader matches by name. The Delta Lake protocol defines an optional column mapping feature that separates the two: every column gets a physical name stored in the files, and the name readers use is a display name that can change without touching the files. The protocol says column mapping supports "the renaming and dropping of columns without having to rewrite all the data" [2]. Whether a given library release does that is a separate question. [The section on Delta Lake](#rename-in-delta-lake) answers it for the pinned release. **Working rule:** Expect an Iceberg column to keep its data through a rename, because the reader matches by field ID. Expect a Delta Lake column to match by name unless column mapping is on. Test a rename on a copy of the table before you rely on it, and read the column names back afterwards. ## Add load_batch after three hours {#add-response-bytes} The first change is the one the case needs first: a column that does not exist in the files already written. In a pipeline, that happens on an ordinary day. The pipeline starts stamping each load with its number, and the table already holds rows that predate the stamp. The example starts from the first three hours, each written as its own commit, so the table holds three snapshots and three data files. An Iceberg snapshot is the table's state after a commit that changes data. Each append in this chapter adds one, and a schema change adds none. In Iceberg, you add a column inside a schema update. The call takes the name and the type, and the table assigns the field ID: ```python with table.update_schema() as update: update.add_column("load_batch", IntegerType()) ``` The example runs the three early commits, adds the column, then appends hours 03 to 05. Each of those loads sets `load_batch` to its hour number, so its rows carry the value. It then counts rows by hour of the day: hours 00 to 02 were written before the add, and hours 03 to 05 after it. Each hour is one write, so the split is also the split by load. Each count is printed with a label, so the text can quote it. Example `add-column-to-existing-files` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Add load_batch after three hours of appends, in Iceberg and in Delta Lake. load_batch holds the number of the hourly load that wrote each row. Rows written before the add have no value for it, so they read as null. Later loads set it. """ import pyarrow as pa from deltalake import DeltaTable, write_deltalake from deltalake.schema import Field, PrimitiveType from pyiceberg.types import IntegerType from lib.lakehouse import ICEBERG_ID, delta_file_count, iceberg_file_count, new_warehouse, read_cut_hour, read_delta, split_nulls def stamp(rows, hour): """Return one hourly load with load_batch set to that load's hour number.""" return rows.append_column("load_batch", pa.array([hour] * rows.num_rows, type=pa.int32())) with new_warehouse() as wh: table = wh.catalog.create_table(ICEBERG_ID, schema=read_cut_hour(0).schema) for hour in range(3): table.append(read_cut_hour(hour)) print(f"iceberg snapshots before the add: {len(table.snapshots())}") print(f"iceberg data files before the add: {iceberg_file_count(table)}") with table.update_schema() as update: update.add_column("load_batch", IntegerType()) print(f"iceberg snapshots after the add: {len(table.snapshots())}") print(f"iceberg data files after the add: {iceberg_file_count(table)}") for hour in range(3, 6): table.append(stamp(read_cut_hour(hour), hour)) iceberg_rows = table.scan().to_arrow() print(f"iceberg rows in the table: {iceberg_rows.num_rows}") split_nulls(wh.con, iceberg_rows, "iceberg", "load_batch", 3, sides=("written before the add", "written after the add")) path = f"{wh.root}/delta/pageviews_hourly" for hour in range(3): write_deltalake(path, read_cut_hour(hour), mode="append") print(f"delta version before the add: {DeltaTable(path).version()}") print(f"delta data files before the add: {delta_file_count(DeltaTable(path))}") DeltaTable(path).alter.add_columns([Field("load_batch", PrimitiveType("integer"))]) print(f"delta version after the add: {DeltaTable(path).version()}") print(f"delta data files after the add: {delta_file_count(DeltaTable(path))}") for hour in range(3, 6): write_deltalake(path, stamp(read_cut_hour(hour), hour), mode="append") delta_rows = read_delta(path) print(f"delta rows in the table: {delta_rows.num_rows}") split_nulls(wh.con, delta_rows, "delta", "load_batch", 3, sides=("written before the add", "written after the add")) ``` Output: ```text iceberg snapshots before the add: 3 iceberg data files before the add: 3 iceberg snapshots after the add: 3 iceberg data files after the add: 3 iceberg rows in the table: 2999 iceberg rows written before the add: 1499 iceberg rows written before the add with load_batch null: 1499 iceberg rows written after the add: 1500 iceberg rows written after the add with load_batch null: 0 delta version before the add: 2 delta data files before the add: 3 delta version after the add: 3 delta data files after the add: 3 delta rows in the table: 2999 delta rows written before the add: 1499 delta rows written before the add with load_batch null: 1499 delta rows written after the add: 1500 delta rows written after the add with load_batch null: 0 ``` Before the add, the table has 1499 rows, and 1499 of them read null for `load_batch`. After the add and the three later appends, the table has 2999 rows. The 1500 rows written after the add have 0 nulls in `load_batch`. The snapshot count stays at 3 across the schema change, and the table keeps 3 data files. The change touched the metadata and no data file. The two halves of the example hold the same rows and the same nulls, so a disagreement between them would point at a reader, not at the data. Delta Lake adds a column through its `alter` object. The call takes a field, and the table's version goes up by one: ```python DeltaTable(path).alter.add_columns([Field("load_batch", PrimitiveType("integer"))]) ``` The same example does the Delta half. It writes the three early hours as three appends, adds the column, and appends hours 03 to 05. Delta counts versions from 0, so the first write is version 0. Every version number this chapter prints is Delta's real one. The version moves from 2 to 3, and the table keeps 3 data files. Before the add there are 1499 rows, all null for `load_batch`. After the add and the three later appends the table has 2999 rows, and the 1500 rows written after the add have 0 nulls. The example's table has no column mapping. Both formats give the same result for this case. The old rows stay as they were written, and the column reads as null for them. The writer did not have to touch a single old file. **Working rule:** Add a column with the format's own call, and never by rewriting the files. Expect the rows written before the add to read null for the column, and check that they do. ### Old files read as null {#old-files-read-as-null} A null in an added column can look like missing data, and a reader who sees it may assume the old files are wrong. They are not. The old files hold the values the pipeline wrote, and the column they lack is filled on read. The rule the reader follows is the one [the columns section](#columns-tracked-by-id) quotes from the spec: with no `initial-default`, no value for the field means null. That makes the null count a useful check. The example counts nulls by hour of the day, and each hour is one write, so the counts show the pattern to expect: every row written before the add is null, and no row written after it is. In Iceberg the output reads 1499 nulls among 1499 rows. In Delta Lake it reads 1499 among 1499. Both match the rule. If a count falls outside that pattern, a writer sent a different set of columns, or sent the column with nulls, and the cause is in the pipeline, not in the format. Look at the writer for the rows that break the pattern before you query the column, because a query over a column that is partly null will return an answer that looks reasonable. **Working rule:** After an add, count nulls in the added column by hour of the day, where each hour is one write. Rows written before the add should be null, and rows written after it should not be. Any other pattern means a writer sent a different set of columns, or sent nulls in the column, so find that writer before you trust the column. ## Rename a column by its ID without rewriting files {#rename-column-without-rewrite} With the field ID in place, a rename is one call in Iceberg. The call changes the name the schema reports, and keeps the ID: ```python with table.update_schema() as update: update.rename_column("page_title", "article_title") ``` The example writes three hours, reads the field ID of `page_title`, renames the column, and reads the ID again. It counts the data files before and after the call, and it reads the table through DuckDB with the new name in the query. Example `rename-by-id` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Rename page_title to article_title in Iceberg, then read old files under the new name. Iceberg keeps each column's field ID across a rename, so the three files written under page_title are read under article_title without a rewrite. """ import pyarrow as pa from lib.lakehouse import ( ICEBERG_ID, iceberg_file_count, new_warehouse, read_cut_hour, scan_arrow, snapshot_labels, sql_rows, ) with new_warehouse() as wh: table = wh.catalog.create_table(ICEBERG_ID, schema=read_cut_hour(0).schema) for hour in range(3): table.append(read_cut_hour(hour)) before = table.schema().find_field("page_title").field_id print(f"page_title field ID before the rename: {before}") print(f"data files before the rename: {iceberg_file_count(table)}") with table.update_schema() as update: update.rename_column("page_title", "article_title") print(f"article_title field ID after the rename: {table.schema().find_field('article_title').field_id}") print(f"data files after the rename: {iceberg_file_count(table)}") rows = scan_arrow(table) print(f"rows: {rows.num_rows}") count = sql_rows(wh.con, "SELECT count(article_title) FROM t", t=rows)[0][0] print(f"article_title not null: {count}") labels = snapshot_labels([snapshot.snapshot_id for snapshot in table.snapshots()]) first = table.snapshots()[0].snapshot_id print(f"{labels[first]} columns: {', '.join(scan_arrow(table, snapshot_id=first).column_names)}") ``` Output: ```text page_title field ID before the rename: 3 data files before the rename: 3 article_title field ID after the rename: 3 data files after the rename: 3 rows: 1499 article_title not null: 1499 snapshot 1 columns: hour_utc, domain_code, page_title, views, response_bytes ``` The field ID of `page_title` is 3 before the rename, and the field ID of `article_title` is 3 after it. The table keeps 3 data files on both sides of the call, so no file was rewritten. The DuckDB query counts 1499 non-null `article_title` values, out of 1499 rows. Every row counts, including the rows that were written under `page_title`, because the reader matches the files by ID and reads them under `article_title`. The dashboards need the name, and the table does not have to be rewritten to give it to them. The data stays as it was written, and the next read sees the renamed column. The example also reads the first snapshot's columns, and they are `hour_utc`, `domain_code`, `page_title`, `views` and `response_bytes`. A snapshot reads with the schema it was written with, so an older snapshot still reports the old name. [Chapter 2](#snapshots-and-time-travel) covers reading older snapshots in general, and the point here is narrower: a rename does not reach back into the history. The cost comparison is what makes this the right call for Iceberg. A rename that keeps the ID touches the metadata and nothing else, so its cost does not grow with the size of the table. The same rename done by copying the table costs one read and one write of every row. [The section on Delta Lake](#rename-in-delta-lake) shows what that copy costs in that format. **Working rule:** Rename an Iceberg column when its name is wrong, and check the field ID before and after the call. The field ID must be the same, and the data file count must not change. Tell every reader of the table about the rename, because the current schema has only `article_title`, and an older snapshot still has `page_title`. ### Delta Lake has no rename at this pin {#rename-in-delta-lake} The same rename in Delta Lake has no call at this pin. The Delta Lake table alteration object has no rename method at deltalake 1.6.6, and its methods are listed in the package [3]. The one that looks like a rename is `set_column_metadata`, because it takes a dictionary of field metadata, and nothing stops you from passing a `name` key. The call returns without error when you pass a `name`. It renames nothing on any table, with or without column mapping, and that is the trap this section shows. The trap example creates a table with column mapping on, which is the protocol's route to a rename, and then sends the call: ```python DeltaTable(path).alter.set_column_metadata("page_title", {"name": "article_title"}) ``` Example `delta-rename-trap` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Rename a Delta Lake column with set_column_metadata, then read the names back. The call returns without error and writes a commit, but the column keeps its name. A query that uses the new name fails. """ from deltalake import DeltaTable, write_deltalake from lib.lakehouse import new_warehouse, read_cut_hour, read_delta, safe_error, sql_rows with new_warehouse() as wh: path = f"{wh.root}/delta/pageviews_hourly" write_deltalake(path, read_cut_hour(0), configuration={"delta.columnMapping.mode": "name"}) print(f"version before the rename: {DeltaTable(path).version()}") DeltaTable(path).alter.set_column_metadata("page_title", {"name": "article_title"}) print("set_column_metadata returned") table = DeltaTable(path) print(f"version after the rename: {table.version()}") page = next(field for field in table.schema().fields if field.name == "page_title") print(f"name in page_title metadata after the rename: {page.metadata['name']}") print(f"columns read back: {', '.join(read_delta(path).column_names)}") try: sql_rows(wh.con, "SELECT article_title FROM t", t=read_delta(path)) except Exception as exc: print(safe_error(exc, wh)) ``` Output: ```text version before the rename: 0 set_column_metadata returned version after the rename: 1 name in page_title metadata after the rename: article_title columns read back: hour_utc, domain_code, page_title, views, response_bytes BinderException: Binder Error: Referenced column "article_title" not found in FROM clause! Candidate bindings: "page_title", "domain_code", "hour_utc", "views", "response_bytes" LINE 1: SELECT article_title FROM t ^^^^^^^^^^^^^ ``` The call returns, and the table's version moves from 0 to 1, so the library wrote a commit. The `name` key is in the field's metadata. The column is still `page_title`: the names read back through the library are `hour_utc`, `domain_code`, `page_title`, `views` and `response_bytes`. The DuckDB query that asks for `article_title` fails with a `BinderException`, and the error lists `page_title` among the columns it could bind. The trap is that the call looks like the rename and succeeds at the call. Nothing in the return value says the name is unchanged. Only a read of the names shows the truth. Column mapping gives no rename at this pin. The protocol's column mapping section describes a rename [2], but deltalake 1.6.6 has no call that performs one. The example creates the table with `delta.columnMapping.mode` set to `name`, and the call still changes no name. That shows what `set_column_metadata` does on a mapped table, and nothing about a rename with mapping, because the library has no call that makes one. Example `delta-rename-by-rewrite` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Rename a Delta Lake column by rewriting the table, and read the old version back. The rewrite gives the current version the new name and writes every data file again. Older versions keep the names they were written with. Delta numbers versions from 0. """ from deltalake import DeltaTable, write_deltalake from lib.lakehouse import new_warehouse, read_cut_hour, read_delta with new_warehouse() as wh: path = f"{wh.root}/delta/pageviews_hourly" for hour in range(3): write_deltalake(path, read_cut_hour(hour), mode="append") print(f"data files before the rewrite: {len(DeltaTable(path).file_uris())}") before = DeltaTable(path).version() print(f"version before the rewrite: {before}") print(f"columns before the rewrite: {', '.join(read_delta(path, version=before).column_names)}") renamed = read_delta(path).rename_columns(["hour_utc", "domain_code", "article_title", "views", "response_bytes"]) write_deltalake(path, renamed, mode="overwrite", schema_mode="overwrite") print(f"data files after the rewrite: {len(DeltaTable(path).file_uris())}") after = DeltaTable(path).version() print(f"version after the rewrite: {after}") print(f"columns after the rewrite: {', '.join(read_delta(path).column_names)}") print(f"rows after the rewrite: {read_delta(path).num_rows}") ``` Output: ```text data files before the rewrite: 3 version before the rewrite: 2 columns before the rewrite: hour_utc, domain_code, page_title, views, response_bytes data files after the rewrite: 1 version after the rewrite: 3 columns after the rewrite: hour_utc, domain_code, article_title, views, response_bytes rows after the rewrite: 1499 ``` A rewrite is the way to get the name at this pin. The second example reads the whole table through the library, renames the column in memory, and writes the result back with `mode="overwrite"` and `schema_mode="overwrite"`, which replace the data and the schema together. The table had 3 data files before the rewrite and has 1 after it. The rewrite makes version 3, which reads `article_title`, and 1499 rows are in it. The rewrite costs the size of the table, because every row is read and written again. The example's table is small, and the file count falls from 3 to 1. A production table that holds millions of rows pays the same per-row cost, and a team should plan the rewrite as a job of its own. The version before the rewrite, 2, still reads `page_title`, which the example also prints. The rewrite therefore changes the current version only. Older versions keep the names they were written with. Before you run a rewrite, list the readers that ask for the column by name. The switch happens in one commit, so every reader of the current version moves at the same moment, while a job that pins an older version keeps the old name. Move those jobs with the readers, and record the version number the rewrite replaced until they have moved. **Working rule:** Choose Delta Lake column names when the table is created, because at this pin a later rename is a rewrite of the whole table. Do not use `set_column_metadata` to rename a column: it changes the version and the metadata, and the column keeps its name. After any schema change, read the column names back through the library, as the trap example does, and check them before a query depends on them. ## What goes wrong in production {#schema-production-failures} Schema changes fail quietly more often than they fail loudly. Four failures in this chapter are worth knowing, and each one has a way to notice it. The first is the no-op rename. A `set_column_metadata` call succeeds, a commit appears in the history, and the column keeps its old name. The failure shows on the first query that uses `article_title`: the DuckDB query fails with a `BinderException`. A scheduled job that runs such a query fails on its first run after the change, so the error arrives close to its cause. Read the names back after any change, as the trap example does, before any query runs. The second is the stale name in history. A rewrite gives the current version the name `article_title`, but the earlier versions keep the names they were written with, and the rewrite example shows it: version 2 still reads `page_title`. An Iceberg snapshot does the same: the first snapshot still reports `page_title`. A query that reads an earlier version or snapshot by `article_title` finds no such column. Reads by version or snapshot number need the names that were current at that point, so a job that pins a version number also pins its schema. The third is the null that looks like missing data. An added column reads as null in every row written before the add. A null count on that column is correct, and it is a fault only when it falls outside the before-and-after pattern the add example shows. Count nulls by hour, as the add example does, where each hour is one write, and investigate any row that breaks the pattern. The fourth is the schema change that does not show in the snapshot list. The Iceberg add and rename both left the snapshot count where it was, and the file count too. A tool that looks for schema changes by counting snapshots will miss them. Compare the schema itself, by field ID and name, when you need to know what changed. **Working rule:** After any schema change, do three things before a downstream query runs: read the column names back through the library, count nulls in each added column by hour, where each hour is one write, and compare the schema, not the snapshot count, with what the change was meant to do. ## Decisions this chapter made {#schema-decisions} The table in this section sets the two formats side by side for the changes in this chapter. Each cell comes from an example in this chapter. | Change | Iceberg | Delta Lake | | --- | --- | --- | | Add a column | Works. Old rows read null, and the snapshot count stays at 3 | Works. Old rows read null, and the version goes up by one | | Rename a column | Works. The field ID stays the same, and no data file is rewritten | No rename call. `set_column_metadata` commits and changes no name; a rewrite gives the name | | Read an older snapshot or version | Shows the older name | Shows the older name | - Add a column to a table that already holds data, in both formats. Old rows read it as null, and the add example shows the counts by hour that confirm the pattern. We recommend this over any copy of the table, because it writes no old file. - Rename an Iceberg column with `rename_column`, and check that its field ID is the same before and after. We recommend it for any Iceberg column whose name is wrong, because its cost does not depend on the table's size. - Choose Delta Lake column names when the table is created. At the pinned release, a rename there is a rewrite of the whole table, and that is the cost to plan for when a name is wrong. - Do not rename a Delta Lake column with `set_column_metadata`. The call succeeds and changes nothing about the name. - Do not count on column mapping for a rename at this pin. The library has no call that performs one, so turning mapping on gains no rename. - Read the column names back after every schema change, in both formats, before a query depends on them. ### Sources {#schema-sources} 1. Apache Iceberg Table Spec, Schema Evolution (Column Projection), https://iceberg.apache.org/spec/, read Nov 6, 2026. Quoted: "Renaming an existing field must change the name, but not the field ID." and "Return `null` in all other cases." 2. Delta Transaction Log Protocol, Column Mapping section, tag v4.4.1, https://raw.githubusercontent.com/delta-io/delta/v4.4.1/PROTOCOL.md, read Nov 6, 2026. Quoted: "the renaming and dropping of columns without having to rewrite all the data." 3. deltalake 1.6.6, the methods of `DeltaTable.alter` in the installed package: `add_columns`, `add_constraint`, `add_feature`, `drop_column_not_null`, `drop_constraint`, `set_column_metadata`, `set_table_description`, `set_table_name` and `set_table_properties`. None renames a column. To list them, run `dir(DeltaTable(path).alter)`. # Change partitioning without rewriting old files By the end of this chapter you can change the partitioning of a table in Iceberg, Delta Lake and DuckLake, and count the files a query plans before and after the change. Partitioning decides which files a query opens, so a change to it changes what every reader pays. What it does to the files already written depends on the format. The case's table, `wiki.pageviews_hourly`, is partitioned by day. Each example in this chapter builds its table from the cut again, with the original columns and not the ones [chapter 3](#columns-tracked-by-id) added. This chapter changes how the rows are grouped. The [three hourly loads](#append-each-hour) from the start are already on disk. The dashboards read one hour at a time, so the team wants hour partitions for every load after hour 02. They do not want to copy the three loads that exist, and they need to know what the change does to them. The change has two costs. The first is the work to make it, which depends on whether the format leaves the old files alone. The second is what every later query pays, because a query that spans the change plans files from both eras. This chapter measures both with counts from each library, so the decision rests on what the tables do, not on a description of the format. ## Partition specs are versioned too {#partition-specs-are-versioned} A partition spec says how a table groups its rows: one group per day of `hour_utc`, for example. In Iceberg the spec is part of the table's metadata, and the manifest that lists each data file records the spec it was written under. Changing the spec makes a second spec with its own ID. Later writes use the second spec, and the files already on disk keep the spec they were written under. The [Iceberg specification](https://iceberg.apache.org/spec/) says that a changed spec is added to the table's list of partition specs under a unique spec ID, and that each manifest records the ID of the spec used to write it (`partition_spec_id`). That gives the table two partitioning eras side by side. A query reads the spec of each file it opens, so one scan can cover files under several specs. The planner does this for you, which means the cost of the change is visible only in the plan and the file counts, not in the rows the query returns. A query written against `hour_utc` keeps its text when the spec changes, which is the point of the design. What differs is the set of files the planner opens. The rest of this chapter shows both. The change is also a metadata change. Iceberg writes a spec with its own ID and does not touch the data files. The file counts in [the next section](#files-written-under-each-spec) show it. A spec change also leaves the old spec in place as history. The table keeps a record of every spec it has had, so a reader can always find out which fields a file was written under. This matters for a team that changes partitioning more than once: each file answers the question "how was I split?" with its own spec, and nobody has to remember the order of the changes. The record is metadata, so keeping it costs little, but it does mean that the table's list of specs grows with each change. The other two formats handle partitioning differently, and the differences decide what a change costs. Delta Lake takes its partition columns by name, so the writer creates them as columns. DuckLake keeps its partitioning keys in its catalog. Each of the two has its own part in [the section on the other formats](#partitioning-in-other-formats), after the Iceberg steps that the rest of the chapter depends on. **Working rule:** Change an Iceberg partition spec when your readers' filters change and the current spec plans too many files for them. If the current spec already plans few files for those filters, leave it alone. When you do change it, leave the old files where they are. Record the change, with the date, the old and replacement fields, and the readers' filters that prompted it, in the table's runbook. Keep the plan counts from before and after the change with it, as [the section on what a query scans](#check-what-a-query-scans) shows. ## Move from daily to hourly partitions {#daily-to-hourly-partitions} Iceberg partitions by a transform of a column. The day transform on `hour_utc` puts each row under its calendar day, and the hour transform puts it under its hour. The pyiceberg library computes these transforms with the `pyiceberg-core` package. The day transform imports that package when it writes, so an install with only `pyiceberg` fails at the first append under a day or hour partition (pyiceberg 0.12.0 source, [`pyiceberg/transforms.py`](https://raw.githubusercontent.com/apache/iceberg-python/pyiceberg-0.12.0/pyiceberg/transforms.py), `DayTransform.pyarrow_transform`). The pins in this book therefore include `pyiceberg-core` at 0.10.1, next to `pyiceberg` at 0.12.0. The case's table starts with a day spec. The code appends hours 00 to 02, changes the spec, and appends hours 03 to 05. The change is one `update_spec()` block that removes the day field and adds the hour field: ```python with table.update_spec() as update: update.remove_field("ts_day") update.add_field("hour_utc", HourTransform(), "ts_hour") ``` Both calls go in one block, so the table gets one spec change with the day field gone. The example lists the specs in order and the transform of each field. The field names are the ones this book uses: `ts_day` and `ts_hour`, so a spec label says which transform each field applies. The source column stays `hour_utc`. The day field is a transform of that column, and the hour field is another, so a reader's filter on the column keeps working across both eras. Example `day-then-hour-spec` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Partition an Iceberg table by day, then change the spec to hour and load three more hours. Each partition spec is labelled in the order it was made. Every data file keeps the spec it was written under. """ from collections import Counter from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.transforms import DayTransform, HourTransform from lib.lakehouse import create_iceberg_table, new_warehouse, read_cut_hour, safe_error, scan_arrow def main(): try: with new_warehouse() as wh: day_spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name="ts_day")) table = create_iceberg_table(wh, partition_spec=day_spec) for hour in range(3): table.append(read_cut_hour(hour)) with table.update_spec() as update: update.remove_field("ts_day") update.add_field("hour_utc", HourTransform(), "ts_hour") for hour in range(3, 6): table.append(read_cut_hour(hour)) specs = sorted(table.specs()) labels = {spec_id: f"spec {number}" for number, spec_id in enumerate(specs, start=1)} for spec_id in specs: for field in table.specs()[spec_id].fields: print(f"{labels[spec_id]} field {field.name}: transform {field.transform}") files = Counter(table.inspect.data_files().column("spec_id").to_pylist()) for spec_id in specs: print(f"{labels[spec_id]} data files: {files[spec_id]}") print(f"rows in table: {scan_arrow(table).num_rows}") except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text spec 1 field ts_day: transform day spec 2 field ts_hour: transform hour spec 1 data files: 3 spec 2 data files: 3 rows in table: 2999 ``` The output has two specs. Spec 1 has one field, `ts_day`, with the day transform. Spec 2 has one field, `ts_hour`, with the hour transform. The table's rows are all there after the change: `2999` rows, which is the three hours before the change and the three after it. The spec change is the only step that touches the metadata alone. The loads on either side of it write data files as usual. [The next subsection](#files-written-under-each-spec) shows which spec each file belongs to. A mistake here shows up later, not in the change itself. If the spec list is wrong, every load after the change writes files under the wrong partitioning, and the counts from the next section are the first place you see it. Checking the spec list right after the change costs one line of output and saves a rewrite. **Working rule:** In Iceberg, make the spec change in one `update_spec()` block, so the old field is removed in the same change that adds the replacement. Split the two edits only if a step has to run between them; this chapter has no such step. Check the spec list after the change, and check that the field names and transforms are the ones you meant before you load any data under the second spec. ### Files written under each spec {#files-written-under-each-spec} Each data file belongs to the spec it was written under, so counting files by spec tells you which era each file is from. The example reads the spec ID of every data file, then prints the count for each spec, under the labels "spec 1" and "spec 2". The spec IDs themselves do not appear in the output, so it reads the same on every run. The output shows `3` data files under spec 1 and `3` under spec 2. The three day-partitioned loads kept their files through the change, and the three hour-partitioned loads wrote their own. The spec change itself wrote nothing. If the count under spec 1 had dropped after the change, something would have rewritten those files, and you would need to find out what before you trusted the old era's data. **Working rule:** After any spec change, count the data files under each spec and compare the counts with the loads you made. A count that moves when no rewrite was planned means the change did more than you wanted, so stop and check before the next load. ## Check what a query scans {#check-what-a-query-scans} A partition spec is a way to skip files. Before a query reads a file, the planner looks at the file's partition values and its column bounds, and drops files that cannot match the filter. Your query is only as cheap as the files it does not plan. So the way to check a change is to count the files a typical filter plans, before and after the change, rather than to compare the rows. Planning a scan returns the list of files without reading their rows, so the count is cheap enough to run before every change. The example plans three scans. Two filters each ask for one hour: hour 01 of the case's day, which was loaded before the change, and hour 04, which was loaded after it. The third has no filter, which is a full scan. It counts the planned files for each, before the spec change from day to hour partitions (see [moving to hourly partitions](#daily-to-hourly-partitions)) and after the three hour-partitioned loads. Example `plan-files-by-partition` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Count the files an Iceberg scan plans for two hours and for the whole table, before and after a spec change. The filters are hour 01 of the case's day, loaded before the change, and hour 04, loaded after it. The spec changes from day to hour between the two counts. """ from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.transforms import DayTransform, HourTransform from lib.lakehouse import create_iceberg_table, hour_filter, new_warehouse, read_cut_hour, safe_error def plan_counts(table): """Return the files planned for hour 01, for hour 04 and for no filter, in that order.""" one_hour = len(table.scan(row_filter=hour_filter(1)).plan_files()) hour_four = len(table.scan(row_filter=hour_filter(4)).plan_files()) everything = len(table.scan().plan_files()) return one_hour, hour_four, everything def main(): try: with new_warehouse() as wh: day_spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name="ts_day")) table = create_iceberg_table(wh, partition_spec=day_spec) for hour in range(3): table.append(read_cut_hour(hour)) one_hour, hour_four, everything = plan_counts(table) print(f"before the change, one-hour filter plans: {one_hour}") print(f"before the change, hour 04 filter plans: {hour_four}") print(f"before the change, no filter plans: {everything}") with table.update_spec() as update: update.remove_field("ts_day") update.add_field("hour_utc", HourTransform(), "ts_hour") for hour in range(3, 6): table.append(read_cut_hour(hour)) one_hour, hour_four, everything = plan_counts(table) print(f"after the change, one-hour filter plans: {one_hour}") print(f"after the change, hour 04 filter plans: {hour_four}") print(f"after the change, no filter plans: {everything}") except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text before the change, one-hour filter plans: 1 before the change, hour 04 filter plans: 0 before the change, no filter plans: 3 after the change, one-hour filter plans: 1 after the change, hour 04 filter plans: 1 after the change, no filter plans: 6 ``` | Stage | One-hour filter | Hour 04 filter | No filter | | --- | --- | --- | --- | | Before the change, three day-partitioned loads | 1 | 0 | 3 | | After the change, three hour-partitioned loads added | 1 | 1 | 6 | The one-hour filter plans 1 file before the change and 1 file after it. Hour 01 was loaded before the change, so the one file in both plans is that hour's day-partitioned file. Before the change, all three files share one day, so the planner can skip the other two only by their `hour_utc` bounds. After the change, the three hour-partitioned files are skipped as well. The hour 04 filter runs the other way. It plans 0 files before the change, because no day-partitioned file holds hour 04, and 1 file after it, the hour-partitioned file that holds that hour. The example does not show which part of the plan skips a file, by partition values or by column bounds, so the counts show what the plan does, not how it does it. The useful result is that each filter reaches the file of its own hour, in whichever era that hour was loaded. A plan count is a count of files, not of bytes. Two plans with the same count can read different amounts of data, because files differ in size. The count is still the right first check, because it is the number the planner decides on. When the counts agree and the query is still slow, look at file sizes and at the query itself. The no-filter plan grows from 3 files to 6. Those are the day-partitioned files plus the hour-partitioned ones. A full scan reads both eras, and it pays for every file the table has written. That is the cost of keeping old files in place, and it is the reason to run the full scan's count as well as the filtered ones. **Working rule:** Plan the filters your readers use, and the full scan, before and after every partition change, and keep the counts with the runbook entry. If a filter that planned one file starts to plan many, the change has broken the pruning for that filter, and you should change the spec or the filter before the next load under it. ## Partitioning in the other formats {#partitioning-in-other-formats} The same change looks different in the other two formats, and the difference is about the cost, not the result. In each format the rows read back are the same. The files and the plan are what change. The table below sums up the three. | Format | How the change is made | Files already written | What to check after | | --- | --- | --- | --- | | Iceberg | One `update_spec()` block | Kept, and still under their old spec | File count per spec; plan counts | | Delta Lake | Full overwrite with `schema_mode="overwrite"` | Written again in a later version; old version kept | Row count; file count; partition columns of the old version | | DuckLake | `ALTER TABLE ... SET PARTITIONED BY` | Kept; the setting applies to later loads | File count after one load; the partition key | **Delta Lake** partitions by plain columns. Its partition option takes column names, not expressions. The protocol defines `partitionColumns` as [an array containing the names of columns](https://raw.githubusercontent.com/delta-io/delta/v4.4.1/PROTOCOL.md) by which the data is partitioned. A generated column can compute `day` or `hour` inside the table, but it raises the table's writer protocol, so this book computes both in the writer before it writes. The example does that in `with_partition_columns`, which adds the two string columns from `hour_utc`. The table is first partitioned by `day`. A change to `hour` is refused unless the write is a full overwrite that also replaces the schema: ``` DeltaError: Generic error: Specified table partitioning does not match table partitioning: expected: ["day"], got: ["hour"]. To change partition columns, use full table overwrite with schema overwrite and no replaceWhere predicate. ``` The example first makes that write with `mode="overwrite"` alone, and prints the error. It then makes the same write with `schema_mode="overwrite"`, which is the setting the library's API at the pins uses in place of the older `overwrite_schema` flag. The rows are read into memory before the overwrite starts, because an overwrite replaces every row the table holds. Writing the rows back is the only way to keep them. Example `delta-partition-overwrite` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Change a Delta Lake table's partition columns from day to hour, which rewrites the table. The table holds three hourly loads with day and hour as plain columns. Delta numbers versions from 0. """ import pyarrow.compute as pc from deltalake import DeltaTable, write_deltalake from lib.lakehouse import delta_file_count, new_warehouse, read_cut_hour, read_delta, safe_error def with_partition_columns(rows): """Add the day and hour columns that Delta partitions by. Delta partitions only by plain columns.""" rows = rows.append_column("day", pc.strftime(rows["hour_utc"], format="%Y-%m-%d")) return rows.append_column("hour", pc.strftime(rows["hour_utc"], format="%H")) def main(): try: with new_warehouse() as wh: path = f"{wh.root}/delta/pageviews_hourly" for hour in range(3): write_deltalake(path, with_partition_columns(read_cut_hour(hour)), partition_by=["day"], mode="append") table = DeltaTable(path) print(f"partition columns before: {table.metadata().partition_columns}") print(f"version before: {table.version()}") print(f"data files before: {delta_file_count(table)}") rows = read_delta(path) try: write_deltalake(path, rows, partition_by=["hour"], mode="overwrite") except Exception as exc: print(f"change without schema mode: {safe_error(exc, wh)}") write_deltalake(path, rows, partition_by=["hour"], mode="overwrite", schema_mode="overwrite") table = DeltaTable(path) print(f"partition columns after: {table.metadata().partition_columns}") print(f"version after: {table.version()}") print(f"data files after: {delta_file_count(table)}") print(f"rows after: {read_delta(path).num_rows}") old = DeltaTable(path, version=2) print(f"partition columns in version 2: {old.metadata().partition_columns}") print(f"rows in version 2: {read_delta(path, version=2).num_rows}") except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text partition columns before: ['day'] version before: 2 data files before: 3 change without schema mode: DeltaError: Generic error: Specified table partitioning does not match table partitioning: expected: ["day"], got: ["hour"]. To change partition columns, use full table overwrite with schema overwrite and no replaceWhere predicate. partition columns after: ['hour'] version after: 3 data files after: 3 rows after: 1499 partition columns in version 2: ['day'] rows in version 2: 1499 ``` Delta counts versions from 0. The three loads are versions 0 to 2, so the table is at version 2 before the change. The rewrite makes version 3. It holds the same 1499 rows, in 3 files. That count matches the three hours, one file for each hour partition. Version 2 still reads back, with its `day` partition, and it holds 1499 rows. The example reads it with `read_delta`, which is the book's reader for Delta tables. The Delta change is a rewrite of the table's current files, not a metadata edit. The example's table is small. A large table needs the same rewrite, at its own cost, so plan it as a maintenance task with a read window. The example also reads every row into memory before it writes, so the rewrite needs room for the whole table at once. For a table that does not fit, a copy into a new location is the safer route, and [chapter 9](#check-copy-with-digest) shows how to check such a copy. **DuckLake** stores partitioning as a table setting in its catalog, and a change applies to later loads only. The [DuckLake documentation](https://ducklake.select/docs/stable/duckdb/advanced_features/partitioning) says the partitioning keys of a table affect only the data written after the change. The example sets the table to partition by `hour(hour_utc)`, loads one more hour, and counts the data files before and after that load. The helper attaches the catalog with inlining off, so each load writes a data file, and the file count shows what each load did. Example `ducklake-partition-alter` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Set a DuckLake table's partitioning to hour, then load one more hour and count the files each load wrote. DuckLake applies the new partitioning to new data only. Inlining is off, so every load is a data file. """ from lib.lakehouse import (DUCKLAKE_ALIAS, NAMESPACE, TABLE, connect_ducklake, create_ducklake_table, ducklake_file_count, ducklake_file_paths, new_warehouse, read_cut_hour, safe_error) def load(con, hour): con.register("batch", read_cut_hour(hour)) con.execute(f"INSERT INTO {DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE} SELECT * FROM batch") con.unregister("batch") def main(): try: with new_warehouse() as wh: con = connect_ducklake(wh.con) create_ducklake_table(con, wh.root) for hour in range(3): load(con, hour) print(f"data files before the change: {ducklake_file_count(con, DUCKLAKE_ALIAS, TABLE)}") before = ducklake_file_paths(con, DUCKLAKE_ALIAS, TABLE) con.execute(f"ALTER TABLE {DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE} SET PARTITIONED BY (hour(hour_utc))") load(con, 3) print(f"data files after one load under hour: {ducklake_file_count(con, DUCKLAKE_ALIAS, TABLE)}") after = ducklake_file_paths(con, DUCKLAKE_ALIAS, TABLE) print(f"files kept from before the change: {len(before & after)}") for key, transform in con.execute( "SELECT partition_key_index, transform FROM __ducklake_metadata_lake.ducklake_partition_column ORDER BY 1" ).fetchall(): print(f"partition key {key}: transform {transform}") rows = con.execute(f"SELECT count(*) FROM {DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE}").fetchall() print(f"rows in table: {rows[0][0]}") except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text data files before the change: 3 data files after one load under hour: 4 files kept from before the change: 3 partition key 0: transform hour rows in table: 1999 ``` The table has 3 data files after the three loads, and 4 after the hour load. The catalog lists one partition key, with the transform `hour`. The table holds 1999 rows. The load after the change is one file. Of the files listed after that load, 3 were already listed before it, so the three earlier loads were not rewritten, which is what the documentation's rule means in practice. **Working rule:** In Iceberg and DuckLake, a partition change is a change to metadata. Old files stay in place and later loads use the partitioning the change set, so check the file counts after the first load. In Delta Lake, a partition change is a full overwrite that rewrites the table's current files, so plan it as a rewrite and check the row count and the file count after it. ## What goes wrong in production {#partition-production-failures} Five failures come up often enough to name. The first is a missing package. At the pins set out in [the section on moving to hourly partitions](#daily-to-hourly-partitions), the day and hour transforms need `pyiceberg-core`, and the transforms import it when they write. An environment that installs only `pyiceberg` can create the table and then fails at the first append under a time transform. You notice it at the first scheduled load, not when the spec is changed, which is why the pin belongs in the requirements file and not in someone's memory of the install. The second is over-partitioning. A partition grain finer than the filters need gives each query more files to open, and each file holds fewer rows. Hourly loads of a few hundred rows each give files that are small in absolute terms. This is a rule of thumb, not a measurement: the cost depends on your storage and your engine. The check is the plan count in [the section on what a query scans](#check-what-a-query-scans), which shows how many files a query opens under each spec. You notice the problem when the full-scan count keeps growing with every load while the rows per file stay small, and the fix is compaction, which [chapter 6](#compact-iceberg-by-rewriting) covers, or a coarser grain for the later loads. The third is a high-cardinality key. An identity partition on `page_title` makes one partition per title, and each load writes a file for every title it contains. The table here has a few thousand rows, but a real page table has many more titles than that, and the file count would follow the titles, not the time. Partition by time, which the filters use, and leave the titles as a column that queries filter on. The fourth is judging the change by its result. The rows are the same before and after a partition change, so a check that compares the rows passes whether or not the change did anything useful. The examples in this chapter print the rows only as a check that nothing was lost. The counts are what show the change. The fifth is a rewrite that runs against a table readers use. The example runs alone and does not test readers during the write. For a table that readers depend on, we recommend writing the rewrite to a copy, checking the copy's counts and rows, and then pointing readers at it. [Chapter 9](#check-copy-with-digest) uses the same check between formats. This recommendation is a judgement, not a measured result. **Working rule:** Choose the partition grain from the filters your readers use and the size of the files you expect, not from how often data arrives. Before you run a partition change on a table that serves readers, run it on a copy, and compare the plan counts and the row count on both sides. ## Decisions this chapter made {#partition-decisions} - We partition the Iceberg table by the day and hour transforms of `hour_utc`, and pin `pyiceberg-core` beside `pyiceberg`. The transforms keep the time column as the partition source, so a filter on time reads the same field the writer used. We recommend this for any table that filters on time. - We change the Delta Lake table by a full overwrite with `schema_mode="overwrite"`, and we write its partition columns as plain string columns, `day` and `hour`, computed by the writer. We accept the rewrite, because in this library a partition change needs a full overwrite. - We count files and plans through each library's own listing: `plan_files()` for Iceberg, `file_uris()` for Delta Lake and `ducklake_list_files` for DuckLake. The examples never print a file name, because the names carry a random ID that differs on every run. - We turn off DuckLake's inlining of small inserts for every example that counts files, so a count of files is a count of data files. - We choose the hour grain from the dashboards' filters, which read one hour at a time, and not from how often loads arrive. - We show the Delta failure first, then the fix, so the schema setting is explained by the error that asks for it. - We print the DuckLake partition key from the catalog's partition table, not from file names, so the output does not change when the file names do. - We keep the field and spec labels in the output and never print a spec ID, a snapshot ID or a file name, because each of those differs from one run to the next. # Correct rows with deletes, updates and merges By the end of this chapter you can delete, upsert and merge rows in Iceberg, Delta Lake and DuckLake, say what each write does to the files, and count the snapshots a pyiceberg upsert adds. The table so far only grows. Every write in [chapter 1](#append-each-hour) added rows, and this chapter changes rows that are already committed. Each example starts from a fresh warehouse, so each one runs on its own, and each one prints counts and labels that a reader can compare with the text. ## Late data arrives as a correction {#late-data-as-correction} The first delivery of hour 02 reached the table in [chapter 1](#append-each-hour) with the counts the source sent at the time. A later delivery of the same hour carries corrected counts for some pages, and one page that the first delivery left out. The pages in the correction are the same pages as before, so the table needs to change rows in place. Appending the corrected hour would store every page twice, once with each count. A row in this table is identified by its key: `hour_utc`, `domain_code` and `page_title` together. A correction is a set of rows with keys. For each key, the table either already holds a row, which should change, or it does not, which should be added. Writing a correction is therefore a question about keys, and the write answers it with one rule: a key must identify one row in the batch you write. pyiceberg and DuckLake both enforce that rule. This chapter does not test a repeated key in a deltalake merge. A correction usually arrives as the whole hour, re-delivered, and not as a few rows. Sending the whole hour keeps the batch self-contained, because the write decides row by row what changed. The cost is that every row of the hour is compared, not only the corrected ones. Because every row is compared, pyiceberg can count only the rows whose values differ, and a rerun of the same batch changes nothing. The examples build the correction in code. The helper `hour_corrections` halves the counts of 20 page titles in hour 02, rounding down. The late row is a copy of the hour's first row, with a page title that the first delivery never had. The pageviews rows are the real cut from [chapter 1](#land-three-hours). Only the correction batch is generated. **Working rule:** Before you write a correction, check that its batch holds each key once. pyiceberg and DuckLake refuse a batch with a repeated key, each in its own way, and a retry does not fix the data. The [production section](#corrections-production-failures) shows the failures. ## Delete rows, and what each format writes {#delete-rows-iceberg-and-delta} A delete removes the rows that match a predicate. The rows go away in every format, but the files that held them can be handled in two ways. In copy-on-write, the writer rewrites each data file that held a matching row and leaves out the deleted rows. In merge-on-read, the writer leaves the data file alone and writes a delete file that records which rows are gone. Readers apply the delete file when they scan. The two libraries in this chapter take the predicate in different forms. pyiceberg takes an expression object, built here from `EqualTo` and `LessThan`. deltalake takes a SQL string, which reads like the `WHERE` clause a reader already knows. The example uses both forms, so you can compare them on the same rows. The example deletes hour 02 rows with fewer than 180 views. The number is a choice of this chapter. The cut in [chapter 1](#land-three-hours) keeps only the most-viewed rows in each hour, so a predicate for fewer than two views matches nothing. pyiceberg warns with `UserWarning: Delete operation did not match any records` and writes no snapshot, as its [source](https://raw.githubusercontent.com/apache/iceberg-python/pyiceberg-0.12.0/pyiceberg/table/__init__.py) shows. A threshold of 180 catches a few of the least-viewed rows in hour 02, which is enough to show what a delete writes. Example `delete-low-views` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Delete the rows below 180 views from hour 02 in Iceberg, in a Delta Lake copy and in DuckLake.""" from datetime import datetime from deltalake import DeltaTable from pyiceberg.expressions import And, EqualTo, LessThan from lib.lakehouse import DUCKLAKE_ALIAS, NAMESPACE, TABLE, ducklake_file_count, load_three_formats, new_warehouse, read_delta, safe_error, scan_arrow HOUR_02 = datetime(2024, 6, 3, 2) LOW_VIEWS = 180 def main(): with new_warehouse() as wh: iceberg, delta_path, lake = load_three_formats(wh, range(6)) iceberg_before = scan_arrow(iceberg).num_rows delta_before = read_delta(delta_path).num_rows table = f"{DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE}" lake_before = lake.execute(f"SELECT count(*) FROM {table}").fetchall()[0][0] files_before = ducklake_file_count(lake, DUCKLAKE_ALIAS, TABLE) iceberg.delete(And(EqualTo("hour_utc", HOUR_02), LessThan("views", LOW_VIEWS))) summary = iceberg.current_snapshot().summary print("Iceberg deleted rows:", iceberg_before - scan_arrow(iceberg).num_rows) print("Iceberg remaining rows:", scan_arrow(iceberg).num_rows) print("Iceberg data files rewritten:", summary["deleted-data-files"]) delta = DeltaTable(delta_path) metrics = delta.delete(f"hour_utc = '2024-06-03T02:00:00' AND views < {LOW_VIEWS}") print("Delta Lake deleted rows:", delta_before - read_delta(delta_path).num_rows) print("Delta Lake remaining rows:", read_delta(delta_path).num_rows) print("Delta Lake data files rewritten:", metrics["num_removed_files"]) lake.execute(f"DELETE FROM {table} WHERE hour_utc = TIMESTAMP '2024-06-03 02:00:00' AND views < {LOW_VIEWS}") lake_after = lake.execute(f"SELECT count(*) FROM {table}").fetchall()[0][0] print("DuckLake deleted rows:", lake_before - lake_after) print("DuckLake remaining rows:", lake_after) print("DuckLake data files before:", files_before) print("DuckLake data files after:", ducklake_file_count(lake, DUCKLAKE_ALIAS, TABLE)) print("DuckLake delete files written:", lake.execute( f"SELECT count(*) FROM ducklake_list_files('{DUCKLAKE_ALIAS}', '{TABLE}', schema => '{NAMESPACE}') " "WHERE delete_file IS NOT NULL").fetchall()[0][0]) if __name__ == "__main__": try: main() except Exception as exc: # the redacted error is what a reader sees print(safe_error(exc)) raise SystemExit(1) ``` Output: ```text Iceberg deleted rows: 4 Iceberg remaining rows: 2995 Iceberg data files rewritten: 1 Delta Lake deleted rows: 4 Delta Lake remaining rows: 2995 Delta Lake data files rewritten: 1 DuckLake deleted rows: 4 DuckLake remaining rows: 2995 DuckLake data files before: 6 DuckLake data files after: 6 DuckLake delete files written: 1 ``` The example loads six hours into an Iceberg table, a Delta Lake table and a DuckLake table, then deletes the rows under the threshold from each. Iceberg's delete removes 4 rows and leaves 2995. Iceberg rewrote 1 data file, the one that held hour 02. The Delta Lake delete removes 4 rows, the same rows, and leaves 2995. It rewrote 1 file. Both formats rewrote the file for hour 02 and left the other five hours alone, because a file is the unit a copy-on-write delete replaces. A delete that touches many hours rewrites many files, so the cost of a delete depends on where its matching rows sit. The DuckLake `DELETE` removes 4 rows and leaves 2995. It rewrites no data file. The data files stay at 6 before the delete and 6 after it, and the delete writes 1 delete file. The cost of the Iceberg and Delta Lake deletes follows the size of the files they touch, not the number of rows they remove. 4 rows went, and 1 whole file was rewritten to remove them. A table whose matches are spread across many files pays for each of those files. Each library reports its own counts, and this example computes the deleted rows as the rows before minus the rows after, so the deleted and remaining counts agree by construction. The rewritten-file counts come from the library: the Iceberg snapshot summary's `deleted-data-files` and the `num_removed_files` metric that deltalake returns from `delete`. ### Copy-on-write and merge-on-read {#copy-on-write-and-merge-on-read} The two write styles trade cost between the writer and the reader. Copy-on-write makes the write expensive and the read cheap: a reader gets the rows as the files hold them, with nothing to apply. Merge-on-read makes the write cheap, because it writes only the positions of the deleted rows, and it moves the cost to reads until something rewrites the data files. The three formats in this chapter differ as follows. | Format | Library at the pins | How the library writes a delete at the pins | Shown in this chapter | | --- | --- | --- | --- | | Iceberg | pyiceberg 0.12.0 | Copy-on-write. `delete` rewrites the data files that contain matching rows and writes no delete files. With `write.delete.mode` set to merge-on-read, pyiceberg warns "Merge on read is not yet supported, falling back to copy-on-write" ([source](https://raw.githubusercontent.com/apache/iceberg-python/pyiceberg-0.12.0/pyiceberg/table/__init__.py)). | Yes, in the delete example | | Delta Lake | deltalake 1.6.6 | Copy-on-write. The file with matching rows is rewritten, and the log records the old file as removed. | Yes, in the delete example | | DuckLake | DuckDB 2.0.0 | Merge-on-read. `DELETE` writes a delete file and leaves the data file in place. | Yes, in the delete example | The DuckLake row is shown by the delete example, which counts the delete file and the data files that stay in place. The [DuckLake documentation on deletion vectors](https://ducklake.select/docs/stable/duckdb/advanced_features/deletion_vectors) describes an experimental option that records deleted row positions in `.puffin` files. That option is off by default, and this chapter does not turn it on. The cost of reading through delete files is not measured here. A reader of a merge-on-read table pays that cost on every scan until the data files are rewritten, so measure it on the reader's own table. Delta Lake also has deletion vectors, which record deleted row positions in a file. That makes them Delta's merge-on-read option, not a third write style, and this book does not use them. At the pins, deltalake 1.6.6 does not write a deletion vector: `delete` on a table with the feature enabled still rewrites the file. The [Delta protocol](https://raw.githubusercontent.com/delta-io/delta/v4.4.1/PROTOCOL.md) requires the deletion-vectors feature in a table's reader and writer features when the table uses deletion vectors, and enabling the feature adds it to both. `QueryBuilder`, which this book uses for every Delta read, reads such a table. `to_pyarrow_table` and `to_pyarrow_dataset` refuse it, because the table declares the feature. So this book keeps the feature off, and its Delta deletes are copy-on-write. **Working rule:** Expect copy-on-write deletes to cost a file rewrite for each file that holds a match. Prefer predicates whose matches sit in few files, such as one hour, and batch the deletes that you know about into one write. Merge-on-read suits a table where writes are frequent and reads can pay for the delete files until something rewrites them. ## Upsert a corrected hour {#upsert-corrected-hour} An upsert updates the row whose key matches and inserts the row whose key does not. It is the right write for a correction, because the corrected hour holds the keys that the table already has, and the late row holds a key the table does not have yet. A plain append would store the corrected rows next to the originals, so every page in the hour would appear twice. The next two subsections show pyiceberg's method and DuckLake's SQL `MERGE` with the same batch. The key finds the row that a change applies to, and it decides when two incoming rows are the same row. Too narrow a key collapses two real rows into one, and too wide a key inserts a second copy. The key here is `hour_utc`, `domain_code` and `page_title`, the identity the cut gives each row in [chapter 1](#land-three-hours). The batches in the examples hold each key once. ### Upserts in pyiceberg {#upserts-in-pyiceberg} pyiceberg's `upsert` takes an Arrow table of rows and the columns that form the key, passed as `join_cols`. It returns the count of rows it updated and the count it inserted. A row counts as updated only when a non-key column changed, so a row that is identical to the table is not counted as updated. Copy-on-write still rewrites such a row when its file also holds a changed row. Example `redeliver-hour-upsert` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Upsert the corrected hour 02 into the Iceberg table on its key, then upsert it again.""" import pyarrow as pa from lib.lakehouse import KEY, hour_corrections, load_three_formats, new_warehouse, read_cut_hour, safe_error, snapshot_labels def main(): with new_warehouse() as wh: iceberg, _, _ = load_three_formats(wh, range(6)) before = {snapshot.snapshot_id for snapshot in iceberg.snapshots()} corrected = hour_corrections(read_cut_hour(2)) # A late row: a page the first delivery of hour 02 did not have. late = corrected.slice(0, 1).set_column( corrected.schema.get_field_index("page_title"), "page_title", pa.array(["Late_arrival_example"], type=pa.string()), ) batch = pa.concat_tables([corrected, late]) result = iceberg.upsert(batch, join_cols=list(KEY)) print("rows updated:", result.rows_updated) print("rows inserted:", result.rows_inserted) iceberg.refresh() labels = snapshot_labels([snapshot.snapshot_id for snapshot in iceberg.snapshots()]) added = [snapshot for snapshot in iceberg.snapshots() if snapshot.snapshot_id not in before] for snapshot in added: label = labels[snapshot.snapshot_id] print(label, snapshot.summary.operation.value) print(f"{label} records added:", snapshot.summary.get("added-records") or 0) print(f"{label} records deleted:", snapshot.summary.get("deleted-records") or 0) print("snapshots added:", len(added)) again = iceberg.upsert(batch, join_cols=list(KEY)) print("rows updated on second upsert:", again.rows_updated) print("rows inserted on second upsert:", again.rows_inserted) if __name__ == "__main__": try: main() except Exception as exc: # the redacted error is what a reader sees print(safe_error(exc)) raise SystemExit(1) ``` Output: ```text rows updated: 20 rows inserted: 1 snapshot 7 overwrite snapshot 7 records added: 480 snapshot 7 records deleted: 500 snapshot 8 append snapshot 8 records added: 20 snapshot 8 records deleted: 0 snapshot 9 append snapshot 9 records added: 1 snapshot 9 records deleted: 0 snapshots added: 3 rows updated on second upsert: 0 rows inserted on second upsert: 0 ``` The example loads six hours, builds the corrected hour with the late row, and upserts it on the key. The result reports 20 updated rows and 1 inserted row. The snapshot list then shows what the call wrote: 3 snapshots, labelled by the operation each one records, with the records each one added and deleted. The first is an overwrite that removes the hour-02 file and writes back its unchanged rows. It deletes 500 rows and adds 480. The next two are appends. The second adds 20 changed rows, and the third adds 1 row, the late row. The labels come from `snapshot_labels`, so the numbers are the commit order, not the snapshot IDs, which change on every run. The count of snapshots is the part to remember. One `upsert` call is not one commit. In the [pyiceberg 0.12.0 source](https://raw.githubusercontent.com/apache/iceberg-python/pyiceberg-0.12.0/pyiceberg/table/__init__.py), `upsert` calls `overwrite` only when some rows changed, and `append` for the new rows only when there are some. `overwrite` writes two snapshots, a delete and an append. The library records the first as `overwrite`, because it also writes rows back. So a batch with changed and new rows adds three snapshots, one with only changed rows adds two, and one with only new rows adds one. Count them in the snapshot list, not from the library's call. The example runs the same upsert a second time and prints 0 updated rows and 0 inserted rows. The second run finds every key already at its corrected value, so both counts are zero. A correction that is sent twice does not change the table twice, and that is what makes a retry safe. **Working rule:** Use `upsert` with `join_cols` set to the key, and read its counts from the result. Count the snapshots from the snapshot list, and do not assume one commit per call. Run the same batch again when you are unsure whether the first write landed, and check that the second run reports no updates. ### MERGE in DuckLake {#merge-in-ducklake} DuckLake takes the same correction as a `MERGE` statement in SQL. The statement names the target table, the source batch and the key in its `ON` clause. Each `WHEN` clause says what to do with a matched or unmatched row. The statement below is the one the example runs. ```sql MERGE INTO lake.wiki.pageviews_hourly AS t USING batch AS s ON t.hour_utc = s.hour_utc AND t.domain_code = s.domain_code AND t.page_title = s.page_title WHEN MATCHED AND t.views <> s.views THEN UPDATE SET views = s.views WHEN NOT MATCHED THEN INSERT (hour_utc, domain_code, page_title, views, response_bytes) VALUES (s.hour_utc, s.domain_code, s.page_title, s.views, s.response_bytes) ``` The `WHEN NOT MATCHED` branch also inserts `response_bytes`, which is 0 in every row of the cut, so the late row's value is 0 too. The `AND t.views <> s.views` condition matters. Without it, the `MERGE` would update and count every matched row, so a rerun would count them all again. With it, a matched row whose counts already agree is left alone and not counted. The examples do not run the statement without the condition, so the first half of that sentence is a reading of the statement, not a measured count. Example `merge-correction-ducklake` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """MERGE the corrected hour 02, and its late row, into the DuckLake copy, then run the same MERGE again.""" import pyarrow as pa from lib.lakehouse import hour_corrections, load_three_formats, new_warehouse, read_cut_hour, safe_error MERGE = """ MERGE INTO lake.wiki.pageviews_hourly AS t USING batch AS s ON t.hour_utc = s.hour_utc AND t.domain_code = s.domain_code AND t.page_title = s.page_title WHEN MATCHED AND t.views <> s.views THEN UPDATE SET views = s.views WHEN NOT MATCHED THEN INSERT (hour_utc, domain_code, page_title, views, response_bytes) VALUES (s.hour_utc, s.domain_code, s.page_title, s.views, s.response_bytes) """ def main(): with new_warehouse() as wh: _, _, lake = load_three_formats(wh, range(6)) corrected = hour_corrections(read_cut_hour(2)) late = corrected.slice(0, 1).set_column( corrected.schema.get_field_index("page_title"), "page_title", pa.array(["Late_arrival_example"], type=pa.string()), ) lake.register("batch", pa.concat_tables([corrected, late])) print("source rows:", lake.execute("SELECT count(*) FROM batch").fetchall()[0][0]) matched = lake.execute( "SELECT count(*) FROM lake.wiki.pageviews_hourly AS t JOIN batch AS s " "ON t.hour_utc = s.hour_utc AND t.domain_code = s.domain_code AND t.page_title = s.page_title" ).fetchall()[0][0] print("source rows matched:", matched) print("rows changed by the merge:", lake.execute(MERGE).fetchall()[0][0]) print("rows changed by the same merge again:", lake.execute(MERGE).fetchall()[0][0]) print("rows in table:", lake.execute("SELECT count(*) FROM lake.wiki.pageviews_hourly").fetchall()[0][0]) if __name__ == "__main__": try: main() except Exception as exc: # the redacted error is what a reader sees print(safe_error(exc)) raise SystemExit(1) ``` Output: ```text source rows: 501 source rows matched: 500 rows changed by the merge: 21 rows changed by the same merge again: 0 rows in table: 3000 ``` The example registers the corrected hour and the late row as a batch of 501 rows, then runs the merge on the DuckLake copy of the table. Of those, 500 match a key already in the table, so the one that does not match is the late row. The merge reports 21 rows changed. The example then runs the same statement again, and it reports 0 changed rows. The table ends with 3000 rows: the six hours and the late row. The changed count adds the two kinds of change together, so the split needs the source size. The rows inserted are the source rows minus the matched rows, here 501 minus 500. The rows updated are the changed rows minus the rows inserted. A count of matched rows alone gives neither number. The merge runs on the DuckLake copy. The examples attach DuckLake with inlining off, as [chapter 1](#files-metadata-and-commit) explains, so each load writes data files rather than rows in the catalog. **Working rule:** Write a `MERGE` with a `WHEN MATCHED AND` condition that compares the columns you update, so that a rerun changes nothing. Make sure the source holds each key once before you run it. DuckLake refuses a merge that matches one target row to two source rows, and the [production section](#corrections-production-failures) shows the error. ## Merge a changed label into the domains table {#merge-domains-table} The second table in this book, `wiki.domains`, maps each domain code to a language. A dimension table holds labels, not counts, and a merge suits a few labels that change at once. The point is that the same statement shape works on a dimension table, and the [rule about conditions](#merge-in-ducklake) applies there too. An `UPDATE` would change the label. The `MERGE` shape also has a `WHEN NOT MATCHED` clause, which inserts a domain code the table does not hold yet, so a feed that adds a domain can use the same statement. A rerun that finds the label already set changes nothing. This table lives only in DuckLake, so the merge runs there. The pageviews table is merged in DuckLake in [MERGE in DuckLake](#merge-in-ducklake). In a pipeline, the label you merge here is the label that [chapter 8's join](#join-tables-from-two-catalogs) returns. Example `merge-domains-label` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """MERGE a changed language label into the domains table in DuckLake, then run the same MERGE again.""" import pyarrow as pa from lib.lakehouse import attach_ducklake, connect_ducklake, print_rows, read_domains, new_warehouse, safe_error MERGE = """ MERGE INTO lake.wiki.domains AS t USING labels AS s ON t.domain_code = s.domain_code WHEN MATCHED AND t.language <> s.language THEN UPDATE SET language = s.language WHEN NOT MATCHED THEN INSERT (domain_code, language) VALUES (s.domain_code, s.language) """ def main(): with new_warehouse() as wh: lake = connect_ducklake(wh.con) attach_ducklake(lake, "lake", wh.root, inlining_row_limit=0) lake.execute("CREATE SCHEMA lake.wiki") lake.execute("CREATE TABLE lake.wiki.domains (domain_code VARCHAR, language VARCHAR)") lake.register("dims", read_domains()) lake.execute("INSERT INTO lake.wiki.domains SELECT * FROM dims") lake.register("labels", pa.table({"domain_code": ["fr"], "language": ["French (France)"]})) print("rows changed by the merge:", lake.execute(MERGE).fetchall()[0][0]) print("rows changed by the same merge again:", lake.execute(MERGE).fetchall()[0][0]) print_rows(lake.execute("SELECT domain_code, language FROM lake.wiki.domains").fetchall()) if __name__ == "__main__": try: main() except Exception as exc: # the redacted error is what a reader sees print(safe_error(exc)) raise SystemExit(1) ``` Output: ```text rows changed by the merge: 1 rows changed by the same merge again: 0 de German en English fr French (France) ja Japanese ``` The example creates the domains table in DuckLake, loads the four rows from the book's own data, and merges one changed label. The label for `fr` changes from `French` to `French (France)`. The merge reports 1 changed row. A second run of the same merge reports 0. The sorted table follows the output, with one row per domain code. **Working rule:** Merge a dimension table with the same `WHEN MATCHED AND` condition as a fact table, and run the merge a second time to check that it changes nothing. A label change reaches every query that joins to the dimension, so review the label you are about to merge. ## What goes wrong in production {#corrections-production-failures} Three things go wrong with corrections in practice: a repeated key, a loop of single-row deletes, and a snapshot count that surprises whoever reads the table's history. Each one is cheap to check before the write and expensive to find afterwards, so the checks belong in the job that writes the batch. **A repeated key in the batch.** A correction sent twice, or two rows with the same key from a bad join, makes the batch invalid. The example sends one corrected row twice. pyiceberg refuses the upsert before it writes anything: Example `duplicate-source-keys` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Show what an upsert or a MERGE does when the batch has two rows for one key.""" import pyarrow as pa from lib.lakehouse import KEY, hour_corrections, load_three_formats, new_warehouse, read_cut_hour, safe_error, scan_arrow MERGE = """ MERGE INTO lake.wiki.pageviews_hourly AS t USING batch AS s ON t.hour_utc = s.hour_utc AND t.domain_code = s.domain_code AND t.page_title = s.page_title WHEN MATCHED AND t.views <> s.views THEN UPDATE SET views = s.views WHEN NOT MATCHED THEN INSERT (hour_utc, domain_code, page_title, views, response_bytes) VALUES (s.hour_utc, s.domain_code, s.page_title, s.views, s.response_bytes) """ def main(): with new_warehouse() as wh: iceberg, _, lake = load_three_formats(wh, range(6)) corrected = hour_corrections(read_cut_hour(2)) # A corrected row, sent twice: the batch holds two rows with one key. before = read_cut_hour(2)["views"].to_pylist() changed = [i for i, after in enumerate(corrected["views"].to_pylist()) if after != before[i]] doubled = pa.concat_tables([corrected, corrected.slice(changed[0], 1)]) try: iceberg.upsert(doubled, join_cols=list(KEY)) print("Iceberg upsert: no error") except Exception as exc: print("Iceberg upsert:", safe_error(exc, wh)) print("Iceberg rows after the failed upsert:", scan_arrow(iceberg).num_rows) lake.register("batch", doubled) try: lake.execute(MERGE) print("DuckLake merge: no error") except Exception as exc: print("DuckLake merge:", safe_error(exc, wh)) print("DuckLake rows after the failed merge:", lake.execute("SELECT count(*) FROM lake.wiki.pageviews_hourly").fetchall()[0][0]) if __name__ == "__main__": try: main() except Exception as exc: # the redacted error is what a reader sees print(safe_error(exc)) raise SystemExit(1) ``` Output: ```text Iceberg upsert: ValueError: Duplicate rows found in source dataset based on the key columns. No upsert executed Iceberg rows after the failed upsert: 2999 DuckLake merge: InvalidInputException: Invalid Input Error: MERGE INTO command cannot affect the same target row more than once. A target row matched more than one source row; ensure the source rows are deduplicated or the ON condition is unique. DuckLake rows after the failed merge: 2999 ``` The output shows the refusal and the table's row count after it. The Iceberg upsert raises a `ValueError` and the table still holds 2999 rows. The DuckLake merge raises an `InvalidInputException` that names the cause, a target row that matched more than one source row, and the table still holds 2999 rows. In both runs the table kept every row it held before the write. A pipeline that catches the error and retries the same batch fails again, because the batch is the problem. Deduplicate the source by key before the write. This chapter does not test a Delta Lake merge with repeated keys, so it makes no claim about how deltalake handles one. **A loop of single-row deletes.** Copy-on-write makes each delete a commit that rewrites the files that hold its matches. Ten deletes in a loop rewrite the file that holds their matches ten times and leave ten snapshots or versions behind. This is a rule of thumb from the cost model, not a measurement in this chapter. Collect the rows to delete, then write one delete with a predicate that covers them. **A snapshot count that surprises the reader.** A single upsert with changed rows and new rows adds 3 snapshots in the [run above](#upserts-in-pyiceberg). A reader who expects one commit per change will find the history longer than the change log. The snapshots stay in the table's metadata until they are expired, which [chapter 7](#snapshots-keep-files-alive) covers, and each one keeps the files it refers to. **How you'd notice.** Each of these shows up in a number you already log. The rows-updated and rows-inserted counts from pyiceberg's result, and the changed-rows count from a merge, should match the batch you sent. A rerun that reports changes means the batch differs from what you think it is. A snapshot count that grows faster than the snapshots you expect per load (up to three for an upsert) points at a non-idempotent retry, such as an append sent twice. The refusals show up as exceptions, so a job that catches them and continues will look healthy while its table stays behind. **Working rule:** Check the batch for repeated keys before any write, and treat a refusal as a bug in the batch, not a transient fault. Gather deletes into one write. Count snapshots from the snapshot list when you audit a correction. ## Decisions this chapter made {#corrections-decisions} - Corrections are matched on the key `hour_utc`, `domain_code` and `page_title`, and a batch must hold each key once before it is written. - The delete example uses a threshold of 180 views, because the cut keeps only the most-viewed rows in each hour, so a threshold of two matches nothing. - Iceberg and Delta Lake deletes are copy-on-write, and their examples show the files they rewrite. DuckLake `DELETE` writes a delete file and leaves the data files in place, and the same example shows that. - Deletion vectors are not used. At the pins, deltalake 1.6.6 does not write them, and enabling the feature makes `to_pyarrow_table` and `to_pyarrow_dataset` refuse the table. - Upserts and merges are checked by running them twice: the second run changes no rows. - Snapshot counts are read from the snapshot list. The chapter does not treat one call as one commit. - The DuckLake examples turn inlining off, so that the counts describe data files. - The corrections are built in code, and the late row is generated. The examples show the write path, not the history of a real correction. - The duplicate-key example shows the refusal itself, not a description of it, so the failure is in the output a reader sees. - The chapter has no Delta Lake upsert or merge example, so it makes no claim about deltalake's merge beyond the delete. - The examples keep the late row and the correction in the same batch, so one call shows both an update and an insert. # Compact the small files that hourly appends leave By the end of this chapter you can count the data files a table holds, merge them into fewer files in Delta Lake, DuckLake and Iceberg, and say what each merge leaves on disk. This chapter works on the pageviews table from the [first chapter](#first-table), loaded once an hour into each of the three formats. Each load is one commit, and each commit writes its own data files. A data file is a Parquet file that holds rows of the table. ## Hourly appends leave small files {#hourly-appends-leave-small-files} Every commit that appends rows writes at least one data file. A table that takes one load an hour gets 24 data files a day, and each file holds one hour of rows, about 500 of them here. A query has to open every file its filter does not skip, and each open costs time. The table's metadata also grows with every file it lists, because each format keeps a record of every data file that the current version uses. Many small files therefore cost reads and planning even when the total number of rows is modest. The count of files is the number to watch, because it is the one that grows with time. The size of each file matters too, but this book prints no file sizes, since a size in bytes changes from one run to the next. Counting files is also the honest check on a compaction: the rows should be the same before and after, and the count should drop. The fix is compaction: rewriting many small data files into fewer, larger ones. A compaction is a commit like any other. It writes the replacement files, lists them in the metadata, and leaves the old files on disk for as long as the table's history still refers to them. [Chapter 7](#expiry-and-cleanup) covers when those old files can go. The three formats have three different calls for it. The table below shows what each call did to the six hourly files that this chapter builds. | Format | Call | Files before | Files after | | --- | --- | --- | --- | | Delta Lake | `optimize.compact()` | 6 | 1 | | DuckLake | `ducklake_merge_adjacent_files` | 6 | 1 | | Iceberg | `table.overwrite(table.scan().to_arrow_batch_reader())` | 6 | 1 | The counts in the table are the ones each example printed. Each example loads the six hours of the cut, one load per hour, into a table with no partitions, so each format starts with one data file per load. The partitioned tables from [chapter 4](#partition-evolution) would behave differently, because each partition is compacted on its own in the formats that support it. This chapter keeps the table unpartitioned so that the count has one cause. DuckLake needs one extra sentence before its numbers mean anything. DuckLake keeps inserts of 10 rows or fewer in its catalog by default, a feature it calls data inlining, so those rows write no data file at all ([DuckLake documentation: Data Inlining](https://ducklake.select/docs/stable/duckdb/advanced_features/data_inlining)). The hourly loads here have about 500 rows each, so the default would write them as files anyway. The DuckLake examples still attach with inlining off, so that a file count never depends on how many rows a load happens to have. **Working rule:** Count the data files of each table after each load, and schedule compaction from the rate at which files arrive and the file size your readers need. A load every hour gives 24 files a day per table, so a daily compaction is a reasonable first choice. This is a rule of thumb, not a measurement. Compact more often only when your own read times or file counts show that you need to. ## Compact with optimize.compact() {#compact-in-delta-lake} Delta Lake is the first of the three formats in the table above. Its Python library, `deltalake`, puts compaction on the table's `optimize` property. Its `compact()` method rewrites the table's small data files into larger ones and returns a dictionary of metrics. The dictionary holds counts of the files added and removed, and size fields that this book does not print. The example below loads the six hours into a Delta table, one append each, then counts the data files the table lists, compacts, and counts again. Each count comes from the table's own list of files for its current version. Example `file-counts-before-after` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Compact six small Delta Lake appends into one file, and count files before and after.""" from deltalake import DeltaTable, write_deltalake from lib.lakehouse import delta_file_count, new_warehouse, read_cut_hour, read_delta def main(): with new_warehouse() as wh: path = str(wh.root / "delta" / "pageviews_hourly") for hour in range(6): write_deltalake(path, read_cut_hour(hour), mode="append") table = DeltaTable(path) version_before = table.version() print(f"files before compaction: {delta_file_count(table)}") metrics = table.optimize.compact() print(f"files added by compaction: {metrics['numFilesAdded']}") print(f"files removed by compaction: {metrics['numFilesRemoved']}") table = DeltaTable(path) print(f"files after compaction: {delta_file_count(table)}") print(f"rows after compaction: {len(read_delta(path))}") before = read_delta(path, version=version_before) print(f"rows at the version before compaction: {len(before)}") parquet = list((wh.root / "delta").rglob("*.parquet")) print(f"parquet files on disk after compaction: {len(parquet)}") if __name__ == "__main__": main() ``` Output: ```text files before compaction: 6 files added by compaction: 1 files removed by compaction: 6 files after compaction: 1 rows after compaction: 2999 rows at the version before compaction: 2999 parquet files on disk after compaction: 7 ``` Before the call, the table lists 6 data files. The metrics say what the call did: 1 file added and 6 files removed. After it, the table lists 1 file, and a read of the current version returns 2999 rows. Those are the same rows as before, in fewer files. The compaction did not remove the old files. The example reads the version that existed before the call, and that version still returns all 2999 rows. The version still points at the old files, so they must stay on disk. The directory holds 7 Parquet files after the call: the six old ones and the new one. Delta Lake removes files that no retained version uses in a separate step called `vacuum`, which [chapter 7](#expiry-and-cleanup) shows. Until that step runs, a compaction adds files on disk rather than taking any away. The example reads through the library's query builder, as every Delta read in this book does. Its purpose is the file count, so it checks the rows by their count only. **Working rule:** Compact a Delta table on the schedule from the first section (daily for hourly loads), and check the file count afterwards. Read the retention rules in chapter 7 before you remove anything: a compaction adds a commit and leaves the old files in place, so it adds storage until the vacuum runs. ## Compact in DuckLake {#compact-in-ducklake} DuckLake has compaction as a maintenance function that you call from SQL. The function is one call, and it returns a row that says what it did for each table it merges. The next part describes the function and the one case it skips. ### Merge adjacent files {#merge-adjacent-files} `ducklake_merge_adjacent_files` takes a catalog, and optionally a table name and a schema name, which the call needs when the table is not in the `main` schema. Without a table name it merges the files of every table. The DuckLake documentation says that the function can "merge multiple Parquet files into a single Parquet file that holds data inserted by multiple snapshots", and it adds that the merge works "without needing to expire snapshots" and "does not immediately delete the old files" ([DuckLake documentation: Merge Files](https://ducklake.select/docs/stable/duckdb/maintenance/merge_adjacent_files)). The last two points matter for the rest of this chapter. Because the merge does not expire snapshots, every older snapshot still refers to the old files. The example below attaches the DuckLake catalog with inlining off, inserts the six hours one insert each, and calls the merge once. The table lives in the `wiki` schema, so the call names it. The example reads the snapshot number before the merge, so it can read the table as it was at that point. Example `compact-ducklake` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Merge six small DuckLake data files into one, and read a snapshot from before the merge.""" from lib.lakehouse import ( DUCKLAKE_ALIAS, NAMESPACE, TABLE, connect_ducklake, create_ducklake_table, ducklake_file_count, new_warehouse, read_cut_hour, ) def main(): with new_warehouse() as wh: lake = connect_ducklake(wh.con) create_ducklake_table(lake, wh.root) table = f"{DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE}" for hour in range(6): lake.register("hour", read_cut_hour(hour)) lake.execute(f"INSERT INTO {table} SELECT * FROM hour") lake.unregister("hour") before = lake.execute(f"SELECT max(snapshot_id) FROM ducklake_snapshots('{DUCKLAKE_ALIAS}')").fetchall()[0][0] print(f"data files before merge: {ducklake_file_count(lake, DUCKLAKE_ALIAS, TABLE)}") result = lake.execute( f"CALL ducklake_merge_adjacent_files('{DUCKLAKE_ALIAS}', '{TABLE}', schema => '{NAMESPACE}')" ).fetchall() print(f"files merged: {result[0][2]}") print(f"files created: {result[0][3]}") print(f"data files after merge: {ducklake_file_count(lake, DUCKLAKE_ALIAS, TABLE)}") print(f"rows after merge: {lake.execute(f'SELECT count(*) FROM {table}').fetchall()[0][0]}") old = lake.execute(f"SELECT count(*) FROM {table} AT (VERSION => {before})").fetchall()[0][0] print(f"rows at the snapshot before merge: {old}") parquet = list((wh.root / f"{DUCKLAKE_ALIAS}_data").rglob("*.parquet")) print(f"parquet files on disk after merge: {len(parquet)}") if __name__ == "__main__": main() ``` Output: ```text data files before merge: 6 files merged: 6 files created: 1 data files after merge: 1 rows after merge: 2999 rows at the snapshot before merge: 2999 parquet files on disk after merge: 7 ``` The call returns one row with four values: the schema, the table, the number of files merged and the number of files created. The first two values name what the call worked on, so a merge aimed at the wrong table shows in the output before anything else is read. Here the row reports 6 files merged and 1 file created. The data-file list goes from 6 files to 1. The old snapshot still reads. A count taken at the snapshot before the merge returns 2999 rows, the same as the current table. The Parquet files on disk number 7: the six files the inserts wrote, and the one the merge wrote. The six old files stay on disk until the cleanup in [chapter 7](#expiry-and-cleanup) removes them. The merge skips a file that has a delete file attached. A delete file is a record of deleted rows, which DuckLake writes instead of rewriting the data file when rows are deleted. [Chapter 5](#corrections-and-merges) writes delete files in DuckLake, so this limit applies there. The DuckLake documentation names `ducklake_rewrite_data_files` as the call that rewrites those files, and its default `delete_threshold` is 0.95, which rewrites a file only when that share of its rows is deleted ([DuckLake documentation: Rewrite Files with Deletes](https://ducklake.select/docs/stable/duckdb/maintenance/rewrite_data_files)). The default is a value you should choose for your own data, not accept without a look. **Working rule:** Run `ducklake_merge_adjacent_files` after the inserts for a period are done, and read the file count it returns. A merge does not remove deleted rows, so for a table with delete files, call `ducklake_rewrite_data_files` with a `delete_threshold` that you chose for that table, and check the result. ## Rewrite the table to compact it {#compact-iceberg-by-rewriting} Compaction in Iceberg, at the pinned version of pyiceberg, is a rewrite by overwrite. This chapter rewrites the whole table, which is the case for an unpartitioned table. You read the table with a scan and overwrite the table with the rows it returns. The overwrite writes the rows as replacement data files, and the old files drop out of the current snapshot. This is the only compaction path in the book for Iceberg, and it is a workaround for a missing call, which the next part explains. The example below loads the six hours with six appends, counts the data files in the current snapshot, overwrites the table with its own scan, read as a stream of record batches, and counts again. Example `iceberg-rewrite-by-overwrite` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Rewrite an Iceberg table by overwriting it with its own scan, and count data files before and after.""" from lib.lakehouse import ( create_iceberg_table, iceberg_file_count, new_warehouse, read_cut_hour, scan_arrow, snapshot_labels, ) def main(): with new_warehouse() as wh: table = create_iceberg_table(wh) for hour in range(6): table.append(read_cut_hour(hour)) last_append = table.current_snapshot().snapshot_id print(f"data files before rewrite: {iceberg_file_count(table)}") table.overwrite(table.scan().to_arrow_batch_reader()) print(f"data files after rewrite: {iceberg_file_count(table)}") print(f"rows after rewrite: {len(scan_arrow(table))}") snapshots = sorted(table.snapshots(), key=lambda snapshot: snapshot.sequence_number) labels = snapshot_labels([snapshot.snapshot_id for snapshot in snapshots]) operations = [snapshot.summary.operation.value for snapshot in snapshots] print("snapshot operations in order: " + ", ".join(operations)) print(f"{labels[last_append]} rows: {len(scan_arrow(table, snapshot_id=last_append))}") parquet = list((wh.root / "warehouse").rglob("*.parquet")) print(f"parquet files on disk after rewrite: {len(parquet)}") if __name__ == "__main__": main() ``` Output: ```text data files before rewrite: 6 data files after rewrite: 1 rows after rewrite: 2999 snapshot operations in order: append, append, append, append, append, append, delete, append snapshot 6 rows: 2999 parquet files on disk after rewrite: 7 ``` The counts go from 6 to 1. A reader of the table sees the same rows, read the same way, from fewer files. The rewrite is one commit that adds two snapshots. The operations, in commit order, are six appends and then `delete` and `append`: the `delete` snapshot removes every data file from the table, and the `append` snapshot adds the rewritten files. The current snapshot moves once, so no reader sees the table empty, but the history keeps both snapshots. Deleting whole files in this way is the copy-on-write behaviour that [chapter 5](#corrections-and-merges) describes for Iceberg deletes. The snapshot from before the rewrite still reads every row. The example labels snapshots in commit order, so snapshot 6 is the last append before the rewrite. Read at that snapshot, the table still returns 2999 rows. The Parquet files on disk number 7: the six old files and the one the rewrite wrote. [Chapter 7](#expiry-and-cleanup) covers how to expire the old snapshots and count what they still hold. **Working rule:** Rewrite an Iceberg table by overwrite only when no other writer commits to the table during the rewrite. On an unpartitioned table, stream the scan as this example does. On a partitioned table, the scan must be read with `to_arrow()`, so the whole scan has to fit in the memory of the job. A streamed reader can be read only once: if the commit fails, build a new reader for the next attempt. Check the file count and the row count after each rewrite, and keep the snapshot from before it until you have checked the result. ### Why pyiceberg has no rewrite call {#why-no-rewrite-call} The pyiceberg API documentation lists `overwrite` among its table operations ([PyIceberg API documentation](https://py.iceberg.apache.org/api/)). Its one sentence on compaction is in the note on fast appends, and it is about metadata: the page says that metadata compaction is planned. It says nothing about data files. Its only maintenance call is `expire_snapshots`, which chapter 7 covers. The source of pyiceberg 0.12.0, in `pyiceberg/table/__init__.py`, has no method that compacts or rewrites data files. Its write operations are `append`, `overwrite`, `delete`, `upsert`, `dynamic_partition_overwrite` and `add_files`. The others write data files. `add_files` registers files that already exist and writes none. `dynamic_partition_overwrite`, and `overwrite` with an `overwrite_filter`, can replace the rows of one partition, so a partitioned table can be rewritten one partition at a time. A compaction is therefore a rewrite built from those operations. The rewrite has a memory cost that the file counts do not show, and the cost depends on how the scan is read. `table.scan().to_arrow()` reads the whole current snapshot into one Arrow table in memory before the overwrite starts. This example passes `table.scan().to_arrow_batch_reader()` instead. pyiceberg documents that a streamed overwrite is microbatched by `write.target-file-size-bytes`, so its memory stays bounded. Streaming works only on unpartitioned tables, which this chapter uses. A partitioned table has to be read with `to_arrow()`, and then memory must hold all of its rows at once. The bound is pyiceberg's documented behaviour, not a measurement made here. The rewrite also reads every file and writes every file, including the ones that were already large enough. A table whose files are mostly the right size gets no benefit from it. This is the reason to use it only when the file count has grown and the table fits in memory. **Working rule:** Check the pyiceberg release notes at each upgrade for a compaction call, because a call would replace the overwrite rewrite in this chapter. Until then, treat the overwrite as a batch job, run it when no other writer is committing, and for a partitioned table budget memory for the whole scan. ## Check each compaction before you trust it {#check-each-compaction} The examples in this chapter print four things, and you should check the same four after any compaction you run. The first is the file count from the table's own listing. The second is the row count of the current version, which must match the count before the compaction. The third is the row count of a version or snapshot from before the compaction, which must still read. The fourth is the number of Parquet files under the table's location, which should exceed the file count by the number of old files that are waiting for cleanup. The first two checks show that the compaction did what it claims and changed no rows. The third shows that the history still works, which matters for any reader that pins an older version. The fourth shows what the compaction left on disk. The first and the fourth come from different places, and the book's examples print both, so that a mismatch is visible in the output rather than hidden in one number. In the examples here, the fourth count is 7 for the DuckLake table and 7 for the Iceberg table, each one more than the files that remain listed. A compaction that passes the first three checks and shows that count is not at fault. The extra files are the state that chapter 7 cleans up. Each library exposes its own file listing. The Delta example counts the URIs of the current version, the DuckLake example asks DuckLake's catalog for the files of the table, and the Iceberg example counts the entries of the table's file inspection. The three numbers are read from metadata, so they do not depend on the directory listing. The directory listing is the second source. When the two disagree, stop and find out why before you run the next compaction. When a check fails, stop the job and leave the table as it is. Do not run a second compaction to fix the first one, because each run is a commit, and the second run makes the history harder to read. Read the table's history for the commits in order, find the one that changed the row count, and decide from there. The state before the fault is easier to examine while the table still holds it. **Working rule:** After every compaction, compare the row count with the count from before it, and read a version from before it. A compaction that changes the row count is a fault to fix before you run anything else on that table. ## What goes wrong in production {#compaction-production-failures} **Storage grows before it shrinks.** Every compaction in this chapter writes replacement files and leaves the old ones on disk. The counts show it: after the DuckLake merge the table lists 1 data file while 7 Parquet files sit on disk, and after the Iceberg rewrite the table lists 1 while 7 sit on disk. A compaction that runs without an expiry step leaves the storage bill higher than before. Plan the expiry and the cleanup together with the compaction, and measure the disk after both. You notice this as a disk that grows after each run while the table's row count and file count stay flat. **Deleted rows stay until a rewrite handles them.** A DuckLake merge skips any file that has a delete file, as [the merge section](#merge-adjacent-files) shows. A table with deletes that is only merged keeps those files as they are. Use `ducklake_rewrite_data_files` for them, with a threshold you chose. Measure the rows before and after, because the merge alone will not show you that the deleted rows are still in the files. You notice the problem when the count of files that have delete files stays the same after each merge. **Concurrent writers are untested here.** This chapter's examples run one writer at a time and do not test a compaction that overlaps a commit. Test the overlap on your own engine, and schedule compaction in a window where no writer commits until you have that test. The Iceberg rewrite is the most exposed of the three, because it reads the whole table and then writes it back. A failed overlap would show as an error from the writer or from the compaction, and the book has not reproduced one. **Too often costs commits, too seldom costs reads.** Each compaction is a commit, and each commit adds metadata. Compacting after every load doubles the commits per load. Compacting once a month lets small files build up for a month. You notice the first as a commit count that climbs faster than the loads, and the second as query times that rise as the file count grows. This is a trade-off you measure on your own table, not a rule the examples prove. **Six files to one is a count goal, not a size goal.** A single very large file gives a reader less parallel work and fewer files to prune by file-level statistics. The examples aim at a file count, and they print no file sizes. This is a judgement about readers, and your readers' query times should set the target. You notice a problem as query times that get worse after a compaction that produced one very large file. **Working rule:** Schedule compaction next to the expiry and cleanup, not apart from them. Run the four checks from [the check section](#check-each-compaction) after each run, and stop the schedule if the row count changes or if a concurrent writer fails. ## Decisions this chapter made {#compaction-decisions} - Each format uses its own library's call, and no example uses a DuckDB extension for Iceberg or Delta Lake. The reads and writes go through the Python libraries and through DuckDB's DuckLake support, so each count comes from the tool that wrote the files. - The DuckLake examples attach with data inlining off. The default keeps inserts of 10 rows or fewer in the catalog, so a small load could leave no file to compact. Turning inlining off makes every count a count of files, whatever the load size. - The file counts come from each library's own file listing. The Parquet count on disk is a second check, and the examples print no file sizes, because sizes in bytes change from run to run. - The tables are unpartitioned, so each count has one cause: the number of loads. The partitioned case is left to the chapter on partition evolution. - Iceberg compaction is a rewrite by overwrite, because pyiceberg 0.12.0 has no compaction call. The example streams the scan, which pyiceberg allows for unpartitioned tables. The chapter names the workaround and the memory cost that remains for partitioned tables. - No example in this chapter expires a snapshot or removes a file. [Chapter 7](#expiry-and-cleanup) does both, and this chapter's counts are the starting point for that work. - The schedule in the working rule is a starting point. The book does not claim a best interval, because the right one depends on how fast a table's files arrive and how large its readers need them to be. # Expire snapshots and remove unreferenced files By the end of this chapter you can expire old snapshots in each of the three formats, remove the data files that only those snapshots still listed, and show that a read of an expired snapshot fails. [Compaction](#hourly-appends-leave-small-files) leaves the same kind of old file: the rewrites left the replaced files on disk, because older snapshots still list them. The examples here use the hour-02 correction from [the corrections chapter](#late-data-as-correction), written as an overwrite of that hour, not as the upsert in its later section, because it leaves exactly one old file per format. Those old files stay until their snapshots are expired and their files are removed. This chapter does both steps in each format, on the same table the earlier chapters built, and it checks each step with a count rather than with a listing you have to trust. ## Snapshots keep files alive {#snapshots-keep-files-alive} Every snapshot records which data files make up the table at that commit. [What a snapshot points to](#what-a-snapshot-points-to) covers how each format stores that list. The point for this chapter is that a commit that replaces rows does not delete the files it replaces. It writes files for the replacement rows, and the snapshots from before the commit still list the old ones, so time travel keeps working. A data file is garbage only when no snapshot you keep lists it. Removing a file takes two steps in every format. Expiry drops old snapshots from the metadata, so a read cannot reach them any more. Cleanup then deletes the files that no remaining snapshot lists. The Iceberg maintenance page puts it in the same order: "Expiring old snapshots removes them from metadata, so they are no longer available for time travel queries," and "data files are not deleted until they are no longer referenced by a snapshot that may be used for time travel or rollback" ([Apache Iceberg, maintenance](https://iceberg.apache.org/docs/latest/maintenance/)). The case shows the effect. The table `wiki.pageviews_hourly` takes three hourly appends, and then hour 02 is corrected by an overwrite of that hour. Each append writes one file. The overwrite writes a file with the corrected hour-02 rows, and it removes the old file from the current snapshot. The snapshot that appended the old file still lists it. Until that snapshot is expired, the old file cannot be deleted without breaking time travel to it. [Upserting a corrected hour](#upsert-corrected-hour) shows the other way to write the same correction, and it leaves the same kind of old file behind. Each format has its own expiry call and its own cleanup call. The table lists them, and the rest of the chapter follows the same order for each format. | Format | Expiry call | Expiry deletes files? | Cleanup call | | --- | --- | --- | --- | | Iceberg (pyiceberg) | `expire_snapshots().by_ids(...)` | No | None in the library: count and remove files yourself | | Delta Lake (deltalake) | Version retention in `vacuum` | Yes, by `vacuum` | `vacuum` | | DuckLake | `ducklake_expire_snapshots` | No | `ducklake_cleanup_old_files`, then `ducklake_delete_orphaned_files` | **Working rule:** Decide which snapshots or versions a reader may still need, and expire the rest before you remove any file. Never remove a file that a snapshot you keep still lists, and check that claim with a count before you delete. ## Expire snapshots with pyiceberg {#expire-snapshots-in-iceberg} In pyiceberg 0.12.0, `table.maintenance.expire_snapshots()` takes snapshot IDs, with `by_ids` for a list or `by_id` for one, or an `older_than` timestamp. It has no argument that keeps the last N snapshots. This chapter chooses IDs from the commit order, as the case does: the three hourly appends are the first three snapshots. The IDs themselves are random 64-bit numbers, so the example prints the labels "snapshot 1", "snapshot 2" and so on, in the same order. The overwrite adds two snapshots, not one. pyiceberg writes a delete snapshot that removes the old file, then an append snapshot that adds the corrected rows. Snapshot 4 is the delete and snapshot 5 is the append. Expiring the three appends therefore leaves the overwrite in place, and that is the state the example checks. The expiry call deletes no data files, so the example counts them. It counts the distinct files that the snapshots list, and the files on disk that no snapshot lists. Here is the example: Example `expire-iceberg-snapshots` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Expire three Iceberg snapshots by ID, then count the data files they kept alive.""" from lib.lakehouse import ( ICEBERG_ID, create_iceberg_table, hour_corrections, hour_filter, new_warehouse, orphan_data_files, read_cut_hour, scan_arrow, snapshot_labels, ) def referenced_files(table): """Distinct data files that some retained snapshot lists. Count them; do not print them.""" return len(set(table.inspect.all_files()["file_path"].to_pylist())) with new_warehouse() as wh: table = create_iceberg_table(wh) for hour in (0, 1, 2): table.append(read_cut_hour(hour)) table.overwrite(hour_corrections(read_cut_hour(2)), overwrite_filter=hour_filter(2)) ids = [snapshot.snapshot_id for snapshot in table.metadata.snapshots] labels = snapshot_labels(ids) print("snapshots before expiry:", len(ids)) print("data files referenced before expiry:", referenced_files(table)) print("data files in the current snapshot:", len(table.inspect.files())) expired = ids[:3] # the three hourly appends, by commit order table.maintenance.expire_snapshots().by_ids(expired).commit() table = wh.catalog.load_table(ICEBERG_ID) print("remaining snapshots:", ", ".join(labels[s.snapshot_id] for s in table.metadata.snapshots)) print("data files referenced after expiry:", referenced_files(table)) print("unreferenced data files left on disk:", len(orphan_data_files(table))) print("rows in the current snapshot:", scan_arrow(table).num_rows) try: scan_arrow(table, snapshot_id=expired[-1]) except ValueError as exc: print(f"read {labels[expired[-1]]}: {type(exc).__name__}") ``` Output: ```text snapshots before expiry: 5 data files referenced before expiry: 4 data files in the current snapshot: 3 remaining snapshots: snapshot 4, snapshot 5 data files referenced after expiry: 3 unreferenced data files left on disk: 1 rows in the current snapshot: 1499 read snapshot 3: ValueError ``` Before the expiry, the table has 5 snapshots, and 4 data files are listed by at least one of them. The current snapshot lists only 3 of those files. The remaining one is the old hour-02 file, which snapshot 3 still lists. Reading the counts this way shows what the overwrite did: the current table is complete, and the history still holds the file the overwrite replaced. The example then expires snapshots 1 to 3, the three appends. The snapshots that remain are snapshot 4, snapshot 5, the delete from the overwrite and its append. Afterwards 3 data files are listed, and the old hour-02 file is no longer listed by any snapshot. It is still on disk: the example reports 1 unreferenced file. The current table still returns 1499 rows, so the expiry changed no data a reader of the current table sees. The last lines show the time travel failure. A scan of snapshot 3 after the expiry raises `ValueError`, which the output shows as "read snapshot 3: ValueError". A reader that pinned snapshot 3 before the expiry gets an error, not the old rows. The error, `Snapshot not found`, names the snapshot's ID, but it does not say that an expiry removed the snapshot. Your own logs have to record which expiry ran and which snapshot the report used. **Working rule:** Expire Iceberg snapshots by ID, chosen from the commit order, and count the files each remaining snapshot lists before and after. Expect no disk space back from the expiry. pyiceberg removes metadata only, so the files it leaves behind stay on disk until you remove them yourself. ### Find orphan files after expiry {#find-orphan-files} A file that no snapshot lists is an orphan. Spark and Java offer a `deleteOrphanFiles` action: it lists the table location, compares the files with the metadata and deletes the difference, and its default age interval is 3 days ([Apache Iceberg, maintenance](https://iceberg.apache.org/docs/latest/maintenance/)). The pyiceberg API page lists no orphan-file step, and the 0.12.0 package has none either. Its maintenance methods expire snapshots ([PyIceberg API](https://py.iceberg.apache.org/api/)). The example's helper `orphan_data_files` does the comparison by hand. It lists the Parquet files under the table location, subtracts the files that every remaining snapshot lists, and returns the rest. The example prints only the count, because the file names carry UUIDs. That count is the line in the output that shows the expired file is still on disk. A listing cannot tell an orphan from a file that a writer has written and not yet committed. The commit links the file to the table, and until then the file looks unreferenced. This is why the age guards matter: [DuckLake's `ducklake_delete_orphaned_files`](#expire-and-clean-up-in-ducklake) keeps young files by default, and the Spark action takes an age interval for the same reason. pyiceberg has no interval, so the check is yours to make. Deleting an orphan is a separate decision, and it belongs after the writers that could commit it have stopped. **Working rule:** Count orphans by listing, and remove them only after you have confirmed that no writer can still commit them and no reader still uses a snapshot that lists them. In pyiceberg, nothing in the library makes that check for you. ## Vacuum a table with deltalake {#vacuum-in-delta-lake} Delta Lake does not delete a file when a commit replaces it. The commit adds a `remove` action to the log, a tombstone that says the file is no longer part of the table, and the file stays on disk. The protocol states that such files "can be lazily deleted by the vacuum command after a user-specified retention period (default 7 days)" ([Delta transaction log protocol, v4.4.1](https://raw.githubusercontent.com/delta-io/delta/v4.4.1/PROTOCOL.md)). Delta counts versions from 0, so the output in this section uses Delta's own version numbers. Vacuum deletes the files whose tombstones are older than the retention. Until then, every version can be read, because every file it lists still exists. The Delta utility page states the cost: "The ability to time travel back to a version older than the retention period is lost after running vacuum" ([Delta Lake, VACUUM](https://docs.delta.io/delta-utility/)). The case has three appends, which make versions 0 to 2, and then the hour-02 correction. The correction is a delete of that hour, which is version 3, followed by an append of the corrected rows, which is version 4. Version 2 still lists the old hour-02 file, and version 3 does not. Nothing has been removed yet, and that is the state the next section starts from. The default retention in deltalake is 168 hours, seven days. A table can set its own in the `delta.deletedFileRetentionDuration` property, and `vacuum` uses that property when you pass no `retention_hours`. That property is the production setting. The library refuses a retention shorter than the property's value unless you turn off `enforce_retention_duration`. A longer retention needs no flag. The retention you choose is a promise about how far back history reaches. Every reader that opens an older version depends on it, and the next section shows the refusal and why this chapter's teaching run turns the check off. **Working rule:** Keep vacuum's default retention, or set a longer one in `delta.deletedFileRetentionDuration`, so that every version a reader may request is still readable for that long after it stops being current. Go shorter only after you have checked the longest read and the longest write in your system, and only by turning `enforce_retention_duration` off as a deliberate, recorded decision. ### Dry run first {#vacuum-dry-run-first} A dry run lists the files that a real vacuum would delete and deletes none of them. Run one before every vacuum, and compare its count with the table's history. The case is three appends, a delete in version 3 and an append in version 4. A short list can be checked by hand: here, each file in the list is one that a delete removed, and the one that matters is the delete in version 3. Here is the example: Example `vacuum-delta-dry-run` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Vacuum a Delta table: dry run first, then the teaching-only zero-hour run, then a read of a removed version.""" import pyarrow as pa from deltalake import DeltaTable, write_deltalake from deltalake.exceptions import DeltaError from lib.lakehouse import new_warehouse, read_cut_hour, read_delta, hour_corrections with new_warehouse() as wh: path = str(wh.root / "delta" / "pageviews_hourly") for hour in (0, 1, 2): write_deltalake(path, read_cut_hour(hour), mode="append") dt = DeltaTable(path) dt.delete("hour_utc = '2024-06-03T02:00:00'") write_deltalake(path, hour_corrections(read_cut_hour(2)), mode="append") dt = DeltaTable(path) print("current version:", dt.version()) print("default retention, dry run lists:", len(dt.vacuum(dry_run=True))) try: dt.vacuum(retention_hours=0, dry_run=True) except DeltaError as exc: print("zero retention without the flag:", type(exc).__name__) # Teaching run only: a zero-hour retention removes files at once. Production keeps the 168-hour default. print("zero retention, dry run lists:", len(dt.vacuum(retention_hours=0, dry_run=True, enforce_retention_duration=False))) print("zero retention, removed:", len(dt.vacuum(retention_hours=0, dry_run=False, enforce_retention_duration=False))) print("rows in version 0:", read_delta(path, version=0).num_rows) try: read_delta(path, version=2) except pa.ArrowInvalid as exc: print("read version 2:", type(exc).__name__) ``` Output: ```text current version: 4 default retention, dry run lists: 0 zero retention without the flag: DeltaError zero retention, dry run lists: 1 zero retention, removed: 1 rows in version 0: 499 read version 2: ArrowInvalid ``` The table is at version 4. With the default retention, the dry run lists 0 files. The file that version 3 removed is too recent to be listed, and vacuum keeps it. The next call sets the retention to zero, and deltalake refuses it. The example shows the refusal as "zero retention without the flag: DeltaError". The library's own message says that the minimum retention is 168 hours, so the check protects the default. The teaching run turns the check off with `enforce_retention_duration=False`. The code marks it as teaching-only, and production does not use it. With the check off, the dry run lists 1 file, and the real vacuum removes 1 file, the old hour-02 file. Zero retention is the right setting for a demo, because it deletes the file at once. In production it is wrong: any reader that started before the vacuum, or that pinned an older version, loses the files it reads in the middle of a query. A short retention also deletes the files of a writer that has written but not yet committed. The Delta utility page warns that "tables can be corrupted when VACUUM deletes files that have not yet been committed" ([Delta Lake, VACUUM](https://docs.delta.io/delta-utility/)). The damage is to the table, not only to one read. The read at the end shows what the vacuum cost. Version 0 still reads, with 499 rows, because every file it lists still exists. Version 2 fails with `ArrowInvalid`, shown as "read version 2: ArrowInvalid". That class is the one the query engine raises for a failed Parquet read, and the message names the missing object. The example prints only the class, because the file names carry UUIDs. Read the message in your own logs before you decide that the table is damaged: it says which object is missing, and so whether a vacuum removed it. A dry run is not a promise about the real run. A writer that commits between the two calls changes the table, and the real run works from the table state it finds. Treat the dry-run count as a check on what you expect, and read the real count next to it. **Working rule:** Run vacuum as a dry run first and read the count. Remove files with a real run only when the count matches what you expected from the table's history, and keep the default retention unless you have checked the longest read in your system. ## Expire and clean up in DuckLake {#expire-and-clean-up-in-ducklake} DuckLake keeps the physical files of a snapshot until that snapshot is expired. The DuckLake documentation says it this way: "Data can only be physically removed from DuckLake by expiring snapshots that refer to the old data" ([DuckLake, expire snapshots](https://ducklake.select/docs/stable/duckdb/maintenance/expire_snapshots)). Expiry does not delete the files at once. The same page says "expiring snapshots does not immediately delete files that are no longer referenced," and the cleanup page gives the second call ([DuckLake, cleanup of files](https://ducklake.select/docs/stable/duckdb/maintenance/cleanup_of_files)). So DuckLake maintenance has two steps, and the example runs both. `ducklake_expire_snapshots` takes a list of snapshot numbers in `versions`, or an `older_than` timestamp, but not both. `ducklake_cleanup_old_files` removes the files that expiry scheduled. With `older_than`, it removes only files scheduled before a given time. With `cleanup_all => true`, it removes every scheduled file, with no age check. Both steps support `dry_run`. A third call, `ducklake_delete_orphaned_files`, handles files that no snapshot lists and that cleanup never scheduled, such as a file a writer left behind after a failed commit. Its default keeps a file that is still young, because a writer may be about to commit it. Its `cleanup_all` option removes such files regardless of age. The [catalog chapter](#catalog-production-failures) shows what that option does when a second catalog shares the data folder. The case: attaching the catalog and creating the schema and the table through the helpers makes three snapshots before any rows. The attach makes snapshot 1, and the two `CREATE` statements make snapshots 2 and 3. This book labels DuckLake's snapshots from 1 in commit order, so label N is DuckLake's snapshot number N−1, because DuckLake counts from 0. The example then inserts hours 00 to 02, deletes hour 02, and inserts the corrected hour 02. That makes 8 snapshots, which the example labels "snapshot 1" up to the last one. The last is the corrected insert. The one before it is the delete, which drops the old hour-02 file. The insert before that, snapshot 6, is the last snapshot that lists it. Example `ducklake-expire-cleanup` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Expire DuckLake snapshots, then clean up the files they released, in two steps.""" import shutil from pathlib import Path import duckdb from lib.lakehouse import ( DUCKLAKE_ALIAS, NAMESPACE, TABLE, connect_ducklake, create_ducklake_table, ducklake_file_count, ducklake_file_paths, load_ducklake_with_correction, new_warehouse, version_labels, ) LAKE = f"{DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE}" def listed(con, call): """Run a DuckLake procedure and return how many rows it lists.""" return con.sql(f"CALL {call}").to_arrow_table().num_rows with new_warehouse() as wh: con = connect_ducklake(wh.con) create_ducklake_table(con, wh.root) load_ducklake_with_correction(con) ids = [r[0] for r in con.execute(f"SELECT snapshot_id FROM ducklake_snapshots('{DUCKLAKE_ALIAS}') ORDER BY 1").fetchall()] labels = version_labels(ids, word="snapshot") print("snapshots before expiry:", len(ids)) print("data files in the current snapshot:", ducklake_file_count(con, DUCKLAKE_ALIAS, TABLE)) stray = Path(sorted(ducklake_file_paths(con, DUCKLAKE_ALIAS, TABLE))[0]) shutil.copy(stray, stray.with_name("stray.parquet")) # written, but no snapshot lists it expired = ids[:-1] # every snapshot before the correction expire = f"ducklake_expire_snapshots('{DUCKLAKE_ALIAS}', versions => {expired}" print("expiry dry run lists:", listed(con, f"{expire}, dry_run => true)")) con.execute(f"CALL {expire})") print("remaining snapshots:", labels[ids[-1]]) cleanup = f"ducklake_cleanup_old_files('{DUCKLAKE_ALIAS}', cleanup_all => true" print("cleanup dry run lists:", listed(con, f"{cleanup}, dry_run => true)")) print("cleanup removed:", listed(con, f"{cleanup})")) orphans = f"ducklake_delete_orphaned_files('{DUCKLAKE_ALIAS}', dry_run => true)" print("orphans listed without cleanup_all:", listed(con, orphans)) print("rows in the current snapshot:", con.sql(f"SELECT count(*) FROM {LAKE}").fetchall()[0][0]) try: con.sql(f"SELECT count(*) FROM {LAKE} AT (VERSION => {expired[-1]})").fetchall() except duckdb.InvalidInputException as exc: print(f"read {labels[expired[-1]]}: {type(exc).__name__}") ``` Output: ```text snapshots before expiry: 8 data files in the current snapshot: 3 expiry dry run lists: 7 remaining snapshots: snapshot 8 cleanup dry run lists: 1 cleanup removed: 1 orphans listed without cleanup_all: 0 rows in the current snapshot: 1499 read snapshot 7: InvalidInputException ``` The current snapshot lists 3 data files. The example copies one of them to another name in the same folder, to stand in for a file a writer has written but not committed. Then it runs the steps in order. The expiry dry run lists 7 snapshots, which are all of them except the corrected insert. The example keeps the correction and expires everything before it, so the table's current state is untouched. The real expiry leaves snapshot 8. The old hour-02 file is released at this step, because snapshot 6, the last snapshot that listed it, is expired. The cleanup dry run lists 1 file, the old hour-02 file. The DELETE matched every row in that file, and no delete file is scheduled with it, so the old file is the only one left to remove. The real cleanup removes 1 file. The orphan check lists 0 orphans. The stray copy is still in the folder, because the default age check kept it: the file is only seconds old. The table still returns 1499 rows. Last, the read of the delete, snapshot 7, fails: "read snapshot 7: InvalidInputException". The class is the one DuckLake raises for a version it no longer has. The example passes `cleanup_all => true` because it has one catalog and no reader. Production uses `older_than` instead, with an age longer than the longest read, so that the cleanup removes only files that have been scheduled for long enough. **Working rule:** Expire DuckLake snapshots first, then run each cleanup call as a dry run and then for real, in that order. Keep the default orphan check. Use `cleanup_all` only when one catalog owns the data folder and no reader still uses the files it removes, and use `older_than` otherwise. ## What goes wrong in production {#maintenance-production-failures} **Removing a file that a reader still uses.** A query that started before a vacuum or a cleanup reads a file that is then gone, and the read fails partway through. You will see it as a read error that names a missing object, in a job that ran during a maintenance window. Schedule maintenance outside the long reads, and keep retention longer than the longest of them. A short retention can also delete a file that a writer has written and not yet committed, which corrupts the table rather than failing one read, so keep retention longer than the longest running write as well. **Retention shorter than the longest query.** The seven-day default in Delta Lake is a margin, not a measurement. A dashboard that reads a version from a month ago is outside it. If a reader must see an older version, the retention has to cover it, or the reader has to copy the data first. **Time travel that fails after maintenance.** The expired snapshot and the vacuumed version fail with `ValueError`, `ArrowInvalid` or `InvalidInputException`, depending on the format, as the examples show. Any job that records a snapshot ID or a version number to read later fails the same way once that snapshot is expired. Keep your own record of which snapshot each report used, so that a rebuild starts from the data and does not depend on reading the old snapshot. **A dry run that differs from the real run.** Two calls can disagree when a writer commits between them. The count from the dry run is what you expected, not what the real run will remove. Read the real count next to the dry-run count, and stop if they differ by more than you can explain. **An audit that needs an older snapshot.** Some reports must reproduce a figure from months earlier. Expiry removes that ability for good, so decide the audit window before you expire, and keep the snapshots or the copied data the audit needs. Expiry is not a substitute for a copy. **A cleanup that deletes a file another catalog needs.** `cleanup_all` and the orphan check with `cleanup_all` delete files by path. If a second catalog points at the same data folder, its live files look like orphans to the first catalog. The [catalog chapter](#catalog-production-failures) shows that failure in a temporary directory. **Working rule:** Before you expire or clean up, list the readers, reports and audits that may open an older snapshot or version, and set the retention to cover the longest of them. Run each call as a dry run, compare its count with what you expected, and run the real call only when the two agree. ## Decisions this chapter made {#maintenance-decisions} - Iceberg snapshots are expired by ID, chosen from the commit order. The chapter does not expire by age, so the result does not depend on the clock the example runs on, and pyiceberg 0.12.0 has no keep-last-N argument. - pyiceberg removes no files, so the Iceberg example counts reachable and unreachable files and leaves the removal to the reader. - The Delta example's zero-retention run is teaching-only. It is marked in the code, and production keeps the default retention. - DuckLake expiry and cleanup are separate calls, and each runs as a dry run before the real call. - `cleanup_all` appears only in single-catalog examples. The catalog chapter covers the case where it removes live files. - The examples print counts, stable labels and error class names. They print no snapshot IDs, timestamps, file names or paths, and every number in the text comes from a bound output line or from the example's own construction. - Every example keeps the table the earlier chapters built, so each count can be checked against the history the reader has already seen. - Each example builds its own temporary warehouse, so no output depends on an earlier example or on the machine that runs it. - The DuckLake example's stray copy stands in for a file a writer has not committed. That shows the orphan rule without waiting for a file to age, because the example runs in seconds. # Register tables in a catalog and query them from DuckDB By the end of this chapter you can register an Iceberg table in a SQL catalog, attach a DuckLake catalog, open a Delta Lake table by its path, and run one DuckDB statement that joins a table from Iceberg with a table from DuckLake. You can also say what each catalog holds, and what happens to a table when its catalog and its data files disagree. The earlier chapters opened each table by its own route: a catalog object for Iceberg, a file for DuckLake, a folder for Delta Lake. Every chapter uses the same table, `wiki.pageviews_hourly`, the hourly view counts of Wikimedia pages. [The first table](#create-catalog-and-table) registered it in an Iceberg SQL catalog under the name `local`. This chapter explains what that name does, and what each format uses in its place. It also adds a second table, `wiki.domains`: a four-row list of domain codes and the language each one is in. It lives in the DuckLake catalog, and [the join section](#join-tables-from-two-catalogs) joins it to the Iceberg pageviews. ## What a catalog holds {#what-a-catalog-holds} A table format stores a table as data files plus metadata files. The metadata names the table's current state: its schema, its partition spec and, directly or through manifest files, the data files of its current snapshot or version. Each commit writes new metadata and moves the table to it. The catalog is the registry that maps a table's name to the place where its current metadata is. A reader that asks the catalog for `wiki.pageviews_hourly` gets a pointer to the metadata, and then reads the data files that the metadata lists. The catalog holds pointers, not the rows. The Parquet files stay in the warehouse folder, and the catalog says where to start reading. Names have a second level as well: `wiki` is a namespace, and every Iceberg table name in this book is a namespace and a table name joined by a dot. Choosing a catalog comes down to two questions: who writes the table, and what reads it. A table that one job writes and one engine reads can live in a file-based catalog and be found by its path. A table that several engines read needs a name each of them can resolve, which is what a SQL catalog or a service provides. We recommend one catalog for each table, and a record next to the table that names its catalog and its owner. A join across catalogs, as [the join section](#join-tables-from-two-catalogs) shows, works in one session. It does not make the two catalogs one. The three formats in this book use three different catalogs: | Format | What names the table | Where the current metadata is | | --- | --- | --- | | Iceberg | A SQL catalog object, here named `local` | The catalog's database, which points at the metadata file | | DuckLake | The catalog file, attached by path | Inside the catalog file | | Delta Lake | Nothing; the table's folder path | In the table's folder | The Iceberg REST catalog is the other kind of catalog, a service instead of a file. Its [REST catalog page](https://iceberg.apache.org/rest-catalog-spec/) describes it as a common API, specified with OpenAPI, for interacting with any Iceberg catalog, so that clients need one implementation for all of them. The same page says the protocol uses change-based commits, so the service can resolve conflicts and retry on its side, and that the catalog service writes the root metadata. This chapter runs no REST catalog. Each one needs a running service, and the examples here run offline with no service. The book does not compare the products that implement the protocol. A catalog can hold more than names. Classifications and retention rules are table metadata too. [*Data Governance as Code*](https://rollingedition.com/data-governance-as-code) covers them, and this book does not teach them. **Working rule:** Before you write to a table, find out which catalog or path names it, and back up that catalog along with the data folder. Parquet files do not name a table. The catalog does, as [the failure section](#catalog-production-failures) shows. ## Register a table in a SQL catalog {#register-iceberg-in-sql-catalog} The pyiceberg `SqlCatalog` keeps its tables in a database. You give it a `uri` for that database and a `warehouse` for the folder where it writes data files and metadata. This book uses a SQLite file for the database and a local folder for the warehouse. The [pyiceberg configuration page](https://py.iceberg.apache.org/configuration/) says the SQL catalog "requires a database for its backend", and that "SQLite is not built for concurrency, you should use this catalog for exploratory or development purposes". The examples here are development examples, so that limit does not change anything in them. A production catalog needs a database that several writers can use. The next example opens the same catalog file a second time, with the same name, and reads the table through that second object. It then registers a second name for the table's metadata, and shows what a catalog object with a different name can see. Example `register-iceberg-table` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Register an Iceberg table in a SQL catalog, then reach it from a second catalog object.""" from lib.lakehouse import (ICEBERG_ID, NAMESPACE, create_iceberg_table, new_warehouse, open_catalog, read_cut_hour, safe_error, scan_arrow) with new_warehouse() as wh: table = create_iceberg_table(wh) for hour in range(3): table.append(read_cut_hour(hour)) # A second object with the same catalog name sees the same tables. same = open_catalog(wh) loaded = same.load_table(ICEBERG_ID) print("rows read through the second object:", scan_arrow(loaded).num_rows) # A loaded table is a snapshot of the catalog as it was when loaded. stale = same.load_table(ICEBERG_ID) table.append(read_cut_hour(3)) print("rows in the loaded object after another append:", scan_arrow(stale).num_rows) print("rows after a reload:", scan_arrow(same.load_table(ICEBERG_ID)).num_rows) # Register the same metadata file under a new name: a pointer, not a copy. same.create_namespace("archive") same.register_table("archive.pageviews_copy", table.metadata_location) print("rows in archive copy:", scan_arrow(same.load_table("archive.pageviews_copy")).num_rows) # An object with another name does not see the namespace. try: open_catalog(wh, name="other").list_tables(NAMESPACE) except Exception as exc: print("other name:", safe_error(exc, wh)) ``` Output: ```text rows read through the second object: 1499 rows in the loaded object after another append: 1499 rows after a reload: 1999 rows in archive copy: 1999 other name: NoSuchNamespaceError: Namespace does not exist: wiki ``` The second object reads 1499 rows of `wiki.pageviews_hourly`, the three hours that [chapter 1](#land-three-hours) landed. pyiceberg stores each table under the catalog name you pass to `SqlCatalog`. An object with the same name finds the same tables. The catalog and the warehouse folder are a pair. The catalog holds the locations of the metadata files, and those locations sit under the folder that the `warehouse` property names. A copy of the catalog file without the warehouse folder, or the folder without the file, gives you pointers to files that are not there. Back up both, and move both together. Each process opens its own catalog object, and none shares the state of another. The example uses two objects in one process to show the effect that a second process would have. A loaded table is a snapshot of the catalog as it was when it was loaded. The example keeps a table object from before a fourth hour is appended through the first object. That object still reads 1499 rows. A reload reads 1999. A process that loads a table at its start and reads it at its end reads the state from the start. It has to reload to see the commits made in between. **Working rule:** Open a SQL catalog file with one name everywhere in your code, and reload a table before you read it when another writer may have committed since you loaded it. Keep SQLite for one writer at a time, and use a database built for concurrent writers before you run several. ### Register a table from its metadata file {#register-from-metadata-file} `register_table(identifier, metadata_location)` adds a name for an existing metadata file. The [pyiceberg API page](https://py.iceberg.apache.org/api/) gives it as the way to register a table from its metadata, and says that passing `overwrite=True` replaces an entry that already exists. The example registers the table's metadata under `archive.pageviews_copy`, in a namespace it creates first. The archive name reads the same rows as the original table: 1999. Both names point at the same metadata, and the registration writes no data files. A registered name is a second route to the same table, not a second table. Commits through one name are the commits of both, because both read the same metadata. Treat a second name as a read route, and not as a backup. **Working rule:** Keep one writer name for each table. Register a metadata file under a second name only for a read you will remove afterwards, and pass `overwrite=True` only when you mean to replace that entry. ## Attach a DuckLake catalog {#attach-ducklake-catalog} DuckLake's catalog is a database file, and DuckDB attaches it with an `ATTACH` statement that starts with `ducklake:` and gives the file's path. The data path, the folder where DuckLake writes Parquet files, is stored in the catalog. The [DuckLake connecting page](https://ducklake.select/docs/stable/duckdb/usage/connecting) says the data storage location only has to be given when a new DuckLake is created, and that the location is loaded from the catalog database after that. The helper `connect_ducklake` loads DuckLake into the DuckDB session with downloads off, and `attach_ducklake` runs the `ATTACH` for a catalog file under a folder. The next example creates `wiki.domains` in a fresh DuckLake catalog, detaches it, and attaches the file again with no data path. It then lists the tables in the `wiki` schema of the file. Example `attach-ducklake` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Attach a DuckLake catalog file, then attach it again from the file alone.""" from lib.lakehouse import (DUCKLAKE_ALIAS, NAMESPACE, attach_ducklake, connect_ducklake, create_ducklake_domains, new_warehouse, print_rows, sql_rows) with new_warehouse() as wh: con = connect_ducklake(wh.con) create_ducklake_domains(con, wh.root) print("rows written:", sql_rows(con, f"SELECT count(*) FROM {DUCKLAKE_ALIAS}.{NAMESPACE}.domains")[0][0]) # Detach, then attach with no DATA_PATH: the catalog file stores the data path. con.execute(f"DETACH {DUCKLAKE_ALIAS}") attach_ducklake(con, DUCKLAKE_ALIAS, wh.root) print("tables in the catalog:") print_rows(sql_rows(con, f"""SELECT schema_name, table_name FROM duckdb_tables() WHERE database_name = '{DUCKLAKE_ALIAS}' AND schema_name = '{NAMESPACE}'""")) print("rows after re-attach:", sql_rows(con, f"SELECT count(*) FROM {DUCKLAKE_ALIAS}.{NAMESPACE}.domains")[0][0]) ``` Output: ```text rows written: 4 tables in the catalog: wiki domains rows after re-attach: 4 ``` The four rows of `wiki.domains` are written, and 4 rows read back after the re-attach. The table list shows one table in the `wiki` schema, `domains`. The second attach needed no data path, because the file already held it. The listing comes from the catalog, not from the data folder. The catalog file holds the list of tables, and the data folder holds the bytes. Losing the first loses the names even when the second is intact, as [the failure section](#catalog-production-failures) shows. The [same connecting page](https://ducklake.select/docs/stable/duckdb/usage/connecting) says that DuckLake inlines small inserts into the catalog database, with a default limit of 10 rows. The helper this chapter uses sets the limit to 0, so every insert in these examples becomes a Parquet file in the data folder. Without that setting, a small table lives in the catalog file, and a copy of the data folder would not hold its rows. That is one reason the catalog file needs a backup of its own. **Working rule:** Back up a DuckLake catalog file with the data folder, and copy the file while no session has it attached. Give the data path when you create a catalog, and attach later with the catalog file alone. ## Find a table by path, with no catalog {#find-delta-by-path} A Delta Lake table in this book is a folder. It has no catalog, and nothing in this book's setup names it apart from its path. The `deltalake` package opens the folder with `DeltaTable(path)`, and every read goes through that path. Delta counts versions from 0, so the three appends in the next example are versions 0, 1 and 2. Example `open-delta-by-path` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Open a Delta Lake table by its path, with no catalog, and read it through QueryBuilder.""" from deltalake import DeltaTable, write_deltalake from lib.lakehouse import new_warehouse, read_cut_hour, read_delta with new_warehouse() as wh: path = str(wh.root / "delta" / "pageviews_hourly") for hour in range(3): write_deltalake(path, read_cut_hour(hour), mode="append") table = DeltaTable(path) print("delta version:", table.version()) print("rows at that version:", read_delta(path).num_rows) print("rows at version 0:", read_delta(path, version=0).num_rows) # The path is the only name a Delta table has here: a wrong path is a missing table. try: DeltaTable(str(wh.root / "delta" / "pageviews_typo")) print("missing path: opened") except Exception as exc: print("missing path:", type(exc).__name__) ``` Output: ```text delta version: 2 rows at that version: 1499 rows at version 0: 499 missing path: TableNotFoundError ``` The table's current version is 2, and it reads 1499 rows. Version 0 holds 499 rows, the first hour alone. The rows here come through `read_delta`, which reads through deltalake's `QueryBuilder` and returns an Arrow table. The [decisions](#catalog-decisions) at the end of this chapter say why the book reads Delta tables that way. The path is the whole name, so a typo in it is a missing table. The example opens a folder that does not exist, and `DeltaTable` raises `TableNotFoundError`. The error names the path it tried, but only once the wrong path has been used, so the check belongs in the code that builds the path. A path is a good name for a table that one job writes and a few jobs read. It needs no service and no setup. It has costs, though. A path does not say which namespace a table is in, so two tables with the same folder name in different places look identical to a reader. An engine that wants to find the table by name has nothing to look up. The Delta tables in this book are small and local, and the path is enough for them. A shared lake read by several engines needs a catalog for its Delta Lake tables, and this book does not teach one. **Working rule:** Keep each Delta table's path in one place your code reads, such as a configuration value, and open the table once at the start of a job to confirm the path. A path that does not exist is an error, not an empty table. ## Join tables from two catalogs {#join-tables-from-two-catalogs} DuckDB can query an Arrow table that a library hands it, and a DuckLake table that is attached to the same session, in one statement. The Iceberg table leaves pyiceberg as an Arrow table through `scan_arrow`. DuckDB registers that table under a name, and the join reads `wiki.domains` through the attached catalog. The book does not load DuckDB's `iceberg` extension for this, so the Iceberg side is the Arrow table and nothing else. Example `join-pageviews-and-domains` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Join an Iceberg table, read through Arrow, with a DuckLake table in one DuckDB statement.""" from lib.lakehouse import (DUCKLAKE_ALIAS, NAMESPACE, connect_ducklake, create_iceberg_table, create_ducklake_domains, new_warehouse, print_rows, read_cut_hour, scan_arrow, sql_rows) with new_warehouse() as wh: table = create_iceberg_table(wh) for hour in range(3): table.append(read_cut_hour(hour)) con = connect_ducklake(wh.con) create_ducklake_domains(con, wh.root) pageviews = scan_arrow(table) joined = f"FROM pageviews p JOIN {DUCKLAKE_ALIAS}.{NAMESPACE}.domains d ON p.domain_code = d.domain_code" print("joined rows:", sql_rows(con, f"SELECT count(*) {joined}", pageviews=pageviews)[0][0]) print("rows with no language:", sql_rows(con, f"""SELECT count(*) FROM pageviews p LEFT JOIN {DUCKLAKE_ALIAS}.{NAMESPACE}.domains d ON p.domain_code = d.domain_code WHERE d.domain_code IS NULL""", pageviews=pageviews)[0][0]) print("top three by views:") print_rows(sql_rows(con, f"""SELECT p.domain_code, d.language, p.page_title, p.views {joined} ORDER BY p.views DESC, p.domain_code, p.page_title, p.hour_utc LIMIT 3""", pageviews=pageviews, keep_order=True), keep_order=True) ``` Output: ```text joined rows: 1499 rows with no language: 0 top three by views: en English Main_Page 64859 en English Main_Page 64564 en English Main_Page 61107 ``` The join matches 1499 pageview rows to a language. The count of rows with no language is 0, so every domain code in the pageviews is in the four-row dimension. The top three rows by views are the same page, `Main_Page` on the English domain. The join key is `domain_code`, a string on both sides. A join on a key that differs in case or in trailing spaces matches nothing: the joined-row count would drop to 0, and the count of rows with no language would equal the table's row count. Check the key on both sides, in the same session, before you trust a join. The query works because both sides meet in one DuckDB session. It does not make the two tables one. Each keeps its own commits, and the Arrow table is a snapshot of the Iceberg table at the moment `scan_arrow` ran. A join over a table that another writer is changing reads whatever state was current at that moment. We recommend a cross-catalog join for lookups like this one: a small dimension table that stays in its own catalog, read by a query over a large table. When one table appears in almost every query that reads another, copy it into the catalog those queries use. Chapter 9 copies a table between formats and [checks each copy](#check-copy-with-digest), and that check is the step to take before you move a table anywhere. **Working rule:** Join across catalogs in one session only when the side you read as Arrow fits in memory. Choose its state before you read it, and count the rows that find no match, which should be zero. ## What goes wrong in production {#catalog-production-failures} The symptoms are usually small: a count that stops moving, a namespace that is missing, or a read error on a table that was read before. Each case below starts from one of them. Three failures in this chapter are shown by an example, so you can recognize them when they happen. A fourth is named and not shown. **The wrong catalog name.** The last line of [register-iceberg-table](#register-iceberg-in-sql-catalog) opens the same file under the name `other`. Listing the namespace then fails with `NoSuchNamespaceError: Namespace does not exist: wiki`. The table is still in the file. The SQL catalog scopes its tables by catalog name, so the second name sees an empty catalog. The message names the namespace, not the name, so the error looks like a missing schema. A job that builds its catalog name from configuration can fail this way after a small edit. You notice it at the first read, not at the write, so log the catalog name at the start of the job. **The stale pointer.** A process that loads a table and reads it much later reads the state from when it loaded. [register-iceberg-table](#register-iceberg-in-sql-catalog) showed 1499 rows where a reload shows 1999. Nothing fails. The job runs and reports fewer rows than the table holds. You notice it when two jobs disagree about a count, so keep the load next to the read. **The catalog file replaced by an empty one.** The next example attaches a fresh, empty catalog file to the data folder of a live catalog, then runs `ducklake_delete_orphaned_files` with `cleanup_all => true`. The [DuckLake cleanup page](https://ducklake.select/docs/stable/duckdb/maintenance/cleanup_of_files) shows `cleanup_all` deleting all orphaned files, and `dry_run` listing them instead. It defines an orphan as a file that the catalog you run it against does not track. A fresh catalog tracks nothing, so every Parquet file in the folder is an orphan to it. Example `ducklake-orphan-cleanup-hazard` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Run cleanup_all against a fresh, empty DuckLake catalog that shares a data path with a live one.""" from lib.lakehouse import (DUCKLAKE_ALIAS, NAMESPACE, connect_ducklake, create_ducklake_domains, new_warehouse, sql_rows) CLEANUP = "CALL ducklake_delete_orphaned_files('fresh', cleanup_all => true{dry})" def parquet_count(root): return len(list((root / f"{DUCKLAKE_ALIAS}_data").rglob("*.parquet"))) with new_warehouse() as wh: con = connect_ducklake(wh.con) create_ducklake_domains(con, wh.root) print("parquet files, live catalog:", parquet_count(wh.root)) # A replacement catalog: a new, empty file that points at the same data folder. con.execute(f"DETACH {DUCKLAKE_ALIAS}") # Attached by hand: attach_ducklake would give fresh its own data folder, not the live one. con.execute(f"ATTACH 'ducklake:{wh.root}/fresh.ducklake' AS fresh (DATA_PATH '{wh.root}/{DUCKLAKE_ALIAS}_data/')") print("files the dry run lists:", len(sql_rows(con, CLEANUP.format(dry=", dry_run => true")))) con.execute(CLEANUP.format(dry="")) print("parquet files after cleanup_all:", parquet_count(wh.root)) # The live catalog still names the table, and its data files are gone. con.execute(f"ATTACH 'ducklake:{wh.root}/{DUCKLAKE_ALIAS}.ducklake' AS live") print("count through the live catalog:", sql_rows(con, f"SELECT count(*) FROM live.{NAMESPACE}.domains")[0][0]) try: print("reading the live catalog:", len(sql_rows(con, f"SELECT domain_code, language FROM live.{NAMESPACE}.domains"))) except Exception as exc: print("reading the live catalog:", type(exc).__name__) ``` Output: ```text parquet files, live catalog: 1 files the dry run lists: 1 parquet files after cleanup_all: 0 count through the live catalog: 4 reading the live catalog: IOException ``` The example runs in a temporary folder, and prints counts only. The live catalog's data is in 1 Parquet file. The dry run with `cleanup_all` lists 1 file, which is the file the real run deletes. After the real run, the folder holds 0 Parquet files. The live catalog still names the table. It answers `count(*)` with 4, from the catalog's statistics, and reading its columns fails with `IOException`. A count check would pass here, so it does not show missing files. The safe order starts where the example did not. Back up the catalog file first. Run the cleanup with `dry_run => true` against the catalog that owns the data folder, and read the list. Run it for real only when every listed file is one you know is orphaned. Never run `cleanup_all => true` against a catalog you have not confirmed. **Two writers on one SQLite catalog.** The pyiceberg documentation says SQLite is not built for concurrency. This chapter does not show what two writers do to the SQL catalog, so the book does not claim that they are safe. The stale-pointer example is the nearest case it does show: an object loaded before a commit reads the older state, and no error is raised. A writer that holds an old object does not write against that old state. At the pinned version of pyiceberg, an append through such an object is retried against the new metadata and commits on top of the newer state. The book makes no claim for other operations, or for two processes on one SQLite catalog. The reload rule in [the SQL catalog section](#register-iceberg-in-sql-catalog) applies to reads. **Working rule:** Run an orphan cleanup with `dry_run => true` first, against the catalog that owns the data folder, and read the list before any real run. If the catalog file is missing or empty, restore it before you run anything that deletes files. ## Decisions this chapter made {#catalog-decisions} - **Iceberg is registered in a pyiceberg SQL catalog on SQLite, with a local warehouse.** The book has one catalog name, `local`, and every example uses it. The pyiceberg documentation describes SQLite for development, and the examples are development examples. - **Iceberg tables reach DuckDB as Arrow tables.** The book does not use DuckDB's `iceberg` extension, so the examples read Iceberg data through `scan_arrow`. - **Delta Lake tables are found by path, and the book teaches no Delta catalog.** The examples open a folder, and a missing folder is an error. - **Delta tables are read through `QueryBuilder` only.** This is a measurement from the book's build, not a primary source. At the pinned versions, the other Arrow conversion methods aborted the process at exit in about 17 of 20 runs. The `QueryBuilder` path aborted in none of 24 runs, including reads of earlier versions. - **DuckLake is loaded by `connect_ducklake`, with downloads off.** Every example runs offline on the pinned DuckDB. - **The cleanup hazard runs in a temporary folder and prints counts only.** The book shows the failure so the order of the safe steps is clear. It does not print file names, which change on every run. - **Each example prints counts and labelled values only.** The examples name no file, path or ID in their output, so the output reads the same on every run and on every machine. - **REST catalogs have no runnable example.** They need a network, so the chapter describes the protocol from its page and makes no claim about any product. # Move a table between formats and check the copy By the end of this chapter you can copy a table from one format into another, check that the copy holds the same rows, and name what the copy leaves behind. You will also know which of those losses to accept and which ones should stop the move. The case reaches this point when a team changes its format. The pageviews table has lived in Apache Iceberg since the earlier chapters. Suppose the analysts now want it in DuckLake, where they query a catalog file through DuckDB. Another team keeps its tables in Delta Lake and wants the same rows in Iceberg. In both moves the source stays where it is, and the move is a copy into a new table. A copy is a one-time write. Most of the work is in the checks around it. The examples in this chapter pin 0.12.0 for pyiceberg, 1.6.6 for deltalake and 2.0.0 for DuckDB, which also runs DuckLake. Each copy example builds its own small source in a temporary warehouse, from the same three hours of the pageviews cut. That is 1499 rows. Each example runs alone, so none depends on the state that an earlier chapter left behind. ## What a copy keeps and loses {#what-a-copy-keeps-and-loses} A copy reads the rows of one table at one moment and writes them into another table, in another format. Each target library writes its own data files, its own metadata and its own history. What survives the trip is what both sides can represent: the rows, and the columns whose types both sides support. Everything else stays with the source, or has to be set again on the target by hand. The losses are easiest to see as a table. Each row names a part of the table, what the copy does with it, and where this chapter shows the result. | Part of the table | After the copy | Where this chapter shows it | | --- | --- | --- | | Rows and their values | Kept | The digest check | | Column names and types | Kept where both formats have the type | The Delta Lake section | | Commit history | Lost. The copy starts a history of its own | Both copy sections | | Partition spec | Lost unless it is set on the target | The DuckLake section | | Timestamps below a microsecond | Depends on the format: rejected, cut short or kept | The timestamp section | The history is a loss that readers often do not notice. Iceberg records changes as snapshots, and Delta Lake records each commit as a version. The chapter on [snapshots and time travel](#read-an-earlier-snapshot) showed how a reader uses them. A copy starts a new history in the target, so the target's history begins at the copy. The source keeps its own history for as long as you keep the source. Nothing in the copy refers back to it. A copy is also different from registration. A [registered metadata file](#register-from-metadata-file) makes a table visible in a second catalog without touching its files, so both catalogs read the same data. A copy writes new files and leaves the source alone, which means two tables now take writes and can drift apart. Choose a copy when the target must own its files and its history. Choose registration when the two catalogs should share one set of files, and accept that the shared files must stay put. The rows survive because the copy moves values, not files. A copy does not move data files. It reads values out of one table and writes new Parquet files for the other. That is why file names and file counts differ between a source and its copy, and why comparing file names or file counts can't check the copy. The check has to compare the rows. **Working rule:** Before a copy, list what the source holds that the target will not keep: its history, its partition spec and any value finer than the target can store. Give each item one decision, either to accept the loss, to set the property by hand on the target, or to stop the move. The runbook should name every decision. ## Copy the pageviews table into DuckLake {#iceberg-to-ducklake} The copy reads the Iceberg table as an Arrow table with `scan_arrow`, a shared helper that calls pyiceberg's `scan().to_arrow()`, then gives those rows to DuckLake. DuckLake creates an empty table with their columns, sets the partition spec on it, and then inserts the rows with `INSERT ... SELECT`. Arrow is the bridge because each side keeps its own library. pyiceberg reads the source, and DuckDB writes the DuckLake table through its own `ducklake` extension. Neither side reads the other's files. The [catalog chapter](#attach-ducklake-catalog) showed how the DuckLake catalog is attached, and the helper in this chapter attaches it the same way. The source is partitioned by the hour of `hour_utc`, which is the layout that [moving from daily to hourly partitions](#daily-to-hourly-partitions) arrives at. The example builds the source with that spec, then appends three hours, one commit each: Example `iceberg-to-ducklake-copy` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Copy an Iceberg table into DuckLake through Arrow, partitioned from the start, then check its layout and history.""" from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.transforms import HourTransform from lib.lakehouse import (copy_to_ducklake, create_iceberg_table, ducklake_files_by_partition, ducklake_partition_transforms, new_warehouse, read_cut_hour, safe_error, scan_arrow, TABLE) def main(): try: with new_warehouse() as wh: spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=HourTransform(), name="ts_hour")) table = create_iceberg_table(wh, partition_spec=spec) for hour in range(3): table.append(read_cut_hour(hour)) arrow = scan_arrow(table) print(f"source rows: {arrow.num_rows}") print(f"source snapshots: {len(table.metadata.snapshots)}") for field in table.spec().fields: print(f"source partition field: {field.name}") target = copy_to_ducklake(wh, arrow, partition_by="hour(hour_utc)") print(f"copy rows: {wh.con.execute(f'SELECT count(*) FROM {target}').fetchall()[0][0]}") print(f"copy partition transforms: {ducklake_partition_transforms(wh.con, table=TABLE, current=True)}") files = ducklake_files_by_partition(wh.con, "lake", TABLE) print(f"copy partition values: {[value for value, _, _ in files]}") print(f"copy files per partition: {[count for _, count, _ in files]}") print(f"copy rows per partition: {[int(rows) for _, _, rows in files]}") changes = wh.con.execute("SELECT changes FROM ducklake_snapshots('lake') ORDER BY snapshot_id").fetchall() for number, (change,) in enumerate(changes, start=1): print(f"copy snapshot {number}: {', '.join(sorted(change))}") except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text source rows: 1499 source snapshots: 3 source partition field: ts_hour copy rows: 1499 copy partition transforms: ['hour'] copy partition values: ['0', '1', '2'] copy files per partition: [1, 1, 1] copy rows per partition: [499, 500, 500] copy snapshot 1: schemas_created copy snapshot 2: schemas_created copy snapshot 3: tables_created copy snapshot 4: tables_altered copy snapshot 5: tables_inserted_into ``` The helper `scan_arrow` reads the source at its current snapshot and returns every row of that snapshot as one Arrow table. The copy therefore holds the rows of one moment, the last commit that the scan saw. Writers that commit after the scan are not in the copy, and nothing in the copy tells you so. That is the cutover problem, which the production section returns to. The output has four parts, and each one answers a question about the copy. The first part describes the source. It prints the row count, the number of snapshots and the partition fields. The source has 3 snapshots, one for each append, and one partition field, `ts_hour`, an hour transform of `hour_utc`. The second part is the copy's row count, which is 1499. The row count is the same as the source's. It is a useful first check, but the digest section below explains why it is not enough on its own. The third part is the copy's layout, and it is where a copy goes wrong without any error. The example creates the copy empty, runs `ALTER TABLE ... SET PARTITIONED BY (hour(hour_utc))` on it, and only then inserts the rows. The [DuckLake partitioning docs](https://ducklake.select/docs/stable/duckdb/advanced_features/partitioning) say the partition keys "only affect new data written to the table", so an `ALTER` after the rows are in place would leave them unpartitioned. The partition list reads the catalog table `ducklake_partition_column`, as the [DuckLake specification](https://ducklake.select/docs/stable/specification/tables/ducklake_partition_column) defines it, through the attached metadata database. No public function returns it at this pin. The helper keeps only the spec still in force, and it lists one transform, `hour`. The list says what the copy was given. The files say what it holds: the copy writes [1, 1, 1] files, and the rows are divided between them as [499, 500, 500]. A copy made in one statement carries no partition spec, so its list is empty and its rows sit in unpartitioned files. The rows are all there either way, so the check has to look at the layout. The fourth part is the copy's snapshot list. Each line gives a snapshot's number and the changes it made. The numbers count from 1 over every snapshot DuckLake records, including the setup ones, so label N is DuckLake's own snapshot number N−1, because DuckLake counts from 0. The first two snapshots create the default schema and the `wiki` schema. The third creates the empty table, the fourth sets its partition spec, and the fifth writes the rows into the table that is already partitioned. The three hourly appends from the source do not appear. The copy's history starts at its own creation, so time travel on the copy reaches back only that far. To read the source as it was after the first hour, you still need the Iceberg table. The example also sets `DATA_INLINING_ROW_LIMIT` to 0 through the attach helper, so the copied rows are written as Parquet files, and the file counts in the third part count those files. **Working rule:** Copy through Arrow with the target's own writer. Create the target empty, set its partition spec with an `ALTER`, and only then insert the rows. Print the target's partition list and its files per partition. Read the copy's snapshots as the copy's own history, and keep the source until the checks in the last two sections have passed. ## Copy the Delta table into a pyiceberg table {#delta-to-iceberg} The Delta source is built the way a team's pipeline would build it: three appends with deltalake, one per hour. Delta counts its versions from 0, so the three appends are versions 0, 1 and 2. The example reads the table's current rows with `read_delta`, a shared helper that reads through deltalake's `QueryBuilder`. It then appends those rows to an Iceberg table made with the target schema: Example `delta-to-iceberg-copy` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Copy a Delta Lake table into Iceberg through Arrow, then compare the copy with its source.""" from deltalake import DeltaTable from lib.lakehouse import (create_iceberg_table, new_warehouse, print_schema, read_delta, safe_error, scan_arrow, write_delta_hours) def main(): try: with new_warehouse() as wh: path = write_delta_hours(wh.root, range(3)) source = read_delta(path) print(f"delta versions: {DeltaTable(path).version() + 1}") print(f"delta rows: {source.num_rows}") print("delta columns (name, type, nullability):") print_schema(source.schema) table = create_iceberg_table(wh) table.append(source) print(f"iceberg snapshots: {len(table.metadata.snapshots)}") print(f"iceberg rows: {scan_arrow(table).num_rows}") print("iceberg columns (name, type, requirement):") for field in sorted(table.schema().fields, key=lambda field: field.name): requirement = "required" if field.required else "optional" print(f"{field.name}\t{field.field_type}\t{requirement}") except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text delta versions: 3 delta rows: 1499 delta columns (name, type, nullability): domain_code string nullable hour_utc timestamp[us] nullable page_title string nullable response_bytes int64 nullable views int64 nullable iceberg snapshots: 1 iceberg rows: 1499 iceberg columns (name, type, requirement): domain_code string optional hour_utc timestamp optional page_title string optional response_bytes long optional views long optional ``` The output starts with the source. The Delta table has 3 versions and 1499 rows. The Iceberg table is built from one append of all of them, so it has 1 snapshot. Three commits in the source became one commit in the copy. The copy's row count, 1499, is the same as the source's. Like the DuckLake copy, this one starts its own history. The rest of the output lists both schemas, one column per line. Read them side by side: | Column | Delta Lake (Arrow type) | Iceberg type | | --- | --- | --- | | `domain_code` | `string` | `string` | | `hour_utc` | `timestamp[us]` | `timestamp` | | `page_title` | `string` | `string` | | `response_bytes` | `int64` | `long` | | `views` | `int64` | `long` | Names come across unchanged, because the copy takes them from the Arrow schema. The types map one to one. The only spelling change is `int64` in Arrow, which Iceberg calls `long`. Both sides use microseconds for `hour_utc`, and the timestamp check below explains what happens when they do not. Names matter more than they look. pyiceberg checks the Arrow schema against the table's schema before it writes. It stops with a mismatch error when a column name differs, or when a type cannot be promoted to the table's type. It accepts a type it can promote, such as `int` to `long`. That check is in the pinned library's `_check_pyarrow_schema_compatible` function in `pyiceberg/io/pyarrow.py`. The Delta table also cannot rename a column at this pin, as [the rename section of the schema chapter](#rename-in-delta-lake) showed. If the two sides disagree on a name, you change the target, not the source. pyiceberg matches the Arrow columns to the table by name, so the target's field IDs come from the schema you write. You write the target schema yourself, rather than letting the first batch decide it. **Working rule:** Write the target schema yourself, with the types you want, and print both schemas after the copy, as the example does. Accept a near-equal type such as `int64` to `long` only after you have checked it in the output. Do not accept a name that differs. ## Check timestamp precision before the copy {#timestamp-precision} Timestamps are a place where the formats disagree, sometimes silently. The Iceberg specification defines nanosecond timestamps as a version 3 type, so a version 2 table has no type for them ([the Iceberg specification](https://iceberg.apache.org/spec/), version 3 section). The example's table is version 2, the default. At this pin, pyiceberg refuses to write version 3, so the nanosecond values can only be kept by a format other than Iceberg. This example writes two values that differ only below a microsecond, 1000000500 and 1000000999 nanoseconds, to each format: Example `timestamp-precision-check` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Write one timestamp with nanoseconds to each format, and read back what each format kept.""" import pyarrow as pa from deltalake import write_deltalake from pyiceberg.schema import Schema from pyiceberg.types import NestedField, TimestampType from lib.lakehouse import copy_to_ducklake, new_warehouse, NAMESPACE, read_delta, safe_error, sql_rows NANOSECONDS = pa.table({"hour_utc": pa.array([1000000500, 1000000999], pa.timestamp("ns"))}) def main(): try: with new_warehouse() as wh: print(f"written distinct values: {len(set(NANOSECONDS['hour_utc'].cast(pa.int64()).to_pylist()))}") table = wh.catalog.create_table( f"{NAMESPACE}.precision", schema=Schema(NestedField(1, "hour_utc", TimestampType(), required=False)), ) try: table.append(NANOSECONDS) except Exception as exc: print(f"iceberg append: {safe_error(exc, wh)}") path = str(wh.root / "precision") write_deltalake(path, NANOSECONDS, mode="overwrite") kept = read_delta(path)["hour_utc"].cast(pa.int64()).to_pylist() print(f"delta read back as microseconds: {kept}") print(f"delta distinct values read back: {len(set(kept))}") target = copy_to_ducklake(wh, NANOSECONDS) rows = sql_rows(wh.con, f"SELECT epoch_ns(hour_utc) FROM {target}") print(f"ducklake epoch nanoseconds read back: {[row[0] for row in rows]}") except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text written distinct values: 2 iceberg append: UnsupportedPyArrowTypeException: Column 'hour_utc' has an unsupported type: timestamp[ns] delta read back as microseconds: [1000000, 1000000] delta distinct values read back: 1 ducklake epoch nanoseconds read back: [1000000500, 1000000999] ``` The first line is the input: the two values are distinct. Then each format answers in its own way. The Iceberg append fails. The error is UnsupportedPyArrowTypeException: Column 'hour_utc' has an unsupported type: timestamp[ns]. The error names the column and the type, so the failure is the clearest of the three cases, and by default nothing is silently changed. pyiceberg's error names a setting, `PYICEBERG_DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE`, that downcasts nanoseconds to microseconds on write. With it set, the same append succeeds and the values read back as whole microseconds, with only a logged warning. That is the same cut that Delta makes, so set it only for a job that has decided to cut the values, and not in an environment that every writer shares. The Delta write succeeds, and the read back shows [1000000, 1000000]. Both values were cut to whole microseconds, and the two distinct values became one. The write raised no error, and the row count is the same. Only a check on the values shows the loss. The distinct count drops from 2 to 1. DuckLake keeps the nanoseconds. Its table stores the two values as they were written, and the example reads them back as [1000000500, 1000000999]. The pageviews table records hours, so the case never reaches the problem. A table of raw request logs does. Its timestamps come from a clock that counts finer than a microsecond, and its copy has to choose a rule for those digits. Ask the question for every timestamp column before the first copy, because the answer is only visible in the values. The copies in this chapter use microsecond timestamps, which all three formats keep, so the copies match. A real table can hold values below a microsecond from its source, and the copy has to decide what to do with them before the first write. **Working rule:** Check timestamp precision with a value below a microsecond before the first copy. Choose the rule in advance. Cast the column to microseconds and count the values the cast changes, reject the copy, or write to a format that keeps nanoseconds. Do not leave the choice to the writer library, because a silent cut gives no error to act on. ## Check each copy with a sorted digest {#check-copy-with-digest} A matching row count does not prove that the values match. A copy can keep the count and change one value, and the count will not show it. A digest does. `digest_rows` sorts the rows, writes each one out as a line of text and hashes the result. The sort makes the digest independent of the order in which the rows come back, so two copies with the same rows get the same digest, however each format returns them. The example runs the check on both copies, and then on a copy that has one view count changed on purpose: Example `verify-copies-by-digest` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Check each copy with a sorted digest, and show that one changed value fails the check.""" from lib.lakehouse import (copy_to_ducklake, create_iceberg_table, digest_rows, new_warehouse, read_cut_hour, read_delta, safe_error, scan_arrow, sql_rows, write_delta_hours) def same(source, copy): return digest_rows(source) == digest_rows(copy) def check_iceberg_to_ducklake(): with new_warehouse() as wh: table = create_iceberg_table(wh) for hour in range(3): table.append(read_cut_hour(hour)) target = copy_to_ducklake(wh, scan_arrow(table)) source_rows = sql_rows(wh.con, "SELECT * FROM source", source=scan_arrow(table)) copy_rows = wh.con.execute(f"SELECT * FROM {target}").fetchall() first = copy_rows[0] changed = [(*first[:3], first[3] + 1, first[4]), *copy_rows[1:]] print(f"iceberg to ducklake rows: {len(source_rows)} and {len(copy_rows)}") print(f"iceberg to ducklake match: {same(source_rows, copy_rows)}") print(f"iceberg to ducklake with one changed view count match: {same(source_rows, changed)}") def check_delta_to_iceberg(): with new_warehouse() as wh: path = write_delta_hours(wh.root, range(3)) table = create_iceberg_table(wh) table.append(read_delta(path)) source_rows = sql_rows(wh.con, "SELECT * FROM source", source=read_delta(path)) copy_rows = sql_rows(wh.con, "SELECT * FROM copy", copy=scan_arrow(table)) print(f"delta to iceberg rows: {len(source_rows)} and {len(copy_rows)}") print(f"delta to iceberg match: {same(source_rows, copy_rows)}") def main(): try: check_iceberg_to_ducklake() check_delta_to_iceberg() except Exception as exc: print(safe_error(exc)) raise SystemExit(1) if __name__ == "__main__": main() ``` Output: ```text iceberg to ducklake rows: 1499 and 1499 iceberg to ducklake match: True iceberg to ducklake with one changed view count match: False delta to iceberg rows: 1499 and 1499 delta to iceberg match: True ``` For the [Iceberg-to-DuckLake copy](#iceberg-to-ducklake), the output prints both row counts, which are equal, and then the digest comparison. It reports `True`. The same check on the Delta-to-Iceberg copy also matches. The changed copy reports `False`. One changed number is enough to change the digest, so a digest that matches is evidence, and a digest that differs is a stop. The digest compares the values as the database returns them, and it has a limit you should know. If both sides are cast to text before hashing, two values of different types can produce the same text. The integer `1` and the text `'1'` are the same text. Compare typed values, as the example does, and print the schemas as the [Delta-to-Iceberg copy](#delta-to-iceberg) did. The digest also says nothing about column names or types. Two tables with the same rows under different column names have the same digest. The schema listings in the earlier sections check the names, and the digest checks the values, so a copy needs both. The digest is not a hash of the files. The rows are sorted by their typed values, each becomes one line of its Python representation, and the joined text is hashed with SHA-256. Two rows with equal values produce equal lines, which is what the comparison needs. A change in one value changes one line and so the whole digest. Also choose the moment of the check. A digest is true for the rows that existed when it ran. Any write to the source after that point makes the copy stale, and the digest from before the write no longer applies. The check belongs after the last write to the source, and before anyone reads from the copy. **Working rule:** Digest both sides after the last write to the source, over the same columns, with typed values. Compare the row counts and the digests together. Do not move readers or delete the source until both match. If writers ran during the copy, repeat the copy and the digest. ## What goes wrong in production {#migration-production-failures} Most failed migrations pass the first check and fail later. These are the failures to look for, and how you would notice each one. - **A cutover that races a writer.** A digest is true at the moment it is taken. The first append after it makes the copy stale, and the digest does not warn you. Stop the writers, copy, digest and then move the readers. If the writers cannot stop, copy in windows and check the rows that arrived after the last window. - **Deleting the source before the checks pass.** The source holds the history that the copy does not. Once the source is gone, time travel into its snapshots and versions goes with it. Keep the source until the digests match and every reader has moved. Then decide how long to keep it, in writing. - **A copy with the rows and without the layout.** A copy made without the `ALTER` would hold every row and match the digest, with no partition set. The partition list and the files per partition are the checks that show the layout is missing, so print both after the copy. - **Timestamps that change without an error.** [Delta Lake](#timestamp-precision) cut the nanosecond values to microseconds and raised no error. The row count stayed the same. A check on the values is the only thing that catches it, so a copy with sub-microsecond data needs that check before it runs. - **Names and types that differ.** pyiceberg refuses an append whose column names differ from the table's, or whose type it cannot promote, and that refusal is the clearest failure. The harder case is a column whose name matches and whose type it accepts, such as a `long` that was an `int`. Print both schemas and read them, rather than trusting that the append succeeded. - **Size and memory.** The helpers in this chapter read a whole table into memory before they write it. That suits 1499 rows. A table of billions of rows needs a copy that reads and writes in batches, and the digest must then be computed over the same batches on both sides, so it stays a check of the whole table. - **Readers that write to both tables.** During a copy, a job that writes to both tables by mistake makes the two tables differ, and the digest will show a mismatch that no single write explains. Name one writer per table in the runbook before the cutover starts. - **A digest over text.** A digest over casts can match when the types differ. The check in this chapter compares typed values, and the runbook should say so, so that a later edit to the check does not quietly drop the types. ## Decisions this chapter made {#migration-decisions} - **Copies go through Arrow, and each target is written by the library that owns its format.** pyiceberg writes Iceberg, DuckDB's `ducklake` extension writes DuckLake, and deltalake writes Delta Lake. This book does not use DuckDB's `iceberg` or `delta` extensions, so no copy in it depends on them. - **Delta tables are read through `read_delta`,** which uses deltalake's `QueryBuilder`. Every read of a Delta table in this chapter takes that path. - **A copy is one-time, starts a new history and is checked before anyone switches to it.** The source stays in place until the digests match and the readers have moved. - **The partition spec is set on the target by hand, with an `ALTER` before the rows are inserted, and printed.** The copy does not bring it across. - **Each copy example builds its own source from the same three hours of the cut.** Every example runs alone, and each check reads only the rows that its own example wrote. - **Checks run on typed values, after the last write, over the same columns.** The row count and the digest are read together, and neither one is used alone to call a copy good. - **The copies use microsecond timestamps, and a separate example checks sub-microsecond values.** Each copy in this chapter keeps its microsecond values, so the digests match. The precision example shows what each format does with a value it cannot keep. - **Metadata-only migration is not used here.** pyiceberg's `Table.add_files` adds Parquet files that already exist to an Iceberg table as data files (`Table.add_files` in `pyiceberg/table/__init__.py`, pyiceberg 0.12.0). Its [API docs](https://py.iceberg.apache.org/api/), which are not pinned to that version, describe this as committing the files without rewriting them. That keeps the data files in place, which a copy does not. This chapter did not run it, and it copies rows instead. # Choose a format from what your tools support By the end of this chapter, you can list every tool that reads or writes your tables, measure what each one supports at its pinned version, and choose a format and its features from those measurements. Across chapters 1 to 9, one table, `wiki.pageviews_hourly`, went through its whole life in three formats: Apache Iceberg, Delta Lake and DuckLake. Each format did the same jobs in a different way, and each library supported a different subset of them. The choice you face is not which format is best in general. It is which format every tool in your team can open, change and clean up, at the versions you run. This chapter sets out how to measure that, shows the measurements for this book's pins, and gives a short guide for the choice. In the running case, the readers of `wiki.pageviews_hourly` are a dashboard that queries through DuckDB, a pipeline that appends through pyiceberg, and an archive job that reads a Delta Lake copy through deltalake. Each reader has its own pin, and each one must open the data the others write. That is the situation the rest of this chapter prepares you for. Three things make the choice harder than it looks. The features that differ most between formats, such as deletes, renames and partition changes, are the ones a team needs first. Each library changes on its own schedule, so a table one tool reads at one version may not open at another. And a check that passes on one machine says nothing about the versions your other tools run. The sections in this chapter deal with each of these in turn. ## Measure support before you choose {#measure-support-before-choosing} A format name does not tell you what a tool supports. Support belongs to a tool at one version: a writer can create a feature that a reader cannot open, and a vendor page describes the vendor's engine, not the library on your laptop. So start from the readers, not the format. Write down every tool that will read the table, with its version, and then test each feature you plan to use on each of those versions. The examples measure in that order: the reader first, then the feature. The `iceberg-format-versions` example creates an Iceberg table in format version 2, deletes rows from it, and asks pyiceberg for format version 3. The `feature-check` example turns on deletion vectors in a Delta Lake table and reads it back. The `ducklake-catalog-version` example prints a DuckLake catalog's version, and the `ducklake-delete-files` and `ducklake-inlined-delete` examples count delete files after a small and a large delete. Each prints one labelled line per result, so you can compare the output with the tables in [the support matrix](#support-matrix-as-of-pins) and with your own readers. The person who owns the table should run the measurements on the machines that run the readers, not on a developer's laptop with different versions. Keep the record in a small table with one row per reader. The columns are the reader, its pinned version, the features your tables need and the date you measured them. A feature with no measurement counts as unsupported, and that rule is the one that prevents most of the failures later in this chapter. | Reader | Pinned version | Features you need | Measured on | | --- | --- | --- | --- | | pyiceberg | 0.12.0 | Format version 2 writes, deletes | Date of your run | Read each printed line as a claim that holds at one version of one tool. A line that says a write was refused is as useful as one that says it worked, because the refusal tells you which format version to drop. The examples print class names, not messages, so the output stays the same when a library rewords an error. **Working rule:** Measure each reader on its own pinned version before you choose a format version or a feature. Write each result down with the version it was measured at. Run the measurement again whenever any reader's version changes, and do not reuse a support table from a vendor page. ## Read what the pinned tools did {#support-matrix-as-of-pins} These tables state what the pinned tools did in this book's runs. Each column is one tool: pyiceberg for Iceberg, deltalake for Delta Lake, and the DuckLake extension in DuckDB for DuckLake. The pins are pyiceberg 0.12.0, deltalake 1.6.6 and DuckDB 2.0.0. DuckDB 2.0.0 is a long-term support release, dated Oct 21, 2026 ([DuckDB release calendar](https://duckdb.org/release_calendar)). DuckLake 1.0 was released on Apr 13, 2026 ([DuckLake release calendar](https://ducklake.select/release_calendar.html)). The catalog at these pins reports version 1.1, and that is the version readers must open. The extension is a separate version: DuckDB reports its build as 417a9117f5, and the extension ships with DuckDB 2.0.0, so its version follows DuckDB's. The last column names the example or source behind each cell. Not shown here means a gap in this book, not a missing feature. Example `ducklake-catalog-version` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Attach a DuckLake catalog and print its catalog version and the version of the DuckLake extension that DuckDB loaded.""" from lib.lakehouse import attach_ducklake, connect_ducklake, new_warehouse def main(): with new_warehouse() as wh: con = connect_ducklake(wh.con) attach_ducklake(con, "probe", wh.root) version = con.execute("SELECT value FROM __ducklake_metadata_probe.ducklake_metadata WHERE key = 'version'").fetchall()[0][0] print("ducklake catalog version:", version) extension = con.execute("SELECT extension_version FROM duckdb_extensions() WHERE extension_name = 'ducklake'").fetchall()[0][0] print("ducklake extension version:", extension) if __name__ == "__main__": main() ``` Output: ```text ducklake catalog version: 1.1 ducklake extension version: 417a9117f5 ``` ### Writing and reading versions {#matrix-writing-and-reading} | Versions and reads | pyiceberg | deltalake | DuckLake | Shown in | | --- | --- | --- | --- | --- | | Write rows as commits | Yes | Yes | Yes | [Land three hours](#land-three-hours) | | Read an earlier commit by number | Yes, by snapshot | Yes, by version | Yes, by snapshot | [Iceberg](#snapshots-in-pyiceberg), [Delta and DuckLake](#versions-in-ducklake-and-delta) | | Create format version 3 | No, raises an error | — | — | [Example](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/iceberg-format-versions.py) | | Rename a column | Yes, by field ID | No, a silent no-op | Not shown here | [Iceberg](#rename-column-without-rewrite), [deltalake](#rename-in-delta-lake) | | Column mapping at create | — | Yes | — | [deltalake](#rename-in-delta-lake) | ### Changing and cleaning up {#matrix-changing-and-cleaning-up} | Changes and cleanup | pyiceberg | deltalake | DuckLake | Shown in | | --- | --- | --- | --- | --- | | Delete rows | Rewrites files | Rewrites files | Writes delete files | [Chapter 5](#delete-rows-iceberg-and-delta) | | Delete files written | None | None | Yes, unless inlined | [Format versions](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/iceberg-format-versions.py), [feature check](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/feature-check.py), [DuckLake](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/ducklake-delete-files.py), [inlined](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/ducklake-inlined-delete.py) | | Upsert or merge | Upsert, several snapshots | Not shown here | `MERGE INTO` | [Upserts](#upserts-in-pyiceberg), [MERGE](#merge-in-ducklake) | | Change partitioning | Old files keep the old spec | Overwrite with schema mode | Change with `ALTER TABLE` | [Chapter 4](#daily-to-hourly-partitions), [other formats](#partitioning-in-other-formats) | | Compact small files | No method; overwrite the table | `optimize.compact()` | Merge adjacent files | [Iceberg](#compact-iceberg-by-rewriting), [deltalake](#compact-in-delta-lake), [DuckLake](#compact-in-ducklake) | | Expire snapshots | By ID or `older_than`; deletes no files | Vacuum by retention | Expire, then clean up | [Iceberg](#expire-snapshots-in-iceberg), [deltalake](#vacuum-dry-run-first), [DuckLake](#expire-and-clean-up-in-ducklake) | Example `ducklake-delete-files` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Load six hours into a DuckLake table with data inlining off, then count the delete files that a small and a large delete write.""" from lib.lakehouse import DUCKLAKE_ALIAS, NAMESPACE, TABLE, connect_ducklake, create_ducklake_table, new_warehouse, read_cut_hour LOW_VIEWS = 180 VERY_LOW_VIEWS = 1000 def delete_files(con): return con.execute(f"SELECT count(*) FROM ducklake_list_files('{DUCKLAKE_ALIAS}', '{TABLE}', schema => '{NAMESPACE}') " "WHERE delete_file IS NOT NULL").fetchall()[0][0] def main(): with new_warehouse() as wh: con = connect_ducklake(wh.con) create_ducklake_table(con, wh.root, inlining_row_limit=0) table = f"{DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE}" for hour in range(6): con.register("hour", read_cut_hour(hour)) con.execute(f"INSERT INTO {table} SELECT * FROM hour") con.unregister("hour") con.execute(f"DELETE FROM {table} WHERE hour_utc = TIMESTAMP '2024-06-03 02:00:00' AND views < {LOW_VIEWS}") print("rows left after the small delete:", con.execute(f"SELECT count(*) FROM {table}").fetchall()[0][0]) print("delete files after the small delete:", delete_files(con)) con.execute(f"DELETE FROM {table} WHERE views < {VERY_LOW_VIEWS}") print("rows left after the large delete:", con.execute(f"SELECT count(*) FROM {table}").fetchall()[0][0]) print("delete files after the large delete:", delete_files(con)) if __name__ == "__main__": main() ``` Output: ```text rows left after the small delete: 2995 delete files after the small delete: 1 rows left after the large delete: 195 delete files after the large delete: 6 ``` Example `ducklake-inlined-delete` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Load six hours into a DuckLake table with DuckLake's default data inlining, then count the delete files that a small and a large delete write.""" from lib.lakehouse import DUCKLAKE_ALIAS, NAMESPACE, TABLE, connect_ducklake, create_ducklake_table, new_warehouse, read_cut_hour LOW_VIEWS = 180 VERY_LOW_VIEWS = 1000 def delete_files(con): return con.execute(f"SELECT count(*) FROM ducklake_list_files('{DUCKLAKE_ALIAS}', '{TABLE}', schema => '{NAMESPACE}') " "WHERE delete_file IS NOT NULL").fetchall()[0][0] def main(): with new_warehouse() as wh: con = connect_ducklake(wh.con) create_ducklake_table(con, wh.root, inlining_row_limit=None) table = f"{DUCKLAKE_ALIAS}.{NAMESPACE}.{TABLE}" for hour in range(6): con.register("hour", read_cut_hour(hour)) con.execute(f"INSERT INTO {table} SELECT * FROM hour") con.unregister("hour") con.execute(f"DELETE FROM {table} WHERE hour_utc = TIMESTAMP '2024-06-03 02:00:00' AND views < {LOW_VIEWS}") print("rows left after the small delete:", con.execute(f"SELECT count(*) FROM {table}").fetchall()[0][0]) print("delete files after the small delete:", delete_files(con)) con.execute(f"DELETE FROM {table} WHERE views < {VERY_LOW_VIEWS}") print("rows left after the large delete:", con.execute(f"SELECT count(*) FROM {table}").fetchall()[0][0]) print("delete files after the large delete:", delete_files(con)) if __name__ == "__main__": main() ``` Output: ```text rows left after the small delete: 2995 delete files after the small delete: 0 rows left after the large delete: 195 delete files after the large delete: 6 ``` ### Catalogs and DuckDB {#matrix-catalogs-and-duckdb} | Catalogs and DuckDB | pyiceberg | deltalake | DuckLake | Shown in | | --- | --- | --- | --- | --- | | Catalog needed | Yes, a SQL catalog | No, found by path | Yes, the catalog file | [Chapter 8](#what-a-catalog-holds), [by path](#find-delta-by-path), [catalog file](#attach-ducklake-catalog) | | DuckDB reads it | Through Arrow | Through Arrow, from `QueryBuilder` | Natively | [Chapter 1](#query-with-duckdb), [Chapter 9](#check-copy-with-digest), [Chapter 8](#attach-ducklake-catalog) | | DuckDB writes it | — | — | Natively, through SQL | [Chapter 8](#attach-ducklake-catalog) | DuckDB's own `delta` extension is a separate path from deltalake. It can write Delta Lake tables, which is covered in the decision guide below. **Working rule:** Read each cell against the example or source it names. A cell with no example behind it is a claim to test on your own readers, not one to rely on. ## pyiceberg writes no version 3 table {#iceberg-v3-and-deletion-vectors} The version 3 section of the Iceberg specification makes binary deletion vectors the row-level delete format. It says that "writers are not allowed to add new position delete files to v3 tables" ([Iceberg spec](https://iceberg.apache.org/spec/)). The same section adds row lineage for new rows, default column values, the variant type and nanosecond timestamps. Example `iceberg-format-versions` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Create a format version 2 Iceberg table, delete rows from it, then ask pyiceberg for format version 3.""" from pyiceberg.expressions import LessThan from lib.lakehouse import create_iceberg_table, new_warehouse, read_cut_hour, scan_arrow LOW_VIEWS = 180 def main(): with new_warehouse() as wh: table = create_iceberg_table(wh, properties={"format-version": "2"}) table.append(read_cut_hour(0)) print("format version:", table.metadata.format_version) before = scan_arrow(table).num_rows table.delete(LessThan("views", LOW_VIEWS)) print("rows deleted:", before - scan_arrow(table).num_rows) contents = table.inspect.files().column("content").to_pylist() print("delete files after the delete:", sum(1 for content in contents if content != 0)) for number, snapshot in enumerate(table.metadata.snapshots, start=1): print(f"snapshot {number} operation:", snapshot.summary.operation.value) try: create_iceberg_table(wh, identifier="wiki.format_three", properties={"format-version": "3"}) except NotImplementedError as exc: print("format version 3 create:", type(exc).__name__) else: print("format version 3 create: created") if __name__ == "__main__": main() ``` Output: ```text format version: 2 rows deleted: 82 delete files after the delete: 0 snapshot 1 operation: append snapshot 2 operation: overwrite format version 3 create: NotImplementedError ``` pyiceberg at this pin writes format version 2. The example prints the format version of its table, which is 2. The delete removed 82 rows and left 0 delete files. The table has two snapshots: the first is an `append`, and the second, the delete, is an `overwrite`. The same example asks for format version 3, and the create raises `NotImplementedError`, so none of the version 3 features is available from this writer. pyiceberg deletes by rewriting the data files it touches at these pins, although format version 2 also allows delete files ([chapter 5](#copy-on-write-and-merge-on-read)). Measure that rewrite on a copy of your own table before you choose. | Feature | Iceberg v3 specification | pyiceberg at these pins | | --- | --- | --- | | Binary deletion vectors | Row-level delete format | Not written | | Position delete files | Legacy, in format version 2 tables only | Not written | | Row lineage, defaults, variant, nanosecond timestamps | Part of version 3 | Not available: no version 3 table | **Working rule:** Write format version 2 with pyiceberg at these pins, and do not promise a version 3 feature to any reader. Choose version 3 only when the writer you use creates it and every reader in your set opens it, and check both on their own pins. ## Deletion vectors need a protocol upgrade {#delta-deletion-vectors} The Delta Lake protocol, at the release tag its link names, says a table with deletion vectors needs reader version 3 and writer version 7, with the `deletionVectors` feature in both feature lists. It also says a reader must implement every feature in the reader list before it reads the table ([Delta protocol](https://raw.githubusercontent.com/delta-io/delta/v4.4.1/PROTOCOL.md)). A feature is a promise that the reader handles it, so a reader without it cannot read the table correctly. Example `feature-check` (Python 3.12.15 · deltalake 1.6.6 · duckdb 2.0.0 · pyarrow 26.0.1 · pyiceberg 0.12.0), passing in Edition 2026.11.1: ```python """Turn on deletion vectors in a Delta Lake table, delete the rows below 180 views, and print the protocol the table declares.""" from deltalake import DeltaTable, write_deltalake from lib.lakehouse import new_warehouse, read_cut_hour, read_delta LOW_VIEWS = 180 def deletion_vector_files(path): """Count files under the table folder that are not Parquet or the Delta log.""" return sum(1 for item in path.rglob("*") if item.is_file() and item.suffix != ".parquet" and "_delta_log" not in item.parts) def main(): with new_warehouse() as wh: path = wh.root / "dv_table" write_deltalake(str(path), read_cut_hour(0), configuration={"delta.enableDeletionVectors": "true"}) table = DeltaTable(str(path)) protocol = table.protocol() print("minimum reader version:", protocol.min_reader_version) print("minimum writer version:", protocol.min_writer_version) print("reader features:", ", ".join(sorted(protocol.reader_features))) before = read_delta(str(path)).num_rows metrics = table.delete(f"views < {LOW_VIEWS}") print("rows deleted:", before - read_delta(str(path)).num_rows) print("data files rewritten:", metrics["num_removed_files"]) print("rows through QueryBuilder:", read_delta(str(path)).num_rows) print("deletion vector files:", deletion_vector_files(path)) if __name__ == "__main__": main() ``` Output: ```text minimum reader version: 3 minimum writer version: 7 reader features: deletionVectors, timestampNtz, variantType rows deleted: 82 data files rewritten: 1 rows through QueryBuilder: 417 deletion vector files: 0 ``` The example enables the deletion vector setting on a table of pageviews, deletes the rows below 180 views and prints what the table declares. The minimum reader version is 3, and the minimum writer version is 7. The reader features are `deletionVectors, timestampNtz, variantType`. The example set only the deletion vector setting, so the library added the other two features on its own. The delete removed 82 rows and rewrote 1 data file. The table holds 0 deletion vector files, and `QueryBuilder` reads 417 rows. So at this pin deltalake declares the feature but deletes by rewriting the file. The setting changes the protocol that readers must support, not the way deletes write. The book reads Delta tables only through `QueryBuilder`, as [chapter 8](#find-delta-by-path) shows. Every feature in the list matters, not only the deletion vector. A reader that lacks `timestampNtz` or `variantType` has the same problem as one that lacks `deletionVectors`. Compare the printed feature lists with each reader's own list after every change. So a deletion vector setting is a protocol change for every reader of the table, and the write succeeds before any reader is tested. Turn it on only when every reader in your set implements the `deletionVectors` feature. Otherwise keep copy-on-write. A reader check, in the sense used here, opens the table through the reader's own interface, before the first production write. For the Delta Lake copy in the case, that check is a read through `QueryBuilder` and a read of the protocol, both in the feature-check example. **Working rule:** Enable deletion vectors only after you have checked each reader's feature support on its own version. Read the table's protocol back after each change, and keep copy-on-write for any table that a reader outside your tested set will open. ## A checklist for a new table {#checklist-for-new-table} Use this list before you create a table that more than one tool will open. Each item closes a risk that [the support matrix](#support-matrix-as-of-pins) and the [deletion vector section](#delta-deletion-vectors) showed. 1. List every reader and writer, with its version. A tool you leave off the list is a reader you have not tested, and it will be the one that fails. 2. For each feature the table needs, measure its support on each tool, as the examples in this chapter do. Measure the reader, not only the writer, because a successful write proves only that the writer can produce the files. 3. Set the format version and the protocol on purpose. Print them back, and read the table from each reader. Do not accept a default you did not choose. 4. Refuse any format version or feature that one reader in the set does not support. A table that one reader cannot open is an outage, however good the features are for the other readers. 5. Promise a rename only where the measured support allows it. Iceberg renames by field ID. deltalake has no rename at these pins, so plan a rewrite instead, and check the column names afterwards. 6. Keep a plain copy-on-write path for deletes unless every reader has been tested with deletion vectors. Copy-on-write costs rewritten files, and it opens everywhere. 7. Plan any upgrade in this order: readers first, then writers, then the protocol. Read the table after each step, and stop if a reader fails. 8. Run the measurements again on every pin change, and record the new versions in the table of readers. **Working rule:** Keep the list with the table's design note, so the next person who adds a reader starts from the same record. A list kept up to date costs less than one rebuilt after a failed read. ## Pick a format from your readers {#choice-decision-guide} Use this guide after you have measured your readers. The first column is the situation, not the format. It maps a measured reader set to a format, and it does not replace the measurements. | If your situation is | Choose | Why | | --- | --- | --- | | pyiceberg and DuckDB read the table | Iceberg, format version 2 | pyiceberg reads it, and DuckDB queries its output | | Several readers, and you need in-place renames | Iceberg | Renames keep the field ID, as [chapter 3](#rename-column-without-rewrite) shows | | DuckDB is the only engine, and you want SQL changes | DuckLake | Deletes, merges and partition changes run as SQL | | Delta tables already exist and deltalake reads them | Keep Delta Lake, with no rename promise | Change partitions by overwrite, and read back after each change | | You want DuckDB to change a Delta table | Not at these pins | DuckDB can only append; use deltalake for updates and deletes | DuckDB's `delta` extension can write Delta Lake tables. The DuckDB documentation for the pinned release, 2.0.0, says it can create tables and append to them, and it lists `UPDATE`, `DELETE`, `MERGE INTO`, `ALTER TABLE` and `DROP TABLE` as not supported ([DuckDB delta docs, Limitations](https://duckdb.org/docs/2.0/core_extensions/delta.html)). No example in this book runs the extension. The examples install nothing at run time and use each format's own library, so the reader sees the same library that writes the table, and nothing downloads while an example runs. Read the table row by row. When two rows point in different directions, the readers decide. First choose the format that every reader in your measured set opens, then choose features that every reader supports. Apply the guide to the running case. The dashboard and the pipeline both open the pageviews table with pyiceberg or DuckDB, which points to Iceberg format version 2. The archive job reads a Delta Lake copy, so the Delta row applies to that copy only, and the copy keeps its deletes as copy-on-write. We recommend Iceberg format version 2 as the default for a table that several tools read, because pyiceberg writes it, DuckDB reads its output, and the examples measure both. Its costs at these pins: every delete rewrites the data files it touches, the table needs a catalog, and pyiceberg has no compaction or orphan-file cleanup call. We recommend DuckLake when DuckDB is the only engine, because the reader and the writer share one SQL surface. Its costs: readers are limited to DuckDB, the catalog version is tied to the DuckDB release, and small deletes are inlined unless you turn inlining off. Choose differently when the checks in this chapter show that a reader cannot do what you need. **Working rule:** Pick the format that every reader in your measured set opens, then choose only the features that set supports. If no format meets every need, split the table or change the readers. Do not promise a feature to a reader that the measurements do not cover. ## What goes wrong in production {#choice-production-failures} Most format failures in a shared lakehouse come from a tool the writer did not test, not from a bug in the format. Each failure in this section is one that an earlier section measured or cited, so you can check for it in your own tables. **Choosing by vendor claims.** A support page describes one engine, often at a different version from yours. Use it to find what to test, not as the answer. Check the claim against your own pins. **Testing one engine only.** A table can be written by one library and read by another that does not support the same feature. The write succeeds in one tool, and the failure appears when a second tool opens the table. Run the reader check for each tool in the set before the first write. **A feature the reader does not implement.** The deletion vector setting adds a reader feature to the table, and the protocol requires readers to implement it. A reader outside your set may open the table and fail on it. Find out from the protocol record, which the feature-check example prints, and not from a later query error. **A rename that does nothing.** At these pins, deltalake's `set_column_metadata` call returns without an error and bumps the version, but the column names stay the same. [Chapter 3](#rename-in-delta-lake) shows the trap. The check is to read the column names after the change. **Upgrading in the wrong order.** A writer that moves ahead of its readers can produce a table that the readers cannot open, and the first sign is a failed read in another team's job. The order in the checklist keeps every reader able to open the table at each step. **Judging a table by its write.** A write that succeeds shows that the writer can produce the files. It does not show that another tool can read them. The deletion vector check in this chapter writes, deletes and then counts the reader's features for exactly this reason. **Counting rows for a DuckLake delete.** Under DuckLake's default inlining, a small delete of a few rows writes no delete file, and the same delete with inlining off writes one. A check that counts rows says the delete worked either way. Count delete files too, as the [DuckLake delete example](https://github.com/rollingedition/book-open-table-formats/blob/997d15495139d2b136021ddc6de32d54d3738c81/examples/ducklake-delete-files.py) does. **A catalog version that moves with DuckDB.** The DuckLake catalog at these pins reports version 1.1, and the extension that opens it reports the build 417a9117f5. The DuckLake documentation names the DuckDB release that version 1.1 requires ([DuckLake docs](https://ducklake.select/docs/stable/)). An older DuckDB may not open that catalog, so an upgrade can break a reader that did not change. **Working rule:** Measure every reader on its own pinned version before you write, upgrade or rely on a vendor claim. When you upgrade a writer or a catalog, keep the old release in place until every reader has opened the result. ## Decisions this chapter made {#choice-decisions} - We recommend Iceberg format version 2 with pyiceberg at these pins. We do not promise format version 3 to any reader, and we set no date for when that changes. The pin decides, and the measurement shows it. - We state the matrix as measured at these pins, and each cell points to an example or a cited source. - We keep deletion vectors out of tables that a reader outside the tested set will open. - We do not promise a rename on a deltalake table at these pins. - We count DuckLake delete files as well as rows. - We treat an unmeasured feature as unsupported, because the failures in this chapter all came from a feature someone assumed. - We state the DuckLake catalog version apart from the extension version. The catalog version decides which readers can open the catalog, and the extension version follows DuckDB. - We chose a measured matrix over a vendor's support page, because the pinned versions decide what opens. - The default of Iceberg format version 2 is a judgement, labelled as one. The measurements behind it are the examples, and they are the part to re-run. - The book runs no DuckDB `delta` or `iceberg` extension. Its examples install nothing at run time and use each format's own library. The DuckDB `delta` extension can write Delta Lake tables, as its documentation describes.