Life of a write query in Iceberg: three files, one pointer swap, and what ACID means here

• The sequence
• What each commit costs
• Optimistic concurrency, measured
  • When retries are not enough
• Isolation, precisely
• What the four letters actually mean here
• What Iceberg does not give you
• What the retry policy actually is
• What a manifest actually records
• Watching a commit happen
• Troubleshooting a write
• What the write path does not do
• Common misconceptions
• A model worth keeping
• Run it yourself
• References
• Trademarks

TL;DR

• Every commit writes exactly three metadata files — a manifest, a manifest list, a metadata.json — then atomically swaps the catalog pointer. Measured across three appends on a Hive catalog: 1 → 4 → 7 → 10 metadata files.
• The swap is a compare-and-swap: commit only if the metadata I started from is still current. That is the entire concurrency control.
• Two threads writing 6 rows each concurrently produced 13 snapshots, 13 rows, A=6 and B=6 — no lost writes, no surfaced errors. 11 of the 13 commits needed a second attempt, absorbed by the retry loop.
• With retries switched off, the same test lost 6 of 12 inserts to CommitFailedException, and under the default serializable isolation a MERGE racing an append failed with ValidationException.
• The snapshot chain stayed linear: every snapshot has a parent, so history is a sequence even though writes were parallel.
• Manifest reuse keeps this affordable — a commit adding one file reports added-data-files=1 while total-records accumulates.

The sequence

A write is four steps, and only the last one is visible to readers.

flowchart TB
    P["1. plan: read current metadata<br/>schema, spec, current snapshot"] --> W["2. executors write data files<br/>and any delete files"]
    W --> M["3. write manifest, manifest list,<br/>new metadata.json"]
    M --> CAS{"4. catalog compare-and-swap<br/>is my base still current?"}
    CAS -->|yes| OK["commit: new snapshot is live"]
    CAS -->|no| R["retry: re-apply onto the new base<br/>and re-run validation"]
    R --> M

Nothing is visible until step 4. Data files written in step 2 exist on storage but belong to no snapshot, so no reader can see them. A job that dies between steps 2 and 4 leaves files nobody references — orphans, cleaned up later by remove_orphan_files, and harmless apart from the storage.

That is also why a failed write never corrupts a table. There is no partial state to observe.

What each commit costs

Creating a table partitioned by days(event_ts) in the Hive catalog and appending one day of events three times, counting files on storage under the table’s metadata/ and data/ prefixes:

after CREATE, before any data   metadata_files=1   data_files=0
after append #1                 metadata_files=4   data_files=1
after append #2                 metadata_files=7   data_files=2
after append #3                 metadata_files=10  data_files=3

Three per commit, every commit. One manifest listing the data files this commit added, one manifest list for the new snapshot, one metadata.json recording the new table state. The one before any data is the initial metadata.json. A Hive catalog keeps the pointer to the current metadata.json in the metastore; a Hadoop catalog has no metastore, so it adds a version-hint.text file as well, and there you would count two before any data.

This is the number that makes streaming tables accumulate metadata: a commit per minute is three metadata files per minute, regardless of how many rows moved.

The snapshot log shows what each commit recorded:

snap=5418759700351153818 parent=None op=append  added-data-files=1 total-data-files=1 total-records=3000
snap=8504705344298678251 parent=5418759700351153818 op=append  added-data-files=1 total-data-files=2 total-records=6000
snap=7965481609336464585 parent=8504705344298678251 op=append  added-data-files=1 total-data-files=3 total-records=9000

added-data-files stays 1 while total-records climbs. Each commit describes only its own addition; the manifest list points at the previous snapshot’s manifests rather than copying them. The manifests table for the current snapshot shows it: three manifests, each written by a different commit.

+-------------------+----------------------+-------------------------+
|added_snapshot_id  |added_data_files_count|existing_data_files_count|
+-------------------+----------------------+-------------------------+
|5418759700351153818|1                     |0                        |
|7965481609336464585|1                     |0                        |
|8504705344298678251|1                     |0                        |
+-------------------+----------------------+-------------------------+

Snapshots share metadata, which is what makes keeping a thousand of them affordable.

The operation field names what happened, and it is the record you consult later: append for added data, overwrite when files were replaced, delete when a delete file was written.

Optimistic concurrency, measured

Iceberg does not lock the table while a job runs. Every writer plans against a snapshot, does its work, and at commit time asks the catalog to swap the pointer only if the metadata it started from is still current. If another writer committed meanwhile, the swap fails with CommitFailedException and the writer retries.

How the catalog implements that check varies. The Hive catalog takes a short metastore lock around the swap itself (engine.hive.lock-enabled, default true), then compares the base metadata.json location it started from with the one the metastore holds, and refuses if they differ. The lock covers the pointer update, not your job.

Two threads, each inserting 6 rows into the same table, running concurrently:

concurrent writer errors: none
final rows   = 13
snapshots    = 13
per-writer   = [('A', 6), ('B', 6), ('seed', 1)]
parent chain is linear (every parent used once): True

Iceberg’s LoggingMetricsReporter prints a CommitReport per commit, including how many attempts it took. Counted over that run:

   2 tableName=ice.blog_write_path.payments, attempts=1
  11 tableName=ice.blog_write_path.payments, attempts=2

Three things worth drawing out.

No rows were lost. Both writers landed all six of their inserts. This is the property that a directory-based table cannot offer, where two writers putting files into the same path simply coexist and hope.

No errors surfaced. Conflicts happened: 11 of the 13 commits lost the race once and succeeded on their second attempt. The retries were handled inside the client, and application code saw success.

History is linear. Thirteen snapshots, each after the first with a distinct parent. Concurrent writes did not create a branching history; they were serialized by the compare-and-swap into one sequence. That linearity is what makes time travel meaningful.

When retries are not enough

Retries are bounded, and a conflict that cannot be resolved surfaces as CommitFailedException. Setting commit.retry.num-retries to 0 on the same two-writer test shows what the retry loop was absorbing:

failed inserts: 6
   org.apache.iceberg.exceptions.CommitFailedException - Cannot commit: Base metadata location 's3a://warehouse/blog_write_path.db/payments_no_retry/metadata/00001-...metadata.json' is not same as the current table metadata location ...
   org.apache.iceberg.exceptions.CommitFailedException - Cannot commit: stale table metadata
final rows   = 7
per-writer   = [('B', 6), ('seed', 1)]

Half the inserts failed, all from one writer, and both messages appear: the Hive catalog’s location check, and stale table metadata from a writer whose refreshed base was already out of date. Which writer loses is a race: a second run failed all six of B’s inserts instead of A’s. Two situations produce this in practice.

Heavy contention: many writers committing faster than retries can settle. The fix is fewer, larger commits — batching a minute of data into one commit rather than sixty.

A genuine conflict: two writers modifying the same files, where re-planning cannot simply be re-applied. Appends rarely conflict this way because they only add files. UPDATE, DELETE and MERGE against overlapping data can, because the rows one writer is rewriting are the rows another is deleting.

There is a third outcome worth knowing: CommitStateUnknownException, when the catalog call did not return a clear answer. Do not clean up after this one. The commit may have succeeded, and deleting its files would corrupt a live table.

Isolation, precisely

Readers get snapshot isolation for free. A query resolves the current metadata once and reads that snapshot throughout, so a commit landing mid-query is invisible to it. There is no read lock and no torn read.

Writers get serializable behaviour on the commit itself, since only one compare-and-swap can succeed against a given base. What that does not give you is serializability of the computation — a writer that read the table, decided something, and then wrote is not protected against the table changing in between unless the commit itself conflicts.

Row-level operations tighten this with write.delete.isolation-level, write.update.isolation-level and write.merge.isolation-level, all serializable by default. At commit time Spark asks Iceberg to validate the changes made since the operation’s scan. serializable checks for conflicting data, new files that could hold rows matching the operation’s filter, and for conflicting deletes. snapshot checks only for conflicting deletes, so a concurrent append of matching rows is allowed through.

Measured with a MERGE that adds 5 to three account balances while another writer appends account 4 before the MERGE commits:

serializable MERGE -> org.apache.iceberg.exceptions.ValidationException: Found conflicting files that can contain records matching true: [s3a://warehouse/blog_write_path.db/accounts_serializable/data/...]
             rows now: [(1, 100.0), (2, 200.0), (3, 300.0), (4, 400.0)]
snapshot     MERGE -> committed
             rows now: [(1, 105.0), (2, 205.0), (3, 305.0), (4, 400.0)]

Under serializable the MERGE was rejected and the table kept the append alone; matching true means the MERGE had no filter on the target, so any new file conflicts. Under snapshot both changes landed. The default is the stricter one, and relaxing it is a decision to make explicitly.

What the four letters actually mean here

The sequence above is the whole implementation of ACID on an Iceberg table, and each letter maps onto a specific part of it.

Atomicity is the pointer swap. Everything before step 4 is invisible: data files exist on storage but belong to no snapshot, so no reader can see them. A write either happened or did not, and there is no window in which half a partition is readable — the failure that makes directory-based tables unsafe to read while they are being written.

Consistency is structural, not semantic. Schema is validated on write, a type cannot change to something existing values would not satisfy, and partition specs are recorded per file so old layouts keep meaning. What is not enforced is domain rules: there are no foreign keys, no unique constraints, no check constraints. A MERGE with duplicate keys in the source fails because the statement would be ambiguous, not because the table enforces uniqueness.

Isolation is covered in its own section below.

Durability is delegated to storage. Once the files are written and the swap returns success, durability is whatever S3, HDFS or your database provides.

What Iceberg does not give you

Worth being explicit, because “ACID” invites database expectations that do not all hold.

No multi-statement transactions. Each commit is atomic; two statements are two commits, and a reader can observe the state between them. There is no BEGIN and COMMIT spanning several writes.

No cross-table transactions. Two tables updated together can be observed half-updated. Some catalogs are adding multi-table commits; the core format does not have them.

No locking. You cannot reserve a table. Concurrency is optimistic by design, and a workload that genuinely needs pessimistic control needs coordination above Iceberg.

No read-your-writes across sessions without re-resolving the catalog. A long query started before a commit will not see it — which is isolation working correctly, and occasionally surprising.

What the retry policy actually is

“It retries” is worth making precise, because the defaults decide how a busy table behaves under contention. SnapshotProducer wraps the commit in a retry loop with exponential backoff, configured by four table properties:

So a contended commit tries five times in total, waiting longer each time, and abandons the attempt after half an hour. Raising num-retries on a hot table trades latency for success rate; lowering total-timeout-ms makes a stuck job fail fast instead of occupying an executor slot for thirty minutes.

Each retry re-applies the same file changes onto the new base; it never re-runs your query. SnapshotProducer.apply() refreshes the table, runs the operation’s validations against the new current state, and writes a new manifest list and metadata.json that add the already-written data files on top of it. For an append that is cheap and almost always succeeds. For a MERGE, UPDATE or DELETE the validation can fail, and a ValidationException is not retried at all: only CommitFailedException is. The statement fails and has to be run again from the start, which is why row-level operations are the ones that suffer under contention.

ALTER TABLE ice.blog_write_path.events SET TBLPROPERTIES (
  'commit.retry.num-retries'      = '10',
  'commit.retry.max-wait-ms'      = '10000',
  'commit.retry.total-timeout-ms' = '300000');

That configuration suits a table with several concurrent appenders: more attempts, a lower ceiling so backoff does not idle, and a five-minute cap so a doomed commit fails while someone is still watching.

What a manifest actually records

Step 3 writes a manifest, and the manifest list describes each manifest with a schema fixed by the spec. The sixteen fields, from ManifestFile in Iceberg 1.11.0:

path                  length              spec_id
manifest_content      sequence_number     min_sequence_number
snapshot_id           added_files_count   existing_files_count
deleted_files_count   added_rows_count    existing_rows_count
deleted_rows_count    partition_summaries key_metadata
first_row_id

first_row_id is the newest: it is the row id handed to the first row of the manifest’s added files, which is how format v3 assigns row lineage.

Three of those do disproportionate work.

spec_id records which partition spec the files in this manifest were written under, which is what lets a table change its partitioning without rewriting history.

manifest_content distinguishes a data manifest from a delete manifest, so a reader can find delete files without scanning every manifest.

partition_summaries holds the lower and upper partition bound per manifest, which is what lets query planning discard an entire manifest unopened.

Commits also merge manifests when they accumulate, governed by two more properties: commit.manifest.min-count-to-merge, default 100, and commit.manifest.target-size-bytes, default 8 MB. That is why a table committing frequently does not grow manifests without bound — up to a point, compaction of metadata happens on its own.

Watching a commit happen

Everything above is observable from SQL. Take the newest commit time before a write, write, and read back only the snapshots that appeared after it:

before = spark.sql("SELECT max(committed_at) c FROM ice.blog_write_path.events.snapshots").collect()[0]["c"]
df.writeTo("ice.blog_write_path.events").append()
new = spark.sql(f"""
    SELECT snapshot_id, operation, summary
    FROM ice.blog_write_path.events.snapshots WHERE committed_at > timestamp'{before}'""").collect()
for r in new:
    print(r["operation"], "added-data-files =", r["summary"].get("added-data-files"),
          "added-records =", r["summary"].get("added-records"))

append added-data-files = 1 added-records = 1000

And the metadata files each commit wrote:

SELECT file, latest_snapshot_id
FROM ice.blog_write_path.events.metadata_log_entries ORDER BY timestamp DESC LIMIT 2;

s3a://warehouse/blog_write_path.db/events/metadata/00004-068b71ab-...metadata.json  3444404348366228180
s3a://warehouse/blog_write_path.db/events/metadata/00003-f78e9f3e-...metadata.json  7965481609336464585

That last table is the one to reach for when the metadata directory is growing faster than you expect: it lists every metadata.json the table has written, including the one from CREATE, so after three appends it held four rows.

Troubleshooting a write

The second row is the one worth internalising. CommitStateUnknownException means Iceberg does not know whether the pointer moved. Cleaning up the files “because the commit failed” deletes data a successful commit is now referencing. Leave them; orphan cleanup handles the other case safely, later.

What the write path does not do

It does not reclaim space. An overwrite writes new files and leaves the old ones referenced by the previous snapshot. Storage is reclaimed by expire_snapshots and nothing else.

It does not compact. Each commit writes what it was given. A hundred small appends produce a hundred small files, and rewrite_data_files is a separate job.

It does not reorder. Data is written in the order it arrives unless a sort order is declared, and unsorted data makes column-bounds pruning useless at read time.

Common misconceptions

“Iceberg locks the table to write.” Nothing is locked while the job runs. The commit is a compare-and-swap that retries; a Hive catalog holds a metastore lock only for the swap itself.

“Concurrent writers will lose data.” Measured: two writers, 6 rows each, all 12 landed. With retries switched off, 6 of them failed loudly with CommitFailedException; none was silently lost.

“A failed write leaves the table inconsistent.” It leaves unreferenced files. The table is exactly as it was.

“Every commit rewrites the metadata.” It writes a new metadata file that reuses the previous snapshot’s manifests.

“ACID on a lake means a database.” It means atomic commits and snapshot isolation over files. Multi-statement and cross-table transactions are not included.

“Retry means my job is slow.” Retries are cheap when commits are infrequent and expensive when they are constant. Commit frequency is the lever.

A model worth keeping

A write is: read the current state, write data files nobody can see yet, describe them in three metadata files, then swap one pointer if nothing changed underneath.

Everything else follows. Atomicity is the swap. Isolation is resolving the pointer once. Concurrency is the retry. Metadata growth is three files per commit, which is why how often you commit matters more than how much you write.

Run it yourself

Every output above comes from two scripts run on the local stack described in Run every demo on this blog locally, on Spark 3.5.9 with Iceberg 1.11.0 and the stack’s Hive Metastore catalog:

git clone https://github.com/rangareddy/datalake-playground.git
cd datalake-playground
SPARK_VERSION=3.5.9 sh run_datalake.sh start
sh run_datalake.sh status

commit_anatomy.py creates the partitioned events table, counts files on storage after each append, and prints the snapshot summaries, the manifests table, one watched commit and the retry properties:

"""What one Iceberg commit writes, counted on storage and in the metadata tables.

Run inside datalake-playground:
  docker cp commit_anatomy.py spark-master:/tmp/
  docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[2]" \
    --driver-memory 1g --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
    /tmp/commit_anatomy.py'
"""
from datetime import datetime, timedelta

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("iceberg-commit-anatomy")
    .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.blog_write_path.events"
jvm = spark._jvm
hconf = spark._jsc.hadoopConfiguration()


def count_files(location, sub):
    """Count files under <table location>/<sub> directly on storage."""
    path = jvm.org.apache.hadoop.fs.Path(f"{location}/{sub}")
    fs = path.getFileSystem(hconf)
    if not fs.exists(path):
        return 0
    it = fs.listFiles(path, True)
    n = 0
    while it.hasNext():
        it.next()
        n += 1
    return n


def batch(day, n=3000):
    """One day of click events, one row every 28.8 seconds."""
    start = datetime(2026, 9, day)
    rows = [(day * 100_000 + i, f"user-{i % 500}", ["view", "click", "buy"][i % 3],
             start + timedelta(seconds=i * 28.8)) for i in range(n)]
    return spark.createDataFrame(rows, "event_id BIGINT, user_id STRING, action STRING, event_ts TIMESTAMP")


sql("CREATE NAMESPACE IF NOT EXISTS ice.blog_write_path")
sql(f"DROP TABLE IF EXISTS {TABLE} PURGE")
sql(f"""CREATE TABLE {TABLE} (event_id BIGINT, user_id STRING, action STRING, event_ts TIMESTAMP)
        USING iceberg PARTITIONED BY (days(event_ts))""")
location = sql(f"DESCRIBE TABLE EXTENDED {TABLE}").where("col_name = 'Location'").collect()[0][1]

print("== files on storage per commit ==")
print(f"after CREATE, before any data   metadata_files={count_files(location, 'metadata'):<3} "
      f"data_files={count_files(location, 'data')}")
for i, day in enumerate((1, 2, 3), start=1):
    batch(day).writeTo(TABLE).append()
    print(f"after append #{i}                 metadata_files={count_files(location, 'metadata'):<3} "
          f"data_files={count_files(location, 'data')}")

print("\n== what each commit recorded ==")
for r in sql(f"""SELECT snapshot_id, parent_id, operation, summary
                 FROM {TABLE}.snapshots ORDER BY committed_at""").collect():
    s = r["summary"]
    print(f"snap={r['snapshot_id']} parent={r['parent_id']} op={r['operation']}  "
          f"added-data-files={s['added-data-files']} total-data-files={s['total-data-files']} "
          f"total-records={s['total-records']}")

print("\n== manifests the current snapshot points at ==")
sql(f"""SELECT added_snapshot_id, added_data_files_count, existing_data_files_count
        FROM {TABLE}.manifests ORDER BY added_snapshot_id""").show(truncate=False)
print("metadata.json files in the log:",
      sql(f"SELECT count(*) FROM {TABLE}.metadata_log_entries").collect()[0][0])

print("\n== watching one more commit ==")
before = sql(f"SELECT max(committed_at) c FROM {TABLE}.snapshots").collect()[0]["c"]
batch(4, 1000).writeTo(TABLE).append()
new = sql(f"""SELECT snapshot_id, operation, summary FROM {TABLE}.snapshots
              WHERE committed_at > timestamp'{before}'""").collect()
for r in new:
    print(r["operation"], "added-data-files =", r["summary"].get("added-data-files"),
          "added-records =", r["summary"].get("added-records"))
sql(f"""SELECT file, latest_snapshot_id FROM {TABLE}.metadata_log_entries
        ORDER BY timestamp DESC LIMIT 2""").show(truncate=False)

print("== retry and merge settings ==")
sql(f"""ALTER TABLE {TABLE} SET TBLPROPERTIES (
        'commit.retry.num-retries'      = '10',
        'commit.retry.max-wait-ms'      = '10000',
        'commit.retry.total-timeout-ms' = '300000')""")
sql(f"SHOW TBLPROPERTIES {TABLE}").where("key LIKE 'commit.%'").show(truncate=False)
spark.stop()

concurrent_writers.py runs the two-writer test with default retries and with retries switched off, then the MERGE against a concurrent append under both isolation levels. It turns on Iceberg’s LoggingMetricsReporter, so grep -oE "attempts=CounterResult\{unit=COUNT, value=[0-9]+\}" on its output counts the attempts per commit:

"""Two writers on one Iceberg table: retries, exhausted retries, and isolation levels.

Run inside datalake-playground:
  docker cp concurrent_writers.py spark-master:/tmp/
  docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[4]" \
    --driver-memory 1g --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
    /tmp/concurrent_writers.py'

Iceberg's LoggingMetricsReporter prints one CommitReport per commit at INFO,
including how many attempts it took; the script turns that logger on so the
retries are countable: grep "attempts=" in the output.
"""
import threading
import time

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf

spark = (
    SparkSession.builder.appName("iceberg-concurrent-writers")
    .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")
jvm = spark._jvm
jvm.org.apache.logging.log4j.core.config.Configurator.setLevel(
    "org.apache.iceberg.metrics.LoggingMetricsReporter", jvm.org.apache.logging.log4j.Level.INFO)
sql = spark.sql
NS = "ice.blog_write_path"
sql(f"CREATE NAMESPACE IF NOT EXISTS {NS}")


def run_writers(table, per_writer=6):
    """Two threads insert per_writer payments each into the same table."""
    errors = []

    def writer(name):
        for i in range(per_writer):
            try:
                sql(f"INSERT INTO {table} VALUES ('{name}', {i}, {10.0 + i})")
            except Exception as e:                      # keep going, record what failed
                j = getattr(e, "java_exception", None)
                errors.append((name, i, j.getClass().getName() if j else type(e).__name__,
                               (j.getMessage() if j else str(e)).splitlines()[0][:160]))

    threads = [threading.Thread(target=writer, args=(n,)) for n in ("A", "B")]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return errors


def report(table):
    rows = sql(f"SELECT count(*) FROM {table}").collect()[0][0]
    snaps = sql(f"SELECT snapshot_id, parent_id FROM {table}.snapshots ORDER BY committed_at").collect()
    ids = {s["snapshot_id"] for s in snaps}
    per = [tuple(r) for r in sql(f"SELECT writer, count(*) FROM {table} GROUP BY writer ORDER BY writer").collect()]
    linear = all(s["parent_id"] in ids for s in snaps[1:]) and \
        len({s["parent_id"] for s in snaps[1:]}) == len(snaps) - 1
    print(f"final rows   = {rows}")
    print(f"snapshots    = {len(snaps)}")
    print(f"per-writer   = {per}")
    print(f"parent chain is linear (every parent used once): {linear}")


def make_payments(table, extra_props=""):
    sql(f"DROP TABLE IF EXISTS {table} PURGE")
    sql(f"CREATE TABLE {table} (writer STRING, seq INT, amount DOUBLE) USING iceberg {extra_props}")
    sql(f"INSERT INTO {table} VALUES ('seed', 0, 1.0)")


print("== 1. two writers, default retries ==")
T1 = f"{NS}.payments"
make_payments(T1)
errors = run_writers(T1)
print("concurrent writer errors:", errors or "none")
report(T1)

print("\n== 2. the same, with commit.retry.num-retries = 0 ==")
T2 = f"{NS}.payments_no_retry"
make_payments(T2, "TBLPROPERTIES ('commit.retry.num-retries' = '0')")
errors = run_writers(T2)
print("failed inserts:", len(errors))
for e in sorted({(c, m) for _, _, c, m in errors}):
    print("  ", e[0], "-", e[1])
report(T2)


print("\n== 3. MERGE against a concurrent append ==")


@udf("double")
def slow_amount(x):
    time.sleep(2)            # keeps the MERGE running while the other writer commits
    return x


def merge_while_appending(level):
    table = f"{NS}.accounts_{level}"
    sql(f"DROP TABLE IF EXISTS {table} PURGE")
    sql(f"""CREATE TABLE {table} (account_id INT, balance DOUBLE) USING iceberg
            TBLPROPERTIES ('write.merge.isolation-level' = '{level}')""")
    sql(f"INSERT INTO {table} VALUES (1, 100.0), (2, 200.0), (3, 300.0)")
    (spark.createDataFrame([(1, 5.0), (2, 5.0), (3, 5.0)], "account_id INT, delta DOUBLE")
          .repartition(1).withColumn("delta", slow_amount("delta"))
          .createOrReplaceTempView("deltas"))
    outcome = {}

    def merge():
        try:
            sql(f"""MERGE INTO {table} t USING deltas d ON t.account_id = d.account_id
                    WHEN MATCHED THEN UPDATE SET t.balance = t.balance + d.delta""")
            outcome["merge"] = "committed"
        except Exception as e:
            j = getattr(e, "java_exception", None)
            outcome["merge"] = (j.getClass().getName() if j else type(e).__name__) + ": " + \
                (j.getMessage() if j else str(e)).splitlines()[0][:150]

    m = threading.Thread(target=merge)
    m.start()
    time.sleep(3)                                     # the MERGE has planned and is running
    sql(f"INSERT INTO {table} VALUES (4, 400.0)")     # a concurrent append commits first
    m.join()
    print(f"{level:<12} MERGE -> {outcome['merge']}")
    print(f"{'':<12} rows now: {[tuple(r) for r in sql(f'SELECT * FROM {table} ORDER BY account_id').collect()]}")


merge_while_appending("serializable")
merge_while_appending("snapshot")
spark.stop()

Snapshot ids and metadata file names are random, so yours will differ; the counts will not. The MERGE test relies on timing: a UDF that sleeps two seconds per row keeps the MERGE running while the append commits three seconds in.

Clean up afterwards. PURGE removes the files a table references, then mc rm removes the empty folders and any metadata left by commits that lost a race:

docker exec spark-master bash -c '$SPARK_HOME/bin/spark-sql --master "local[1]" \
  --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
  --conf spark.sql.catalog.ice=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.ice.type=hive \
  --conf spark.sql.catalog.ice.uri=thrift://hive-metastore:9083 \
  -e "DROP TABLE ice.blog_write_path.events PURGE; DROP TABLE ice.blog_write_path.payments PURGE;
      DROP TABLE ice.blog_write_path.payments_no_retry PURGE;
      DROP TABLE ice.blog_write_path.accounts_serializable PURGE;
      DROP TABLE ice.blog_write_path.accounts_snapshot PURGE;
      DROP NAMESPACE ice.blog_write_path"' 2>/dev/null
docker exec mc /usr/bin/mc rm --recursive --force minio/warehouse/blog_write_path.db/

References

• Iceberg table specification for metadata, snapshots and the commit protocol
• Iceberg configuration for commit.retry.* and isolation-level properties
• SnapshotProducer, the retry loop and backoff quoted above
• ManifestFile for the manifest schema
• Apache Iceberg architecture for the commit contract in source
• Life of a read query in Iceberg for how these files are consumed
• Table maintenance for Iceberg and Hudi for reclaiming what commits leave behind

Trademarks

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