Delta Lake Reference

The transaction log, writes and MERGE, time travel and restore, schema evolution, OPTIMIZE and VACUUM, Change Data Feed, streaming, and interoperability.

1. What Delta Lake Is

Delta Lake is an open table format that adds a transaction log on top of Parquet files. The log turns a folder of files into a table with ACID transactions, MERGE/UPDATE/DELETE, schema enforcement, time travel and streaming support. It is the default table format on Databricks and is open source (Linux Foundation) for Spark, Flink, Trino and Python.

s3://lake/silver/transactions/
├── _delta_log/
│   ├── 00000000000000000000.json      ← commit 0: schema, protocol, files added
│   ├── 00000000000000000001.json      ← commit 1: files added / removed
│   ├── …
│   ├── 00000000000000000010.checkpoint.parquet   ← snapshot of the log every 10 commits
│   └── _last_checkpoint
├── event_date=2026-10-25/part-00000-….snappy.parquet
└── event_date=2026-10-26/part-00000-….snappy.parquet

• Every write adds a new numbered JSON commit listing the files it added and removed. Readers replay the log (starting from the latest checkpoint) to find the current set of files.
• Optimistic concurrency: writers prepare files, then try to create the next commit number atomically. If another writer got there first, Delta checks for logical conflicts and either retries or fails.
• Old files are not deleted when data changes; they are only removed from the log. That is what makes time travel possible, and why VACUUM exists.
• The log also stores per-file statistics (min/max/null counts for the first 32 columns by default), used for data skipping.

2. Setup & Writing Data

On Databricks Delta is built in. With open-source Spark, install delta-spark (match its version to your Spark version) and enable the extension:

# pip install delta-spark==3.2.0      (Delta 3.2 ↔ Spark 3.5)
from pyspark.sql import SparkSession
from delta import configure_spark_with_delta_pip

builder = (SparkSession.builder.appName("delta-demo")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog"))
spark = configure_spark_with_delta_pip(builder).getOrCreate()

# write by path or as a catalog table
df.write.format("delta").mode("append").partitionBy("event_date").save("/lake/silver/transactions")
df.write.format("delta").mode("overwrite").saveAsTable("silver.transactions")

# overwrite only one day, atomically (safe, idempotent reruns)
(df.write.format("delta").mode("overwrite")
   .option("replaceWhere", "event_date = '2026-10-25'")
   .save("/lake/silver/transactions"))

spark.read.format("delta").load("/lake/silver/transactions")
spark.read.table("silver.transactions")

-- Databricks SQL
CREATE TABLE silver.transactions (
  txn_id     STRING NOT NULL,
  account_id STRING,
  amount     DECIMAL(18,2),
  event_ts   TIMESTAMP,
  event_date DATE GENERATED ALWAYS AS (CAST(event_ts AS DATE))
) USING DELTA
PARTITIONED BY (event_date);

Open-source Delta with Spark 3.5 rejects GENERATED ALWAYS AS in SQL ("does not support generated columns"). Declare generated columns with the builder API instead:

from delta.tables import DeltaTable

(DeltaTable.createIfNotExists(spark)
   .tableName("silver.transactions")
   .addColumn("txn_id", "STRING", nullable=False)
   .addColumn("account_id", "STRING")
   .addColumn("amount", "DECIMAL(18,2)")
   .addColumn("event_ts", "TIMESTAMP")
   .addColumn("event_date", "DATE", generatedAlwaysAs="CAST(event_ts AS DATE)")
   .partitionedBy("event_date")
   .execute())

3. MERGE: Upserts & Deletes

from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "silver.customers")

(target.alias("t")
   .merge(updates.alias("s"), "t.customer_id = s.customer_id")
   .whenMatchedUpdateAll(condition="s.updated_at > t.updated_at")   # only newer data wins
   .whenNotMatchedInsertAll()
   .whenNotMatchedBySourceDelete()      # optional: delete rows missing from a full snapshot (Delta 2.3+)
   .execute())

MERGE INTO silver.customers AS t
USING updates AS s
ON t.customer_id = s.customer_id
WHEN MATCHED AND s.op = 'D'               THEN DELETE
WHEN MATCHED AND s.updated_at > t.updated_at THEN UPDATE SET *
WHEN NOT MATCHED AND s.op <> 'D'          THEN INSERT *;

• Deduplicate the source first (one row per key, e.g. latest by updated_at). If two source rows match the same target row, MERGE fails with "multiple source rows matched".
• Add partition or clustering columns to the ON condition (e.g. AND t.event_date = s.event_date) so MERGE scans fewer files.
• For SCD Type 2 with MERGE, see the staged-union pattern in the Data Warehouse reference.

4. Time Travel, History & Restore

DESCRIBE HISTORY silver.transactions;                 -- version, timestamp, operation, user, metrics

SELECT * FROM silver.transactions VERSION AS OF 42;
SELECT * FROM silver.transactions TIMESTAMP AS OF '2026-10-25 06:00:00';

RESTORE TABLE silver.transactions TO VERSION AS OF 42;   -- undo a bad load (itself a new commit)

spark.read.format("delta").option("versionAsOf", 42).load(path)
spark.read.format("delta").option("timestampAsOf", "2026-10-25").load(path)
DeltaTable.forName(spark, "silver.transactions").restoreToVersion(42)

Time travel needs both the log entry and the data files. After VACUUM removes old files, versions that referenced them can no longer be read.

5. Schema Enforcement & Evolution

Delta rejects writes whose schema does not match the table (schema enforcement), so a bad upstream change fails loudly instead of corrupting the table. Allow changes explicitly:

-- column mapping needs a protocol upgrade (reader v2, writer v5) in the same statement
ALTER TABLE silver.customers SET TBLPROPERTIES (
  'delta.minReaderVersion' = '2',
  'delta.minWriterVersion' = '5',
  'delta.columnMapping.mode' = 'name');

ALTER TABLE silver.customers RENAME COLUMN name TO full_name;
ALTER TABLE silver.customers DROP COLUMN legacy_flag;

6. OPTIMIZE, Z-ORDER & VACUUM

OPTIMIZE silver.transactions;                                  -- compact small files (bin-packing)
OPTIMIZE silver.transactions WHERE event_date >= '2026-10-01';  -- only recent partitions
OPTIMIZE silver.transactions ZORDER BY (account_id);            -- co-locate rows for data skipping

VACUUM silver.transactions;                    -- delete unreferenced files older than 7 days
VACUUM silver.transactions RETAIN 168 HOURS;
VACUUM silver.transactions DRY RUN;            -- list what would be deleted

• Streaming and frequent small batches create many small files; compact them regularly. Databricks can do this automatically (optimized writes, auto compaction, predictive optimization).
• For new tables, liquid clustering (CLUSTER BY) replaces partitioning + Z-ORDER; see the Spark & Databricks reference.
• Do not lower VACUUM retention below the longest-running reader or stream; it can delete files they still need. A safety check blocks retention under 7 days unless you disable it.

7. Change Data Feed

Change Data Feed (CDF) records row-level changes so downstream jobs can process only what changed, instead of re-reading the table.

ALTER TABLE silver.customers SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- SQL: changes since version 5
SELECT * FROM table_changes('silver.customers', 5);

changes = (spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", 5)
    .table("silver.customers"))
# extra columns: _change_type (insert | update_preimage | update_postimage | delete),
#                _commit_version, _commit_timestamp

Typical uses: propagating silver changes to gold incrementally, feeding CDC to downstream systems, and auditing. Changes are only captured from the version where CDF was enabled.

8. Delta with Structured Streaming

# Delta as a streaming source and sink (bronze → silver)
(spark.readStream.table("bronze.transactions")
    .filter("amount IS NOT NULL")
    .writeStream
    .option("checkpointLocation", "/chk/silver_transactions")
    .trigger(availableNow=True)                # process all new data, then stop
    .toTable("silver.transactions"))

• A Delta streaming source reads new commits in order; maxFilesPerTrigger / maxBytesPerTrigger limit batch size.
• By default a stream fails if the source table has updates or deletes; set skipChangeCommits to ignore them, or read the Change Data Feed instead.
• Idempotent upserts from a stream: use foreachBatch with a MERGE inside.

9. Constraints, Generated Columns & Deletion Vectors

ALTER TABLE silver.transactions ALTER COLUMN txn_id SET NOT NULL;
ALTER TABLE silver.transactions ADD CONSTRAINT positive_amount CHECK (amount > 0);
-- writes that violate a constraint fail the whole transaction

ALTER TABLE silver.transactions SET TBLPROPERTIES ('delta.enableDeletionVectors' = true);

• Generated columns (GENERATED ALWAYS AS) derive values such as a date from a timestamp, and let filters on the source column prune partitions.
• Deletion vectors mark deleted rows in a small side file instead of rewriting the whole Parquet file (merge-on-read). DELETE, UPDATE and MERGE become much faster; files are physically rewritten later by OPTIMIZE or REORG.
• Identity columns (GENERATED ALWAYS AS IDENTITY) give surrogate keys on Databricks and recent Delta versions.
• Enabling some features upgrades the table protocol, so older readers and writers may no longer open the table. Check engine versions before enabling.

10. Interoperability: UniForm, delta-rs & Sharing

• UniForm writes Iceberg metadata alongside the Delta log, so Iceberg engines (Athena, Snowflake, Trino) can read the same files without copying:

CREATE TABLE gold.daily_sales (...) TBLPROPERTIES (
  'delta.columnMapping.mode'             = 'name',
  'delta.enableIcebergCompatV2'          = 'true',
  'delta.universalFormat.enabledFormats' = 'iceberg'
);

UniForm is built into Databricks. On open-source Spark, also add the io.delta:delta-iceberg_2.12 package (same version as delta-spark); without it the write fails with DELTA_MISSING_ICEBERG_CLASS.

# delta-rs: read and write Delta from plain Python, no Spark or JVM
# pip install deltalake
from deltalake import DeltaTable, write_deltalake

dt = DeltaTable("s3://lake/silver/transactions")
print(dt.version(), dt.schema())
pdf = dt.to_pandas(columns=["txn_id", "amount"])
write_deltalake("./local_table", pdf, mode="append")

Delta Sharing is an open protocol for sharing live Delta tables with other organizations or tools (pandas, Spark, Power BI) without copying data.

11. Common Interview Questions