upsert duplicates rows on a partitioned table (regression vs 0.11.1)

[paulcaron16k]

Apache Iceberg version

main (reproduced at 154288fb; the affected files are byte-identical at 7d0f5031, which is version = "0.12.0")

Please describe the bug 🐞

On a partitioned table, upsert leaves the row it replaced and duplicates a row it never touched. No error is raised.

0.11.1 is correct. An unpartitioned table is correct on both, so the partition spec is required to reproduce.

Reproduction

import tempfile, pathlib, datetime as dt
import pyarrow as pa
from pyiceberg.catalog.sql import SqlCatalog
from pyiceberg.schema import Schema
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.transforms import DayTransform
from pyiceberg.types import NestedField, StringType, IntegerType, TimestampType

wh = pathlib.Path(tempfile.mkdtemp())
cat = SqlCatalog("c", uri=f"sqlite:///{wh}/c.db", warehouse=f"file://{wh}")
cat.create_namespace("db")
schema = Schema(
    NestedField(1, "k", StringType(), required=False),
    NestedField(2, "v", IntegerType(), required=False),
    NestedField(3, "ts", TimestampType(), required=False),
)
spec = PartitionSpec(
    PartitionField(source_id=3, field_id=1000, transform=DayTransform(), name="ts_day")
)
t = cat.create_table("db.t", schema=schema, partition_spec=spec,
                     properties={"format-version": "2"})

arrow = pa.schema([pa.field("k", pa.string()), pa.field("v", pa.int32()),
                   pa.field("ts", pa.timestamp("us"))])
base = dt.datetime(2026, 1, 6, 12)
def rows(pairs):
    return pa.table({"k": [p[0] for p in pairs],
                     "v": pa.array([p[1] for p in pairs], type=pa.int32()),
                     "ts": [base] * len(pairs)}, schema=arrow)

t.append(rows([("a", 1), ("b", 1)]))
t.refresh()
t.upsert(rows([("a", 2)]), join_cols=["k"])
t.refresh()

arr = t.scan().to_arrow()
print(sorted(zip(arr["k"].to_pylist(), arr["v"].to_pylist())))
print("data files:", len(t.inspect.files().to_pylist()))

version	output	data files
0.11.1	[('a', 2), ('b', 1)] — correct	2
main	[('a', 1), ('a', 2), ('b', 1), ('b', 1)]	3

Two problems in one result: the superseded ('a', 1) survives beside its replacement, and ('b', 1) — untouched by the upsert — is duplicated.

Suspected cause

_OverwriteFiles._existing_manifests keeps a manifest its evaluator does not match, whole:

manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
...
if not manifest_evaluators[manifest_file.partition_spec_id](manifest_file):
    existing_files.append(manifest_file)
    continue

The predicate comes from _build_delete_files_partition_predicate, built from the deleted files' partition values. For this upsert the join column is k, which is not a partition field, so the resulting predicate cannot select the manifest holding the rows being replaced. That manifest is appended intact, while the rewritten file and the new row are added alongside it, so a reader sees both copies.

I have not tried to determine whether the right fix is to widen the predicate, to fall back to a full scan when the predicate cannot be projected onto a spec, or to match deleted entries directly — happy to test a patch.

Knock-on effect

The symptom most users will hit first is the next upsert, which fails in upsert_util.get_rows_to_update:

ValueError: Target table has duplicate rows, aborting upsert

That guard is doing its job. It is detecting real duplicates — just ones the previous upsert created.

Why this may be worth looking at before 0.12

version on main is already 0.12.0, and the failure is silent: a merge-style ingest into a partitioned table accumulates duplicate rows with no error until a later upsert happens to trip the duplicate check.

Willingness to contribute

Would be interested in contributing a fix to Apache Iceberg.

[paulcaron16k]

Retested against 7d0f5031 (version = "0.12.0"). Still reproduces, and I was able to narrow it usefully.

It is specific to temporal transforms

Same script, only the partition transform changed. append([("a",1),("b",1)]) then upsert([("a",2)]):

partition spec	result
identity(ts)	✅ [('a', 2), ('b', 1)]
truncate(k, 1)	✅ [('a', 2), ('b', 1)]
bucket(k, 4)	✅ [('a', 2), ('b', 1)]
year(ts)	❌ [('a', 1), ('a', 2), ('b', 1), ('b', 1)]
month(ts)	❌ same
day(ts)	❌ same
hour(ts)	❌ same

So it is not "partitioned" in general, and not "non-identity" either — truncate and bucket are both fine. Only the four temporal transforms fail, which are the ones whose partition value is an integer ordinal derived from a timestamp source column.

That fits the predicate I mentioned above. Traced during a run partitioned by day(ts):

Or(EqualTo(term=Reference(name='ts'), literal=LongLiteral(20455)), ...)

The source column (ts, a timestamp) compared against a partition ordinal (20455, days since epoch). For identity the two coincide, and for truncate/bucket the projection is evidently still usable; for the temporal transforms the resulting predicate cannot select the manifest, so _existing_manifests keeps it whole.

Minimum condition

One manifest containing both a row being replaced and a row that is not.

• append 1 row, upsert it → ✅ correct. Every entry in the manifest is deleted, so the manifest is dropped entirely and the stale path is never taken.
• append 2 rows, upsert 1 → ❌ one entry survives, so the manifest has to be rewritten, the evaluator does not match, and it is appended verbatim with both entries — beside the newly written ones.

Minimal reproduction

import datetime as dt
import tempfile

import pyarrow as pa
from pyiceberg.catalog.sql import SqlCatalog
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.transforms import DayTransform
from pyiceberg.types import IntegerType, NestedField, StringType, TimestampType

warehouse = tempfile.mkdtemp()
catalog = SqlCatalog("c", uri=f"sqlite:///{warehouse}/c.db", warehouse=f"file://{warehouse}")
catalog.create_namespace("db")

table = catalog.create_table(
    "db.t",
    schema=Schema(
        NestedField(1, "k", StringType(), required=False),
        NestedField(2, "v", IntegerType(), required=False),
        NestedField(3, "ts", TimestampType(), required=False),
    ),
    partition_spec=PartitionSpec(PartitionField(3, 1000, DayTransform(), "p")),
    properties={"format-version": "2"},
)

schema = pa.schema(
    [pa.field("k", pa.string()), pa.field("v", pa.int32()), pa.field("ts", pa.timestamp("us"))]
)
when = dt.datetime(2026, 1, 6, 12)


def batch(pairs):
    return pa.table(
        {
            "k": [k for k, _ in pairs],
            "v": pa.array([v for _, v in pairs], type=pa.int32()),
            "ts": [when] * len(pairs),
        },
        schema=schema,
    )


table.append(batch([("a", 1), ("b", 1)]))          # one manifest, two rows
table.refresh()
table.upsert(batch([("a", 2)]), join_cols=["k"])   # replace one of them
table.refresh()

result = table.scan().to_arrow()
print(sorted(zip(result["k"].to_pylist(), result["v"].to_pylist())))
# 0.11.1  -> [('a', 2), ('b', 1)]
# 0.12.0  -> [('a', 1), ('a', 2), ('b', 1), ('b', 1)]

Happy to test a patch against these cases if that helps.