Delta Lake: Transactions, Schema Evolution and Time Travel

Plain Parquet files in a data lake have no transactions: a failed job can leave half-written files, two jobs can overwrite each other, and readers can see partial results. Delta Lake adds a transaction log next to the Parquet files so that a folder of files behaves like a table with commits, versions and a schema. This lesson shows the log itself, the guarantees it gives, time travel, and how schemas are enforced and evolved, with examples you can run on a laptop.

Setting up a local Delta session

Install pyspark and delta-spark with pip (matching versions: delta-spark 4.4 targets Spark 4.2). configure_spark_with_delta_pip adds the Delta JARs to the session; the two config lines register Delta’s SQL extensions and catalog so MERGE, RESTORE, DESCRIBE HISTORY and delta. path tables work in SQL.

import json, os, re, tempfile
from pyspark.sql import SparkSession
from delta import configure_spark_with_delta_pip
from delta.tables import DeltaTable

builder = (
    SparkSession.builder.master("local[2]").appName("delta-transactions")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .config("spark.sql.shuffle.partitions", "2")
)
spark = configure_spark_with_delta_pip(builder).getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

base = tempfile.mkdtemp()          # every table in this lesson lives under a temp dir
orders = f"{base}/orders"

def error_class(e):
    """Return the [ERROR_CLASS] tag from a Spark or Delta exception message."""
    found = re.search(r"\[[A-Z_]+\]", str(e))
    return found.group(0) if found else type(e).__name__

Delta Lake architecture

A Delta table has three parts:

1. Data files: ordinary Parquet files. Delta never edits a Parquet file in place; a change writes new files and marks old ones as removed.
2. The transaction log (_delta_log/): an ordered series of commit files, 00000000000000000000.json, 00000000000000000001.json and so on. Each commit records actions: files added, files removed, schema and table properties (metaData), the reader and writer features required (protocol), and audit information (commitInfo).
3. Checkpoints: every so often (by default every 10 commits, set by delta.checkpointInterval) the writer saves the whole table state as a Parquet checkpoint, so a reader loads the latest checkpoint and replays only the JSON commits after it.

The table you query is the result of replaying the log: start from the checkpoint, apply each later commit’s add and remove actions, and the surviving add entries are the files that make up the current version. Nothing else, not a directory listing and not a metastore, decides which files are in the table.

Delta is a table format and protocol, not a server. Any engine that implements the protocol (Spark, Trino, Flink, DuckDB, Polars and others through Delta Kernel or delta-rs) can read the same table. Not every engine supports every feature, which is why the log records the protocol version and table features a reader needs.

The transaction log (_delta_log)

Write two small commits and look at what lands on disk:

spark.createDataFrame(
    [(1, "Ana", 10.0), (2, "Ben", 20.0)], "order_id INT, customer STRING, amount DOUBLE"
).write.format("delta").save(orders)                          # version 0

spark.createDataFrame(
    [(3, "Cy", 30.0)], "order_id INT, customer STRING, amount DOUBLE"
).write.format("delta").mode("append").save(orders)           # version 1

log = f"{orders}/_delta_log"
print(sorted(f for f in os.listdir(log) if f.endswith(".json")))
for line in open(f"{log}/00000000000000000000.json"):
    action, body = next(iter(json.loads(line).items()))
    print(action, body.get("operation") or body.get("path") or body.get("schemaString", "")[:40] or body)

Output (file names shortened):

['00000000000000000000.json', '00000000000000000001.json']
commitInfo WRITE
metaData {"type":"struct","fields":[{"name":"orde
protocol {'minReaderVersion': 1, 'minWriterVersion': 2}
add part-00000-810677d5-...-c000.snappy.parquet
add part-00001-99e701db-...-c000.snappy.parquet

Version 0 carries the schema (metaData.schemaString), the protocol and one add per data file. Each add also holds file statistics: row count and per-column minimum, maximum and null count for the first 32 columns by default. Readers use those statistics to skip files that cannot match a filter, without opening them. Version 1 holds only commitInfo and one add, because an append changes nothing else.

DESCRIBE HISTORY reads the same log as a table:

(spark.sql(f"DESCRIBE HISTORY delta.`{orders}`")
      .select("version", "operation", "operationParameters.mode").show())

+-------+---------+-------------+
|version|operation|         mode|
+-------+---------+-------------+
|      1|    WRITE|       Append|
|      0|    WRITE|ErrorIfExists|
+-------+---------+-------------+

Pitfalls

• Never edit or delete files under _delta_log or data files by hand, and never let a tool write plain Parquet files into the table directory. Files the log does not reference are invisible to readers and are later deleted by VACUUM.
• Statistics are collected only for the first 32 columns (delta.dataSkippingNumIndexedCols). If you filter on a column further right, move it earlier or set delta.dataSkippingStatsColumns.

ACID transactions on Delta

• Atomicity. A writer first writes its Parquet files, which are invisible because no commit mentions them, then creates the next commit file. Creating N.json must be atomic and must fail if the file already exists (“put-if-absent”). A job that dies before committing leaves only orphan files that no reader sees.
• Consistency. Every commit is checked against the table’s schema, NOT NULL columns and CHECK constraints before it is written.
• Isolation. A reader picks one version and reads exactly the files of that version, so it never sees a half-applied write (snapshot isolation). Writers use optimistic concurrency control, described below.
• Durability. Committed files live in object storage or HDFS, so they survive the cluster.

Constraints are part of consistency and are worth using on curated tables:

spark.sql(f"ALTER TABLE delta.`{orders}` ADD CONSTRAINT amount_non_negative CHECK (amount >= 0)")
try:
    spark.createDataFrame([(4, "Di", -5.0)], "order_id INT, customer STRING, amount DOUBLE") \
         .write.format("delta").mode("append").save(orders)
except Exception as e:
    print(error_class(e))
print(spark.read.format("delta").load(orders).count())

[DELTA_VIOLATE_CONSTRAINT_WITH_VALUES]
3

The whole append is rejected and the table still has three rows.

How optimistic concurrency works

1. The writer records the version it read, say version 7.
2. It does its work and writes new data files.
3. It tries to create 00000000000000000008.json.
4. If another writer created version 8 first, the writer reads the winning commits and checks whether they conflict with what it read and changed. If they do not, for example two blind appends, it retries as version 9 automatically. If they do, the commit fails with a subclass of DeltaConcurrentModificationException.

Exception	Typical cause
ConcurrentAppendException	Another writer added files in the data this transaction read, for example two MERGE jobs on the same partition
ConcurrentDeleteReadException	Another writer removed a file this transaction read
ConcurrentDeleteDeleteException	Both transactions removed the same file, for example two DELETEs or a DELETE racing OPTIMIZE
MetadataChangedException	Another writer changed the schema or table properties
ProtocolChangedException	Another writer upgraded the protocol or created the table at the same time

Appends do not conflict with each other. UPDATE, DELETE, MERGE and OPTIMIZE can. To reduce conflicts, make concurrent jobs touch disjoint data and put the separating column in the condition (WHERE region = 'EU', or t.event_date = s.event_date in a MERGE) so Delta can prove the files are disjoint. Retry a failed job only if it is idempotent.

In interviews

“How does Delta give ACID on S3?” A strong answer names the log, atomic commit-file creation, snapshot reads from a version, and optimistic concurrency with conflict detection. Add that ACID covers one table: Delta has no multi-table transactions, so a pipeline that writes two tables must be designed to be safely re-run.

Time travel and versioning

Old data files stay on disk until VACUUM removes them, so earlier versions are still readable:

v0 = spark.read.format("delta").option("versionAsOf", 0).load(orders)
print(v0.count(), spark.read.format("delta").load(orders).count())

# SQL forms; timestampAsOf / TIMESTAMP AS OF take a date or timestamp
spark.sql(f"SELECT order_id, amount FROM delta.`{orders}` VERSION AS OF 1 ORDER BY order_id").show()

2 3
+--------+------+
|order_id|amount|
+--------+------+
|       1|  10.0|
|       2|  20.0|
|       3|  30.0|
+--------+------+

RESTORE rolls the table back by writing a new commit whose files equal the old version’s. History is kept, so a restore can itself be undone:

spark.sql(f"DELETE FROM delta.`{orders}` WHERE order_id = 2")      # the "bad" change
spark.sql(f"RESTORE TABLE delta.`{orders}` TO VERSION AS OF 2") \
     .select("num_removed_files", "num_restored_files").show()
print(sorted(r.order_id for r in spark.read.format("delta").load(orders).collect()))
(spark.sql(f"DESCRIBE HISTORY delta.`{orders}`").select("version", "operation")
      .orderBy("version").show())

+-----------------+------------------+
|num_removed_files|num_restored_files|
+-----------------+------------------+
|                0|                 1|
+-----------------+------------------+

[1, 2, 3]
+-------+--------------+
|version|     operation|
+-------+--------------+
|      0|         WRITE|
|      1|         WRITE|
|      2|ADD CONSTRAINT|
|      3|        DELETE|
|      4|       RESTORE|
+-------+--------------+

Order 2 sat alone in one small file, so the DELETE simply removed that file from the table, and the restore re-added it (one restored file, nothing to remove). Version 2 is the target because it is the last good state that still includes the constraint added above; restoring to version 1 would have dropped it.

Where time travel stops

Two retention settings decide how far back you can go:

Property	Default	Controls
delta.logRetentionDuration	30 days	How long commit history is kept; older entries are cleaned up when checkpoints are written
delta.deletedFileRetentionDuration	7 days	How old a removed file must be before VACUUM may delete it

A version is readable only while both its log entries and all of its data files exist. After VACUUM deletes files older than seven days, any version that needed them fails to read (RESTORE to it fails too), even though DESCRIBE HISTORY still lists it. Time travel is therefore an operational safety net for days, not an archive: for “what did this look like at month end” for years, write snapshots or keep history in the data (a Type 2 dimension, an append-only fact).

Use cases

• Debugging and audit: compare today with yesterday (EXCEPT between two versions) to see what a job changed.
• Rollback: RESTORE after a bad load, then fix and re-run the job.
• Reproducibility: pin a training set or a report to a version number.

RESTORE counts as a data change, so a downstream streaming reader of the table sees the restored files as new data and can produce duplicates. Plan downstream handling before you restore a streamed table.

Schema enforcement

On write, Delta compares the incoming DataFrame with the table schema. Writes are rejected when the data has a column the table does not, or a type that cannot be safely cast. Missing nullable columns are allowed and become NULL.

extra_col = spark.createDataFrame([(5, "Ed", 50.0, "SPRING")],
                                  "order_id INT, customer STRING, amount DOUBLE, coupon STRING")
try:
    extra_col.write.format("delta").mode("append").save(orders)
except Exception as e:
    print(error_class(e))

bad_type = spark.createDataFrame([("A6", "Flo", 60.0)], "order_id STRING, customer STRING, amount DOUBLE")
try:
    bad_type.write.format("delta").mode("append").save(orders)
except Exception as e:
    print(error_class(e))

[DELTA_METADATA_MISMATCH]
[DELTA_FAILED_TO_MERGE_FIELDS]

Both writes fail before any commit (the second because order_id arrives as a string), so the table is unchanged. Compare this with a plain Parquet directory, where both files would be written and the breakage would surface in a reader days later.

Schema evolution

When a change is intended, opt in for that write or change the table explicitly.

Add columns on write with mergeSchema

extra_col.write.format("delta").mode("append").option("mergeSchema", "true").save(orders)
spark.read.format("delta").load(orders).orderBy("order_id").show()

+--------+--------+------+------+
|order_id|customer|amount|coupon|
+--------+--------+------+------+
|       1|     Ana|  10.0|  NULL|
|       2|     Ben|  20.0|  NULL|
|       3|      Cy|  30.0|  NULL|
|       5|      Ed|  50.0|SPRING|
+--------+--------+------+------+

Older rows read the new column as NULL; no file is rewritten. The ways to evolve, and what each is for:

Mechanism	What it changes
.option("mergeSchema", "true") on a write	Adds new columns (and nested fields) from the incoming data
MERGE WITH SCHEMA EVOLUTION or DeltaTable.merge(...).withSchemaEvolution()	Same, for MERGE
.option("overwriteSchema", "true") with mode("overwrite")	Replaces the schema and data entirely
ALTER TABLE ... ADD COLUMNS	Adds columns explicitly, reviewed in code
Column mapping + RENAME COLUMN / DROP COLUMN	Renames or drops without rewriting files
Type widening + ALTER COLUMN ... TYPE	Widens a type, for example INT to BIGINT, without rewriting files

Renames and drops need column mapping

Parquet files identify columns by name, so a plain Delta table cannot rename a column without rewriting every file:

try:
    spark.sql(f"ALTER TABLE delta.`{orders}` RENAME COLUMN coupon TO coupon_code")
except Exception as e:
    print(error_class(e))

spark.sql(f"ALTER TABLE delta.`{orders}` SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name')")
spark.sql(f"ALTER TABLE delta.`{orders}` RENAME COLUMN coupon TO coupon_code")
spark.sql(f"ALTER TABLE delta.`{orders}` DROP COLUMN customer")
print(spark.read.format("delta").load(orders).columns)
spark.sql(f"DESCRIBE DETAIL delta.`{orders}`").select("minReaderVersion", "minWriterVersion").show()

[DELTA_UNSUPPORTED_RENAME_COLUMN]
['order_id', 'amount', 'coupon_code']
+----------------+----------------+
|minReaderVersion|minWriterVersion|
+----------------+----------------+
|               2|               7|
+----------------+----------------+

With column mapping, each column gets a stable physical name and ID stored in the schema, and the logical name is just metadata, so a rename or drop is a log-only change. The cost is a protocol upgrade: readers must support column mapping (reader version 2), and engines or connectors that do not will refuse the table. The upgrade is not reversible in place. Column mapping also limits change data feed reads across renames and drops (see the MERGE and change data feed lesson). A dropped column’s data stays in the files until they are rewritten, which matters if you dropped it to remove personal data.

Type widening

Delta 4.0 made type widening generally available. After enabling the delta.enableTypeWidening property, ALTER COLUMN ... TYPE can widen supported types (for example BYTE/SHORT/INT to BIGINT, FLOAT to DOUBLE, a decimal to higher precision) without rewriting files:

spark.sql(f"ALTER TABLE delta.`{orders}` SET TBLPROPERTIES ('delta.enableTypeWidening' = 'true')")
spark.sql(f"ALTER TABLE delta.`{orders}` ALTER COLUMN order_id TYPE BIGINT")
print(spark.read.format("delta").load(orders).schema.simpleString())

struct<order_id:bigint,amount:double,coupon_code:string>

Narrowing a type, or changing between unrelated types (string to int), is not schema evolution: rewrite the table with overwriteSchema, or use an expand-and-contract migration, as covered in schema evolution patterns.

Pitfalls

1. Setting mergeSchema on every job, or the session-wide spark.databricks.delta.schema.autoMerge.enabled, so an upstream typo (custmer_id) silently becomes a new column. Keep automatic merge for raw landing tables.
2. Using overwriteSchema to “fix” a mismatch and wiping data you meant to keep.
3. Enabling column mapping, type widening or deletion vectors without checking that every reader (BI connector, older Spark, another engine) supports the new protocol.

Practice questions

Two jobs append to the same Delta table at the same time. What happens?

Both succeed. Each writes its files and tries to create the next commit. The second to arrive finds the version taken, checks the winning commit, sees that blind appends cannot conflict, and commits as the following version. Conflicts arise when a transaction read data that another changed, such as two MERGE or DELETE operations touching the same files.

A job died halfway through writing. Do readers see partial data?

No. The data files it wrote are not referenced by any commit, so no reader includes them. They are orphans that VACUUM removes once they are older than the retention period. Re-running the job is safe if the job itself is idempotent.

Why can a version listed in DESCRIBE HISTORY fail to load?

History lives in the log (kept for 30 days by default), but the version’s data files may have been deleted by VACUUM (which may delete removed files older than 7 days by default). Time travel needs both the log entries and the files.

What is the difference between schema enforcement and schema evolution?

Enforcement is the default check that rejects writes whose columns or types do not match the table. Evolution is an explicit opt-in (mergeSchema, ALTER TABLE, column mapping, type widening) that changes the table schema. Enforcement protects curated tables from accidental drift; evolution is how you make intended changes.

Why does renaming a column require column mapping?

Without it, Parquet files are read by column name, so renaming the column in the schema would orphan the existing data in those files. Column mapping stores a stable physical name per column, so the logical name can change in the log alone. It requires a protocol upgrade that older readers cannot handle.

How would you undo a bad overwrite from this morning?

Find the last good version with DESCRIBE HISTORY, check it with a VERSION AS OF query, then RESTORE TABLE ... TO VERSION AS OF n. It must be inside the retention window. Warn owners of streaming consumers, because the restore appears as a data change to them.

Key takeaways

• The _delta_log is the table: replaying commits from the latest checkpoint gives the current set of files.
• Commits are atomic file creations; readers use one version, and writers use optimistic concurrency with conflict detection.
• ACID is per table. Pipelines spanning tables still need idempotent re-runs.
• Time travel and RESTORE work only while both the log entries and the data files are retained.
• Schema enforcement is on by default. Evolve deliberately: mergeSchema for additions, column mapping for renames and drops, type widening for safe type changes.