Time travel and rollback in Iceberg: undoing a bad write in one statement

• Overview
• The core problem and architecture
  • Why a Hive table cannot be undone
  • The metadata tree a snapshot points into
  • What time travel and rollback each move
• Under the hood: engine internals and file layouts
  • Write: what one commit leaves on storage
  • Read: how a query resolves a snapshot
  • Rollback: moving main and what history keeps
  • Expiry: the only step that deletes files
• Hands-on demo: datalake-playground and PySpark
  • Environment
  • Script 1: time travel and rollback
  • Script 2: incremental reads, write-audit-publish and retention
• Real-world scenarios and production failures
  • Finding the snapshot before the bad write
  • A dashboard keeps serving the bad rows after the rollback
  • An incremental consumer silently skips deletes and updates
  • The branch write that went to main
  • Write-audit-publish instead of rollback
  • Expiry removed the snapshot you needed
• Optimisation, tuning knobs and best practices
  • Retention settings
  • Retention is your storage bill and your undo button at once
  • Catalog cache
  • Checklist
• Summary and key takeaways
• References
• Trademarks

TL;DR

• Reading an old state is a query: VERSION AS OF <snapshot_id> returned 2 rows on a table whose current state had 3.
• Undoing a bad write is one call. After a bad load took an orders table from 3 rows to 5, rollback_to_snapshot returned it to 3, and wrote one metadata file and no data.
• A rollback does not erase history. The history table kept the abandoned snapshot with is_current_ancestor=false, and TIMESTAMP AS OF a moment inside the bad window still returns the bad rows.
• A rollback is not instant for every reader. A session that had already cached the table kept serving the bad state until REFRESH TABLE.
• Your recovery window is exactly your snapshot retention. Expiry is a recovery policy, not a cleanup chore.

Overview

A nightly load replays yesterday’s file and a table of orders now double-counts revenue. The fix you want is “put the table back the way it was at 01:00”, in one statement, without a restore from backup, and without rewriting a byte of data. On Iceberg that fix exists because the table never overwrites its own past.

This post follows that incident through the engine. It covers what a snapshot is on disk and how VERSION AS OF and TIMESTAMP AS OF resolve to one. It shows what rollback_to_snapshot changes, and what it leaves behind. Then it covers the four production traps found while re-running this post:

• Stale readers that keep serving the rolled-back state.
• Incremental consumers that skip deletes without a word.
• A branch write that lands on main.
• A retention setting that quietly deletes your undo button.

Every output below comes from the scripts in the hands-on demo, run on the local datalake stack.

The core problem and architecture

Why a Hive table cannot be undone

A Hive table is a directory. INSERT OVERWRITE on a partition deletes the old files and writes new ones. Nothing records what the partition held a minute earlier. Recovery means a backup, an HDFS snapshot taken in advance, or re-running every upstream job. Readers that list the directory during the overwrite can also see a half-written partition, because there is no atomic switch between “old files” and “new files”.

Iceberg’s design decision is the inverse. Files are immutable, metadata is immutable, and the only mutable thing is one pointer in the catalog. A commit writes new files and new metadata next to the old, then swaps the pointer. The old metadata is still on storage, so the old state is still addressable.

The metadata tree a snapshot points into

 catalog (Hive Metastore)
   metadata_location ──┐
                       v
 metadata/00003-....metadata.json  <- current table metadata
   current-snapshot-id: S3
   snapshots:    S1, S2, S3        <- every retained snapshot
   snapshot-log: S1@t1, S2@t2, S3@t3
   refs:         main -> S3
        │
        │  each snapshot names one manifest list
        v
 metadata/snap-S3-...avro          <- manifest list for S3
   ├── manifest for S1's files ──> data/order_date=2026-09-20/...parquet
   ├── manifest for S2's files ──> data/order_date=2026-09-21/...parquet
   └── manifest for S3's files ──> data/order_date=2026-09-21/...parquet

Read it top down.

1. The catalog stores one thing: the path of the current metadata.json.
2. That file lists every retained snapshot, a complete description of the table at one commit. It also keeps a snapshot log of which snapshot was current when, and the refs: named pointers such as the main branch.
3. Each snapshot names a manifest list. The list names manifests, and each manifest lists data files with their partition values and column statistics.

A new snapshot reuses the manifests it did not change. That makes snapshots cheap: S3 above adds one manifest and points at the two it inherited. It does not make them free, as the storage section shows.

What time travel and rollback each move

                 time travel reads here
                          │
 S1 ──> S2 ──> S3         v
              ▲  (bad)   S1: 2 rows
              │
     main ────┘  after rollback_to_snapshot(S2)

Time travel changes which snapshot a query plans from, for that query only. Rollback changes which snapshot main points at, for everyone. Neither touches a data file.

Under the hood: engine internals and file layouts

This section follows the data path: a write creates a snapshot, a read resolves one, a rollback moves the pointer, and expiry deletes what nothing references.

Write: what one commit leaves on storage

Three appends and three pointer moves on the demo’s orders table left this under s3a://warehouse/shop.db/orders/:

orders/
├── data/
│   ├── order_date=2026-09-20/00000-2-...-00001.parquet    S1
│   └── order_date=2026-09-21/
│       ├── 00000-5-...-00001.parquet                      S2
│       └── 00000-19-...-00001.parquet                     S3 (bad load)
└── metadata/
    ├── 00000-....metadata.json    CREATE TABLE
    ├── 00001-....metadata.json    append     -> S1
    ├── 00002-....metadata.json    append     -> S2
    ├── 00003-....metadata.json    append     -> S3
    ├── 00004-....metadata.json    rollback_to_snapshot(S2)
    ├── 00005-....metadata.json    set_current_snapshot(S3)
    ├── 00006-....metadata.json    set_current_snapshot(S2)
    ├── snap-3657...-1-d851....avro    manifest list, S1
    ├── snap-5270...-1-c7b9....avro    manifest list, S2
    ├── snap-2349...-1-6c70....avro    manifest list, S3
    ├── d851...-m0.avro                manifest, S1
    ├── c7b9...-m0.avro                manifest, S2
    └── 6c70...-m0.avro                manifest, S3

Each append wrote three metadata files: a metadata.json, a manifest list and one manifest. Each pointer move wrote one metadata.json and nothing else. That asymmetry is the whole cost model of rollback: it is a metadata commit, so its cost does not depend on table size.

Read: how a query resolves a snapshot

VERSION AS OF <id> looks the id up in the snapshot list. An unknown id fails at analysis with IllegalArgumentException: Cannot find snapshot with ID 42. A string such as VERSION AS OF 'audit' resolves a branch or tag in refs.

TIMESTAMP AS OF <ts> walks the snapshot log, not the snapshot list. It picks the snapshot that was current at that moment. A timestamp before the first commit fails with IllegalArgumentException: Cannot find a snapshot older than 2020-01-01T00:00:00+00:00.

The consequence is easy to miss. The snapshot log records the bad snapshot as current between its commit and the rollback. So after the rollback, TIMESTAMP AS OF a moment inside that window still returns the 5 bad rows. That is correct as an audit of what readers saw. It is wrong as a recovery target, so recover by snapshot id.

Once resolved, the plan is the same. Both of these produced an identical physical plan, with the partition filter pushed into the same BatchScan:

EXPLAIN SELECT * FROM ice.shop.orders WHERE order_date = DATE'2026-09-20';
EXPLAIN SELECT * FROM ice.shop.orders VERSION AS OF 3657524344527679724
        WHERE order_date = DATE'2026-09-20';

== Physical Plan ==
*(1) ColumnarToRow
+- BatchScan ice.shop.orders[order_id#19L, customer_id#20L, amount#21, status#22, order_date#23] ice.shop.orders (branch=null) [filters=order_date IS NOT NULL, order_date = 20716, groupedBy=] RuntimeFilters: []
== Physical Plan ==
*(1) ColumnarToRow
+- BatchScan ice.shop.orders[order_id#41L, customer_id#42L, amount#43, status#44, order_date#45] ice.shop.orders (branch=null) [filters=order_date IS NOT NULL, order_date = 20716, groupedBy=] RuntimeFilters: []

Only the expression ids differ.

There is no slow path for old data. Planning starts from a different manifest list and prunes the same way.

Rollback: moving main and what history keeps

rollback_to_snapshot writes a new metadata.json whose main ref names the target. It reports both ends of the move:

+--------------------+-------------------+
|previous_snapshot_id|current_snapshot_id|
+--------------------+-------------------+
|23499723825440384   |5270420720687731067|
+--------------------+-------------------+

The history metadata table is the snapshot log, one row per pointer move:

+-----------------------+-------------------+-------------------+-------------------+
|made_current_at        |snapshot_id        |parent_id          |is_current_ancestor|
+-----------------------+-------------------+-------------------+-------------------+
|2026-09-28 13:15:19.839|3657524344527679724|NULL               |true               |
|2026-09-28 13:15:22.058|5270420720687731067|3657524344527679724|true               |
|2026-09-28 13:15:26.352|23499723825440384  |5270420720687731067|false              |
|2026-09-28 13:15:30.086|5270420720687731067|3657524344527679724|true               |
+-----------------------+-------------------+-------------------+-------------------+

Four entries for three snapshots. The bad snapshot is still listed, flagged is_current_ancestor=false: it happened, and it is no longer in the lineage of the current state. The last row records main being pointed back at S2.

Rollback only moves backwards along the lineage. Rolling “forward” to the abandoned snapshot fails:

org.apache.iceberg.exceptions.ValidationException - Cannot roll back to snapshot,
not an ancestor of the current state: 23499723825440384

set_current_snapshot has no ancestry check. It moved main to the abandoned snapshot (5 rows) and back again (3 rows). That is how a rollback is undone, for as long as the abandoned snapshot has not been expired. rollback_to_timestamp is the time-based form; with no snapshot before the timestamp it fails with Cannot roll back, no valid snapshot older than: <millis>.

Expiry: the only step that deletes files

After the rollback the table references 2 data files, but storage holds 3. files lists what the current snapshot reads; all_data_files lists what any retained snapshot reads. The difference is the bad load, kept alive only by the abandoned snapshot.

expire_snapshots removes snapshots that fall outside the retention policy. It then deletes every manifest list, manifest and data file that no surviving snapshot references. Branches and tags count as references: a tagged snapshot survives expiry regardless of age.

Hands-on demo: datalake-playground and PySpark

Environment

Everything runs on the local lakehouse stack described in Run every demo on this blog locally, which puts Spark, a Hive Metastore and MinIO as S3 on one Docker network. Its compose files and start script are in the public datalake-playground repo, and the images are pulled on the first start. The scripts were run with its default Spark line, Spark 3.5.9 (Scala 2.12, Java 17) and Iceberg 1.11.0, whose runtime jar ships in the image under $ICEBERG_HOME. On the stack’s Spark 4.1.3 line (SPARK_VERSION=4.1.3) the same scripts print the same counts, files and exceptions; only the snapshot ids differ.

git clone https://github.com/rangareddy/datalake-playground.git
cd datalake-playground
SPARK_VERSION=3.5.9 sh run_datalake.sh start    # pulls the images, once
sh run_datalake.sh status                       # every service Up (healthy)

The repo’s README lists the services, ports and versions. The scripts below need three of them: spark-master, hive-metastore at thrift://hive-metastore:9083, and minio at http://minio:9000, which backs the s3a://warehouse/ bucket.

Outside the stack, the same scripts run on any Spark 3.5 with --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0 and a reachable metastore and object store. Iceberg 1.11.0 needs Java 17, so use apache/spark:3.5.9-java17-python3 rather than the Java 11 apache/spark:3.5.9-python3.

Script 1: time travel and rollback

Save this as time_travel_rollback.py. It builds an orders table from two good loads, reads it as of each, makes a bad load, and recovers.

"""Time travel and rollback on an Iceberg orders table.

Run inside datalake-playground:
  docker cp time_travel_rollback.py spark-master:/tmp/
  docker exec spark-master bash -c \
    '$SPARK_HOME/bin/spark-submit --master "local[2]" \
       --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
       /tmp/time_travel_rollback.py'
"""
import time
from datetime import date
from decimal import Decimal

from pyspark.sql import SparkSession
from pyspark.sql.types import (DateType, DecimalType, LongType, StringType,
                               StructField, StructType)

spark = (
    SparkSession.builder.appName("iceberg-time-travel")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.ice", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.ice.type", "hive")
    .config("spark.sql.catalog.ice.uri", "thrift://hive-metastore:9083")
    .config("spark.sql.catalog.ice.warehouse", "s3a://warehouse/")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql
TABLE = "ice.shop.orders"

SCHEMA = StructType([
    StructField("order_id", LongType(), False),
    StructField("customer_id", LongType(), False),
    StructField("amount", DecimalType(10, 2), False),
    StructField("status", StringType(), False),
    StructField("order_date", DateType(), False),
])


def load(rows):
    """Append one batch of orders, as a nightly job would."""
    spark.createDataFrame(rows, SCHEMA).writeTo(TABLE).append()
    time.sleep(1.2)  # keep commit timestamps distinct for TIMESTAMP AS OF


def count(suffix=""):
    return sql(f"SELECT count(*) FROM {TABLE} {suffix}").collect()[0][0]


def current_snapshot():
    # The current snapshot is whatever `main` points at. Never "the newest
    # committed_at": after a rollback that is the abandoned snapshot.
    return sql(f"SELECT snapshot_id FROM {TABLE}.refs WHERE name = 'main'").collect()[0][0]


sql("CREATE NAMESPACE IF NOT EXISTS ice.shop")
sql(f"DROP TABLE IF EXISTS {TABLE} PURGE")
sql(f"""
    CREATE TABLE {TABLE} (
        order_id BIGINT, customer_id BIGINT, amount DECIMAL(10,2),
        status STRING, order_date DATE)
    USING iceberg
    PARTITIONED BY (order_date)
""")

# Two good nightly loads
load([(1001, 501, Decimal("120.50"), "PLACED", date(2026, 9, 20)),
      (1002, 502, Decimal("75.00"), "PLACED", date(2026, 9, 20))])
load([(1003, 503, Decimal("310.25"), "PLACED", date(2026, 9, 21))])
good_1 = sql(f"SELECT snapshot_id FROM {TABLE}.snapshots ORDER BY committed_at").collect()[0][0]
good_2 = current_snapshot()
good_1_ts = sql(f"SELECT committed_at FROM {TABLE}.snapshots "
                f"WHERE snapshot_id = {good_1}").collect()[0][0]

print("== 1. time travel ==")
print("current rows                 :", count())
print("VERSION AS OF first snapshot :", count(f"VERSION AS OF {good_1}"))
print("TIMESTAMP AS OF its commit   :", count(f"TIMESTAMP AS OF '{good_1_ts}'"))

# A bad load: a replayed order and a negative amount
load([(1003, 503, Decimal("310.25"), "PLACED", date(2026, 9, 21)),
      (1004, 504, Decimal("-1.00"), "PLACED", date(2026, 9, 21))])
bad = current_snapshot()
bad_ts = sql(f"SELECT committed_at FROM {TABLE}.snapshots "
             f"WHERE snapshot_id = {bad}").collect()[0][0]
print("\n== 2. the bad write ==")
print("rows after bad load:", count())
sql(f"""
    SELECT snapshot_id, parent_id, operation,
           summary['added-records'] AS added, summary['total-records'] AS total
    FROM {TABLE}.snapshots ORDER BY committed_at
""").show(truncate=False)

print("== 3. rollback ==")
sql(f"CALL ice.system.rollback_to_snapshot('shop.orders', {good_2})").show(truncate=False)
print("rows after rollback:", count())
sql(f"""
    SELECT made_current_at, snapshot_id, parent_id, is_current_ancestor
    FROM {TABLE}.history ORDER BY made_current_at
""").show(truncate=False)

newest = sql(f"SELECT snapshot_id FROM {TABLE}.snapshots "
             f"ORDER BY committed_at DESC LIMIT 1").collect()[0][0]
print("newest by committed_at:", newest, "(the abandoned bad snapshot)")
print("main points at        :", current_snapshot())
print("TIMESTAMP AS OF the bad commit still reads:",
      count(f"TIMESTAMP AS OF '{bad_ts}'"), "rows")

print("\n== 4. going forward again ==")
try:
    sql(f"CALL ice.system.rollback_to_snapshot('shop.orders', {bad})").collect()
except Exception as e:  # rollback only moves to an ancestor
    print("rollback_to_snapshot:", e.java_exception.getClass().getName(), "-",
          e.java_exception.getMessage())
sql(f"CALL ice.system.set_current_snapshot('shop.orders', {bad})").collect()
print("set_current_snapshot to bad -> rows:", count())
sql(f"CALL ice.system.set_current_snapshot('shop.orders', {good_2})").collect()
print("set_current_snapshot to good -> rows:", count())

print("\n== 5. files ==")
print("data files in current snapshot:", sql(f"SELECT count(*) FROM {TABLE}.files").collect()[0][0])
print("data files in any snapshot    :", sql(f"SELECT count(*) FROM {TABLE}.all_data_files").collect()[0][0])
print("metadata.json files written   :", sql(f"SELECT count(*) FROM {TABLE}.metadata_log_entries").collect()[0][0])
print("table location                :", sql(f"DESCRIBE TABLE EXTENDED {TABLE}")
      .where("col_name = 'Location'").collect()[0][1])

spark.stop()

Run it with the command in its docstring. The output, with Spark’s log lines removed:

== 1. time travel ==
current rows                 : 3
VERSION AS OF first snapshot : 2
TIMESTAMP AS OF its commit   : 2

== 2. the bad write ==
rows after bad load: 5
+-------------------+-------------------+---------+-----+-----+
|snapshot_id        |parent_id          |operation|added|total|
+-------------------+-------------------+---------+-----+-----+
|3657524344527679724|NULL               |append   |2    |2    |
|5270420720687731067|3657524344527679724|append   |1    |3    |
|23499723825440384  |5270420720687731067|append   |2    |5    |
+-------------------+-------------------+---------+-----+-----+

== 3. rollback ==
+--------------------+-------------------+
|previous_snapshot_id|current_snapshot_id|
+--------------------+-------------------+
|23499723825440384   |5270420720687731067|
+--------------------+-------------------+

rows after rollback: 3
newest by committed_at: 23499723825440384 (the abandoned bad snapshot)
main points at        : 5270420720687731067
TIMESTAMP AS OF the bad commit still reads: 5 rows

== 4. going forward again ==
rollback_to_snapshot: org.apache.iceberg.exceptions.ValidationException - Cannot
roll back to snapshot, not an ancestor of the current state: 23499723825440384
set_current_snapshot to bad -> rows: 5
set_current_snapshot to good -> rows: 3

== 5. files ==
data files in current snapshot: 2
data files in any snapshot    : 3
metadata.json files written   : 7
table location                : s3a://warehouse/shop.db/orders

The history table printed in step 3 is the one shown in Rollback: moving main. To see the files on storage, list the table’s prefix through the stack’s MinIO client:

docker exec mc mc ls --recursive dl/warehouse/shop.db/orders/

Snapshot ids are random 64-bit numbers, so yours will differ. The counts will not.

Script 2: incremental reads, write-audit-publish and retention

Save this as wap_retention.py and run it the same way. It uses its own table and drops it at the end.

"""Incremental reads, write-audit-publish and retention on an Iceberg orders table.

Run inside datalake-playground:
  docker cp wap_retention.py spark-master:/tmp/
  docker exec spark-master bash -c \
    '$SPARK_HOME/bin/spark-submit --master "local[2]" \
       --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
       /tmp/wap_retention.py'
"""
from datetime import date
from decimal import Decimal

from pyspark.sql import SparkSession
from pyspark.sql.types import (DateType, DecimalType, LongType, StringType,
                               StructField, StructType)

spark = (
    SparkSession.builder.appName("iceberg-wap-retention")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.ice", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.ice.type", "hive")
    .config("spark.sql.catalog.ice.uri", "thrift://hive-metastore:9083")
    .config("spark.sql.catalog.ice.warehouse", "s3a://warehouse/")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql
TABLE = "ice.shop.orders_wap"

SCHEMA = StructType([
    StructField("order_id", LongType(), False),
    StructField("customer_id", LongType(), False),
    StructField("amount", DecimalType(10, 2), False),
    StructField("status", StringType(), False),
    StructField("order_date", DateType(), False),
])


def orders(*rows):
    return spark.createDataFrame(list(rows), SCHEMA)


def count(suffix=""):
    return sql(f"SELECT count(*) FROM {TABLE} {suffix}").collect()[0][0]


def main_snapshot():
    return sql(f"SELECT snapshot_id FROM {TABLE}.refs WHERE name = 'main'").collect()[0][0]


def changes_since(start_exclusive, end_inclusive):
    return (spark.read.format("iceberg")
            .option("start-snapshot-id", str(start_exclusive))
            .option("end-snapshot-id", str(end_inclusive))
            .load(TABLE))


sql("CREATE NAMESPACE IF NOT EXISTS ice.shop")
sql(f"DROP TABLE IF EXISTS {TABLE} PURGE")
sql(f"""
    CREATE TABLE {TABLE} (
        order_id BIGINT, customer_id BIGINT, amount DECIMAL(10,2),
        status STRING, order_date DATE)
    USING iceberg
    PARTITIONED BY (order_date)
""")

orders((1001, 501, Decimal("120.50"), "PLACED", date(2026, 9, 20)),
       (1002, 502, Decimal("75.00"), "PLACED", date(2026, 9, 20))).writeTo(TABLE).append()
watermark = main_snapshot()  # what a downstream job stored after its last run

print("== 1. incremental reads ==")
orders((1003, 503, Decimal("310.25"), "PLACED", date(2026, 9, 21))).writeTo(TABLE).append()
orders((1004, 504, Decimal("42.00"), "PLACED", date(2026, 9, 21))).writeTo(TABLE).append()
now = main_snapshot()
print("appended since watermark:",
      sorted(r.order_id for r in changes_since(watermark, now).collect()))

sql(f"DELETE FROM {TABLE} WHERE order_id = 1004")
sql(f"UPDATE {TABLE} SET status = 'SHIPPED' WHERE order_id = 1001")
after = main_snapshot()
sql(f"""SELECT snapshot_id, operation FROM {TABLE}.snapshots
        ORDER BY committed_at DESC LIMIT 2""").show(truncate=False)
print("rows the incremental read returns across that delete and update:",
      changes_since(now, after).count())

print("\n== 2. the branch write that lands on main ==")
sql(f"ALTER TABLE {TABLE} CREATE BRANCH audit")
before = (count(), count("VERSION AS OF 'audit'"))
orders((1005, 505, Decimal("18.00"), "PLACED", date(2026, 9, 22))) \
    .writeTo(TABLE).option("branch", "audit").append()
print("main, audit before:", before, "after writeTo(...).option('branch'):",
      (count(), count("VERSION AS OF 'audit'")))
sql(f"ALTER TABLE {TABLE} DROP BRANCH audit")
sql(f"DELETE FROM {TABLE} WHERE order_id = 1005")

print("\n== 3. write-audit-publish ==")
sql(f"ALTER TABLE {TABLE} CREATE BRANCH audit")
orders((1006, 506, Decimal("99.99"), "PLACED", date(2026, 9, 22)),
       (1007, 507, Decimal("-5.00"), "PLACED", date(2026, 9, 22))) \
    .writeTo(f"{TABLE}.branch_audit").append()
print("main rows:", count(), " audit rows:", count("VERSION AS OF 'audit'"))

bad_rows = sql(f"SELECT count(*) FROM {TABLE} VERSION AS OF 'audit' "
               f"WHERE amount <= 0").collect()[0][0]
print("audit check, non-positive amounts:", bad_rows)
if bad_rows:
    sql(f"DELETE FROM {TABLE}.branch_audit WHERE amount <= 0")
    print("fixed on the branch, audit rows now:", count("VERSION AS OF 'audit'"))
sql(f"CALL ice.system.fast_forward('shop.orders_wap', 'main', 'audit')").show(truncate=False)
print("main rows after publish:", count())
sql(f"ALTER TABLE {TABLE} DROP BRANCH audit")

print("== 4. retention ==")
month_end = main_snapshot()
sql(f"ALTER TABLE {TABLE} CREATE TAG `close-2026-09` AS OF VERSION {month_end} RETAIN 365 DAYS")
for i in range(3):
    orders((2000 + i, 600 + i, Decimal("10.00"), "PLACED", date(2026, 10, 1))).writeTo(TABLE).append()
oldest = sql(f"SELECT snapshot_id FROM {TABLE}.history ORDER BY made_current_at LIMIT 1").collect()[0][0]
print("snapshots before expiry:", sql(f"SELECT count(*) FROM {TABLE}.snapshots").collect()[0][0])

print("expire with the defaults (nothing is 5 days old yet):")
sql(f"CALL ice.system.expire_snapshots(table => 'shop.orders_wap')") \
    .select("deleted_data_files_count", "deleted_manifest_lists_count").show()

sql(f"""ALTER TABLE {TABLE} SET TBLPROPERTIES (
        'history.expire.max-snapshot-age-ms' = '1',
        'history.expire.min-snapshots-to-keep' = '2')""")
print("expire with max age 1 ms, keep at least 2:")
sql(f"CALL ice.system.expire_snapshots(table => 'shop.orders_wap')") \
    .select("deleted_data_files_count", "deleted_manifest_lists_count").show()
sql(f"""SELECT s.snapshot_id, r.name AS ref
        FROM {TABLE}.snapshots s LEFT JOIN {TABLE}.refs r ON s.snapshot_id = r.snapshot_id
        ORDER BY s.committed_at""").show(truncate=False)
print("tag still readable:", count("VERSION AS OF 'close-2026-09'"), "rows")
try:
    count(f"VERSION AS OF {oldest}")
except Exception as e:
    print("oldest snapshot:", type(e).__name__, "-", str(e).splitlines()[0])

sql(f"DROP TABLE {TABLE} PURGE")
spark.stop()

Output:

== 1. incremental reads ==
appended since watermark: [1003, 1004]
+-------------------+---------+
|snapshot_id        |operation|
+-------------------+---------+
|7603322076997799954|overwrite|
|6084086043880926635|delete   |
+-------------------+---------+
rows the incremental read returns across that delete and update: 0

== 2. the branch write that lands on main ==
main, audit before: (3, 3) after writeTo(...).option('branch'): (4, 3)

== 3. write-audit-publish ==
main rows: 3  audit rows: 5
audit check, non-positive amounts: 1
fixed on the branch, audit rows now: 4
+--------------+-------------------+-------------------+
|branch_updated|previous_ref       |updated_ref        |
+--------------+-------------------+-------------------+
|main          |8708299415798259725|7848268967418525682|
+--------------+-------------------+-------------------+
main rows after publish: 4

== 4. retention ==
snapshots before expiry: 12
expire with the defaults (nothing is 5 days old yet):
+------------------------+----------------------------+
|deleted_data_files_count|deleted_manifest_lists_count|
+------------------------+----------------------------+
|                       0|                           0|
+------------------------+----------------------------+
expire with max age 1 ms, keep at least 2:
+------------------------+----------------------------+
|deleted_data_files_count|deleted_manifest_lists_count|
+------------------------+----------------------------+
|                       4|                           9|
+------------------------+----------------------------+
+-------------------+-------------+
|snapshot_id        |ref          |
+-------------------+-------------+
|7848268967418525682|close-2026-09|
|1428465888305999758|NULL         |
|2351970890801618167|main         |
+-------------------+-------------+
tag still readable: 4 rows
oldest snapshot: IllegalArgumentException - Cannot find snapshot with ID 9433222583856300

Steps 1 and 2 each show a trap; the next section explains both. Step 3 is the flow that avoids needing a rollback at all. Step 4 shows what retention keeps: 12 snapshots became 3. Those are the two newest, which is the min-snapshots-to-keep floor, plus the tagged month-end snapshot, which aged out but survived because a tag referenced it.

Real-world scenarios and production failures

Finding the snapshot before the bad write

Rolling back means naming a snapshot, and “the one before the bad write” has to be found rather than guessed.

SELECT made_current_at, snapshot_id, parent_id, is_current_ancestor
FROM ice.shop.orders.history
ORDER BY made_current_at DESC LIMIT 20;

SELECT snapshot_id, operation,
       summary['added-records']   AS added,
       summary['deleted-records'] AS deleted,
       summary['total-records']   AS total
FROM ice.shop.orders.snapshots;

The operation column narrows it immediately. A table that should only ever be appended to, showing an overwrite or delete, has found its culprit. The summary map carries the record deltas, so a load that added twice its usual count is visible without reading data. The target is the bad snapshot’s parent_id.

Do not pick the target with min(snapshot_id) or max(snapshot_id). Snapshot ids are random, not sequential. And do not order snapshots by committed_at to find the current state: after a rollback, the newest committed_at is the abandoned snapshot, as script 1 printed. The current snapshot is SELECT snapshot_id FROM <table>.refs WHERE name = 'main'.

A dashboard keeps serving the bad rows after the rollback

The on-call engineer runs the rollback and checks the count: 3. The dashboard’s long-running Spark session still shows 5.

SparkCatalog wraps its catalog in a CachingCatalog by default (cache-enabled=true). The cache holds each loaded Table object, and a cached table is not refreshed before each query. Entries expire after cache.expiration-interval-ms, 30 seconds by default, measured since last access. A session that queries more often than every 30 seconds therefore never expires its entry. Polling every 10 seconds, the dashboard catalog read 5 rows for over a minute after the rollback in a separate test run. It only saw 3 after sitting idle for 35 seconds.

This script reproduces it with two catalogs in one session: ice with the default cache, and ops with it off.

"""A long-running session keeps reading a table after another job rolled it back.

Run inside datalake-playground:
  docker cp stale_reader.py spark-master:/tmp/
  docker exec spark-master bash -c \
    '$SPARK_HOME/bin/spark-submit --master "local[2]" \
       --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
       /tmp/stale_reader.py'
"""
from pyspark.sql import SparkSession

CATALOG = {
    "": "org.apache.iceberg.spark.SparkCatalog",
    ".type": "hive",
    ".uri": "thrift://hive-metastore:9083",
    ".warehouse": "s3a://warehouse/",
}
builder = (
    SparkSession.builder.appName("iceberg-stale-reader")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
)
# `ice` is a dashboard's catalog with the default table cache.
# `ops` is the on-call engineer's catalog with the cache off.
for name in ("ice", "ops"):
    for key, value in CATALOG.items():
        builder = builder.config(f"spark.sql.catalog.{name}{key}", value)
builder = builder.config("spark.sql.catalog.ops.cache-enabled", "false")
spark = builder.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql


def rows(catalog):
    return sql(f"SELECT count(*) FROM {catalog}.shop.orders_cache").collect()[0][0]


sql("CREATE NAMESPACE IF NOT EXISTS ice.shop")
sql("DROP TABLE IF EXISTS ice.shop.orders_cache PURGE")
sql("CREATE TABLE ice.shop.orders_cache (order_id BIGINT, amount DECIMAL(10,2)) USING iceberg")
sql("INSERT INTO ice.shop.orders_cache VALUES (1001, 120.50), (1002, 75.00), (1003, 310.25)")
good = sql("SELECT snapshot_id FROM ice.shop.orders_cache.refs WHERE name = 'main'").collect()[0][0]
sql("INSERT INTO ice.shop.orders_cache VALUES (1003, 310.25), (1004, -1.00)")

print("dashboard sees          :", rows("ice"))
sql(f"CALL ops.system.rollback_to_snapshot('shop.orders_cache', {good})").collect()
print("on-call sees            :", rows("ops"))
print("dashboard sees          :", rows("ice"))
sql("REFRESH TABLE ice.shop.orders_cache")
print("dashboard after REFRESH :", rows("ice"))

sql("DROP TABLE ice.shop.orders_cache PURGE")
spark.stop()

dashboard sees          : 5
on-call sees            : 3
dashboard sees          : 5
dashboard after REFRESH : 3

Three runs out of three gave the same result. The fix is on the reader side: run REFRESH TABLE after a rollback, and set cache-enabled=false on catalogs used by long-lived sessions that must never serve a rolled-back state. Treat “announce the rollback” as a step in the runbook, not an afterthought.

An incremental consumer silently skips deletes and updates

Downstream jobs often read only what changed since their last run with start-snapshot-id and end-snapshot-id. That read is an append scan: it walks the snapshots between the two ids and keeps only those whose operation is append. In script 2, a delete and an overwrite (the UPDATE) between the two ids produced 0 rows and no error.

A consumer that mirrors this table into a warehouse will therefore never learn that order 1004 was deleted or that order 1001 shipped. If you need deletes and updates, use the create_changelog_view procedure. It builds a view with _change_type, _change_ordinal and _commit_snapshot_id columns. Over the same kind of history it returned a DELETE row for the deleted order, and a DELETE plus an INSERT for the updated one. Do not assume the batch incremental read covers them.

Two more rules for the watermark itself. Store the main snapshot id from refs, never “the latest snapshot”. And expect this error after a rollback that abandoned the snapshot the consumer stored:

IllegalArgumentException: Starting snapshot (exclusive) 4149797747189483283
is not a parent ancestor of end snapshot 1822237857672733606

That consumer must be reset to an ancestor and re-read.

The branch write that went to main

Script 2 step 2 created a branch audit, then wrote with df.writeTo(table).option("branch", "audit").append(). The row landed on main: main went from 3 to 4 and audit stayed at 3, with no warning.

In Iceberg’s Spark module, branch is a read option (SparkReadOptions). SparkWriteOptions has no such key, so the DataFrameWriterV2 path ignores it. It has to: writeTo resolves the target table through the catalog before options are consulted. The branch has to be part of the identifier instead.

The format("iceberg") form works because IcebergSource turns the branch option into a branch_audit table selector. With spark.wap.branch set, unqualified reads in that session also read the branch, which is the point of write-audit-publish and a surprise if you forget to unset it.

Write-audit-publish instead of rollback

A rollback is visible to everyone who queried between the bad commit and the fix. Write-audit-publish removes that window.

 main:   M1 ──────────────────────────────> M1 ──> A2   (fast_forward)
          \                                        ▲
 audit:    └──> A1 (write) ──> A2 (fix bad rows) ──┘
                 │
                 └── checks run here; readers of main never see A1

1. Create a branch from main.
2. Write the load to <table>.branch_audit.
3. Run checks against VERSION AS OF 'audit'.
4. Fix or discard on the branch.
5. fast_forward('<table>', 'main', 'audit') moves main to the branch head.

In script 2 the audit found one non-positive amount, deleted it on the branch, and published 4 rows. Readers of main saw 3 rows, then 4, and never the bad row. fast_forward requires main to be an ancestor of the branch; if main moved in the meantime it fails with IllegalArgumentException: Cannot fast-forward: main is not an ancestor of audit, and the load must be redone on a fresh branch.

Expiry removed the snapshot you needed

After expire_snapshots, a snapshot that is gone fails the same way as a bogus id: IllegalArgumentException: Cannot find snapshot with ID <id>. There is no undo, because the data files that only it referenced are deleted. This is the failure that turns a one-statement fix into a restore from source systems, and it is why the retention numbers below deserve a decision, not a default.

Optimisation, tuning knobs and best practices

Retention settings

These table properties drive expire_snapshots when it is called without explicit arguments. The defaults are from Iceberg’s TableProperties.

Two settings, because age alone is a poor policy for a table whose commit rate varies. min-snapshots-to-keep guarantees a floor: a table that went quiet for a week still has restore points rather than one.

ALTER TABLE ice.shop.orders SET TBLPROPERTIES (
  'history.expire.max-snapshot-age-ms'   = '604800000',   -- 7 days
  'history.expire.min-snapshots-to-keep' = '50');

ALTER TABLE ice.shop.orders CREATE TAG `close-2026-09`
  AS OF VERSION 5270420720687731067 RETAIN 365 DAYS;

CALL ice.system.expire_snapshots(table => 'shop.orders');

Retention is your storage bill and your undo button at once

Every retained snapshot keeps alive the data files only it references. For an append-only table that is little extra, because newer snapshots reference the same files. For a table rewritten by MERGE, UPDATE or compaction, each rewrite leaves the old files owned by the old snapshots. Seven days of retention then means paying for up to seven days of rewritten files. Compare files against all_data_files to see the gap before choosing a number:

SELECT (SELECT sum(file_size_in_bytes) FROM ice.shop.orders.all_data_files)
     - (SELECT sum(file_size_in_bytes) FROM ice.shop.orders.files)
       AS bytes_held_only_by_old_snapshots;

Decide retention as a recovery policy. “How long before we would certainly have noticed a bad write?” is the right question. The storage follows from the answer rather than driving it. Commit frequency matters too: a table that commits every minute writes 3 metadata files per commit: 180 an hour, over 4,000 a day. Its snapshot list also grows inside every new metadata.json. Expire it on a schedule that matches.

Catalog cache

Disabling the cache makes every table load refresh from the catalog: a metastore round trip, plus a metadata.json read when the table has changed. For a dashboard issuing a few queries a minute, that is negligible. For a job planning thousands of queries per second against the same table, keep the cache and use REFRESH TABLE when needed.

Checklist

• Set max-snapshot-age-ms and min-snapshots-to-keep per table from your detection time, not left at the 5-day, 1-snapshot defaults.
• Tag month-end, audit and model-training snapshots with an explicit RETAIN.
• Run expire_snapshots on a schedule, and alert when all_data_files bytes outgrow files bytes.
• Recover by snapshot id from history or snapshots, never by TIMESTAMP AS OF a time inside the bad window.
• Read the current snapshot from refs WHERE name = 'main'.
• After a rollback, REFRESH TABLE in every long-lived session, or run those sessions with cache-enabled=false.
• Write to branches with the .branch_<name> identifier, never writeTo(...).option("branch", ...).
• Use write-audit-publish for loads that can be checked before readers see them.
• Do not rely on start-snapshot-id reads to capture deletes or updates.

Summary and key takeaways

Every commit leaves a snapshot, and the catalog points at one of them. Time travel reads a different one for one query. Rollback points main at a different one for everyone, by writing a single metadata.json. Expiry deletes the ones you no longer keep, and the files only they referenced.

The common misconceptions, and what the runs showed:

The retention setting is the most consequential number on a production table. It is your storage bill and your undo button at the same time.

References

• Iceberg Spark queries for VERSION AS OF, TIMESTAMP AS OF and incremental reads
• Iceberg Spark procedures for rollback_to_snapshot, set_current_snapshot, fast_forward and expire_snapshots
• Iceberg branching and tagging for named references and write-audit-publish
• BaseIncrementalAppendScan and CachingCatalog for the append-only scan and the table cache
• Branching, tagging and write-audit-publish in Iceberg and Table maintenance for Iceberg and Hudi for the neighbouring mechanics

Trademarks

Apache Iceberg, Apache Spark, Apache Hive, Apache and the Apache feather logo are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.