Latest Results
feat(iceberg): add overwrite_filter for static partition overwrite (#7419)
## Changes Made
### `overwrite_filter`: static partition overwrite
`write_iceberg(mode="overwrite")` always replaced the entire table
(`tx.delete(delete_filter=ALWAYS_TRUE)`), so there was no way to replace
a single partition. `overwrite_filter` scopes the delete to the rows
matching a predicate, and the delete and the append still land in a
single Iceberg transaction:
```python
df.write_iceberg(table, mode="overwrite", overwrite_filter="dt = '2024-01-01'")
```
The filter accepts an Iceberg predicate string, a Daft expression, or a
PyIceberg `BooleanExpression`:
```python
df.write_iceberg(table, mode="overwrite", overwrite_filter=col("dt") == "2024-01-01")
df.write_iceberg(table, mode="overwrite", overwrite_filter=EqualTo("dt", "2024-01-01"))
```
It is normalized and bound against the table schema *before* the write
executes, so an invalid predicate fails before any data files are staged
rather than leaving orphans at the warehouse location. Unlike the
read-side conversion helpers, an unconvertible Daft expression raises
instead of degrading to a best-effort result, since widening the
predicate to `AlwaysTrue` would delete the whole table.
The same option is available through the catalog API (`Table.overwrite`,
`Catalog.write_table`).
An empty DataFrame plus a filter drops a partition without writing
anything.
### Validating that written rows match the filter
By default the rows being written must fall inside the filter, so an
overwrite cannot leave behind rows its delete did not cover. The check
follows Iceberg's
`OverwriteFiles.validateAddedFilesMatchOverwriteFilter`: a written file
is accepted when the partition-level strict projection proves every row
matches, or failing that when `_StrictMetricsEvaluator` proves it from
the file's column statistics.
Both tests are needed. Partition values are stored whole while column
bounds are truncated (16 bytes by default), so a long partition value
only passes the partition test, and a predicate over a non-partition
column only passes the metrics test.
Both are conservative, and statistics only ever widen, so the check can
refuse a write it cannot prove but never accepts an unmatched row. Two
cases are not provable and need `validate_overwrite_filter=False`:
equality over a non-partition column whose values exceed the bounds
truncation length, and a predicate over a column whose table properties
disable statistics. Iceberg's own evaluator has the same limitation.
The check runs before the transaction opens, so a rejected write leaves
the table unchanged.
### Fix: `partitioning` column reported null for every partition
The `partitioning` struct column of the DataFrame returned by
`write_iceberg` was built with `getattr(data_file.partition, field,
None)`. `data_file.partition` is a `pyiceberg.typedef.Record`, which
since pyiceberg 0.9 uses `__slots__ = ("_data",)` with positional access
only, so the lookup always returned `None`. Every reported partition
value was null on the pinned pyiceberg. The same code appeared in three
places: the ADD path, the DELETE path, and
`_write_iceberg_with_checkpoint`.
Values are now read by index against the partition spec the file was
actually written with, matching how the read path already handles the
same record in `IcebergScanOperator._iceberg_record_to_partition_spec`.
Files read back from a manifest resolve through `table.specs()` by their
own `spec_id`; files just built by the writer have no `spec_id`
(pyiceberg keeps it outside the record and only fills it in on read) and
fall back to the current spec.
Keys are now Iceberg partition field names rather than source column
names, which matches the read side's partitioning keys and Iceberg's
directory layout, and also fixes a latent collision: two partition
fields over one source column (e.g. `identity(dt)` and `truncate(4,
dt)`) previously collapsed into a single key, appending twice per file
and producing mismatched array lengths.
## Testing
New tests in `tests/io/iceberg/test_iceberg_writes.py` cover the three
filter input forms, partition-scoped delete reporting, filters below
partition granularity falling back to copy-on-write, hidden partitioning
(`day(ts)` spec with a `ts` range predicate), unpartitioned tables, null
partition values, empty-DataFrame partition deletes, catalog-level
dispatch, and the rejection paths (append mode, unknown column,
unconvertible expression, bad type). The truncation behavior is pinned
explicitly, including that a violating write sharing the first 16 bytes
with the filter is still refused.
`test_iceberg_writes_checkpoint.py` now pins the reported partition
values rather than only the column's presence; its docstring previously
described the `getattr` behavior as a guarantee.
Verified locally with `DAFT_RUNNER=native` (`tests/io/iceberg/` +
`tests/catalog`, 453 passed) and `DAFT_RUNNER=ray`
(`test_iceberg_writes_checkpoint.py`, 20 passed), plus doctests.
## Related Issues
None found. Latest Branches
0%
0%
ConeyLiu:test/runtime-concurrency-in-morsel-plans 0%
atovk:fix/iceberg-partition-evolution-row-filter © 2026 CodSpeed Technology