Delta Lake: High-Performance ACID Table Storage over Cloud Object Stores

In one breath — the whole paper, compressed

Cloud object stores are cheap and enormous, but they are key-value stores with no cross-key consistency and very slow metadata operations, so a table kept as a directory of Parquet files has no atomicity, no isolation, and a LIST bill that can exceed the query itself. Delta Lake keeps the authoritative record of which objects belong to a table in a write-ahead log stored in the same bucket as the data, as a sequence of numbered JSON records that are periodically compacted into Parquet checkpoints. A transaction commits by atomically creating the next numbered log record, so writers use optimistic concurrency and readers reconstruct a consistent snapshot from a checkpoint plus the log tail, all without any always-running server. Because the log also carries per-object min/max statistics, record counts and null counts, query planning becomes a columnar scan of one checkpoint file rather than millions of LIST calls and Parquet footer reads. On top of this the paper builds time travel, UPSERT/MERGE/DELETE, exactly-once streaming ingest, Z-order clustering, SSD caching and schema evolution, and reports Delta Lake running at thousands of Databricks customers processing exabytes of data per day.

Before this paper — the world it landed in

Between roughly 2014 and 2016 the standard way to hold a large analytical dataset on S3 was just a bunch of Parquet files, optionally arranged into Hive-style partition directories such as mytable/date=2020-01-01/. That worked for append-only scans and little else: a job that had to touch several objects exposed partial state to readers, a crashed job left the table corrupt, and S3's LIST is eventually consistent, so even a successful writer's own objects might not show up. The alternative was a closed-world engine such as Snowflake or Hive ACID that kept the table's true contents in its own strongly consistent metadata service, which meant an always-on service, a bottleneck at millions of partitions, extra connector work per engine and lock-in to one provider. Meanwhile GDPR and ordinary data repair were forcing table-wide updates onto datasets that had been designed as immutable. The authors report that in the first few years of Databricks' cloud service, around half of all support escalations were data corruption, consistency or performance problems caused by cloud storage strategies.

The problem — what was actually breaking

Cloud object stores provide no atomicity across multiple keys, so a query that rewrites several Parquet objects exposes partial updates to concurrent readers and leaves the table corrupted if it crashes.
The popular object stores are only eventually consistent per key and give no guarantee across keys, so a client can see some of a transaction's new objects but not others, and S3's LIST may not return an object that was just PUT.
Metadata operations are expensive at scale: S3's LIST returns at most 1000 keys per call and each call takes tens to hundreds of milliseconds, so listing a dataset of millions of objects sequentially takes minutes.
Per-object min/max statistics live in Parquet footers, so data skipping costs one extra high-latency read per object, and on an object store these skipping checks can take longer than the actual query.
Object stores implement no warehouse management functionality at all: no table versioning, no way to undo a crashed update job, and no audit log of who changed what.
Custom storage engines that fix this with a separate strongly consistent metadata service must run a highly available service, funnel every I/O through it, demand more engineering per connector for engines like Spark, TensorFlow and PyTorch, and tie users to one provider.

Core ideas — the contributions, and why they work

The transaction log lives in the object store

Delta Lake's central move is to record which objects constitute a table in a write-ahead log placed in the same bucket as the data, under a _delta_log prefix, rather than in an external catalog. The log, not a directory listing, is the source of truth about the table's contents, and that is what makes multi-object change atomic: adding, removing or replacing any number of Parquet objects is one new log record. Because the log is just more objects, no server needs to be running to hold table state, so compute can be launched only when a query runs and the table is exactly as available as the underlying object store. This single property separates Delta Lake both from the directory-of-files approach and from closed-world engines that depend on a metadata service.

Open Parquet data, minimal connector surface

Table contents stay in ordinary Apache Parquet objects, optionally arranged in Hive-style partition directories, each named with a GUID chosen by its writer. The authors picked Parquet because it is column-oriented, offers diverse compression, supports nested types for semi-structured data, and already had performant implementations in many engines; ORC would likely have worked similarly, but Parquet had the most mature Spark support. The consequence is that any engine which can read Parquet needs only a small connector to discover which objects to read, instead of a full storage-engine port. For engines where even that was too much, Delta Lake writes symlink manifest files, a mechanism originally added to Hive for symbolic links, so Presto, Athena, Redshift and Snowflake can read a consistent snapshot as an external table.

Checkpoints turn metadata into a columnar table

Replaying every JSON log record from the beginning would be as slow as listing, so clients periodically compact the log into a Parquet checkpoint, by default every 10 transactions. The checkpoint drops provably redundant actions: an add cancelled by a later remove, older adds for an object superseded by the newest, superseded txn records for the same appId, and stale metaData and protocol actions. What remains is a column-oriented file with one add record per object still in the table, carrying that object's record count and per-column min/max and null counts. Finding the objects relevant to a selective query therefore becomes a vectorized scan of a Parquet file, which the authors report is nearly always faster than LIST operations plus reading each object's footer on an object store.

Optimistic concurrency on one atomic put

Every write transaction reduces to a single atomic step: having read version r, create the log record with ID r+1, and fail if another client already created it. Data objects are written first, in parallel, under GUID names that nothing references until the commit, so a crash before the commit leaves only orphaned objects and never a visible half-written table. A losing writer simply retries and, depending on the query's semantics, can reuse the data objects it already wrote and re-attempt the commit at a higher ID. This is textbook optimistic concurrency control, but its entire implementation cost against the object store is one put-if-absent, which is why the design needs no lock manager and, on most clouds, no coordinator at all.

Immutability buys time travel and caching

Data objects and log records are never modified in place, and removals are logged as timestamped tombstones so physical deletion is deferred past a per-table retention interval. Reading an old version is therefore just reconstructing state at an older log record ID, exposed as SQL AS OF timestamp and VERSION AS OF commit_id; a user can undo a bad pipeline run with a MERGE of the table against its own past, and MLflow can automatically record which table version trained a model. The same immutability makes local caching sound, because no object's contents can change under a cached copy, so Databricks caches both data and log objects on cluster SSDs with no invalidation protocol at all. Two features that would each need substantial machinery in a mutable store fall out of one design choice.

Transactional data layout optimization

Because the mapping from table to objects is transactional, a background process can rewrite the physical layout while queries keep running. OPTIMIZE compacts small objects toward a default target size of 1 GB each and computes any missing statistics, and AUTO OPTIMIZE does this automatically for newly written data on the Databricks service. ZORDER BY reorders records along a Morton space-filling curve over several attributes, so each object spans a narrow value range in every chosen dimension rather than in just one, which multiplies the effect of min/max skipping for workloads that filter on different columns at different times. Crucially these rewrites set dataChange to false, telling streaming consumers that the record moves no data and can be skipped entirely.

One storage system for stream, lake and warehouse

Write compaction, exactly-once txn records and cheap log tailing together let a Delta table act as a message queue as well as a table: producers commit small objects at low latency, consumers LIST forward from the last log record ID they processed, and a background job coalesces the small objects later without disturbing either side. Combined with ACID updates, statistics-driven skipping and SSD caching, one object-store table can serve the ETL, streaming and BI roles that previously demanded a message bus, a data lake and a data warehouse with separate copies of the data. The paper names this the lakehouse: standard DBMS management functions applied directly to low-cost object storage. The authors report that many customers collapsed multi-system pipelines into Delta tables, reducing both storage cost and maintenance overhead.

How it works — the mechanism, concretely

Table layout on the object store

A Delta table is a directory, or a set of objects sharing a key prefix, holding Parquet data objects plus a _delta_log subdirectory. Data objects may sit in Hive-style partition subdirectories such as date=2020-01-01/, and each is named with a GUID generated by its writer. Inside _delta_log are log records named with zero-padded increasing integers, 000001.json, 000002.json and so on, occasional checkpoints such as 000003.parquet, and a _last_checkpoint file whose content is the ID of the most recent checkpoint. The zero padding is deliberate: it makes the lexicographic LIST that object stores offer behave like a numeric range scan, so a client can ask for everything after a given ID.

The action vocabulary of a log record

Each .json log record holds an array of actions to apply to the previous version of the table. metaData completely overwrites the schema, partition column names, data file format and options such as marking a table append-only, and must appear in a table's first version. add names a data object and may carry its record count and per-column min/max and null counts; a later add for the same path replaces the older statistics, which is how existing tables get upgraded to richer statistics. remove carries a timestamp and stays in the log and checkpoints as a tombstone until the retention threshold passes and the object is physically deleted, which is what lets concurrent readers keep executing against stale snapshots. protocol raises the Delta protocol version required to read or write the table, commitInfo records provenance such as which user performed the operation, and txn stores an application-supplied (appId, version) pair.

Checkpoint construction and discovery

Any client may write a checkpoint covering the log up to a chosen record ID, writing 000003.parquet for records through 000003.json; Databricks clients do this every 10 transactions by default. The checkpoint keeps one add per object still in the table, the remove tombstones not yet past retention, and the coalesced txn, protocol and metaData state. After the write completes, the client updates _last_checkpoint if its ID is newer than the one already there. This step is purely an optimization: a client that dies mid-checkpoint, or that writes the Parquet file but never updates _last_checkpoint, corrupts nothing, because readers fall back to an older checkpoint plus a longer log tail.

Read protocol under eventual consistency

A reader (1) fetches _last_checkpoint to learn a recent checkpoint ID, (2) issues a LIST starting at that ID, or 0 if absent, to find newer .json and .parquet objects, (3) replays the checkpoint plus those log records to compute the set of objects with adds but no matching removes and their statistics, (4) uses the statistics to prune to the objects the query needs, and (5) reads those objects in parallel across the cluster. Every step tolerates staleness: a stale _last_checkpoint only costs a longer LIST; a LIST returning 000004.json and 000006.json but not 000005.json still fixes the largest ID as the target version and waits for the gap to become visible; a worker that cannot yet see a data object named in the log simply retries after a short delay. Reconstruction is parallelized, in the Spark connector by reading the checkpoint Parquet and the log objects with Spark jobs.

Write protocol and the commit point

A writer (1) finds a recent log record ID r using the read protocol's first two steps, (2) reads table version r if the transaction needs it, (3) writes its new data objects in parallel under fresh GUID names in the correct data directories, (4) attempts to create the r+1 .json log object only if no other client has written it, and (5) optionally writes a checkpoint and then updates _last_checkpoint. Step 4 is the transaction: it is the only step that must be atomic, and it is what makes the write serializable. Nothing written in step 3 is visible to any reader until step 4 succeeds, since readers learn about objects only through the log, so aborting at any earlier point costs only orphaned objects. If step 4 fails because another client took r+1, the transaction retries and, depending on its semantics, may reuse the data objects it already wrote.

Atomic log creation per storage system

The commit needs put-if-absent, which not every large-scale store offers, so step 4 is implemented differently per backend. Google Cloud Storage and Azure Blob Store expose atomic put-if-absent directly, so those are used as is. On distributed filesystems such as HDFS, and on Azure Data Lake Storage, an atomic rename of a temporary file to 000004.json serves the same purpose and fails if the target already exists. Amazon S3 has neither, so Databricks service deployments run a separate lightweight coordination service that ensures only one client can add a record with each log ID; it sits only on the log-write path, not on reads and not on data operations, so its load is low. The open source Spark connector instead assigns distinct log record IDs from in-memory state in the Spark driver, keeping concurrent operations correct within one Spark cluster, and a pluggable LogStore class lets users supply their own strongly consistent coordination mechanism.

Isolation levels and the transaction rate ceiling

Because only one transaction can own each log record ID, all write transactions are serializable and the serial order is simply increasing log record ID. Readers following the read protocol get snapshot isolation; a client needing a serializable read can run a read-write transaction that performs a dummy write. Connectors also cache the highest log record ID they have seen per table, so a client reads its own writes and observes a monotonic sequence of versions even under snapshot isolation. The ceiling is the latency of the atomic put, tens to hundreds of milliseconds, capping commits at several per second, and as in any optimistic scheme a higher offered rate turns into commit failures; snapshot-isolation reads contend with nothing and scale freely. Transactions are scoped to a single table, since each table has its own log.

What the paper showed — measurements and proofs

Querying a small table of 33,000,000 rows spread over many partitions on S3 from 16-node i3.2xlarge clusters: hosted Hive needed over an hour at 10,000 partitions and hosted Presto over an hour at 100,000; Databricks Runtime listing raw Parquet took 450 seconds at 100,000 partitions; Delta Lake took 108 seconds at 1,000,000 partitions, and 17 seconds with the log cached on SSD.
On a 100-object synthetic network-flow table with uniformly random 32-bit IPs and 16-bit ports, a global sort on (sourceIP, sourcePort, destIP, destPort) skipped 99% of objects for a sourceIP filter but 0% for the other three fields, averaging 25%; Z-ordering the same four fields skipped at least 43% in every dimension and 54% on average.
A real 500 TB network traffic dataset at the information security customer described in the paper, Z-ordered on similar fields, was able to skip 93% of the data in the table for multi-attribute queries.
TPC-DS power test on 1 TB of data in S3 with one master and eight i3.2xlarge workers, averaged over three runs: Databricks with Delta 0.93 hours, Databricks with Parquet 0.99, third-party Spark on Parquet 1.44, third-party Presto on Parquet 3.76.
Loading 400 GB of TPC-DS store_sales from CSV on one master and eight i3.2xlarge workers took about the same time into Delta as into Parquet, showing that per-object statistics collection adds no significant overhead over the rest of the data loading work.
In production Delta Lake runs at thousands of Databricks customers processing exabytes per day, around half the service's overall workload, with the largest instances managing billions of objects; the authors report that support issues about cloud storage fell from about half of all escalations to nearly none, with speedups as high as 100x on very high-dimensional datasets.

Limits and trade-offs — conceded and discovered

The paper concedes that transactions are serializable only within a single table, because each table has its own transaction log; sharing one log across tables would remove the limit but would increase contention on the single append point.
The paper concedes that the write transaction rate is capped by object store put latency of tens to hundreds of milliseconds, allowing only several transactions per second, and that optimistic concurrency turns any higher offered rate into commit failures; the authors argue that large batched writes make this acceptable in practice.
The paper concedes that streaming latency is bounded by the underlying object store, so millisecond-scale streaming is out of reach and a few seconds is the realistic target, which is why Delta Lake replaces a message bus only where that latency is tolerable.
The paper concedes that the only index is the per-object min/max statistics, with a Bloom filter based index merely prototyped at the time, so highly selective point lookups still depend on how well the layout was Z-ordered.
The claim that no server is needed carries an asterisk on S3, the most popular target: Databricks runs a separate commit coordination service there, and the open source connector only serializes commits within one Spark driver. Later developments softened this, since S3 gained strong read-after-write consistency in late 2020 and a conditional-write primitive in 2024, and the ecosystem converged on catalog services anyway.

What it became — the systems that inherited it

Delta Lake was open sourced in 2019 under Apache 2 and, together with Apache Hudi and Apache Iceberg, established the table format as its own layer of the stack: an open log of add and remove actions over Parquet, readable by many engines and owned by none. The paper's lakehouse framing became the industry's dominant story for analytics architecture, elaborated by the same group in the CIDR 2021 lakehouse paper and adopted as product positioning across Databricks, Snowflake, Google BigQuery and AWS. Iceberg took the same log-in-the-object-store idea with a different manifest structure and won broad neutral adoption, to the point that Databricks acquired Tabular in 2024 and shipped Delta UniForm so one set of Parquet objects can present both Delta and Iceberg metadata. Individual mechanisms outlived their origin too: time travel with AS OF syntax, MERGE on lake tables, dataChange-style incremental log tailing as a streaming source, and statistics-carrying manifests as a replacement for LIST are now expected features in Trino, Flink, DuckDB, Spark and Apache Paimon. The S3 coordination workaround the paper describes has largely been retired, first by S3's move to strong read-after-write consistency in December 2020 and then by conditional writes in 2024, while catalog services such as Unity Catalog and the Iceberg REST catalog reintroduced a coordinator precisely to get the cross-table transactions the paper listed as future work.

In the paper’s words — verbatim

“The core idea of Delta Lake is simple: we maintain information about which objects are part of a Delta table in an ACID manner, using a write-ahead log that is itself stored in the cloud object store.”

§1

“However, Delta Lake takes 108 seconds even with 1 million partitions, and only 17 seconds if the log is cached on SSDs.”

§6.1

“Delta Lake is implemented solely as a storage format and a set of access protocols for clients, making it simple to operate and highly available, and giving clients direct, high-bandwidth access to the object store.”

§9

Vocabulary — as this paper uses it

Delta table: A directory, or a set of objects sharing a key prefix, holding Parquet data objects plus a _delta_log subdirectory. The log, not a directory listing, defines which objects belong to the table.
Transaction log (_delta_log): The ordered sequence of zero-padded JSON records, each an array of actions applied to the previous table version, that serves as the table's write-ahead log and its only source of truth.
Checkpoint: A Parquet file summarizing the log up to a given record ID with redundant actions removed, written every 10 transactions by default so readers do not have to replay the entire log.
add and remove actions: The log actions that attach or detach a single data object. An add may carry the object's record count and per-column min/max and null counts; a remove is a timestamped tombstone kept until the retention threshold expires.
Data skipping statistics: The per-object min/max values, null counts and record counts stored in the log instead of in Parquet footers, letting the planner eliminate objects with one columnar scan of the checkpoint rather than one high-latency read per object.
dataChange flag: A boolean on add and remove actions; setting it to false marks a commit that only rearranges existing data or adds statistics, so streaming consumers can skip compaction and Z-order rewrites.
txn action: An application-supplied (appId, version) pair committed atomically in the same log record as the data changes, used by Structured Streaming to make writes idempotent and achieve exactly-once semantics.
Z-ordering: Reordering records along a Morton space-filling curve over several columns so that each object spans a narrow value range in every chosen dimension, multiplying the effectiveness of min/max skipping for multi-attribute filters.
Lakehouse: The paper's name for the resulting architecture: standard DBMS management functions such as transactions, versioning and audit logs applied directly to tables in low-cost cloud object storage.

On the timeline — where this sits in the story

Year2016–2020, mainstream in 2020s

MilestoneApache Hudi, Apache Iceberg, Delta Lake, and lakehouse architecture

Problem solvedData lakes were often just files on object storage: weak transactions, poor schema evolution, hard concurrency, and fragile metadata.

How it solved itLakehouse table formats added ACID-like semantics, schema evolution, time travel, indexing/metadata, and table abstraction over object storage.

Why it matteredMerged warehouse-like reliability with lake-like open storage economics.

DDIACh. 4: Storage & Retrieval Ch. 12: Stream Processing

View on the timeline