Partial deletes can bring back rows removed by earlier delete files

Aberta
  • Algo errado?
#4,291 0 comentários 0 reações 0 responsáveis Ver no GitHub

Mantenedores costumam responder em até 1 dia

@ghoshp83 já está trabalhando nisso.

Desde 5/11/2026.

  • #4292 de @ghoshp83 — aberto

Avaliação

Dificuldade
4/5
Tempo estimado
3-5 dias
Facilidade para iniciantes
25/100
Tipo de issue
Bug
Clareza
Claramente especificada
Status de atividade
Estagnada
Stack de tecnologia
python
Domínio
backend, databases

Direção de pesquisa

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.

Escrita pelo modelo de indexação a partir do texto da issue.

Descrição

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
Linguagem predominante
Python
Estrelas
1.2k
Forks
654
Merge médio
2d 11h
PRs com merge (30d)
122

Preparar o ambiente

  • Sem Dockerfile nem arquivo Docker Compose
  • Tem um modelo de pull request
  • Sem guia de contribuição

Primeiros passos

  1. Leia a issue inteira e depois o guia de contribuição do projeto.
  2. Comente na issue dizendo que vai assumir — evita que duas pessoas façam o mesmo trabalho.
  3. Faça um fork do repositório e trabalhe em uma branch.
  4. Abra um pull request que referencie o número da issue.

Mais de apache/iceberg-python

Todas as issues de apache/iceberg-python

Issues semelhantes

Mais issues de Python

Receba novas issues na sua caixa de entrada

Um resumo curto de issues do GitHub para quem está começando.