lakehouse

How it is built

Architecture overview

A lakehouse table is a directory of immutable columnar files in an object store plus a small, versioned metadata log that says which files make up the table right now. Nothing is updated in place. A writer adds new data files, then commits by publishing one new version of the metadata; readers pick up a version and see exactly its files, so they are isolated from writers without locks. That one idea (table state is a value you replace atomically, not a folder you mutate) gives ACID commits, time travel, schema change and row-level deletes on top of storage that only knows how to put and get whole objects. The three open table formats, Apache Iceberg, Delta Lake and Apache Hudi, differ in how the metadata is laid out and committed, but share the same pieces. The term comes from the CIDR 2021 paper that argued the warehouse and the data lake should merge into one platform on open file formats (Armbrust et al.).

Parts:

• Data files — columnar files with per-column statistics, stored as immutable objects — columnar-file-format (no page yet), Object storage
• Table state — snapshots, manifest lists and manifests (or a transaction log) that name the live files — table-snapshot-manifests (no page yet)
• Commit protocol — the single atomic step that publishes a new version, with conflict checks — conditional-writes (no page yet), Optimistic concurrency control
• Scan planning — shrinks a query to the files and row groups that can match — partition-pruning (no page yet), zone-maps (no page yet), Bloom filter
• Row-level change — deletes and updates without rewriting whole files — merge-on-read-deletes (no page yet), Roaring bitmaps
• Schema management — adds, drops and renames columns without rewriting data — schema-evolution (no page yet)
• Table maintenance — merges small files into large ones, expires old snapshots — Bin packing

 engines: batch (Spark, Flink) · SQL query engine · streaming ingestion
        │ read: load current version        │ write: stage files, then commit
        ▼                                    ▼
 ┌──────────────── catalog (pointer to the current metadata) ───────────────┐
 │  table ─► metadata v17 ──swap: expect v17, write v18──► metadata v18      │
 └───────────────────────────────┬───────────────────────────────────────────┘
                                 ▼
   metadata file ─► snapshot ─► manifest list ─► manifests ─► data files (+ delete files)
   (schema, partition spec)      (partition stats)   (per file: path, size, row count, min/max)
                                 │
 object store:   data/part-0001.parquet  part-0002.parquet ...   metadata/*.json *.avro
 background:     compaction (bin pack) · expire snapshots · delete orphan files

Iceberg and Delta make the same cut in different places. Iceberg publishes new metadata by swapping a pointer held in a catalog. Delta appends numbered JSON entries to a _delta_log directory and relies on the object store to refuse a second writer of the same file name.

How the parts fit together

Data files

Object storage holds everything, data and metadata, and asks for little: Iceberg's specification requires only in-place write (files are not moved or altered once written), seekable reads and deletes, and says these are compatible with object stores and need no random-access writes (source: Iceberg table spec, file system operations). columnar-file-format (no page yet) is the unit the engines scan: Parquet puts rows into row groups, each column of a group into a chunk of encoded pages, and ends the file with a footer holding schema and statistics. A scan reads the footer first with a ranged GET, then only the column chunks it needs. A file of hundreds of megabytes is the target (Iceberg's write.target-file-size-bytes defaults to 536,870,912 bytes, 512 MB; source: Iceberg configuration docs), because each file costs a request, a manifest entry and a task.

The contract with the table layer: a data file is never changed after it is written, and it carries enough statistics (row count, per-column minimum, maximum and null count) for the table layer to copy into its metadata so a query can skip it without opening it.

Table state

table-snapshot-manifests (no page yet) records which files are in the table. Iceberg's tree: a metadata JSON file holds the schema, partition specs and a list of snapshots; a snapshot points at a manifest list; the list names manifests, with partition statistics and file counts for each, so whole manifests can be skipped; a manifest has one row per data file with its partition values and column bounds (source: Iceberg table spec). Delta replaces the tree with a log: version n is a file named with the next available version number zero-padded to 20 digits, containing the add and remove actions of that commit, and a checkpoint Parquet file periodically folds the log into full state; by default clients write a checkpoint every 10 transactions (source: Delta protocol; Armbrust et al., VLDB 2020). Hudi keeps a timeline of instants and groups data into file groups with slices per commit (source: Hudi docs, table types).

All three give snapshot isolation: a query resolves one version at start and never sees partial commits. They also give time travel, because an old snapshot is just an old pointer, provided its files have not been deleted yet. Planning a scan costs a constant number of reads of the metadata tree instead of listing every object, which was the cost the design set out to remove (Iceberg goal: planning with O(1) remote calls rather than O(n) in the table size).

Commit protocol

Commit is where object-store semantics bite. conditional-writes (no page yet) are the primitive: create a name only if it does not exist (put-if-absent) or replace only if the current version is the one you read (compare-and-swap). Without it, two writers both believe they made version 18. S3 now offers this directly: an If-None-Match header refuses the write if the key exists, and If-Match compares the ETag (source: AWS S3 user guide, conditional writes); Google Cloud Storage and Azure Blob Storage supported atomic put-if-absent when the Delta paper was written. Delta's documentation says that on S3, the default mode allows concurrent writes only from a single Spark driver because S3 lacked mutual exclusion, and a multi-cluster mode uses DynamoDB to provide it (source: Delta Lake storage docs). Iceberg leaves it to the catalog: the writer creates a new metadata file and the catalog atomically replaces the pointer, which the Glue catalog does with optimistic locking and the filesystem-based catalog does with a DynamoDB lock table (source: Iceberg AWS docs).

On top of the primitive sits Optimistic concurrency control. A writer assumes no one else will commit, does its work (writes data files, builds new metadata), and at the end checks that the base it read is still current. If not, it reloads and validates: Iceberg says an update can be retried "under well-defined conditions", for example a change that rewrites files can be re-applied if all rewritten files are still in the table (source: Iceberg table spec, optimistic concurrency). Two appends never conflict and just rebase; a compaction and an update touching the same files do. The isolation level is chosen by what the writer validates. This is a good fit for the workload: commits are rare compared with reads, conflicts are rare, and a failed commit wastes only the staged files, which maintenance later deletes as orphans. Each commit is a few objects written and one conditional step, so the commit rate per table is bounded by that round trip (order of a few per second); streaming ingestion therefore batches many records into each commit.

Scan planning

Three layers prune work, from coarse to fine.

partition-pruning (no page yet) skips whole sets of files by partition value. Iceberg uses hidden partitioning: the partition spec is derived from columns through transforms (identity, bucket[N], truncate[W], year, month, day, hour), and queries are planned from predicates on the data columns, not on partition columns the user must know about (source: Iceberg table spec, partitioning). The spec lists evolving partition schemes as a goal, so an old and a new layout can coexist.

zone-maps (no page yet) are the min/max statistics: Iceberg manifests keep lower_bounds and upper_bounds per column per file (source: Iceberg table spec, manifests), Delta's add action carries numRecords, minValues, maxValues and null counts in its stats field (source: Delta protocol). A file whose range cannot contain the predicate value is dropped without being read. Sorting or clustering data by the filter column (Z-order, Hilbert, Delta liquid clustering) makes the ranges narrow enough to prune.

A Bloom filter catches what min/max cannot: an equality predicate on a high-cardinality column whose values are spread across the whole range. Parquet defines a split block Bloom filter per column chunk for exactly this case, where dictionaries are too large and min/max is too wide (source: Parquet bloom filter spec). It answers "definitely not here" or "probably here" and is checked after file-level statistics have already narrowed the set.

Row-level change

Files are immutable, so a DELETE or UPDATE has two designs, the same ones Hudi names copy-on-write and merge-on-read. Copy-on-write rewrites each affected file with the changed rows removed: reads stay simple, writes cost the whole file (source: Hudi docs, table types). merge-on-read-deletes (no page yet) writes a small delete file and merges it at read time. Iceberg v2 added position delete files and equality deletes; v3 adds binary deletion vectors, which "identify deleted rows of a file by encoding deleted positions in a bitmap" (source: Iceberg table spec, delete formats). Delta's deletion vectors work the same way, stored inline or as sidecar files in the RoaringBitmapArray format (source: Delta protocol).

Roaring bitmaps are why this is cheap. A deletion vector is a set of 64-bit row positions, but most fit in 32 bits, so Iceberg keeps a map from the high 32 bits to a 32-bit Roaring bitmap of the low bits. Sparse sets stay as small arrays, dense sets as bitmaps, long runs as run lists. The reader loads the vector for each data file it opens and masks rows. The contract: deletion vectors are keyed by data file, so planning knows which delete applies to which file, and Delta also notes that after a deletion vector is added, statistics may be outdated and numRecords must stay equal to the physical count. Compaction folds vectors back into rewritten files.

Schema management

schema-evolution (no page yet) must let columns change without touching terabytes of Parquet. Iceberg's goals state safe add, drop, reorder and rename including nested fields, which works because fields have stable numeric IDs in the table metadata rather than being matched by name or position (source: Iceberg table spec, goals and schemas). Delta provides the same through column mapping (source: Delta protocol). The compatibility rule is the usual one: adding an optional column is backward compatible, dropping a required one is not, and widening a type must be explicit. The schema lives in the table metadata, so a schema registry can govern the producers while the table format records every version used by every data file.

Table maintenance

Streaming and small batch commits produce small files, and small files make planning and scans slow. Compaction reads small files and rewrites them to the target size, a bin-packing problem: group files up to a size so that output files fill their bin. Iceberg's rewrite_data_files defaults to the binpack strategy and also supports sort, and Delta's OPTIMIZE is described as bin-packing and as idempotent (source: Iceberg Spark procedures; Delta optimizations docs). Maintenance is itself a transaction: compaction commits a "replace these files with those" snapshot, validated like any other commit. Expiring snapshots, which Iceberg recommends doing regularly, is what actually frees data files and keeps metadata small (source: Iceberg maintenance docs); Delta's VACUUM keeps removed files for a retention period (default 7 days) so concurrent readers and time travel still work (source: Delta protocol).

Data and control flow

Append of one micro-batch:

1. The ingestion job collects a few seconds to minutes of rows, sorts or clusters them, writes Parquet files of the target size to the object store, collecting row counts and column bounds.
2. It reads the current table metadata (one pointer read plus the metadata file) to learn schema and partition spec.
3. It builds new manifests listing the added files and a new snapshot (Iceberg), or one _delta_log entry with add actions (Delta).
4. It commits with the conditional step: swap the catalog pointer from v17 to v18, or create 00000000000000000018.json if absent.
5. If the step fails, it reloads, checks the new commits do not conflict with its own, and retries; data files are reused. Time goes to the data upload; the commit is a handful of small requests.

A query with a filter:

1. The query engine or batch engine asks the data catalog for the table's current metadata location and loads it.
2. It evaluates the predicate against partition specs and manifest-list statistics (skipping manifests), then against per-file bounds (skipping files).
3. For the remaining files, it issues ranged reads for the Parquet footer, applies row-group statistics and Bloom filters, and then reads only the needed column chunks.
4. For each file with a deletion vector, it loads the vector and drops the masked rows.
5. Results stream out; the query never holds a lock and sees one snapshot throughout.

Nightly maintenance: a scheduler (the workflow engine) runs compaction by partition, expires snapshots older than the retention window, removes orphan files, and the data quality checks run against the newest snapshot before it is exposed to consumers.

Scaling

1. Small files and metadata size. First limit in practice. Levers: write larger batches, compaction jobs, manifest rewriting, clustering, a checkpoint or root manifest that bounds what readers load.
2. Commit rate per table. Every commit is a conditional write and a metadata write; concurrent writers retry. Levers: fewer, bigger commits, partition the writers so they touch different files, split a hot table into several.
3. Catalog throughput. All commits and loads pass through the pointer store. Levers: a scalable catalog service, caching metadata in the engines, a REST catalog in front of a database.
4. Object-store request rate. S3 documents at least 3,500 write and 5,500 read requests per second per prefix (source: AWS S3 user guide, performance), so spread data files over prefixes and limit listing.
5. Scan cost. Layout (partition, sort, cluster) and statistics decide how many files are opened; add Bloom filters on lookup columns.
6. Delete debt. Many deletion vectors slow reads; compact rewrites them away.

Failure modes

Real implementations

The alternative: a columnar warehouse owns its storage and gives tight performance behind one engine; a lakehouse trades some of that for open files many engines can read. Compare data versioning, which also versions datasets but for different consumers.

Sources

• M. Armbrust et al., Delta Lake: High-Performance ACID Table Storage over Cloud Object Stores, VLDB 2020. paper
• M. Armbrust, A. Ghodsi, R. Xin, M. Zaharia, Lakehouse: A New Generation of Open Platforms that Unify Data Warehousing and Advanced Analytics, CIDR 2021. paper
• Apache Iceberg: table spec, configuration, AWS integration, Spark procedures, maintenance.
• Delta Lake: transaction log protocol, storage configuration, optimizations.
• Apache Hudi: table types, tech specs.
• Apache Parquet: file format, Bloom filter.
• Amazon S3: conditional writes, performance.