From f75e992ff5229e6c83b2ce49af2e7062ca549754 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Thu, 27 Aug 2026 21:03:21 +0200 Subject: [PATCH 1/3] perf(table): avoid cloning schema history for position deletes --- table/inspect_position_deletes.go | 11 +--- table/inspect_position_deletes_bench_test.go | 53 +++++++++++++++++++ ...nspect_position_deletes_regression_test.go | 30 ++++++++++- 3 files changed, 83 insertions(+), 11 deletions(-) diff --git a/table/inspect_position_deletes.go b/table/inspect_position_deletes.go index 0ee9bc383..b84bf5079 100644 --- a/table/inspect_position_deletes.go +++ b/table/inspect_position_deletes.go @@ -808,22 +808,13 @@ func positionDeletesPartitionType( positionDeleteContentOffsetID: {}, positionDeleteContentSizeID: {}, } - for _, schema := range metadata.Schemas() { - fields, err := iceberg.IndexByID(schema) - if err != nil { - return nil, nil, err - } - for id := range fields { - used[id] = struct{}{} - } - } for _, field := range base.FieldList { used[field.ID] = struct{}{} } fields := make([]iceberg.NestedField, len(base.FieldList)) idByOld := make(map[int]int, len(base.FieldList)) - nextID := 1 + nextID := max(1, metadata.LastColumnID()+1) for index, field := range base.FieldList { for { if _, exists := used[nextID]; !exists { diff --git a/table/inspect_position_deletes_bench_test.go b/table/inspect_position_deletes_bench_test.go index 4d08e3c2b..c342d1d1b 100644 --- a/table/inspect_position_deletes_bench_test.go +++ b/table/inspect_position_deletes_bench_test.go @@ -168,6 +168,59 @@ func TestPositionDeleteAppenderDoesNotReadUnpartitionedPartition(t *testing.T) { require.Zero(t, file.partitionCalls) } +var positionDeletesPartitionTypeBenchmarkSink *iceberg.StructType + +func BenchmarkPositionDeletesPartitionType(b *testing.B) { + for _, schemaCount := range []int{1, 16, 128, 1024} { + b.Run(fmt.Sprintf("schemas=%d/fields=64", schemaCount), func(b *testing.B) { + metadata := benchmarkPositionDeletesMetadata(b, schemaCount, 64) + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + partitionType, _, err := positionDeletesPartitionType(metadata) + if err != nil { + b.Fatal(err) + } + positionDeletesPartitionTypeBenchmarkSink = partitionType + } + }) + } +} + +func benchmarkPositionDeletesMetadata(b *testing.B, schemaCount, fieldsPerSchema int) Metadata { + b.Helper() + + schemas := make([]*iceberg.Schema, schemaCount) + lastColumnID := 0 + for schemaID := range schemaCount { + fields := make([]iceberg.NestedField, fieldsPerSchema) + for fieldIndex := range fieldsPerSchema { + fieldID := schemaID*fieldsPerSchema + fieldIndex + 1 + fields[fieldIndex] = iceberg.NestedField{ + ID: fieldID, Name: fmt.Sprintf("field_%d", fieldID), + Type: iceberg.PrimitiveTypes.Int32, Required: true, + } + lastColumnID = fieldID + } + schemas[schemaID] = iceberg.NewSchema(schemaID, fields...) + } + + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{}, + }) + lastPartitionID := 1000 + + return &metadataV2{commonMetadata: commonMetadata{ + FormatVersion: 2, + LastColumnId: lastColumnID, + SchemaList: schemas, + CurrentSchemaID: 0, + Specs: []iceberg.PartitionSpec{spec}, + DefaultSpecID: spec.ID(), + LastPartitionID: &lastPartitionID, + }} +} + func BenchmarkPositionDeleteAppender(b *testing.B) { for _, fieldCount := range []int{1, 5} { for _, rowCount := range []int{1_000, 100_000} { diff --git a/table/inspect_position_deletes_regression_test.go b/table/inspect_position_deletes_regression_test.go index 97be95eac..b7e5a5346 100644 --- a/table/inspect_position_deletes_regression_test.go +++ b/table/inspect_position_deletes_regression_test.go @@ -217,8 +217,36 @@ func TestPositionDeletesPartitionTypeAvoidsHistoricalSchemaFieldIDs(t *testing.T LastPartitionID: &lastPartitionID, }} - partitionType, partitionIDs, err := positionDeletesPartitionType(metadata) + partitionType, partitionIDs, err := positionDeletesPartitionType(metadataWithoutSchemaHistory{Metadata: metadata}) require.NoError(t, err) require.Equal(t, map[int]int{1000: 3}, partitionIDs) require.Equal(t, 3, partitionType.FieldList[0].ID) } + +type metadataWithoutSchemaHistory struct { + Metadata +} + +func (metadataWithoutSchemaHistory) Schemas() []*iceberg.Schema { + panic("position delete partition type should use LastColumnID instead of schema history") +} + +func TestPositionDeletesPartitionTypeSkipsReservedIDsAfterLastColumnID(t *testing.T) { + currentSchema := simpleSchema() + spec := partitionedSpec() + lastPartitionID := 1000 + metadata := &metadataV2{commonMetadata: commonMetadata{ + FormatVersion: 2, + LastColumnId: positionDeleteContentSizeID - 1, + SchemaList: []*iceberg.Schema{currentSchema}, + CurrentSchemaID: 0, + Specs: []iceberg.PartitionSpec{spec}, + DefaultSpecID: spec.ID(), + LastPartitionID: &lastPartitionID, + }} + + partitionType, partitionIDs, err := positionDeletesPartitionType(metadataWithoutSchemaHistory{Metadata: metadata}) + require.NoError(t, err) + require.Equal(t, map[int]int{1000: positionDeleteSpecID + 1}, partitionIDs) + require.Equal(t, positionDeleteSpecID+1, partitionType.FieldList[0].ID) +} From f5052e8a2b98d51fe480e3b545c379f415029b66 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Fri, 28 Aug 2026 01:00:25 +0200 Subject: [PATCH 2/3] fix(table): guard position delete partition field IDs --- table/inspect_files.go | 8 +- table/inspect_position_deletes.go | 37 ++++++++-- ...nspect_position_deletes_regression_test.go | 74 +++++++++++++++++++ 3 files changed, 110 insertions(+), 9 deletions(-) diff --git a/table/inspect_files.go b/table/inspect_files.go index c014c6c72..264c845c9 100644 --- a/table/inspect_files.go +++ b/table/inspect_files.go @@ -297,7 +297,13 @@ func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter. // union of partition fields from every spec, which lets metadata tables // represent live files written before partition evolution. func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) { - currentSchema := metadata.CurrentSchema() + return inspectPartitionTypeWithSchema(metadata, metadata.CurrentSchema()) +} + +func inspectPartitionTypeWithSchema( + metadata Metadata, + currentSchema *iceberg.Schema, +) (*iceberg.StructType, error) { specs := metadata.PartitionSpecs() sort.Slice(specs, func(left, right int) bool { return specs[left].ID() > specs[right].ID() diff --git a/table/inspect_position_deletes.go b/table/inspect_position_deletes.go index b84bf5079..370a767b2 100644 --- a/table/inspect_position_deletes.go +++ b/table/inspect_position_deletes.go @@ -28,6 +28,7 @@ import ( "github.com/apache/arrow-go/v18/arrow/compute" "github.com/apache/arrow-go/v18/arrow/scalar" "github.com/apache/iceberg-go" + iceinternal "github.com/apache/iceberg-go/internal" iceio "github.com/apache/iceberg-go/io" "github.com/apache/iceberg-go/table/dv" tblutils "github.com/apache/iceberg-go/table/internal" @@ -793,7 +794,8 @@ func validatePositionDeleteFilePathValues(values arrow.TypedArray[string], arr a func positionDeletesPartitionType( metadata Metadata, ) (*iceberg.StructType, map[int]int, error) { - base, err := inspectPartitionType(metadata) + currentSchema := metadata.CurrentSchema() + base, err := inspectPartitionTypeWithSchema(metadata, currentSchema) if err != nil { return nil, nil, err } @@ -811,21 +813,40 @@ func positionDeletesPartitionType( for _, field := range base.FieldList { used[field.ID] = struct{}{} } - fields := make([]iceberg.NestedField, len(base.FieldList)) idByOld := make(map[int]int, len(base.FieldList)) - nextID := max(1, metadata.LastColumnID()+1) + maxFieldID := int64(math.MaxInt32) + nextID := int64(1) + wrapped := false + lastColumnID := metadata.LastColumnID() + if lastColumnID >= 0 && int64(lastColumnID) < maxFieldID { + nextID = int64(lastColumnID) + 1 + } else if int64(lastColumnID) >= maxFieldID { + wrapped = true + } for index, field := range base.FieldList { for { - if _, exists := used[nextID]; !exists { - break + if nextID > maxFieldID { + if wrapped { + return nil, nil, fmt.Errorf("%w: no field ID available for position delete partition field %q", + iceberg.ErrInvalidSchema, field.Name) + } + nextID = 1 + wrapped = true + } + fieldID := int(nextID) + if _, exists := used[fieldID]; !exists { + if _, exists := currentSchema.FindFieldByIDRef(fieldID, iceinternal.SchemaRef{}); !exists { + break + } } nextID++ } - idByOld[field.ID] = nextID - field.ID = nextID + fieldID := int(nextID) + idByOld[field.ID] = fieldID + field.ID = fieldID fields[index] = field - used[nextID] = struct{}{} + used[fieldID] = struct{}{} nextID++ } diff --git a/table/inspect_position_deletes_regression_test.go b/table/inspect_position_deletes_regression_test.go index b7e5a5346..605198895 100644 --- a/table/inspect_position_deletes_regression_test.go +++ b/table/inspect_position_deletes_regression_test.go @@ -20,6 +20,8 @@ package table import ( "bytes" "context" + "fmt" + "math" "testing" "time" @@ -223,6 +225,34 @@ func TestPositionDeletesPartitionTypeAvoidsHistoricalSchemaFieldIDs(t *testing.T require.Equal(t, 3, partitionType.FieldList[0].ID) } +func TestPositionDeletesPartitionTypeAvoidsCurrentSchemaFieldIDs(t *testing.T) { + currentSchema := iceberg.NewSchema(0, iceberg.NestedField{ + ID: 2, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true, + }) + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{2}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{}, + }) + lastPartitionID := 1000 + metadata := &metadataV2{commonMetadata: commonMetadata{ + FormatVersion: 2, + UUID: uuid.New(), + LastColumnId: 1, + SchemaList: []*iceberg.Schema{currentSchema}, + CurrentSchemaID: 0, + Specs: []iceberg.PartitionSpec{spec}, + DefaultSpecID: spec.ID(), + LastPartitionID: &lastPartitionID, + }} + + partitionType, partitionIDs, err := positionDeletesPartitionType(metadataWithoutSchemaHistory{Metadata: metadata}) + require.NoError(t, err) + require.Equal(t, map[int]int{1000: 3}, partitionIDs) + require.Equal(t, 3, partitionType.FieldList[0].ID) + require.NotPanics(t, func() { + _ = PositionDeletesSchema(currentSchema, partitionType, metadata.Version()) + }) +} + type metadataWithoutSchemaHistory struct { Metadata } @@ -250,3 +280,47 @@ func TestPositionDeletesPartitionTypeSkipsReservedIDsAfterLastColumnID(t *testin require.Equal(t, map[int]int{1000: positionDeleteSpecID + 1}, partitionIDs) require.Equal(t, positionDeleteSpecID+1, partitionType.FieldList[0].ID) } + +func TestPositionDeletesPartitionTypeFallsBackWhenFieldIDsReachLimit(t *testing.T) { + const partitionFieldCount = 193 + + currentFields := make([]iceberg.NestedField, partitionFieldCount) + partitionFields := make([]iceberg.PartitionField, partitionFieldCount) + for index := range partitionFieldCount { + fieldID := index + 1 + currentFields[index] = iceberg.NestedField{ + ID: fieldID, Name: fmt.Sprintf("field_%d", fieldID), + Type: iceberg.PrimitiveTypes.Int32, Required: true, + } + partitionFields[index] = iceberg.PartitionField{ + SourceIDs: []int{fieldID}, FieldID: 1000 + index, + Name: fmt.Sprintf("partition_%d", fieldID), Transform: iceberg.IdentityTransform{}, + } + } + currentSchema := iceberg.NewSchema(0, currentFields...) + spec := iceberg.NewPartitionSpec(partitionFields...) + lastPartitionID := 1000 + partitionFieldCount - 1 + metadata := &metadataV2{commonMetadata: commonMetadata{ + FormatVersion: 2, + UUID: uuid.New(), + LastColumnId: iceberg.MaxStructFieldID, + SchemaList: []*iceberg.Schema{currentSchema}, + CurrentSchemaID: 0, + Specs: []iceberg.PartitionSpec{spec}, + DefaultSpecID: spec.ID(), + LastPartitionID: &lastPartitionID, + }} + + partitionType, partitionIDs, err := positionDeletesPartitionType(metadataWithoutSchemaHistory{Metadata: metadata}) + require.NoError(t, err) + require.Len(t, partitionType.FieldList, partitionFieldCount) + require.Equal(t, iceberg.MaxStructFieldID+1, partitionIDs[1000]) + require.Equal(t, 194, partitionIDs[1000+partitionFieldCount-1]) + for _, field := range partitionType.FieldList { + require.GreaterOrEqual(t, field.ID, 1) + require.LessOrEqual(t, field.ID, math.MaxInt32) + } + require.NotPanics(t, func() { + _ = PositionDeletesSchema(currentSchema, partitionType, metadata.Version()) + }) +} From b1c166a8c88cd7a5546efa296663ea12c964a1f8 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Fri, 28 Aug 2026 15:43:12 +0200 Subject: [PATCH 3/3] perf(table): avoid cloning partition source types --- table/inspect_files.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/table/inspect_files.go b/table/inspect_files.go index 264c845c9..0a7fb95ed 100644 --- a/table/inspect_files.go +++ b/table/inspect_files.go @@ -31,6 +31,7 @@ import ( "github.com/apache/arrow-go/v18/arrow/memory" "github.com/apache/arrow-go/v18/arrow/scalar" "github.com/apache/iceberg-go" + iceinternal "github.com/apache/iceberg-go/internal" iceio "github.com/apache/iceberg-go/io" "github.com/google/uuid" ) @@ -323,7 +324,10 @@ func inspectPartitionTypeWithSchema( for idx, field := range spec.Fields() { active := true for _, sourceID := range field.SourceIDs { - if _, ok := currentSchema.FindTypeByID(sourceID); !ok { + // This is an existence check only. FindTypeByID clones the + // complete nested type on every call, which is wasteful while + // building the union of position-delete partition fields. + if _, ok := currentSchema.FindFieldByIDRef(sourceID, iceinternal.SchemaRef{}); !ok { active = false break