perf(table): avoid cloning schema history for position_deletes partition IDs

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"
 )
@@ -297,7 +298,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()
@@ -317,7 +324,10 @@ func inspectPartitionType(metadata Metadata) (*iceberg.StructType, error) {
 		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

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
 	}
@@ -808,33 +810,43 @@ 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
+	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++
 	}
 

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} {

table/inspect_position_deletes_regression_test.go

@@ -20,6 +20,8 @@ package table
 import (
 	"bytes"
 	"context"
+	"fmt"
+	"math"
 	"testing"
 	"time"
 
@@ -217,8 +219,108 @@ 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)
 }
+
+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
+}
+
+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)
+}
+
+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())
+	})
+}