Skip to content

Merge Operations Examples

This page walks through end-to-end merge scenarios using the public merge() method on both backends. It focuses on strategy selection, composite keys, and interpreting the MergeResult.

Both backends require the datasets extra. See the extras matrix. For the shared contract, see Dataset Handlers.

PyArrow upsert

 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
import pyarrow as pa
from fsspeckit.datasets.pyarrow import PyarrowDatasetIO

io = PyarrowDatasetIO()

customers = pa.table({
    "customer_id": [1, 2, 3],
    "email": ["alice@example.com", "bob@example.com", "carlie@example.com"],
    "status": ["active", "active", "inactive"],
})
io.write_dataset(customers, "customers/", mode="overwrite")

updates = pa.table({
    "customer_id": [2, 3, 4],
    "email": ["bob.new@example.com", "charlie@example.com", "diana@example.com"],
    "status": ["active", "active", "active"],
})

result = io.merge(
    updates,
    "customers/",
    strategy="upsert",
    key_columns=["customer_id"],
)

print(f"Inserted: {result.inserted}, Updated: {result.updated}")
# Inserted: 1 (id 4), Updated: 2 (ids 2 and 3)

DuckDB update

The DuckDB backend takes a connection. The same strategies apply:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from fsspeckit.datasets.duckdb import DuckDBDatasetIO, create_duckdb_connection

conn = create_duckdb_connection()
io = DuckDBDatasetIO(conn)

io.write_dataset(products, "products/", mode="overwrite")

price_updates = pa.table({
    "product_id": [101, 102, 999],
    "name": ["Widget A", "Widget B", "Unknown"],
    "price": [21.99, 32.99, 1.00],
})

result = io.merge(
    price_updates,
    "products/",
    strategy="update",
    key_columns=["product_id"],
)

print(f"Updated: {result.updated}")  # 2 (ids 101 and 102; 999 ignored)

Choosing a strategy

Strategy Inserts new keys? Rewrites existing files? Typical use
insert Yes (new files) No Append-only loads, event logs
update No Yes (only affected files) Dimension updates, price changes
upsert Yes (new files) Yes (only affected files) CDC, synchronization

All three strategies require key_columns and reject null keys. insert and upsert work on an empty target; update requires an existing target dataset.

Composite-key merge

A composite key identifies a row by several columns. Pass them as a list:

1
2
3
4
5
6
result = io.merge(
    order_updates,
    "orders/",
    strategy="upsert",
    key_columns=["order_id", "line_number"],
)

Key matching is vectorized in PyArrow: composite keys are compared as StructArrays, with semi-join and anti-join operations, falling back to string-based key serialization for heterogeneous type combinations. See Multi-Key API.

Interpreting the result

Every merge() returns a MergeResult. Audit the operation from its fields rather than re-reading the dataset:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
result = io.merge(updates, "customers/", strategy="upsert", key_columns=["customer_id"])

# What changed
print(result.inserted, result.updated, result.deleted)

# Reconciliation check
assert result.target_count_before + result.inserted - result.deleted == result.target_count_after

# File-level audit
print(f"Rewritten: {len(result.rewritten_files)}")
print(f"Inserted: {len(result.inserted_files)}")
print(f"Preserved: {len(result.preserved_files)}")

Key column requirements

  • Key columns must be present in both source and target.
  • Keys cannot contain null values; the merge raises if any are found.
  • For a composite key, the full column combination must identify a row uniquely. Use dedup_order_by (on deduplicate_parquet_dataset_pyarrow) to control which duplicate survives during deduplication.

Error handling

Common failures and their fixes:

  • Key column not found: verify the column names match the source schema exactly.
  • Null values in key columns: clean the source so keys are non-null.
  • update on a missing target: update requires an existing dataset. Use insert or upsert to seed a new one.