Project: Kiosko's native upsert

Description

This project closes module 6. You recalled the three ways Kiosko already solved P002's change (lesson 2). You installed the Iceberg runtime for Spark and documented, with real evidence, how far it gets in this environment (lesson 3). You learned MERGE INTO's general syntax (lesson 4) and applied it to Kiosko's concrete case (lesson 5). You really ran table.upsert(), with no caveats at all (lesson 6). And you built the criterion for choosing among Iceberg's three techniques (lesson 7). Only one step is left: bringing this module's executable result together in a single script, with automated asserts — and, alongside it, Spark's MERGE INTO's complete reference, documented for what it is: representative, not executed in this environment.

Connection to the module. This project doesn't introduce any new concept — it's the final integration of the seven previous lessons. Unlike this guide's other modules' closing projects, this one can't integrate all five techniques into a single executed script: the real incompatibility documented in lesson 3 is still in effect. What this project delivers, honestly, is the result that does run end to end — table.upsert() — and the complete reference for what would run with Spark, in an environment where the versions do match.

An analogy: the result that really got archived, and the blueprint for the one that didn't

An architect presenting a two-building project, one already built and one still on paper because of a permit issue, doesn't hide the second one — they present it for what it is: a complete, verified blueprint, ready to be built the moment the permit gets resolved. This project does exactly that: table.upsert()'s building is built, with asserts confirming it brick by brick; MERGE INTO via Spark's is a complete, verified blueprint — the exact same syntax from lesson 5 — ready to be built the day iceberg-spark-runtime publishes a variant compatible with pyspark==4.2.0.

The material: a new working directory

kiosko_native_upsert_project/
└── kiosko_native_upsert_project.py       (this project: brings the executable pieces together)

With PyIceberg installed in your environment (pip install "pyiceberg[sql-sqlite,pyarrow]", module 1, lesson 4). This project is self-contained: it creates kiosko.dim_product_upsert_demo from scratch, so it doesn't depend on any file from this module's previous lessons — and it doesn't touch kiosko.dim_product, kiosko.fact_orders, kiosko.dim_store, or kiosko.fact_orders_at_scale, the real tables modules 1 through 5 already built, if you run this project in the same directory where you have them.

The reference solution, verified — the part that does run

# kiosko_native_upsert_project.py -- module 6's closing project
# table.upsert() end to end, with automated assert
import os

import pyarrow as pa
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import DoubleType, NestedField, StringType

DIM_PRODUCT_SCHEMA = Schema(
    NestedField(field_id=1, name="product_id", field_type=StringType(), required=True),
    NestedField(field_id=2, name="product_name", field_type=StringType(), required=True),
    NestedField(field_id=3, name="category", field_type=StringType(), required=True),
    NestedField(field_id=4, name="unit_cost", field_type=DoubleType(), required=True),
)
PA_SCHEMA = pa.schema([
    pa.field("product_id", pa.string(), nullable=False),
    pa.field("product_name", pa.string(), nullable=False),
    pa.field("category", pa.string(), nullable=False),
    pa.field("unit_cost", pa.float64(), nullable=False),
])
DIM_PRODUCT_V1 = [
    {"product_id": "P001", "product_name": "Bottled Water 600ml", "category": "beverages", "unit_cost": 0.40},
    {"product_id": "P002", "product_name": "Energy Bar", "category": "snacks", "unit_cost": 0.60},
    {"product_id": "P003", "product_name": "Instant Coffee Sachet", "category": "beverages", "unit_cost": 0.35},
    {"product_id": "P004", "product_name": "Phone Charger Cable", "category": "electronics", "unit_cost": 2.10},
]
DIM_PRODUCT_V2 = [
    {"product_id": "P001", "product_name": "Bottled Water 600ml", "category": "beverages", "unit_cost": 0.40},
    {"product_id": "P002", "product_name": "Energy Bar", "category": "health-snacks", "unit_cost": 0.68},
    {"product_id": "P003", "product_name": "Instant Coffee Sachet", "category": "beverages", "unit_cost": 0.35},
    {"product_id": "P004", "product_name": "Phone Charger Cable", "category": "electronics", "unit_cost": 2.10},
]


def main() -> None:
    print("=== Kiosko: the native upsert, end to end ===\n")

    warehouse_path = os.path.abspath("kiosko_warehouse")
    catalog_db_path = os.path.abspath("kiosko_catalog.db")
    os.makedirs(warehouse_path, exist_ok=True)
    catalog = load_catalog(
        "kiosko", type="sql",
        uri=f"sqlite:///{catalog_db_path}", warehouse=f"file://{warehouse_path}",
    )
    catalog.create_namespace("kiosko")
    print("Step 1/6 -- 'kiosko' catalog ready")

    table = catalog.create_table("kiosko.dim_product_upsert_demo", schema=DIM_PRODUCT_SCHEMA)
    pa_table_v1 = pa.Table.from_pylist(DIM_PRODUCT_V1, schema=PA_SCHEMA)
    table.append(pa_table_v1)
    rows_after_v1 = table.scan().to_arrow().num_rows
    print(f"Step 2/6 -- kiosko.dim_product_upsert_demo created and loaded with V1: {rows_after_v1} rows "
          f"(P002 = snacks/0.60)")

    pa_table_v2 = pa.Table.from_pylist(DIM_PRODUCT_V2, schema=PA_SCHEMA)
    result = table.upsert(pa_table_v2, join_cols=["product_id"])
    print(f"Step 3/6 -- table.upsert(full_V2) run: rows_updated={result.rows_updated}, "
          f"rows_inserted={result.rows_inserted}")

    final_rows = sorted(table.scan().to_arrow().to_pylist(), key=lambda r: r["product_id"])
    p002 = next(r for r in final_rows if r["product_id"] == "P002")
    print(f"Step 4/6 -- final state: {len(final_rows)} rows, "
          f"P002 = {p002['category']}/{p002['unit_cost']}")

    history_len = len(table.history())
    ops = [row["operation"] for row in table.inspect.snapshots().select(["operation"]).to_pylist()]
    print(f"Step 5/6 -- table.history() has {history_len} entries, operations: {ops}")

    others_unchanged = all(
        r["category"] == next(o for o in DIM_PRODUCT_V1 if o["product_id"] == r["product_id"])["category"]
        and r["unit_cost"] == next(o for o in DIM_PRODUCT_V1 if o["product_id"] == r["product_id"])["unit_cost"]
        for r in final_rows if r["product_id"] != "P002"
    )
    print(f"Step 6/6 -- P001, P003, P004 unchanged: {others_unchanged}\n")

    print("=== Final verification ===\n")

    assert rows_after_v1 == 4
    assert result.rows_updated == 1, "the upsert must update exactly 1 row (P002)"
    assert result.rows_inserted == 0, "the upsert must not insert any new row"
    assert len(final_rows) == 4, "the grain must stay one row per product"
    assert p002["category"] == "health-snacks" and p002["unit_cost"] == 0.68
    assert others_unchanged, "P001, P003, and P004 must not have changed"
    assert history_len == 3, "append(V1) + upsert (partial overwrite + append) = 3 entries"
    assert ops == ["append", "overwrite", "append"], "the upsert must resolve as overwrite+append"

    print("All verifications passed:")
    print("  - kiosko.dim_product_upsert_demo: 4 rows, one per product, no history columns")
    print("  - table.upsert(full_V2, join_cols=['product_id']) detected only P002's real change")
    print("  - rows_updated=1, rows_inserted=0 -- with nobody calculating the delta by hand")
    print("  - P001, P003, P004 confirmed unchanged")
    print("  - table.history() confirms 3 snapshots: append + overwrite (partial) + append")


if __name__ == "__main__":
    main()

What to expect (verified by running the real python3 kiosko_native_upsert_project.py, end to end, in a new directory):

=== Kiosko: the native upsert, end to end ===

Step 1/6 -- 'kiosko' catalog ready
Step 2/6 -- kiosko.dim_product_upsert_demo created and loaded with V1: 4 rows (P002 = snacks/0.60)
Step 3/6 -- table.upsert(full_V2) run: rows_updated=1, rows_inserted=0
Step 4/6 -- final state: 4 rows, P002 = health-snacks/0.68
Step 5/6 -- table.history() has 3 entries, operations: ['append', 'overwrite', 'append']
Step 6/6 -- P001, P003, P004 unchanged: True

=== Final verification ===

All verifications passed:
  - kiosko.dim_product_upsert_demo: 4 rows, one per product, no history columns
  - table.upsert(full_V2, join_cols=['product_id']) detected only P002's real change
  - rows_updated=1, rows_inserted=0 -- with nobody calculating the delta by hand
  - P001, P003, P004 confirmed unchanged
  - table.history() confirms 3 snapshots: append + overwrite (partial) + append

Eight asserts, none decorative: they confirm the grain held (4 rows, never more), that upsert() detected exactly the real change (rows_updated=1, rows_inserted=0), that the three unchanged rows really didn't change, and that the internal mechanism resolved exactly as lesson 6 explained (append + overwrite + append).

The complete reference — the representative part: MERGE INTO via Spark

For this project to document the module's complete result, not just the half that ran in this environment, here's lesson 5's MERGE INTO, in full, explicitly marked for what it is:

-- What to expect (representative) -- syntax identical to lesson 5, not executed
-- in this environment because of the real incompatibility documented in lesson 3.

CREATE NAMESPACE IF NOT EXISTS local.kiosko;

CREATE TABLE local.kiosko.dim_product (
    product_id STRING, product_name STRING, category STRING, unit_cost DOUBLE
) USING iceberg;

INSERT INTO local.kiosko.dim_product VALUES
    ('P001', 'Bottled Water 600ml',   'beverages',   0.40),
    ('P002', 'Energy Bar',            'snacks',      0.60),
    ('P003', 'Instant Coffee Sachet', 'beverages',   0.35),
    ('P004', 'Phone Charger Cable',   'electronics', 2.10);

CREATE TABLE local.kiosko.dim_product_staging (
    product_id STRING, product_name STRING, category STRING, unit_cost DOUBLE
) USING iceberg;

INSERT INTO local.kiosko.dim_product_staging VALUES
    ('P002', 'Energy Bar', 'health-snacks', 0.68);

MERGE INTO local.kiosko.dim_product AS target
USING local.kiosko.dim_product_staging AS source
ON target.product_id = source.product_id
WHEN MATCHED THEN UPDATE SET
    target.product_name = source.product_name,
    target.category     = source.category,
    target.unit_cost    = source.unit_cost
WHEN NOT MATCHED THEN INSERT (product_id, product_name, category, unit_cost)
VALUES (source.product_id, source.product_name, source.category, source.unit_cost);

What to expect (representative): exactly the same business result the upsert() script above confirmed — four rows, P002 with category='health-snacks', unit_cost=0.68, P001/P003/P004 intact. The difference, already documented in depth in this module's lessons 4, 5, and 7: this SQL runs distributed, over Spark, and needs the JVM; the script above runs in a single Python process, with neither.

Diagram: where you came from, where you landed

flowchart LR
    A["Modules 1-5:\nkiosko.dim_product on V2\nvia overwrite() + time travel"] --> B["Lesson 2:\n3 techniques recalled,\nDuckDB, dbt, this guide"]
    B --> C["Lesson 3:\nSpark + runtime installed,\nreal incompatibility documented"]
    C --> D["Lesson 4-5:\nMERGE INTO -- real syntax,\nrepresentative output"]
    D --> E["Lesson 6:\ntable.upsert() -- really\nexecuted, rows_updated=1"]
    E --> F["Lesson 7:\nchoice criterion,\n5 factors"]
    F --> G["This project:\nupsert executed + assert,\nMERGE documented alongside it"]
    G --> H["Module 7:\ncatalogs, maintenance,\nDelta Lake by contrast"]

Closing the module's promise, point by point

What lesson 1 promised	Evidence this module delivered it
Recalling the three ways Kiosko already solved P002's change	Lesson 2: literally quoted code and output from data-modeling, dbt, and module 3
Iceberg's native MERGE INTO via Spark SQL	Lessons 3-5: environment really installed, syntax verified against the official documentation, real incompatibility documented with evidence
table.upsert() as a 100% Python alternative	Lesson 6 and this project: UpsertResult(rows_updated=1, rows_inserted=0), really executed, no caveats
Criterion for choosing among the techniques	Lesson 7: five factors, comparison table, decision tree
Both catalogs (kiosko/local) declared distinct, with no interoperability claimed	Explicitly stated since lesson 1, respected in every technical lesson of this module

This project didn't touch kiosko.dim_product, kiosko.fact_orders, kiosko.dim_store, or kiosko.fact_orders_at_scale — those tables stay exactly as modules 1 through 5 left them. What this project delivers is exactly what this module promised: two new ways to apply P002's change without rebuilding the complete table, one executed end to end with automated verification, the other documented with the same honesty the rest of this guide applies to any result it couldn't really execute.

Common mistakes

Running this project on a catalog that already has kiosko.dim_product_upsert_demo from lesson 6. What happens: someone runs this project in the same directory where they already completed lesson 6, and catalog.create_table(...) fails because the table is already registered. Why it happens: this project deliberately repeats the full creation from scratch, so it's self-contained and reproducible without depending on the exact state lesson 6 left behind. How to spot it: if you see TableAlreadyExistsError when running kiosko_native_upsert_project.py, you already have a catalog with that table registered in the same directory. How to fix it: run this project in a new working directory, separate from where you did lesson 6 — exactly as this lesson's "The material" suggests.

Trying to run this lesson's MERGE INTO SQL block, expecting it to work. What happens: someone copies the SQL from this lesson's "The complete reference" section into a real spark-sql, expecting to see it run with no issues. Why it happens: the SQL is written completely and correctly — it isn't pseudocode — so it seems reasonable that it would run. How to spot it: if your environment has exactly this guide's same version combination (pyspark==4.2.0 with iceberg-spark-runtime-4.0_2.13:1.11.0), you're going to run into the same IncompatibleClassChangeError documented in lesson 3. How to fix it: if you have access to an environment with a compatible version combination — for example, pyspark==4.0.x with this same runtime, see lesson 3's Exercise 3 — this SQL should run with no changes. If not, treat it for what this lesson declares it to be: a reference verified against the official documentation, not a demonstration executed in this environment.

Exercises

Exercise 1 — Run the full project yourself, from scratch. In a new directory, run python3 kiosko_native_upsert_project.py. Confirm you see the six steps complete and the final message with the five verifications.

See solution

If PyIceberg is installed in your environment, the output should exactly reproduce this lesson's structure: six numbered steps, followed by the final verification with the five success messages. The whole script runs in under a second — there's no at-scale dataset in this project, unlike module 5's project.

Exercise 2 — Break an assert on purpose, and watch it fail. Temporarily change DIM_PRODUCT_V2 so P001 also has a different unit_cost (for example, 0.45 instead of 0.40), run the script again, and observe which assert fails first. Then revert the change.

See solution

The first assert to fail should be assert result.rows_updated == 1, "the upsert must update exactly 1 row (P002)" — because now there are two rows with a different value (P001 and P002), so result.rows_updated would be 2, not 1. This exercise confirms, with direct evidence, that table.upsert() really does compare each row independently — it isn't a coincidence that only P002 got counted as updated in the original run, it's the direct result of only P002 having a different value between V1 and V2.

Exercise 3 — Explain, in your own words, why this project documents Spark's MERGE INTO instead of simply omitting it. In 3-4 sentences, justify why including code that couldn't really run in this environment is consistent with the rest of this guide's discipline.

See solution

Omitting Spark's MERGE INTO would leave the project incomplete against what module 1 promised: two new techniques, not one. This guide's discipline — stated since the DESIGN doc — was never "only show what works with no friction," but "show exactly what was really verified and what wasn't, with no ambiguity" — the same rule the "What to expect" blocks with a never-hardcoded snapshot_id already applied, or the explicit warning about the two distinct catalogs. Documenting MERGE INTO as a reference verified against the official documentation, with the real incompatibility explained in lesson 3, is more honest and more useful than faking an execution that never happened, or deleting half the module because one piece of external infrastructure didn't cooperate.

Summary and next step: this module's close

With this project you close module 6. You integrated the executable result — table.upsert(), with asserts confirming every claim — alongside the complete, honest reference for what MERGE INTO via Spark would have produced, in an environment with compatible versions. Kiosko now has two more ways to apply a partial change to an Iceberg table without rebuilding the complete state — added to the three it already knew from data-modeling, dbt, and this very guide's module 3.

Where you go next. Module 7 — Catalogs, maintenance, and Delta Lake by contrast — names the production catalogs this guide never implemented (REST, AWS Glue Catalog, Unity Catalog, Polaris), and returns to kiosko.dim_product — the real table, with its snapshots accumulated since module 3 — to safely prune it: expiring old snapshots, compacting small files, without losing the time travel you do need. And it closes with Delta Lake, named once, by contrast.

Resources

• PyIceberg — official documentation (quickstart), the complete catalog, table, append(), and upsert() flow this project integrates. py.iceberg.apache.org. In English.
• PyIceberg — API reference, table.upsert(), UpsertResult, table.inspect.snapshots(). py.iceberg.apache.org/api. In English.
• Apache Iceberg — official documentation, "Spark Writes," MERGE INTO section, source of this project's reference SQL block. iceberg.apache.org/docs/latest/spark-writes/#merge-into. In English.
• GitHub — apache/iceberg#15238, the real, documented incompatibility that explains why this project's SQL section is representative. github.com/apache/iceberg/issues/15238. In English.
• This guide's DESIGN doc — the full map of the eight modules, including the module 7 that follows. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.