Partial deletes can bring back rows removed by earlier delete files

Aperta
  • Qualcosa non va?
#4,291 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@ghoshp83 ci sta già lavorando.

Dal 5/11/2026.

  • #4292 di @ghoshp83 — aperta

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
25/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Ferma
Stack tecnologico
python
Ambito
backend, databases

Direzione di ricerca

Start with Transaction.delete and trace how its filtered scan plans data files and attaches delete files; then inspect _build_metrics_evaluator in _open_manifest and _filter_manifest_entries in update/validate.py. The reproduction in the issue shows position and equality delete cases, plus a missed concurrent-delete conflict. Done means partial deletes and retries preserve rows removed by applicable earlier delete files; an open pull request already addresses this issue.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

kind:bug

Apache Iceberg version: 0.12.0 (latest release). main has the same code.

Please describe the bug 🐞

A partial Table.delete can drop earlier delete files of the data files it rewrites. The rows those delete files removed then come back.

Transaction.delete plans the files to rewrite with the delete filter: self._scan(row_filter=delete_filter).plan_files(). Planning with a filter also leaves out the delete files whose column bounds can't match it (_build_metrics_evaluator, applied to delete entries in _open_manifest). The rewrite then reads each planned file in full, with AlwaysTrue(). The delete files that planning left out are not applied, so the new data file holds their rows again.

Leaving out those delete files is right for a filtered read. A rewrite of the whole file needs every delete file that applies to it.

overwrite(..., overwrite_filter=...), dynamic_partition_overwrite and upsert go through Transaction.delete too.

Reproduction below. PyIceberg doesn't write delete files, so the helper commits one with its low-level writers.

  1. A position delete that stores the deleted row, which the spec allows. It has bounds on k = 2. A delete of k = 1 brings the row back.
  2. An equality delete on k = 2, with bounds on k. PyIceberg refuses to read equality deletes, but the filter k = 1 leaves this one out, so the delete goes ahead. The new data file holds (2, 'b') and has a higher sequence number than the equality delete. So that delete no longer applies to it.
Reproduction 
import struct
import tempfile
import uuid

import pyarrow as pa
import pyarrow.parquet as pq
from pyiceberg.catalog.sql import SqlCatalog
from pyiceberg.expressions import EqualTo
from pyiceberg.manifest import (
    DataFile,
    DataFileContent,
    FileFormat,
    ManifestContent,
    ManifestEntry,
    ManifestEntryStatus,
    ManifestWriterV2,
    write_manifest_list,
)
from pyiceberg.schema import Schema
from pyiceberg.table.refs import SnapshotRefType
from pyiceberg.table.snapshots import (
    Operation,
    Snapshot,
    SnapshotSummaryCollector,
    Summary,
    update_snapshot_summaries,
)
from pyiceberg.table.update import AddSnapshotUpdate, SetSnapshotRefUpdate
from pyiceberg.typedef import Record
from pyiceberg.types import LongType, NestedField, StringType

# PyIceberg doesn't write delete files, so this commits one with its low-level writers


class DeleteManifestWriter(ManifestWriterV2):
    def content(self):
        return ManifestContent.DELETES

    @property
    def _meta(self):
        return {**super()._meta, "content": "deletes"}


def field(name, type_, field_id):
    return pa.field(name, type_, nullable=False, metadata={"PARQUET:field_id": str(field_id)})


def add_delete_file(table, content, rows, bounds, equality_ids=None):
    """Commit one delete file with `rows` (a pyarrow table) and these column bounds."""
    path = f"{table.location()}/data/delete-{uuid.uuid4()}.parquet"
    pq.write_table(rows, path.removeprefix("file://"))
    delete_file = DataFile.from_args(
        content=content,
        file_path=path,
        file_format=FileFormat.PARQUET,
        partition=Record(),
        record_count=rows.num_rows,
        file_size_in_bytes=len(table.io.new_input(path)),
        lower_bounds=bounds,
        upper_bounds=bounds,
        equality_ids=equality_ids,
    )
    delete_file.spec_id = table.spec().spec_id
    parent = table.current_snapshot()
    snapshot_id = uuid.uuid4().int >> 65
    sequence_number = table.metadata.last_sequence_number + 1
    manifest = table.io.new_output(f"{table.location()}/metadata/{uuid.uuid4()}-m0.avro")
    with DeleteManifestWriter(
        spec=table.spec(),
        schema=table.schema(),
        output_file=manifest,
        snapshot_id=snapshot_id,
        avro_compression="deflate",
    ) as writer:
        writer.add(
            ManifestEntry.from_args(
                status=ManifestEntryStatus.ADDED, snapshot_id=snapshot_id, data_file=delete_file
            )
        )
    manifest_list = f"{table.location()}/metadata/snap-{snapshot_id}.avro"
    with write_manifest_list(
        format_version=2,
        output_file=table.io.new_output(manifest_list),
        snapshot_id=snapshot_id,
        parent_snapshot_id=parent.snapshot_id,
        sequence_number=sequence_number,
        avro_compression="deflate",
    ) as list_writer:
        list_writer.add_manifests([writer.to_manifest_file(), *parent.manifests(table.io)])
    collector = SnapshotSummaryCollector()
    collector.add_file(delete_file, schema=table.schema())
    summary = update_snapshot_summaries(
        Summary(Operation.DELETE, **collector.build()), previous_summary=parent.summary
    )
    snapshot = Snapshot(
        snapshot_id=snapshot_id,
        parent_snapshot_id=parent.snapshot_id,
        sequence_number=sequence_number,
        manifest_list=manifest_list,
        summary=summary,
        schema_id=table.schema().schema_id,
    )
    ref = SetSnapshotRefUpdate(
        ref_name="main", type=SnapshotRefType.BRANCH, snapshot_id=snapshot_id
    )
    table.catalog.commit_table(table, (), (AddSnapshotUpdate(snapshot=snapshot), ref))
    return table.catalog.load_table(table.name())


def long(x):
    return struct.pack("<q", x)


tmp = tempfile.mkdtemp()
catalog = SqlCatalog("lake", uri=f"sqlite:///{tmp}/c.db", warehouse=f"file://{tmp}")
catalog.create_namespace("ns")
schema = Schema(NestedField(1, "k", LongType()), NestedField(2, "v", StringType()))
data = pa.table({"k": [1, 2, 3], "v": ["a", "b", "c"]})

# 1. A position delete that stores the deleted row, as the spec allows
table = catalog.create_table(("ns", "pos"), schema=schema)
table.append(data)
data_path = next(iter(table.scan().plan_files())).file.file_path
row = pa.struct([field("k", pa.int64(), 1), field("v", pa.string(), 2)])
deletes = pa.Table.from_pylist(
    [{"file_path": data_path, "pos": 1, "row": {"k": 2, "v": "b"}}],
    schema=pa.schema(
        [
            field("file_path", pa.string(), 2147483546),
            field("pos", pa.int64(), 2147483545),
            field("row", row, 2147483544),
        ]
    ),
)
bounds = {2147483546: data_path.encode(), 2147483545: long(1), 1: long(2), 2: b"b"}
table = add_delete_file(table, DataFileContent.POSITION_DELETES, deletes, bounds)
print(table.scan().to_arrow().sort_by("k").to_pylist())
# [{'k': 1, 'v': 'a'}, {'k': 3, 'v': 'c'}]

# The filtered plan leaves out the delete file, as its bounds exclude k = 1
print([len(t.delete_files) for t in table.scan(row_filter=EqualTo("k", 1)).plan_files()])
# [0]
print([len(t.delete_files) for t in table.scan().plan_files()])
# [1]

table.delete(EqualTo("k", 1))
table = catalog.load_table(("ns", "pos"))
print(table.scan().to_arrow().sort_by("k").to_pylist())
# [{'k': 2, 'v': 'b'}, {'k': 3, 'v': 'c'}]: expected [{'k': 3, 'v': 'c'}]

# 2. An equality delete of k = 2
table = catalog.create_table(("ns", "eq"), schema=schema)
table.append(data)
deletes = pa.Table.from_pylist([{"k": 2}], schema=pa.schema([field("k", pa.int64(), 1)]))
table = add_delete_file(
    table, DataFileContent.EQUALITY_DELETES, deletes, {1: long(2)}, equality_ids=[1]
)
# PyIceberg refuses to read equality deletes, but the filtered plan leaves it out
table.delete(EqualTo("k", 1))
table = catalog.load_table(("ns", "eq"))
for manifest in table.current_snapshot().manifests(table.io):
    for entry in manifest.fetch_manifest_entry(table.io):
        file = entry.data_file
        rows = pq.read_table(file.file_path.removeprefix("file://")).to_pylist()
        print(entry.sequence_number, file.content.name, rows)
# 3 DATA [{'k': 2, 'v': 'b'}, {'k': 3, 'v': 'c'}]
# 2 EQUALITY_DELETES [{'k': 2}]
# The new data file holds (2, 'b'). The equality delete has a lower sequence number,
# so it no longer applies to that file.

Output:

[{'k': 1, 'v': 'a'}, {'k': 3, 'v': 'c'}]
[0]
[1]
[{'k': 2, 'v': 'b'}, {'k': 3, 'v': 'c'}]
3 DATA [{'k': 2, 'v': 'b'}, {'k': 3, 'v': 'c'}]
2 EQUALITY_DELETES [{'k': 2}]

The conflict check on a retry filters delete entries the same way (_filter_manifest_entries in update/validate.py). From reading the code, a concurrent delete of a row outside the filter can be missed too.

A possible fix is to keep the filter for picking data files, and attach every delete file that applies to each of them, for example from an unfiltered delete index.

(With AI) The equality-delete case is reachable with tables written by Iceberg Java. Its equality-delete writer records column bounds with the table's metrics config (MetricsConfig.forTable). A sink in upsert mode writes only the key columns to its delete files. A delete filtered on the key can then leave out every equality delete file and bring back rows that were updated earlier. If any equality delete file passes the filter, the delete fails on the unsupported delete instead. Position deletes from current writers have no bounds on data columns, so they are not affected.

(Found via polars dev, with AI).

Willingness to contribute
  • I can contribute a fix for this bug independently
  • I would be willing to contribute a fix for this bug with guidance from the Iceberg community
  • I cannot contribute a fix for this bug at this time
Lingua principale
Python
Stelle
1.2k
Fork
654
Merge medio
2g 11h
PR unite (30g)
122

Preparare l'ambiente

  • Nessun Dockerfile né file Docker Compose
  • Ha un modello di pull request
  • Nessuna guida per i contributori

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/iceberg-python

Tutte le issue di apache/iceberg-python

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.