This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new b7232461d fix(table): skip empty historical-delete manifest merges
(#2057)
b7232461d is described below
commit b7232461d62427a480038c752b46c47f5acfd21b
Author: Stefan Wang <[email protected]>
AuthorDate: Wed Oct 7 04:55:51 2026 -0700
fix(table): skip empty historical-delete manifest merges (#2057)
---
table/snapshot_producers.go | 62 +++++++++---
table/snapshot_producers_test.go | 199 +++++++++++++++++++++++++++++++++++++++
2 files changed, 246 insertions(+), 15 deletions(-)
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index 9cb0b1dd9..ed6bd6c3b 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -507,42 +507,72 @@ func (m *manifestMergeManager) createManifest(specID int,
bin []iceberg.Manifest
return nil, err
}
- wr, path, counter, fileCloser, err := m.snap.newManifestWriter(spec)
- if err != nil {
- return nil, err
- }
- defer internal.CheckedClose(fileCloser, &err)
+ var wr *iceberg.ManifestWriter
+ var path string
+ var counter *internal.CountingWriter
+ var fileCloser io.Closer
writerClosed := false
+ // Close the ManifestWriter before the underlying file so a mid-loop
+ // write failure flushes into an open file, not a closed one.
defer func() {
- if !writerClosed {
+ if wr != nil && !writerClosed {
internal.CheckedClose(wr, &err)
}
+ if fileCloser != nil {
+ internal.CheckedClose(fileCloser, &err)
+ }
}()
+ ensureWriter := func() error {
+ if wr != nil {
+ return nil
+ }
+
+ var writerErr error
+ wr, path, counter, fileCloser, writerErr =
m.snap.newManifestWriter(spec)
+
+ return writerErr
+ }
+
for _, manifest := range bin {
- for entry, err := range m.snap.iterManifestEntries(manifest,
false) {
- if err != nil {
- return nil, err
+ for entry, iterErr := range
m.snap.iterManifestEntries(manifest, false) {
+ if iterErr != nil {
+ return nil, iterErr
}
switch {
case entry.Status() == iceberg.EntryStatusDELETED &&
entry.SnapshotID() == m.snap.snapshotID:
+ if iterErr = ensureWriter(); iterErr != nil {
+ return nil, iterErr
+ }
// only files deleted by this snapshot should
be added to the new manifest
- err = wr.Delete(entry)
+ iterErr = wr.Delete(entry)
case entry.Status() == iceberg.EntryStatusADDED &&
entry.SnapshotID() == m.snap.snapshotID:
+ if iterErr = ensureWriter(); iterErr != nil {
+ return nil, iterErr
+ }
// added entries from this snapshot are still
added, otherwise they should be existing
- err = wr.Add(entry)
+ iterErr = wr.Add(entry)
case entry.Status() != iceberg.EntryStatusDELETED:
+ if iterErr = ensureWriter(); iterErr != nil {
+ return nil, iterErr
+ }
// add all non-deleted files from the old
manifest as existing files
- err = wr.Existing(entry)
+ iterErr = wr.Existing(entry)
}
- if err != nil {
- return nil, err
+ if iterErr != nil {
+ return nil, iterErr
}
}
}
+ // Java's merge manager writes zero-count manifests; this path omits
bins
+ // whose entries are all deletes from earlier snapshots.
+ if wr == nil {
+ return nil, nil
+ }
+
// close the writer to force a flush and ensure counter.Count is
accurate
writerClosed = true
if err := wr.Close(); err != nil {
@@ -577,7 +607,9 @@ func (m *manifestMergeManager) mergeGroup(firstManifest
iceberg.ManifestFile, sp
if err != nil {
return nil, err
}
- output = append(output, created)
+ if created != nil {
+ output = append(output, created)
+ }
}
return output, nil
diff --git a/table/snapshot_producers_test.go b/table/snapshot_producers_test.go
index 45fe9518a..8346fe377 100644
--- a/table/snapshot_producers_test.go
+++ b/table/snapshot_producers_test.go
@@ -578,6 +578,205 @@ func TestManifestMergeManagerClosesWriterOnError(t
*testing.T) {
require.ErrorIs(t, err, errLimitedWrite)
}
+func TestManifestMergeManagerClosesWriterBeforeFileOnReadFailure(t *testing.T)
{
+ spec := iceberg.NewPartitionSpec()
+ schema := simpleSchema()
+ trackIO := newTrackingIO()
+ txn := createTestTransaction(t, trackIO, spec)
+ sp := newFastAppendFilesProducer(OpAppend, txn, trackIO, nil, nil)
+ df := newTestDataFile(t, spec, "file://data-1.parquet", nil)
+ manifest := writeTestManifestWithContent(t, trackIO, spec, schema,
sp.snapshotID,
+ "table-location/metadata/manifest-1.avro",
iceberg.ManifestContentData,
+ []iceberg.ManifestEntry{
+ iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
&sp.snapshotID, nil, nil, df),
+ })
+ missingManifest := iceberg.NewManifestFile(2,
"table-location/metadata/missing.avro", 1,
+ int32(spec.ID()), sp.snapshotID).Build()
+
+ mgr := manifestMergeManager{snap: sp}
+ _, err := mgr.createManifest(spec.ID(),
[]iceberg.ManifestFile{manifest, missingManifest})
+ require.ErrorIs(t, err, fs.ErrNotExist)
+ require.NotContains(t, err.Error(), "write after close",
+ "buffered entries must flush before the underlying file closes")
+ require.Equal(t, 1, trackIO.GetWriterCount())
+ require.Empty(t, trackIO.GetUnclosedWriters())
+}
+
+func TestManifestMergeReturnsReadErrorBeforeCreatingWriter(t *testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ schema := simpleSchema()
+ trackIO := newTrackingIO()
+ txn := createTestTransaction(t, trackIO, spec)
+ sp := newFastAppendFilesProducer(OpAppend, txn, trackIO, nil, nil)
+
+ oldSnapshotID := sp.snapshotID - 1
+ sequenceNumber := int64(1)
+ df := newTestDataFile(t, spec, "file://deleted.parquet", nil)
+ deletedManifest := writeTestManifestWithContent(t, trackIO, spec,
schema, oldSnapshotID,
+ "table-location/metadata/historical-delete.avro",
iceberg.ManifestContentData,
+ []iceberg.ManifestEntry{
+ iceberg.NewManifestEntry(iceberg.EntryStatusDELETED,
&oldSnapshotID, &sequenceNumber, nil, df),
+ })
+ missingManifest := iceberg.NewManifestFile(2,
"table-location/metadata/missing.avro", 1,
+ int32(spec.ID()), sp.snapshotID).Build()
+
+ mgr := manifestMergeManager{snap: sp}
+ _, err := mgr.createManifest(spec.ID(),
[]iceberg.ManifestFile{deletedManifest, missingManifest})
+ require.ErrorIs(t, err, fs.ErrNotExist)
+ require.Zero(t, trackIO.GetWriterCount())
+}
+
+func TestManifestMergeSkipsHistoricalDeletedOnlyManifest(t *testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ schema := simpleSchema()
+ trackIO := newTrackingIO()
+ txn := createTestTransaction(t, trackIO, spec)
+ sp := newFastAppendFilesProducer(OpAppend, txn, trackIO, nil, nil)
+
+ oldSnapshotID := sp.snapshotID - 1
+ sequenceNumber := int64(1)
+ df := newTestDataFile(t, spec, "file://deleted.parquet", nil)
+ manifestFile := writeTestManifestWithContent(t, trackIO, spec, schema,
oldSnapshotID,
+ "table-location/metadata/historical-delete.avro",
iceberg.ManifestContentData,
+ []iceberg.ManifestEntry{
+ iceberg.NewManifestEntry(iceberg.EntryStatusDELETED,
&oldSnapshotID, &sequenceNumber, nil, df),
+ })
+
+ mgr := manifestMergeManager{snap: sp}
+ created, err := mgr.createManifest(spec.ID(),
[]iceberg.ManifestFile{manifestFile})
+ require.NoError(t, err)
+ require.Nil(t, created)
+ require.Zero(t, trackIO.GetWriterCount())
+}
+
+func TestManifestMergeKeepsCurrentSnapshotDeletedEntries(t *testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ schema := simpleSchema()
+ txn, wfs := createTestTransactionWithMemIO(t, spec)
+ sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+
+ sequenceNumber := int64(1)
+ df := newTestDataFile(t, spec, "file://deleted.parquet", nil)
+ manifestFile := writeTestManifestWithContent(t, wfs, spec, schema,
sp.snapshotID,
+ "mem://default/table-location/metadata/current-delete.avro",
iceberg.ManifestContentData,
+ []iceberg.ManifestEntry{
+ iceberg.NewManifestEntry(iceberg.EntryStatusDELETED,
&sp.snapshotID, &sequenceNumber, nil, df),
+ })
+
+ mgr := manifestMergeManager{snap: sp}
+ created, err := mgr.createManifest(spec.ID(),
[]iceberg.ManifestFile{manifestFile})
+ require.NoError(t, err)
+ require.NotNil(t, created)
+
+ var entries []iceberg.ManifestEntry
+ for entry, err := range created.Entries(wfs, false) {
+ require.NoError(t, err)
+ entries = append(entries, entry)
+ }
+ require.Len(t, entries, 1)
+ require.Equal(t, iceberg.EntryStatusDELETED, entries[0].Status())
+ require.Equal(t, sp.snapshotID, entries[0].SnapshotID())
+ require.Equal(t, df.FilePath(), entries[0].DataFile().FilePath())
+}
+
+func TestManifestMergeGroupDropsEmptyMergedBin(t *testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ schema := simpleSchema()
+ txn, wfs := createTestTransactionWithMemIO(t, spec)
+ sp := newFastAppendFilesProducer(OpAppend, txn, wfs, nil, nil)
+
+ oldSnapshotID := sp.snapshotID - 1
+ sequenceNumber := int64(1)
+ manifests := make([]iceberg.ManifestFile, 0, 2)
+ for i := range 2 {
+ df := newTestDataFile(t, spec,
"file://deleted-"+strconv.Itoa(i)+".parquet", nil)
+ manifests = append(manifests, writeTestManifestWithContent(t,
wfs, spec, schema, oldSnapshotID,
+
"mem://default/table-location/metadata/historical-delete-"+strconv.Itoa(i)+".avro",
+ iceberg.ManifestContentData,
+ []iceberg.ManifestEntry{
+
iceberg.NewManifestEntry(iceberg.EntryStatusDELETED, &oldSnapshotID,
&sequenceNumber, nil, df),
+ }))
+ }
+
+ mgr := manifestMergeManager{
+ targetSizeBytes: manifests[0].Length() + manifests[1].Length(),
+ minCountToMerge: 1,
+ mergeEnabled: true,
+ mergeConcurrency: 1,
+ snap: sp,
+ }
+
+ merged, err := mgr.mergeGroup(manifests[0], spec.ID(), manifests)
+ require.NoError(t, err)
+ require.Empty(t, merged)
+}
+
+func TestMergeAppendCommitSkipsHistoricalDeleteOnlyBin(t *testing.T) {
+ spec := iceberg.NewPartitionSpec()
+ txn, wfs := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ parentSnapshotID := txn.meta.newSnapshotID()
+ sequenceNumber := int64(1)
+ historicalManifests := make([]iceberg.ManifestFile, 0, 2)
+ for i := range 2 {
+ dataFile := newTestDataFile(t, spec,
"file://deleted-"+strconv.Itoa(i)+".parquet", nil)
+ historicalManifests = append(historicalManifests,
writeTestManifestWithContent(t, wfs, spec, schema,
+ parentSnapshotID,
"mem://default/table-location/metadata/historical-delete-bin-"+strconv.Itoa(i)+".avro",
+ iceberg.ManifestContentData, []iceberg.ManifestEntry{
+
iceberg.NewManifestEntry(iceberg.EntryStatusDELETED, &parentSnapshotID,
&sequenceNumber, nil, dataFile),
+ }))
+ }
+ manifestListPath :=
"mem://default/table-location/metadata/snap-historical-delete-bin.avro"
+ manifestList, err := wfs.Create(manifestListPath)
+ require.NoError(t, err)
+ require.NoError(t, iceberg.WriteManifestList(
+ 2, manifestList, parentSnapshotID, nil, &sequenceNumber, 0,
historicalManifests))
+ require.NoError(t, manifestList.Close())
+ parentSnapshot := &Snapshot{
+ SnapshotID: parentSnapshotID,
+ SequenceNumber: sequenceNumber,
+ TimestampMs: time.Now().UnixMilli(),
+ ManifestList: manifestListPath,
+ Summary: &Summary{Operation: OpAppend, Properties:
iceberg.Properties{}},
+ }
+ require.NoError(t, txn.meta.AddSnapshot(parentSnapshot))
+ require.NoError(t, txn.meta.SetSnapshotRef(MainBranch,
parentSnapshotID, BranchRef))
+ meta, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ mergeTxn := New(Identifier{"db", "tbl"}, meta, "metadata.json",
func(context.Context) (iceio.IO, error) {
+ return wfs, nil
+ }, nil).NewTransaction()
+ if mergeTxn.meta.props == nil {
+ mergeTxn.meta.props = make(iceberg.Properties)
+ }
+ mergeTxn.meta.props[ManifestMergeEnabledKey] = "true"
+ mergeTxn.meta.props[ManifestMinMergeCountKey] = "2"
+ mergeTxn.meta.props[ManifestTargetSizeBytesKey] = strconv.FormatInt(
+
historicalManifests[0].Length()+historicalManifests[1].Length(), 10)
+ mergeAppend := newMergeAppendFilesProducer(OpAppend, mergeTxn, wfs,
nil, nil)
+ newFile := newTestDataFile(t, spec, "file://new-data.parquet", nil)
+ mergeAppend.appendDataFile(newFile)
+ updates, requirements, err := mergeAppend.commit(context.Background())
+ require.NoError(t, err)
+ require.NoError(t, mergeTxn.apply(updates, requirements))
+ meta, err = mergeTxn.meta.Build()
+ require.NoError(t, err)
+
+ outputManifests, err := meta.CurrentSnapshot().Manifests(wfs)
+ require.NoError(t, err)
+ require.Len(t, outputManifests, 1)
+ require.Equal(t, int32(1), outputManifests[0].AddedDataFiles())
+ var outputEntries []iceberg.ManifestEntry
+ for entry, entryErr := range outputManifests[0].Entries(wfs, false) {
+ require.NoError(t, entryErr)
+ outputEntries = append(outputEntries, entry)
+ }
+ require.Len(t, outputEntries, 1)
+ require.Equal(t, iceberg.EntryStatusADDED, outputEntries[0].Status())
+ require.Equal(t, newFile.FilePath(),
outputEntries[0].DataFile().FilePath())
+}
+
func TestOverwriteFilesExistingManifestsClosesWriterOnError(t *testing.T) {
spec := partitionedSpec()
schema := simpleSchema()
