Partial deletes can bring back rows removed by earlier delete files

Abierto
  • ¿Algo no está bien?
#4,291 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

@ghoshp83 ya está trabajando en esto.

Desde el 5/11/2026.

  • #4292 de @ghoshp83 — abierto

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
25/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Estancado
Stack tecnológico
python
Área
backend, databases

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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
Lenguaje dominante
Python
Estrellas
1.2k
Forks
654
Merge medio
2 d 11 h
PR fusionados (30 d)
122

Preparar el entorno

  • Sin Dockerfile ni archivo de Docker Compose
  • Tiene una plantilla de pull request
  • Sin guía de contribución

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/iceberg-python

Todos los issues de apache/iceberg-python

Issues similares

Más issues de Python

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.