(iceberg-go) branch main updated: fix(table): look up removed-file partition specs by ID (#1816)

[ze...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zeroshade 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 a6f08ca91 fix(table): look up removed-file partition specs by ID (#1816)
a6f08ca91 is described below

commit a6f08ca9102427fd83901dc2cf9438f53f75b311
Author: anchor <co...@gmail.com>
AuthorDate: Sat Aug 15 00:01:45 2026 +0800

    fix(table): look up removed-file partition specs by ID (#1816)
    
    accumulateSummaryDelta resolved partition specs two different ways.
    Added files went through sp.spec(int(df.SpecID())), a by-ID lookup over
    GetSpecByID, while removed data files, delete files and deletion
    vectors indexed the spec slice directly with specs[df.SpecID()].
    
    Spec ID is an identifier, not a slice position. RemovePartitionSpecs
    filters the slice without renumbering the survivors, and specs loaded
    from metadata keep the partition-specs array order, so the two come
    apart on ordinary tables. The index expression is evaluated as a call
    argument, so it fails before removeFile ever runs: a Delete, Overwrite
    or RewriteFiles commit that removes files panics with index out of
    range, or silently attributes the change to the wrong partition when
    the ID stays in bounds but points at the wrong entry.
    
    Use sp.spec() in the three removal loops so both halves of the function
    agree, and drop the now-unused local. Unknown IDs then yield an empty
    unpartitioned spec instead of panicking, matching the add path.
    
    The regression test evolves specs 0/1/2, removes spec 1 so the slice
    becomes [id0, id2], then sends the same spec-2 data file down both
    branches and asserts they report the same changed partition.
    
    Fixes #1815
    
    Signed-off-by: codeAnqiang-ma <27...@users.noreply.github.com>
    Co-authored-by: codeAnqiang-ma <27...@users.noreply.github.com>
    Co-authored-by: Cursor <cu...@cursor.com>
---
 table/snapshot_producers.go      |  7 +++--
 table/snapshot_producers_test.go | 56 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 59 insertions(+), 4 deletions(-)

diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index a9a5ac241..ef2ebc959 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -1018,9 +1018,8 @@ func (sp *snapshotProducer) accumulateSummaryDelta(countDeleteRemoval func(icebe
 		}
 	}
 
-	specs := sp.txn.meta.specs
 	for _, df := range sp.deletedFiles {
-		if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]); err != nil {
+		if err = ssc.removeFile(df, currentSchema, sp.spec(int(df.SpecID()))); err != nil {
 			return nil, err
 		}
 	}
@@ -1028,7 +1027,7 @@ func (sp *snapshotProducer) accumulateSummaryDelta(countDeleteRemoval func(icebe
 		if countDeleteRemoval != nil && !countDeleteRemoval(df) {
 			continue
 		}
-		if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]); err != nil {
+		if err = ssc.removeFile(df, currentSchema, sp.spec(int(df.SpecID()))); err != nil {
 			return nil, err
 		}
 	}
@@ -1036,7 +1035,7 @@ func (sp *snapshotProducer) accumulateSummaryDelta(countDeleteRemoval func(icebe
 		if countDeleteRemoval != nil && !countDeleteRemoval(df) {
 			continue
 		}
-		if err = ssc.removeFile(df, currentSchema, specs[df.SpecID()]); err != nil {
+		if err = ssc.removeFile(df, currentSchema, sp.spec(int(df.SpecID()))); err != nil {
 			return nil, err
 		}
 	}
diff --git a/table/snapshot_producers_test.go b/table/snapshot_producers_test.go
index bcd0c70af..f3e492899 100644
--- a/table/snapshot_producers_test.go
+++ b/table/snapshot_producers_test.go
@@ -26,6 +26,7 @@ import (
 	"io/fs"
 	"runtime"
 	"strconv"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -1062,6 +1063,61 @@ func TestAddedDeleteManifestsUseDeleteFileSpecID(t *testing.T) {
 	require.ElementsMatch(t, []int32{0, 1}, deleteManifestSpecIDs)
 }
 
+// changedPartitionPaths returns the partition paths named by a snapshot
+// summary's changed-partition entries.
+func changedPartitionPaths(summary iceberg.Properties) map[string]bool {
+	paths := make(map[string]bool)
+	for key := range summary {
+		if path, ok := strings.CutPrefix(key, changedPartitionPrefix); ok {
+			paths[path] = true
+		}
+	}
+
+	return paths
+}
+
+// TestAccumulateSummaryDeltaResolvesRemovedFileSpecByID checks that removed
+// files resolve their partition spec by ID, the way added files already do.
+// Removing a non-default spec that precedes the default one leaves a specs
+// slice whose positions no longer line up with spec IDs, so treating a spec ID
+// as a slice index reads the wrong spec or runs off the end of the slice.
+func TestAccumulateSummaryDeltaResolvesRemovedFileSpecByID(t *testing.T) {
+	txn, wfs := createTestTransactionWithMemIO(t, iceberg.NewPartitionSpec())
+	require.NoError(t, txn.meta.SetProperties(iceberg.Properties{WritePartitionSummaryLimitKey: "10"}))
+
+	bucketSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+		SourceIDs: []int{1}, FieldID: 1000, Name: "id_bucket", Transform: iceberg.BucketTransform{NumBuckets: 4},
+	})
+	require.NoError(t, txn.meta.AddPartitionSpec(&bucketSpec, false))
+	truncSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+		SourceIDs: []int{1}, FieldID: 1001, Name: "id_trunc", Transform: iceberg.TruncateTransform{Width: 8},
+	})
+	require.NoError(t, txn.meta.AddPartitionSpec(&truncSpec, false))
+	require.NoError(t, txn.meta.SetDefaultSpecID(-1))
+	require.NoError(t, txn.meta.RemovePartitionSpecs([]int{1}))
+
+	currentSpec, err := txn.meta.CurrentSpec()
+	require.NoError(t, err)
+	require.Equal(t, 2, currentSpec.ID(), "test setup should leave spec 2 as the default")
+	require.Len(t, txn.meta.specs, 2, "removing spec 1 should decouple slice positions from spec IDs")
+
+	df := newTestDataFile(t, *currentSpec, "mem://default/table-location/data/spec-2.parquet",
+		map[int]any{currentSpec.Field(0).FieldID: int32(0)})
+
+	appended := newOverwriteFilesProducer(OpOverwrite, txn, wfs, nil, nil)
+	appended.appendDataFile(df)
+	addedSummary, err := appended.accumulateSummaryDelta(nil)
+	require.NoError(t, err)
+	require.Equal(t, map[string]bool{"id_trunc=0": true}, changedPartitionPaths(addedSummary))
+
+	deleted := newOverwriteFilesProducer(OpOverwrite, txn, wfs, nil, nil)
+	deleted.deleteDataFile(df)
+	removedSummary, err := deleted.accumulateSummaryDelta(nil)
+	require.NoError(t, err)
+	require.Equal(t, changedPartitionPaths(addedSummary), changedPartitionPaths(removedSummary),
+		"adding and removing the same file must report the same changed partition")
+}
+
 // TestCreateManifestClosesUnderlyingFile tests that createManifest properly
 // closes the underlying file writer. This is related to issue #644 and #681.
 func TestCreateManifestClosesUnderlyingFile(t *testing.T) {