Lakehouse Table Formats Scenarios & Design

The reasoning

**Start from the write pattern, not the format.** Two million updates a day over 400 million rows, scattered across all dates, means every 5-minute batch touches rows in hundreds or thousands of files. Under copy-on-write each touched file is rewritten, so the write cost per commit is roughly (files touched × file size), repeated 288 times a day. That points to merge-on-read in whichever format you pick.

**Hudi.** A `MERGE_ON_READ` table keyed on `order_id` with `hoodie.table.ordering.fields` set to the binlog position or `updated_at`, so replays and out-of-order events resolve to the latest version. Upserts need the index to find each key's file group: with updates spread over all dates, bloom pruning is weak; a record-level index or a bucket index avoids scanning. Compaction runs asynchronously so the writer is not blocked. Downstream jobs can use Hudi incremental queries from the timeline. The cost is the operational surface — index choice, compaction and cleaner settings are yours to tune.

**Iceberg.** Spark `MERGE INTO` with `write.merge.mode = merge-on-read` writes position deletes (deletion vectors in v3) plus new rows; a Flink upsert writer writes equality deletes, which are cheap to write and expensive to read until `rewrite_data_files` applies them. It works well when Trino, Snowflake or other engines must read the same table through a shared catalog. **Delta** with deletion vectors and `MERGE` is the natural choice if the platform is already Databricks.

**How to say it in the room:** "On mechanism, all three can do this with merge-on-read plus compaction. Hudi has the most built-in machinery for keyed upserts and incremental reads; Iceberg has the widest engine and catalog support; Delta is the default on Databricks. I would choose on the rest of the stack, then prove it with our own data, because published comparisons come from vendors." That last sentence is the one interviewers listen for.

The reasoning

**Why Iceberg here.** There are no updates, so the upsert machinery that distinguishes Hudi buys little. What matters is (1) three engines reading the same table safely, which needs one catalog they all support — Iceberg's REST catalog, Glue, or the platform's catalog — and (2) planning over five years of files without listing directories, which Iceberg does through manifests with partition ranges and column stats. Snowflake, Trino and Spark all read Iceberg natively. Delta would work well if the platform were Databricks-centred, and Delta UniForm can expose Iceberg metadata for other readers, read-only.

**Layout.** Partition by `day(event_ts)` (hidden, so analysts filter on `event_ts` and still prune). At roughly 165 GB a day, daily partitions give reasonable file counts at Iceberg's default 512 MB target file size. Set a sort order on the most common filter, such as `WRITE ORDERED BY event_type, user_id`, so column stats in manifests are tight and files can be skipped. Avoid identity partitioning on high-cardinality columns; use `bucket(N, user_id)` only if joins or point lookups on `user_id` dominate.

**Operations.** Daily ingestion by many small writers produces small files; run `rewrite_data_files` on recent partitions. Expire snapshots on a schedule so appends do not accumulate metadata forever, and run `rewrite_manifests` if manifests fragment. Keep the one-table-one-catalog rule: every engine must commit through the same catalog.

**When not Iceberg.** If the same table later needs frequent keyed upserts with incremental consumers, reconsider; if the whole organisation is on Databricks with Unity Catalog, Delta is the lower-friction default and UniForm covers external readers.

The reasoning

**Plain Parquet fails the requirement.** Daily corrections to 30 days means rewriting up to 30 partitions every day by `INSERT OVERWRITE`, with no atomic commit: an analyst querying during the rewrite sees missing or doubled rows, and a failed job leaves a partition half replaced. It is acceptable only for append-only data with a single writer.

**Delta Lake.** Spark (Synapse Spark, Databricks, Fabric) writes it with `MERGE INTO` for the corrections; Synapse serverless SQL can query Delta folders. You get atomic commits and time travel on the Spark side. Plan `OPTIMIZE` for small files and `VACUUM` for old versions (7-day default retention).

**Iceberg.** The right answer when several engines outside Azure's Spark estate must write the table, or when the organisation standardises on an Iceberg catalog. On Azure the reader decides it: Synapse serverless SQL reads Parquet, CSV and Delta but not Iceberg, so choosing Iceberg means moving the analysts to Spark or to Fabric (OneLake shortcuts can expose Iceberg tables). Delta UniForm, which generates Iceberg metadata for a Delta table, is an option if Iceberg readers are needed later.

**Say what you would verify.** The real signal in this question is "which engine reads it?". Name the reader, confirm its support for the format's features you rely on (deletion vectors, column mapping), and only then choose.

The reasoning

**Where the data lives.** A native Snowflake table stores data in Snowflake's own storage format; other engines cannot read it without exporting. An Iceberg table stores Parquet and Iceberg metadata in an external volume — your S3, GCS or Azure storage — so Spark can read the same files. For a 50 TB table read nightly by Spark, that avoids a daily export.

**Who owns commits.** With Snowflake as the Iceberg catalog, Snowflake writes the table and other engines read it through a catalog interface. With an external Iceberg REST catalog such as Glue, Unity Catalog or Snowflake Open Catalog, both Spark and Snowflake can write: full DML (`INSERT`, `UPDATE`, `DELETE`, `MERGE`) on externally managed Iceberg tables, including through a catalog-linked database, has been generally available since October 2025. The rule from the catalog question still applies: every engine commits through that one catalog, and none points at metadata files directly.

**What you take on.** Storage cost moves to your cloud bill. File layout, compaction and snapshot expiry become questions you have to answer (Snowflake performs some maintenance on tables it manages). Features specific to native tables may not apply the same way — check time travel, clustering and data-sharing behaviour for Iceberg tables before promising them.

**The decision line:** multi-engine reads or data ownership requirements → Iceberg; Snowflake-only analytics → native. Mixed: keep curated marts native and the large shared fact tables Iceberg.

The reasoning

**Check native support first.** Trino and Athena have Delta Lake connectors; if they cover the features your tables use, no translation is needed. The problems start with features the reader does not support, such as deletion vectors or column mapping on older connector versions.

**Manifest file.** `GENERATE symlink_format_manifest FOR TABLE t` writes a list of the current version's Parquet files that Presto-era engines can read as a plain table. It is a snapshot in time: after each write it must be regenerated (or auto-generation enabled), readers see no time travel, and a reader mid-regeneration can see an inconsistent list. A fallback, not a design.

**Delta UniForm.** Setting `delta.universalFormat.enabledFormats = 'iceberg'` (with `delta.enableIcebergCompatV2 = 'true'` and column mapping) makes Delta generate Iceberg metadata asynchronously after commits, so Iceberg clients read the same Parquet files. The Delta docs list the constraints: Delta Lake 3.1+ for Iceberg (Hudi is in preview from 3.2), **read-only** from the Iceberg side, and under IcebergCompatV2 **deletion vectors cannot be enabled**; Databricks' IcebergCompatV3 (Iceberg v3) removes that restriction, so check which compatibility version your runtime supports. Iceberg readers may lag the latest Delta commit briefly.

**Apache XTable** (incubating) reads a table's metadata in one format and writes metadata for others without copying data. Its own FAQ lists current limits: no support for Hudi or Iceberg merge-on-read tables or for Delta deletion vectors. It is useful for scheduled sync; like UniForm, treat the translated table as read-only.

The reasoning

**On plain Parquet** there is no row-level delete: you find every file that might contain the user and rewrite each without their rows, across seven years of partitions, with no atomic commit. That is why the question is usually the argument for a table format.

**Step 1 — logical delete.** `DELETE FROM events WHERE user_id = 12345`. With copy-on-write, every data file containing the user is rewritten without them; with merge-on-read, Iceberg writes position deletes (deletion vectors in v3) and Hudi writes delete records to logs. Either way the current snapshot no longer shows the rows. Finding the files cheaply needs the planner to prune — a sort or bucket on `user_id` makes this a handful of files instead of 90,000.

**Step 2 — physical delete.** The old files are still referenced by earlier snapshots. In Iceberg, run `expire_snapshots` so snapshots from before the delete are removed and files only they referenced are deleted; with merge-on-read also run `rewrite_data_files` so data files are rewritten without the deleted rows, then expire again. In Delta, `VACUUM` after the retention window (7 days by default). In Hudi, the cleaner removes superseded file slices after `hoodie.clean.commits.retained` commits; MOR tables need compaction first. Then `remove_orphan_files` (Iceberg default `older_than` 3 days) for anything left by failed jobs.

**Step 3 — prove it.** Query every snapshot still retained for the user id, record the snapshot ids and timestamps of the delete and the expiry, and keep that log. Remember copies outside the table: raw landing zones, downstream tables, extracts. A deletion log keyed by user id, replayed against each table, is how teams make this repeatable.

The reasoning

**Stage only what changed.** Compute a hash of the tracked attributes for incoming rows and compare it with the current row's hash; 100,000 changes out of 50 million means the merge should touch 100,000 keys, not the table. Drop incoming rows whose hash equals the current one.

**One MERGE, two actions per changed key.** The classic trick is to union the staged changes with a copy of each change keyed to null, so a single `MERGE` can both close the old row (`WHEN MATCHED AND t.is_current AND t.hash <> s.hash THEN UPDATE SET is_current = false, valid_to = s.effective_ts`) and insert the new version (`WHEN NOT MATCHED THEN INSERT`). Doing it in one statement keeps the change atomic: readers never see a customer with no current row or two.

**Make the write proportional to the change.** Under copy-on-write, 100,000 scattered keys can touch a large share of the files and rewrite them. Clustering on `customer_id` (Z-order or liquid clustering in Delta, a sort order or `bucket(N, customer_id)` in Iceberg) concentrates the keys in fewer files. Merge-on-read — deletion vectors in Delta, `write.merge.mode = merge-on-read` in Iceberg — turns closing a row into a small delete marker plus a new row, with compaction paying the rewrite later.

**Late and repeated data.** Use an `effective_ts` from the source, not load time, and make the batch idempotent: rerunning it must not create a second version. Keep snapshots long enough to reconcile, and remember that merge-on-read means readers carry delete files until compaction runs.

The reasoning

**Keys and ordering.** Record key = `customer_id`. Ordering field (`hoodie.table.ordering.fields`) = a value that increases with every change at the source — the WAL position (`lsn`) is safer than `updated_at`, which can tie or go backwards with clock changes. With an ordering field set, Hudi uses `EVENT_TIME_ORDERING` and keeps the highest value, both within a batch and against the stored row. A connector restart that replays yesterday's events then changes nothing.

**Deletes.** Debezium emits an event with `op = d` (and a tombstone). The writer has to turn it into a Hudi delete for that key — Hudi's Debezium source and payload classes in Hudi Streamer do this, or a custom writer marks the row with Hudi's delete marker. The ordering check applies to deletes too, so a late update after a delete does not resurrect the row as long as the delete carries the higher LSN.

**Table type and services.** Continuous CDC suits merge-on-read with async compaction. Pick an index that does not degrade as the table grows — bucket or record-level for large tables. Keep the cleaner and archival settings long enough for downstream incremental readers to catch up.

**Downstream.** Consumers read with incremental queries from their last instant. If they need before and after images — to reverse an old value in an aggregate — enable `hoodie.table.cdc.enabled` and read in `cdc` mode.

The reasoning

**What the writer does.** With identifier fields set (`order_id` or `payment_id`) and upsert mode on, Flink does not scan the table to find the old row; it writes an **equality delete** "`payment_id = X`" and the new row, committed at each checkpoint. One commit a minute is 10,080 commits a week, each adding small data files and small equality delete files.

**What readers pay.** Equality deletes apply to every data file with a lower sequence number in the same partition. A reader must load the relevant delete files and filter every row against them. As deletes pile up, planning and scanning both grow — the symptom is steady slowdown with no change in data volume.

**The fix.** Run `CALL system.rewrite_data_files(table => 'db.payments', options => map('delete-file-threshold', '10'))` on a schedule scoped to recent partitions, so data files with many deletes are rewritten with the deletes applied. On v2 tables with position deletes, `rewrite_position_delete_files` compacts those. Then `expire_snapshots` so the superseded files and the thousands of per-minute snapshots go away, and `rewrite_manifests` if manifests fragment.

**Reduce the rate.** A checkpoint interval of 5 to 10 minutes instead of 1 cuts commits, files and deletes proportionally, at the cost of freshness — worth asking whether the consumers need minute-level data. Compaction must coexist with the streaming writer; scoping it by partition and enabling `partial-progress.enabled` limits commit conflicts.

The reasoning

**Why the lake can do this.** Every trip update the stream applied is committed to the Hudi table as an instant. The table holds the latest state of each trip; the timeline records when each change landed. Pinot is the serving copy; Hudi is the replayable one.

**Choose the read.** If the corruption is in Pinot only, a **snapshot query** over `trip_date >= current_date - 30` gives the correct current state of every trip in the window. If the bad deploy also wrote bad records into Hudi, find the last good instant, and either `restore` to a savepoint (destructive, for the whole table) or read a **time-travel** view at that instant and repair forward. Make sure the cleaner and archival have kept that far back — `hoodie.clean.commits.retained` defaults to 10 commits, far less than 30 days at streaming commit rates, which is why teams savepoint before risky deploys.

**Rebuild without stopping the stream.** The batch job reads committed data only, so the live writer is unaffected. Write results to new Pinot segments or a shadow table, validate counts and sums against the Hudi source, and swap atomically. Then use an incremental query from the instant the batch read to catch any trips updated during the rebuild.

**Make it routine.** The replay path should be the same code as the daily batch, parameterised by date range, and exercised regularly. A backfill that has never been run is a hope, not a plan.

The reasoning

**The cause.** Every trigger writes at least one file per task per partition it touched, sized by how much data arrived — not by any target. A one-minute trigger with 20 tasks writing to 4 partitions can create 80 files a minute. Iceberg's `write.target-file-size-bytes` (512 MB default) is a ceiling; it cannot make a batch of 3 MB bigger.

**Why it hurts.** Every file is a manifest entry to plan, a footer to read and a task to schedule; queries spend their time opening files. It also bloats metadata: a commit a minute is 130,000 snapshots in three months unless they are expired.

**Stop making them.** Lengthen the trigger if consumers allow it. Set `write.distribution-mode = hash` so rows are shuffled by partition and each partition gets one writer rather than one file per task. Reduce shuffle partitions for the write.

**Fix the backlog, alongside the stream.** Run `rewrite_data_files` with the `binpack` strategy scoped to recent partitions (`where => 'event_date >= current_date - 7'`), with `partial-progress.enabled = true` so work commits in groups and a conflict with the stream loses only one group. Files are chosen by size thresholds relative to the target, and any group with `min-input-files` (default 5) or more files is rewritten. Use the `sort` strategy instead if you also want better clustering. Afterwards `expire_snapshots` removes the superseded small files and old snapshots, and `rewrite_manifests` regroups the manifest entries.

The reasoning

**Built-in file sizing.** For `INSERT` and `UPSERT`, Hudi looks at the partition's files, treats those below `hoodie.parquet.small.file.limit` as small, and assigns new inserts to them until they approach `hoodie.parquet.max.file.size`, estimating record size from previous commits. Each commit rewrites a small file into a bigger one rather than adding a new one. That is copy-on-write paying a little extra per write to keep reads healthy.

**Why it is not working here.** Common causes: the job uses `BULK_INSERT`, which skips small-file handling; `hoodie.parquet.small.file.limit` was set to 0; each 10-minute batch is spread over many partitions, so each partition gets a sliver; or parallelism is so high that every task writes its own tiny file.

**Fix the write path.** Use `INSERT` rather than `BULK_INSERT` for incremental loads, set a sensible small-file limit, and partition so each batch lands in few partitions (for clicks, the event date). Lower the write parallelism for small batches.

**Fix what exists.** Clustering rewrites the small files into larger ones — `hoodie.clustering.plan.strategy.small.file.limit` selects candidates, `...target.file.max.bytes` sets the output size — and can sort by common filter columns at the same time. Run it inline every N commits or asynchronously; it commits as a `REPLACE_COMMIT`, and the cleaner later removes the replaced files.

The reasoning

**Read the symptom.** A snapshot query reads each file group's latest base file **plus its log files** and merges them. A read-optimised query reads **only base files**. Snapshot slowing down while read-optimised stays fast but stale means base files are not being refreshed — logs are piling up. That is compaction.

**Confirm on the timeline.** Look for `COMPACTION` instants: are they being requested? Are they stuck `INFLIGHT`? When did the last one complete? Compare delta commits since the last compaction with the trigger — `hoodie.compact.inline.trigger.strategy` (`NUM_COMMITS`, `TIME_ELAPSED`, …) and its threshold. Common causes: async compaction was never enabled in a batch writer, the compaction job keeps failing on memory, or it was scheduled but nothing executes the plan.

**Fix the schedule and resources.** Streaming writers normally run compaction asynchronously so ingestion is not blocked — as a separate job or a separate thread. Set the trigger so logs stay bounded, for example compact after a fixed number of delta commits. Give the compaction job its own resources; it rewrites file groups and needs memory proportional to base-file size. Log compaction, which merges small log files without rewriting the base, can help in between.

**Watch it continuously.** The operational metrics that would have caught this: delta commits since last compaction, pending compaction plans, and log files per file group. Alert on those rather than on query latency.

The reasoning

**Where 100 TB came from.** Each hourly copy-on-write MERGE rewrites the files containing changed rows and removes them from the current snapshot — but older snapshots still reference them, so they stay. A year is 8,760 snapshots' worth of superseded files. Separately, Spark tasks that failed or were retried wrote files that no commit ever referenced: **orphans**, which snapshot expiry cannot see.

**expire_snapshots.** `CALL system.expire_snapshots(table => 'db.t', older_than => TIMESTAMP '…', retain_last => 10)` removes snapshots older than the cutoff (default 5 days ago, or the table's `history.expire.max-snapshot-age-ms`) while keeping at least `retain_last` (default 1), then deletes data and metadata files no longer reachable from any remaining snapshot. Tags and branches keep their snapshots, so check `refs` before assuming something will expire.

**remove_orphan_files.** Lists the table location and deletes files not referenced by any metadata, older than `older_than` (default 3 days). The age guard matters: a file written by a job that has not committed yet looks like an orphan, so never run it with a short window while writers are active. Check `equal_schemes` when paths mix `s3://` and `s3a://`, or valid files look orphaned.

**The rest of the routine.** `rewrite_data_files` for small files and accumulated deletes; `rewrite_manifests` for fragmented manifests; `write.metadata.delete-after-commit.enabled = true` so old `metadata.json` files do not pile up with frequent commits. Schedule these per table, monitor the `snapshots` and `files` metadata tables, and alert when storage divided by live data grows.

The reasoning

**Freshness itself.** Time since the latest completed `DELTA_COMMIT`, and end-to-end lag from source change to commit. These tell you the SLA is broken; they do not tell you it is about to be.

**Leading indicators on the write path.** Commit duration trending up at constant input size is the classic early signal. Break it down: time in the index lookup (grows with table size for bloom and simple indexes), time writing files, number of file groups touched per commit. A rising lookup share points at the index; rising file groups touched points at key distribution or partitioning.

**Table-service backlog.** Pending compaction plans, delta commits since the last completed compaction, and log files per file group — the merge-on-read backlog that slows readers and, when compaction and writer compete for resources, the writer too. Also cleaner lag (file slices waiting to be cleaned) and failed or rolled-back instants on the timeline.

**Where they come from.** Hudi reports commit metadata on the timeline for every instant and can push metrics through its metrics reporters (Prometheus, Datadog, Graphite and others are supported). Alert on trends — commit duration up 50% week on week — not only on absolute thresholds.

The reasoning

**In-place, no copy.** Iceberg's Spark procedures: `snapshot` creates a light-weight Iceberg table that uses the source's data files without changing the source — writes go to the new table's location — ideal for testing readers. `migrate` replaces the Hive table with an Iceberg table loaded with the source's data files, copying schema, partitioning and properties. It fails if any partition uses an unsupported format or if the table is bucketed, and existing files are read through a name-to-id mapping from the original schema.

**Or build alongside.** `add_files` imports files from a Hive or path-based table into an existing Iceberg table without moving them, partition by partition — useful for a gradual cut-over — but it does not check the files' schema against the table's, and afterwards Iceberg owns those files, so snapshot expiry can delete them. A full rewrite with `CREATE TABLE … AS SELECT` costs 200 TB of compute and storage but gives you a new partition spec, sort order and file sizes from day one.

**Cut over in order.** (1) Snapshot and run a sample of the 40 jobs against it; compare row counts and aggregates per partition. (2) Pause writers, run `migrate` or finish the build, and point the catalog entry at the Iceberg table. (3) Move readers — engines must read through an Iceberg-aware catalog, not the raw path. (4) Move writers to Iceberg writes. (5) Only then change partitioning (for example to hidden `day(event_ts)`), which is metadata-only thanks to partition evolution.

**Risks to name.** Hive jobs that write directly to paths bypass Iceberg metadata and their files are invisible; engines without Iceberg support break at cutover; and `remove_orphan_files` on a migrated location can see files from path-writing jobs as orphans. Inventory every writer before you start.

The reasoning

**What both formats do.** Delta writes a new log entry only if the version it read is still the latest compatible one, checking for conflicting concurrent changes. Iceberg builds a new snapshot and swaps the metadata pointer; before retrying after losing the race it validates the operation's assumptions. For row-level `MERGE`, Iceberg's default serialisable isolation fails the commit if files that could match its condition were added or removed since it started. Neither silently merges two conflicting writes.

**Why billing loses.** The billing MERGE reads the table for 40 minutes. In that time the CRM job commits two or three times, touching customer rows in the same files or partitions. When billing tries to commit, its validation fails and 40 minutes of work is discarded. The short job almost always wins; the long one loses more often as the short one gets more frequent.

**Fixes, in order of preference.** (1) **Separate the data**: if billing updates columns CRM never touches, split the table (customers core vs customer billing) so they never conflict. (2) **Narrow the conflict**: partition or bucket so each job touches distinct files, and make the MERGE condition include the partition predicate so validation only considers that slice. (3) **Shorten the long write**: stage billing changes first and make the MERGE itself fast, so the window is minutes. (4) **Serialise**: pause CRM while billing commits, via orchestration. Snapshot isolation (`write.merge.isolation-level = snapshot`) relaxes some checks but only where you accept the anomaly it allows — know which before switching.

**Retries are not a fix.** `commit.retry.num-retries` (default 4) helps with brief races on cheap commits; a MERGE that fails validation has to redo its read, and blind retry loops just repeat the 40-minute loss.

The reasoning

**Size first.** At 5 billion rows a day and a few hundred bytes per row compressed, a day is on the order of a terabyte. Daily partitions would be large but fine for pruning; hourly (`hour(event_ts)`) gives around 24 partitions a day and is useful if most queries hit recent hours. Because partitioning is hidden, users filter on `event_ts` either way, and partition evolution lets you switch later without rewriting.

**Do not partition on title or device.** Identity partitions on `title_id` would create hundreds of thousands of tiny partitions. Instead set a sort order — `ALTER TABLE … WRITE ORDERED BY title_id, device_type` — so compaction writes files with narrow value ranges, and manifest column stats let a query for one title skip most files. `bucket(N, title_id)` is worth it only if joins on title dominate.

**Writes and files.** Flink commits at each checkpoint; small files are inevitable. Keep the checkpoint interval as long as freshness allows, and compact recent partitions continuously with `rewrite_data_files` using the `sort` strategy so the sort order is applied, targeting the default 512 MB file size. Use `write.distribution-mode = hash` in Spark backfills so each partition gets few writers.

**Metadata and history.** Expire snapshots daily with a retention that covers recovery (and tag important snapshots), rewrite manifests when they fragment, enable `write.metadata.delete-after-commit.enabled`, and run `remove_orphan_files` with its age guard. Schema changes add columns by id without rewrites, which matters for an event schema that grows every quarter.

The reasoning

**Find the boundary.** `SELECT snapshot_id, committed_at, operation, summary FROM db.transactions.snapshots ORDER BY committed_at` shows each commit; the 02:00 one has the bad job's id in its summary. Tag the snapshot before it — `ALTER TABLE db.transactions CREATE TAG good_0159 AS OF VERSION <id>` — so expiry cannot remove it while you work.

**Verify with time travel.** Compare `SELECT … FROM db.transactions VERSION AS OF <good_id>` with the current table: which rows changed, and did the two later commits touch the same rows? The `changes` view from `create_changelog_view` between the two snapshots gives inserted, deleted and updated rows directly.

**Choose the restore.** If the later commits can be replayed, `CALL system.rollback_to_snapshot('db.transactions', <good_id>)` makes the good snapshot current — a metadata operation, instant, and itself reversible because the bad snapshots still exist until expired. Then re-run the two later jobs. If they cannot be replayed, **repair forward**: `MERGE INTO db.transactions t USING (SELECT … FROM db.transactions VERSION AS OF <good_id> WHERE <rows the bad job touched>) g ON … WHEN MATCHED THEN UPDATE SET amount = g.amount`, which keeps later commits intact.

**Downstream.** Anything that read the table since 02:00 consumed bad data — find those readers (incremental consumers track snapshot ids) and re-run them. Delta's equivalent is `RESTORE TABLE … TO VERSION AS OF`; Hudi's is savepoint and restore. Iceberg also offers write-audit-publish with branches, which stops the next bad write reaching `main` at all.

The reasoning

**Locate the step.** Commit metadata and Spark stages split the upsert into tagging (index lookup), and writing. Batch size is unchanged, so write cost is roughly unchanged; tagging is what scales with the table. With a bloom index, Hudi prunes candidate files by key range and bloom filter. Random UUIDs spread across every file's key range, so range pruning does nothing and every file's bloom filter must be checked — lookup cost now grows with a table ten times larger.

**Change the index.** A **record-level index** (`RECORD_LEVEL_INDEX`, partitioned, or `GLOBAL_RECORD_LEVEL_INDEX`) stores an exact key-to-file map in the metadata table, so the lookup is a keyed read rather than a scan of filters. A **bucket index** removes the lookup entirely by hashing keys to file groups — but the bucket count is fixed with the `SIMPLE` engine, so size it for future growth or use consistent hashing on merge-on-read. Changing index type on an existing table needs the metadata-table index built first; plan it as a migration.

**Reduce the write too.** 200,000 random keys touch a large share of files; under copy-on-write each touched file is rewritten. Merge-on-read turns those rewrites into log appends, with compaction later. If the key can be made time-ordered (UUIDv7, or a time prefix), bloom pruning becomes effective again because updates hit recent files.

**Check the rest.** Small files multiply the files the lookup must consider — cluster them. Make sure the partition path is part of the lookup if keys are unique per partition (non-global index), so only the partitions in the batch are searched.

The reasoning

**Metadata.** Iceberg: a `metadata.json` per version, pointing through manifest lists to manifests, with the current pointer held by a catalog. Delta: an ordered log of JSON commits in `_delta_log` with Parquet checkpoints. Hudi: a timeline of instants in `.hoodie/timeline`, with a metadata table holding file listings and indexes.

**Strengths by design.** Iceberg: hidden partitioning and partition evolution, column identity by field id, a REST catalog protocol, and read/write support across Spark, Flink, Trino, Snowflake and others. Delta: tight Spark and Databricks integration, deletion vectors, change data feed, and UniForm to expose Iceberg metadata. Hudi: record keys with several index types, merge-on-read with async compaction, incremental and CDC queries from the timeline, and table services that run inside the writer.

**Choosing.** Decide by workload and ecosystem: heavy keyed upserts with incremental consumers lean towards Hudi; many engines sharing append-heavy tables lean towards Iceberg; a Databricks-centred platform leans towards Delta. Interoperability tools — Delta UniForm, Apache XTable (incubating) — translate metadata, mostly read-only on the target side. Treat published benchmarks from Onehouse, Databricks or Iceberg vendors as vendor claims and test on your own data.

The reasoning

**Why files get small.** Streaming and frequent batch writes commit whatever data arrived, often a few MB per task. Every file costs a metadata entry, an open, a footer read and a task, so millions of small files make queries slow and planning expensive.

**Two meanings.** (1) **Bin-packing/sorting** small files into target-sized ones — Iceberg `rewrite_data_files` (binpack or sort; 512 MB default target), Delta `OPTIMIZE`, Hudi clustering. (2) **Folding pending changes** into data — Iceberg rewriting data files with their delete files applied, Hudi compaction merging log files into new base files on merge-on-read tables.

**How to run it.** Scope it to recently written partitions, schedule it to match the write rate, let it commit incrementally so it coexists with writers, and follow it with snapshot expiry, because the replaced files stay on storage until no snapshot references them.

The reasoning

**Safe almost everywhere:** adding a nullable column (old rows read null), widening `int` to `long` or `float` to `double`, and widening decimal precision. Iceberg supports these as metadata changes; Delta supports adding columns and some type widening; Hudi supports adding columns and widening on write.

**Safe only with column identity:** renaming and dropping. Iceberg reads by field id, so both are metadata-only and a re-added name never picks up old values. Delta needs column mapping enabled for rename and drop. Formats resolving by name can resurrect old data when a name is reused.

**Unsafe:** narrowing types, changing `string` to `int`, or keeping a name while changing its meaning. Add a new column, backfill, move readers, then drop the old one — and put the rules in a data contract so producers cannot make breaking changes unilaterally.

The reasoning

**Mechanics.** An upsert needs a key, a way to find existing rows with that key, and an atomic commit. On a database the index and transaction log provide those. On a lake the table format provides the commit, and the lookup is a join (Iceberg, Delta `MERGE`) or an index (Hudi's record key index).

**Syntax.** `MERGE INTO t USING s ON t.id = s.id WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *` in Spark SQL for Iceberg and Delta; `hoodie.datasource.write.operation = upsert` for Hudi, with a record key and ordering field.

**The two things to get right.** A key that is actually unique in the source batch (deduplicate first, or the merge fails or picks arbitrarily), and an ordering rule so a late, older version cannot overwrite a newer one. The cost choice is copy-on-write versus merge-on-read.

The reasoning

**Why files linger.** Updates and deletes write new files and mark old ones removed in the new version; the old version still references them for time travel and for readers mid-query. Nothing deletes them until a clean-up job decides no version that matters needs them.

**Delta.** `VACUUM t` removes files not referenced by versions newer than the retention threshold (`delta.deletedFileRetentionDuration`, 7 days by default); shortening it below the default is blocked unless a safety check is turned off, because long-running readers can fail.

**Iceberg and Hudi.** Iceberg `expire_snapshots` removes snapshots older than a cutoff (default 5 days, keeping at least `retain_last`) and deletes files only they referenced; `remove_orphan_files` (default `older_than` 3 days) removes debris from failed writes. Hudi's cleaner, `hoodie.clean.policy` defaulting to `KEEP_LATEST_COMMITS` with `hoodie.clean.commits.retained` of 10, removes superseded file slices and runs automatically after commits by default.