Partial deletes can bring back rows removed by earlier delete files

Offen
  • Stimmt etwas nicht?
#4,291 0 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen

Maintainer antworten meist innerhalb von 1 Tag

@ghoshp83 arbeitet bereits daran.

Seit 05.11.2026.

  • #4292 von @ghoshp83 — offen

Bewertung

Schwierigkeit
4/5
Geschätzter Aufwand
3-5 Tage
Anfängerfreundlichkeit
25/100
Issue-Typ
Bug
Klarheit
Klar beschrieben
Aktivitätsstatus
Veraltet
Tech-Stack
python
Bereich
backend, databases

Rechercherichtung

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.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Beschreibung

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
Vorherrschende Sprache
Python
Sterne
1.2k
Forks
654
Ø Merge
2 T. 11 Std.
Gemergte PRs (30 T.)
122

Entwicklungsumgebung

  • Kein Dockerfile und keine Docker-Compose-Datei
  • Hat eine Pull-Request-Vorlage
  • Kein Beitragsleitfaden

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus apache/iceberg-python

Alle Issues in apache/iceberg-python

Ähnliche Issues

Weitere Issues zu Python

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.