intermediateView Python source

A customer changed. Update the table.

Bo upgrades to gold. Cy changes their name to Cyrus. Ed closes their account, and Fay joins. Apply all four changes with one SQL statement, then read the customer table as it was before.

Run it

You need uv. Run all commands below from the repository root (sail-examples/):

uv run pipelines/delta-merge-upsert/main.py

The first run installs the pinned packages. The customer records are included in the script, so there is no dataset to download, account to create or API key to supply.

The script prints the updated table, followed by version 0. In the updated table, Bo is gold, Cyrus has ID 3, and Fay has ID 6. Ed remains visible only in version 0. Temporary table files are removed when the process exits.

Output

The newest capture appears first. Older output stays here when a result changes.

Current output · Sail 0.7.3

Captured on Sail 0.7.2, 0.7.3. Latest capture: 2026-11-05, Python 3.12.14.

After MERGE:
+---+-----+------+
| id| name|  tier|
+---+-----+------+
|  1|  Ana|  gold|
|  2|   Bo|  gold|
|  3|Cyrus|bronze|
|  4|   Di|silver|
|  6|  Fay|silver|
+---+-----+------+

Version 0 (before MERGE):
+---+----+------+
| id|name|  tier|
+---+----+------+
|  1| Ana|  gold|
|  2|  Bo|silver|
|  3|  Cy|bronze|
|  4|  Di|silver|
|  5|  Ed|bronze|
+---+----+------+

Python code

main.py85 lines

View main.py on GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "pysail==0.7.3",
#   "pyspark-client==4.1.3",
#   "pyarrow==26.0.0",
#   "pandas==3.0.6",
# ]
# ///
"""Upsert changes into a Delta Lake table with MERGE INTO, then read the version before it.

Sail reads and writes Delta Lake natively, so no Delta package or JVM jars are needed. One
MERGE statement updates changed rows, inserts new ones and deletes rows flagged for removal.
"""

import atexit
import shutil
import tempfile

from pyspark.sql import DataFrame, SparkSession

CUSTOMERS = [
    (1, "Ana", "gold"),
    (2, "Bo", "silver"),
    (3, "Cy", "bronze"),
    (4, "Di", "silver"),
    (5, "Ed", "bronze"),
]
CHANGES = [
    (2, "Bo", "gold", False),  # tier upgrade
    (3, "Cyrus", "bronze", False),  # name change
    (5, "Ed", "bronze", True),  # account closed
    (6, "Fay", "silver", False),  # new customer
]


def temp_dir(prefix: str) -> str:
    """A temporary directory that is removed when the process exits."""
    path = tempfile.mkdtemp(prefix=prefix)
    atexit.register(shutil.rmtree, path, ignore_errors=True)
    return path


def run(spark: SparkSession) -> dict[str, DataFrame]:
    path = temp_dir("customers-")
    spark.createDataFrame(CUSTOMERS, "id long, name string, tier string").write.format("delta").mode(
        "overwrite"
    ).save(path)
    spark.sql("DROP TABLE IF EXISTS customers")  # safe to run again in the same session
    spark.sql(f"CREATE TABLE customers USING delta LOCATION '{path}'")
    spark.createDataFrame(
        CHANGES, "id long, name string, tier string, deleted boolean"
    ).createOrReplaceTempView("changes")

    spark.sql(
        """
        MERGE INTO customers AS c
        USING changes AS u
        ON c.id = u.id
        WHEN MATCHED AND u.deleted THEN DELETE
        WHEN MATCHED THEN UPDATE SET c.name = u.name, c.tier = u.tier
        WHEN NOT MATCHED AND NOT u.deleted THEN INSERT (id, name, tier) VALUES (u.id, u.name, u.tier)
        """
    )

    current = spark.read.format("delta").load(path).orderBy("id")
    before = spark.read.format("delta").option("versionAsOf", 0).load(path).orderBy("id")
    by_tier = spark.sql("SELECT tier, count(*) AS customers FROM customers GROUP BY tier ORDER BY tier")
    return {"current": current, "version_0": before, "by_tier": by_tier}


if __name__ == "__main__":
    from pysail.spark import SparkConnectServer

    server = SparkConnectServer()
    server.start()
    _, port = server.listening_address
    spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
    results = run(spark)
    print("After MERGE:")
    results["current"].show()
    print("Version 0 (before MERGE):")
    results["version_0"].show()
    spark.stop()
    server.stop()

The query to keep

MERGE INTO customers AS c
USING changes AS u
ON c.id = u.id
WHEN MATCHED AND u.deleted THEN DELETE
WHEN MATCHED THEN UPDATE SET c.name = u.name, c.tier = u.tier
WHEN NOT MATCHED AND NOT u.deleted
  THEN INSERT (id, name, tier) VALUES (u.id, u.name, u.tier)

The ID decides which customer to change. The first matching condition decides whether to delete, update or insert the row.

Sail executes the SQL and reads and writes Delta Lake directly. Delta Lake keeps the table's versions, which is why the script can read the earlier rows. No Java or separate Delta package is needed for this Sail run.

Try one change: downgrade Ana

pipelines/delta-merge-upsert/main.py:

Linemain.py · original lines 29 to 3429 CHANGES = [
+    (1, "Ana", "silver", False),
30     (2, "Bo", "gold", False),  # tier upgrade
31     (3, "Cyrus", "bronze", False),  # name change
32     (5, "Ed", "bronze", True),  # account closed
33     (6, "Fay", "silver", False),  # new customer
34 ]

From the repository root (sail-examples/), run the edited script:

uv run pipelines/delta-merge-upsert/main.py

Output after this change

Current output · Sail 0.7.3

Captured on Sail 0.7.3. Latest capture: 2026-11-06, Python 3.14.10.

After MERGE:
+---+-----+------+
| id| name|  tier|
+---+-----+------+
|  1|  Ana|silver|
|  2|   Bo|  gold|
|  3|Cyrus|bronze|
|  4|   Di|silver|
|  6|  Fay|silver|
+---+-----+------+

Version 0 (before MERGE):
+---+----+------+
| id|name|  tier|
+---+----+------+
|  1| Ana|  gold|
|  2|  Bo|silver|
|  3|  Cy|bronze|
|  4|  Di|silver|
|  5|  Ed|bronze|
+---+----+------+

How it works

1. Create a table. Write the initial customers in Delta format and register its path as customers.
2. Apply the changes. MERGE matches on ID, removes deleted customers, updates existing customers and adds new ones.
3. Read the past. option("versionAsOf", 0) reads the rows from before the merge.

Use your own records

Replace CUSTOMERS and CHANGES while preserving their declared schemas. This small example assumes one change per customer in the batch. For retries, older updates and conflicting versions, see the versioned changes guide.

On Apache Spark

The same code runs on Apache Spark once Delta Lake is added to the session. The checks run it on Spark 4.2 with delta-spark 4.4.1 and compare the tables with Sail's:

spark = (
    SparkSession.builder.config("spark.jars.packages", "io.delta:delta-spark_2.13:4.4.1")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .getOrCreate()
)

Related Sail behavior

• Read and update Delta Lake tables directly

Keep exploring

Data pipelinesManage a Delta Lake table through its life →Evolve a Delta table and inspect its earlier versions. Data pipelinesManage an Apache Iceberg table and read old snapshots →Update an Iceberg table and read its earlier snapshots.