The Complete Production SOP: How to Deploy and Run Delta Lake ACID Transactions & Time Travel Queries in Autonomous AI Agents

---
name: "delta-lake-acid-time-travel"
description: "Build reliable data lakehouses with ACID transactions, schema enforcement, and time travel."
metadata:
  author: "Delta Lake"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "query delta lake time travel"
  - "run delta lake optimize zorder"
  - "merge into delta table"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# Delta Lake ACID Transactions & Time Travel Queries

Manages Delta Lake tables: executing MERGE INTO (upserts), querying historical snapshot versions (AS OF), compacting small files with OPTIMIZE ZORDER BY, and vacuuming stale parquet files safely.

## When to Invoke
Invoke this skill whenever you need to:
- query delta lake time travel
- run delta lake optimize zorder
- merge into delta table

## Execution SOP & Rules
1. Query historical versions: spark.read.format('delta').option('versionAsOf', 5).load('/data/table').
2. Perform upsert: DeltaTable.forPath(spark, '/data/table').alias('t').merge(updates.alias('s'), 't.id = s.id').whenMatchedUpdateAll().whenNotMatchedInsertAll().execute().
3. Compact small files: spark.sql('OPTIMIZE delta.`/data/table` ZORDER BY (customer_id, date)').
4. Prune old commit snapshots: spark.sql('VACUUM delta.`/data/table` RETAIN 168 HOURS').

from pyspark.sql import SparkSession
from delta import configure_spark_with_delta_pip

def get_delta_spark_session(app_name: str = "DeltaLakeAgentEngine") -> SparkSession:
    builder = (
        SparkSession.builder.appName(app_name)
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
        .config("spark.sql.parquet.vorder.enabled", "true")
        .config("spark.databricks.delta.retentionDurationCheck.enabled", "true")
    )
    return configure_spark_with_delta_pip(builder).getOrCreate()

import pytest
from delta.tables import DeltaTable
from lakehouse_session import get_delta_spark_session

def test_delta_acid_and_history():
    spark = get_delta_spark_session("TestSession")
    table_path = "./data/lakehouse/customer_orders"
    
    # Verify table is valid Delta format
    assert DeltaTable.isDeltaTable(spark, table_path)
    
    # Verify transaction history exists
    delta_table = DeltaTable.forPath(spark, table_path)
    history_df = delta_table.history(2)
    latest_op = history_df.select("operation").first()[0]
    assert latest_op in ["MERGE", "OPTIMIZE", "WRITE"]
    
    # Verify time travel query capability
    v0_df = spark.read.format("delta").option("versionAsOf", 0).load(table_path)
    assert v0_df.count() >= 0