Transaction Log

The _delta_log/ Directory

s3://warehouse/nyc-taxi/
├── _delta_log/
│   ├── 00000000000000000000.json   ← initial commit: table creation
│   ├── 00000000000000000001.json   ← version 1: first data write
│   ├── 00000000000000000002.json   ← version 2: second batch
│   └── ...
│   └── 00000000000000000100.json
│   └── 00000000000000000100.checkpoint.parquet  ← checkpoint at v100
│   └── 00000000000000000101.json
│   └── 00000000000000103.checkpoint.17.parquet  ← older checkpoints cleaned up
├── part-00000-...parquet            ← actual data files (in partition dirs)
├── part-00001-...parquet
└── ...

Commits are zero-padded 20-digit JSON files: 00000000000000000001.json. The padding ensures lexicographic ordering matches numerical ordering — S3 list operations return files in sorted order, which matters for consistent log reads. Checkpoint names embed the version they cover: 00000000000000000100.checkpoint.parquet covers through version 100.

JSON Commit File Format

Each commit file is a JSON array of actions. The action types:

// 00000000000000000003.json — version 3, two files added
[
  {
    "add": {
      "path": "part-00000-abc123.snappy.parquet",
      "size": 134217728,
      "partitionValues": {},
      "modificationTime": 1714567890000,
      "dataChange": true,
      "stats": "{\"numRecords\":500000,\"minTrip_distance\":0.0,\"maxTrip_distance\":52.5}",
      "stats_parsed": {
        "numRecords": 500000,
        "minTrip_distance": 0.0,
        "maxTrip_distance": 52.5
      },
      "nullCount": {"trip_distance": 0},
      "distinctCount": {}
    }
  },
  {
    "add": {
      "path": "part-00001-def456.snappy.parquet",
      "size": 146800640,
      "partitionValues": {},
      "modificationTime": 1714567891000,
      "dataChange": true,
      "stats": "{\"numRecords\":500123}",
      "stats_parsed": {"numRecords": 500123},
      "nullCount": {},
      "distinctCount": {}
    }
  }
]

// A file was removed (tombstoned)
[
  {
    "remove": {
      "path": "part-00000-old.gz.parquet",
      "deletionTimestamp": 1714567900000,
      "dataChange": true,
      "extendedFileStatus": "{\"isAdaptive\":false,\"partitionValues\":{}}"
    }
  }
]

// Metadata changed (schema, partition layout, table properties)
[
  {
    "meta": {
      "createdTime": 1714000000000,
      "partitionProvider": "org.apache.spark.sql.sources.Filters$$Type$,
      "schema": {
        "type": "struct",
        "fields": [
          {"name": "vendor_id", "type": "string", "nullable": true, "metadata": {}},
          {"name": "trip_distance", "type": "double", "nullable": true, "metadata": {}}
        ]
      }
    }
  }
]

// Protocol change (reader/writer feature flags)
[
  {
    "protocol": {
      "minReaderVersion": 3,
      "minWriterVersion": 7
    }
  }
]

// Transaction identifiers (for idempotent writes, e.g., from Spark Structured Streaming)
[
  {
    "txn": {
      "appId": "structured-streaming-job-001",
      "version": 42,
      "lastUpdated": 1714567910000
    }
  }
]

The add action is the most common. It records the physical file path, size, modification timestamp, and per-column statistics (null counts, distinct counts, min/max values). These statistics are what make data skipping effective — the query engine can skip entire files without reading them if the predicate doesn't match the file's min/max range. dataChange: false marks files that are added by operations like OPTIMIZE or Z-ORDER that don't change the data, preventing them from being reprocessed by streaming reads.

Checkpoint Files (Parquet)

As the table grows, replaying from version 0 becomes expensive. A checkpoint is a snapshot of the table's state — all active files, remove tombstones, and metadata — written as a Parquet file. Readers can start from the latest checkpoint and only replay the JSON commits after it.

# When are checkpoints written?
# Default: every 10 commits (configurable via delta.checkpointInterval)
# Also written when requested by user: CALL SYSTEM.TIMESTravel(...)

# Checkpoint structure (Parquet, action schema)
$ spark.read.parquet("/warehouse/nyc-taxi/_delta_log/_checkpoint")
  .printSchema
root
 ├── commitInfo: struct (version, timestamp, userData, operation)
 ├── protocol: struct (minReaderVersion, minWriterVersion)
 ├── meta: struct (createdTime, partitionProvider, schema)
 ├── add: array (active data files with stats)
 ├── remove: array (tombstoned files with deletionTimestamp)
 ├── txn: array (transaction identifiers)
 └── cdc: array (change-data-capture actions)

# Checkpoint interval config
spark.sql("SET delta.checkpointInterval = 10")

# Manual checkpoint via SparkSession
spark.sql("ALTER TABLE nyc_taxi SET TBLPROPERTIES ('delta.checkpointInterval' = 10)")

Checkpoints include all active add actions — every data file currently in the table — and all remove actions within the retention window. The Parquet format means the reader can skip columns it doesn't need. Checkpoints are written by the same transaction semantics as JSON commits — atomically once fully materialized.

Optimistic Concurrency

Delta Lake uses optimistic concurrency control (OCC). Writers don't acquire locks; they read the current state, make changes, and commit. If the commit version doesn't match what the writer expected, the commit is rejected and retried.

# Writer A reads current state: latest version = 5
# Writer B reads current state: latest version = 5
# Writer A writes its changes, commits as version 6
# Writer B writes its changes, attempts commit as version 6 → CONFLICT

# What happens on conflict?
# Spark: org.apache.spark.sql.delta.ConcurrentWriteException
# The job retries from the new latest version.

# The commit attempt is atomic on S3:
# - S3 multipart upload: atomic only on complete upload
# - Delta Lake uses preconditions: if the expected version file already
#   exists, S3 returns 409 and the commit fails

# This is why version numbers are zero-padded: prevents race between
# 6.json and 10.json (lexicographic vs numeric ordering)
# 6.json < 10.json is wrong numerically, so all files are zero-padded to 20 digits

# The preconditon check:
# When committing version N, Delta Lake checks that _delta_log/N.json does NOT
# exist before writing it. S3's "x-amz-nothing-exists" precondition makes this
# an atomic check-and-set.

# Conflict detection example
val transactionVersion = deltaLog.update().version  // e.g., 5
val actions = Seq(AddFile(...), RemoveFile(...))
val commitFile = new Path(deltaLog.logPath, f"$transactionVersion%020d.json")

// Atomic commit: write only if precondition passes
val canWrite = s3.checkNotExists(commitFile)  // S3 precondition
if (canWrite) {
  writeCommitFile(commitFile, actions)        // all-or-nothing
} else {
  throw new ConcurrentWriteException()        // someone else got there first
}

This is a huge advantage over悲观 locking: there's no coordinator, no lease management, no single point of failure. The filesystem itself handles the arbitration through atomic rename or preconditions. The tradeoff is that under high contention (many writers trying to commit the same version simultaneously), many commits will fail and retry. Delta Lake's default configuration assumes contention is low; for streaming workloads with micro-batches, the interval between commits gives plenty of headroom.

Time Travel

Because every commit is immutable and versioned, reading an old snapshot is just reading the log up to a specific version or timestamp. Delta Lake exposes this via VERSION AS OF and TIMESTAMP AS OF.

# Read version 42
spark.read.format("delta").option("versionAsOf", "42").load("/warehouse/nyc-taxi")

# Read as of May 1, 2026 00:00 UTC
spark.read.format("delta")
  .option("timestampAsOf", "2026-05-01T00:00:00.000Z")
  .load("/warehouse/nyc-taxi")

# SQL syntax
SELECT * FROM delta.`/warehouse/nyc-taxi` VERSION AS OF 42
SELECT * FROM delta.`/warehouse/nyc-taxi` TIMESTAMP AS OF '2026-05-01'

# How it works under the hood
# 1. VERSION AS OF N:
#    - Find the latest checkpoint at or before version N
#    - Load checkpoint state
#    - Replay JSON commits from checkpoint+1 through version N
# 2. TIMESTAMP AS OF T:
#    - Binary search the JSON commit files for the commit with
#      commitInfo.timestamp <= T (commits record timestamps)
#    - Apply version-as-of logic from that commit

# History retention (default: 30 days for vacuum, unlimited for history)
spark.sql("SET delta.logRetentionDuration = '30 days'")

# View table history
spark.sql("DESCRIBE HISTORY delta.`/warehouse/nyc-taxi`").show()

# Output example:
# +-------+-------------------+------+-----------+-------------------+
# |version|          timestamp|   operation| operationMetrics| userName|
# +-------+-------------------+------+-----------+-------------------+
# |     42|2026-05-01 14:32:10|    UPDATE  |  rowsRemoved=120|  admin|
# |     41|2026-04-30 09:11:02|     INSERT |  numAdded=500000|  etl-job|
# |     40|2026-04-28 22:00:00|  OPTIMIZE  |  filesRemoved=3 |  admin|
# +-------+-------------------+------+-----------+-------------------+

# Restore to an old version (creates a new commit)
spark.sql("RESTORE TABLE nyc_taxi TO VERSION AS OF 38")

# Querying file size changes over time (how much data each version adds)
spark.sql("""
  SELECT version, operation,
         explode(add_files.path) as path,
         add_files.size as size_bytes
  FROM (
    SELECT version, operation, add
    FROM (DESCRIBE HISTORY delta.`/warehouse/nyc-taxi`)
    WHERE add IS NOT NULL
  )
""")

The timestamp binary search works because every JSON commit has a commitInfo block with the timestamp. The search is O(log N) and doesn't require scanning all commits. Time travel is read-only — it doesn't mutate the table's current state. RESTORE is different: it creates a new commit that recomputes the current state from an old version, which is a forward mutation.

The Vacuum Command

Over time, the table accumulates orphan data files from overwritten data, and tombstoned files whose deletion timestamp is past the retention window. VACUUM cleans them up.

# Dry run: see what would be deleted (without deleting)
spark.sql("VACUUM delta.`/warehouse/nyc-taxi` RETAIN 168 HOURS DRY RUN")
# Returns: [path, size_bytes, safeToDelete]

# Actually vacuum (delete files older than retention)
spark.sql("VACUUM delta.`/warehouse/nyc-taxi` RETAIN 168 HOURS")
# Default retention: 7 days (168 hours)

# Why the 7-day minimum?
# Readers running at the time of a commit may still be reading files that
# were tombstoned in that commit. If we vacuum them immediately, the reader
# would get a FileNotFoundException. The retention window gives in-flight
# readers time to finish.

# Disable vacuum (for disaster recovery scenarios where you want to keep all files)
spark.sql("SET delta.enableDeleteByTimeline = false")

# The vacuum algorithm:
# 1. List all files in _delta_log/
# 2. Read the last N commits (where N = retention commits)
# 3. Compute the set of reachable files from those commits
# 4. Files not reachable and older than retention → delete
# 5. Delete the corresponding remove actions from the log (cleanup tombstones)

# Vacuum only cleans data files; it never touches commit JSON files
# (those are immutable and needed for time travel)
# Checkpoint cleanup: old checkpoints are cleaned when a new checkpoint
# covers the same version range

# Deleting from a specific partition doesn't vacuum — it tombstones
spark.sql("DELETE FROM nyc_taxi WHERE pickup_datetime < '2024-01-01'")
# Result: new commit adds a remove action for the affected files
# Files are not deleted until vacuum runs

# Vacuum in the vacuum.yml settings for Databricks:
# vacuum_schedule = "0 0 * * *"  # daily at midnight
# dry_run_before_delete = true
# retention_duration = 168  # hours

Vacuum never deletes a file if there's any uncertainty about whether a reader might need it. The safety margin is the retention window. Note that vacuum operates on data files only — the JSON commit files are immutable and survive vacuum. Checkpoint cleanup is a separate mechanism that removes old checkpoint Parquet files once a newer checkpoint covers the same range.

Comparison: Delta Log vs Apache Iceberg vs Hudi

Delta Lake's JSON commits are human-readable, which is helpful for debugging. Iceberg's metadata is also JSON (table metadata files) but commits are Avro. Hudi's timeline is stored in Avro and includes instant timestamps, state, and actions. All three use the same underlying principle: immutable commit log + checkpoint snapshots for efficient reads.

Frequently Asked Questions