PyIceberg's upsert: the Python-native alternative

Description

This lesson really runs, with no caveats at all, what lessons 4 and 5 couldn't execute in this environment: the same MERGE INTO idea — update if it exists, insert if it doesn't — but as a single Python method, table.upsert(), with not one line of SQL and no need for the JVM. You're going to create a new table, load it with V1, and confirm, with literal output, that table.upsert() detects P002's real change — and only that one — among four rows you hand it complete.

Connection to the module. Lesson 3 documented a real incompatibility between iceberg-spark-runtime-4.0 and Spark 4.2.0, which left lessons 4 and 5 marked representative. table.upsert() doesn't depend on Spark at all — it lives entirely in PyIceberg, the same package you've already used without interruption in modules 1 through 5 — so this lesson returns to this guide's discipline: really executed code, literally copied output.

Why this lesson doesn't touch kiosko.dim_product

Before the code, a necessary clarification. kiosko.dim_product — the real table, the one you built in module 3 — has already been living in its V2 state for three modules: table.overwrite() already applied the change, and snap_v1 is still available to recover the previous state via time travel. Repeating the same change on that same table wouldn't demonstrate anything new — upsert() would compare V2 against V2, and would find no difference to update.

To compare upsert() with this module's other four techniques, on equal footing — starting from V1, just like lesson 5 did (with local.kiosko.dim_product, in a separate catalog) and the DuckDB and dbt techniques (lesson 2) — this lesson reproduces the experiment on a new, dedicated table: kiosko.dim_product_upsert_demo. Same kiosko catalog as always, new table, so as not to touch the real history you already built. It's the same discipline you already saw in each module's closing projects: a new working directory, a state reproducible from scratch.

An analogy: the same counter, now in self-service mode

Lesson 1's bank counter still needed someone staffing it with SQL. table.upsert() is that same window, turned into a self-service kiosk: you hand it the complete state you believe is correct — all four rows, exactly as you know them today — and the system decides, on its own, which of those rows represent a real change and which were already up to date, with no need for you to calculate the delta in advance or write an ON/WHEN MATCHED.

Worked example: upsert(), really executed

Step 1 — Create the demo table, loaded with V1

# upsert_demo.py
import os

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

warehouse_path = os.path.abspath("kiosko_warehouse")
catalog_db_path = os.path.abspath("kiosko_catalog.db")

catalog = load_catalog(
    "kiosko", type="sql",
    uri=f"sqlite:///{catalog_db_path}", warehouse=f"file://{warehouse_path}",
)
catalog.create_namespace("kiosko")

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),
)
table = catalog.create_table("kiosko.dim_product_upsert_demo", schema=dim_product_schema)

dim_product_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},
]
pa_table_v1 = pa.Table.from_pylist(DIM_PRODUCT_V1, schema=dim_product_pa_schema)
table.append(pa_table_v1)

print("V1 loaded. Rows:", table.scan().to_arrow().num_rows)

What to expect (verified by running the real script):

V1 loaded. Rows: 4

Nothing new up to here — the same create_table() + append() you've already used dozens of times since module 1.

Step 2 — table.upsert(), with the complete V2 state

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},
]
pa_table_v2 = pa.Table.from_pylist(DIM_PRODUCT_V2, schema=dim_product_pa_schema)

result = table.upsert(pa_table_v2, join_cols=["product_id"])
print("UpsertResult:", result)
print("rows_updated:", result.rows_updated)
print("rows_inserted:", result.rows_inserted)

Notice something different from lesson 5: here you hand upsert() all four complete rows — not just the delta — exactly like data-modeling and dbt did in lesson 2. The difference is that upsert(), unlike MERGE INTO as you wrote it in lesson 5, does compare internally every non-key value against what already exists, and only counts as "updated" the row where something really changed.

What to expect (verified by running the real script):

UpsertResult: UpsertResult(rows_updated=1, rows_inserted=0)
rows_updated: 1
rows_inserted: 0

Exactly what you predicted in lesson 2's Exercise 3: one row updated — P002, the only one that changed — zero rows inserted — no new product_id. P001, P003, and P004 were present in the four rows you handed it, but upsert() recognized them as identical to what was already on file, and didn't touch them.

Step 3 — Confirm the final state

print("\ntable.scan().to_arrow() after the upsert (sorted by product_id):")
rows = sorted(table.scan().to_arrow().to_pylist(), key=lambda r: r["product_id"])
for row in rows:
    print(f"  {row['product_id']}  {row['product_name']:<22} {row['category']:<14} unit_cost={row['unit_cost']}")

What to expect (verified by running the real script):

table.scan().to_arrow() after the upsert (sorted by product_id):
  P001  Bottled Water 600ml    beverages      unit_cost=0.4
  P002  Energy Bar             health-snacks  unit_cost=0.68
  P003  Instant Coffee Sachet  beverages      unit_cost=0.35
  P004  Phone Charger Cable    electronics    unit_cost=2.1

A note on order. If you run table.scan().to_arrow().to_pylist() without explicitly sorting, P002's row shows up first, not in the P001..P004 order you saw in module 3 — because upsert() wrote a new, separate data file, with only the updated row, and table.scan() doesn't guarantee any order across files. This guide never assumed an implicit order in any previous lesson by accident: table.scan() reflects the physical order in which Iceberg reads its data files, never an ORDER BY — if your code depends on a specific order, sort it yourself, exactly like sorted(...) does in this step.

Step 4 — What table.history() revealed: two snapshots, not one

print("\ntable.history() has", len(table.history()), "entries")
snaps = table.inspect.snapshots().select(["operation", "summary"])
for row in snaps.to_pylist():
    s = dict(row["summary"])
    print(f"  operation={row['operation']:<9} added-records={s.get('added-records','-')} "
          f"deleted-records={s.get('deleted-records','-')} total-records={s.get('total-records','-')}")

What to expect (verified by running the real script; snapshot_id/committed_at are from your own run, different every time):

table.history() has 3 entries
  operation=append    added-records=4 deleted-records=- total-records=4
  operation=overwrite added-records=3 deleted-records=4 total-records=-
  operation=append    added-records=1 deleted-records=- total-records=4

Three entries: Step 1's append() (the four V1 rows), and two more, not one, produced by the single upsert() call in Step 2. This shouldn't entirely surprise you — it's the same lesson you already learned in module 3, lesson 3, now applied to a different mechanism: a high-level PyIceberg operation can internally resolve into several snapshots.

Going deeper: what upsert() does under the hood, with evidence from the source code itself

It's worth understanding why exactly those two entries show up, not one and not three. PyIceberg 0.11.1's implementation of table.upsert() — open source, verifiable — essentially does two steps:

1. It identifies which source rows match a target row and have at least one different non-key value — the same row-by-row comparison you already saw in Step 2, the reason rows_updated=1 and not 4. With those rows — only P002, in this case — it internally calls table.overwrite(rows_to_update, overwrite_filter=...), with a filter that removes only the rows that match by product_id. Since those rows live together, in the same Parquet file, with the three that didn't change, Iceberg has to rewrite that file — it removes the old file's reference entirely (deleted-records=4) and writes a new one, with only the three rows that weren't going to be updated (added-records=3). This is the operation=overwrite entry in the output above.
2. It adds the updated rows as a separate append() — the new P002 row, with its health-snacks/0.68 values (added-records=1). This is the second operation=append entry.

Two snapshots, one for "what stays the same, rearranged" and another for "what changed, added" — a different mechanism from module 3's full overwrite() (which did a delete of everything followed by an append of everything), but with the same underlying lesson: always count snapshots with table.history(), never assume "one call, one snapshot."

Diagram: two rows stay put, one moves

flowchart TD
    A["table.upsert(full_V2, join_cols=['product_id'])"] --> B["Internal step 1:\ncompares V2 against target\nonly P002 has a different value"]
    B --> C["overwrite(only new P002,\nfilter=product_id=='P002')\nrewrites the file: P001,P003,P004 remain"]
    B --> D["append(P002 health-snacks/0.68)\nnew row, new file"]
    C --> E["UpsertResult\nrows_updated=1, rows_inserted=0"]
    D --> E

Common mistakes

Calling table.upsert() without join_cols, and expecting PyIceberg to guess the key. What happens: someone runs table.upsert(pa_table_v2), without the join_cols argument, trusting PyIceberg to use product_id automatically because it's "obviously" the key. Verify it yourself:

table.upsert(pa_table_v2)  # without join_cols

ValueError: Join columns could not be found, please set identifier-field-ids or pass in explicitly.

Why it happens: upsert() can infer the join column automatically, but only if the table's schema explicitly declares identifier-field-ids — a formal marker of "this column uniquely identifies each row" — something this guide's dim_product schema never declared (not in module 3, not here). How to spot it: the error message is explicit and says exactly what's missing — join_cols or identifier-field-ids. How to fix it: pass join_cols=["product_id"] explicitly, like this lesson does — it's safer than depending on a schema configuration you'd have to remember to declare at table-creation time.

Assuming upsert() with the complete state (V2, four rows) is equivalent to module 3's table.overwrite(). What happens: someone sees this lesson passes all four rows to upsert(), just like overwrite() received all four complete rows in module 3, and concludes they're the same operation under a different name. Why it happens: both receive the same kind of input — the complete state. How to spot it: count the snapshots — module 3's overwrite() produced delete + append (removes everything, adds everything); this lesson's upsert() produced overwrite (partial, only rewrites the file that held the row that changed) + append (only the new row). How to fix it: upsert() always does, first, the work of comparing value by value — the same work the explicit WHEN MATCHED AND (...) had to do by hand in DuckDB's MERGE (lesson 2) — and only touches what really changed; overwrite() with no filter never compares anything, it simply replaces 100% of the table, whether something changed or not.

Exercises

Exercise 1 — Reproduce the full experiment yourself, and confirm the three numbers. With PyIceberg installed, run this lesson's four steps in a new directory. Confirm rows_updated=1, rows_inserted=0, and len(table.history()) == 3.

Exercise 2 — Run upsert() a second time, with the same V2, and predict the result before running it. Without changing anything, call table.upsert(pa_table_v2, join_cols=["product_id"]) again, on the table this lesson left behind.

result2 = table.upsert(pa_table_v2, join_cols=["product_id"])
print("UpsertResult (second run):", result2)

Exercise 3 — Explain, in your own words, why upsert() needed to rewrite an entire file (deleted-records=4) to update a single row. In 2-3 sentences, using what you know about Parquet files from module 2, explain why "updating a row" in Iceberg almost never means "touching only that row" at the physical file level.

Summary and next step

In this lesson you really ran table.upsert(), with no caveats at all: you created kiosko.dim_product_upsert_demo from V1, and confirmed that upsert(full_V2, join_cols=["product_id"]) detects, on its own, that only P002 changed — UpsertResult(rows_updated=1, rows_inserted=0) — with no need for you to calculate any delta in advance. You saw, with table.history(), that this single call produced two internal snapshots — a partial overwrite and an append — and why, by reading PyIceberg's own source code.

Before moving on you should be able to: explain why this lesson uses a new table instead of kiosko.dim_product; reproduce the full experiment with the exact three numbers; and explain the difference between upsert() (compares and only touches what changed) and MERGE INTO as you wrote it in lesson 5 (updates any match, without comparing values, unless you add that condition by hand).

Lesson 7 gives the final criterion: when to choose MERGE INTO in SQL, when to choose upsert() in Python — and when neither of the two, and you still need module 3's table.overwrite().

Resources

• PyIceberg — API reference, table.upsert(), UpsertResult, and the join_cols argument. py.iceberg.apache.org/api. In English.
• PyIceberg — PyPI, current version 0.11.1, the same version you installed in module 1 and that runs this lesson with no changes. pypi.org/project/pyiceberg. In English.
• This same guide, module 3, lesson 3 — source of the original finding that one operation can produce more than one snapshot, revisited here for upsert(). 03-changing-p002-with-a-plain-overwrite.md. In Spanish.
• This guide's DESIGN doc — the full map of the eight modules, including module 6's section, and the hard rule to never hardcode a snapshot_id. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.