perf(metadata): avoid re-cloning defensive schema copies

catalog/hive/schema.go

@@ -33,9 +33,10 @@ func schemaToHiveColumns(schema *iceberg.Schema) []*hive_metastore.FieldSchema {
 		return nil
 	}
 
-	columns := make([]*hive_metastore.FieldSchema, 0, len(schema.Fields()))
-	for _, field := range schema.Fields() {
-		columns = append(columns, fieldToHiveColumn(field))
+	fields := schema.Fields()
+	columns := make([]*hive_metastore.FieldSchema, len(fields))
+	for i, field := range fields {
+		columns[i] = fieldToHiveColumn(field)
 	}
 
 	return columns

metadata_columns.go

@@ -17,10 +17,7 @@
 
 package iceberg
 
-import (
-	"math"
-	"slices"
-)
+import "math"
 
 // MaxStructFieldID is the largest field ID a user-supplied schema may assign.
 // Field IDs greater than this are reserved by the spec for metadata columns
@@ -94,7 +91,7 @@ func SchemaWithRowLineageColumns(s *Schema, rowID, lastUpdatedSeq bool) *Schema
 	if s == nil {
 		return nil
 	}
-	fields := slices.Clone(s.Fields())
+	fields := s.Fields()
 
 	hasRowID := false
 	hasSeqNum := false

partitions.go

@@ -543,6 +543,16 @@ func NewPartitionSpecID(id int, fields ...PartitionField) PartitionSpec {
 	return ret
 }
 
+func (ps PartitionSpec) Clone() PartitionSpec {
+	clone := PartitionSpec{id: ps.id, fields: make([]PartitionField, len(ps.fields))}
+	for i, field := range ps.fields {
+		clone.fields[i] = clonePartitionField(field)
+	}
+	clone.initialize()
+
+	return clone
+}
+
 // CompatibleWith returns true if this partition spec is considered
 // compatible with the passed in partition spec. This means that the two
 // specs have equivalent field lists regardless of the spec id.

partitions_test.go

@@ -123,6 +123,23 @@ func TestNewPartitionSpecIDCopiesFields(t *testing.T) {
 	assert.Equal(t, []int{1}, restored.SourceIDs)
 }
 
+func TestPartitionSpecCloneCopiesFields(t *testing.T) {
+	transform := &iceberg.BucketTransform{NumBuckets: 16}
+	spec := iceberg.NewPartitionSpecID(7, iceberg.PartitionField{
+		SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: transform,
+	})
+
+	clone := spec.Clone()
+	field := clone.Field(0)
+	field.SourceIDs[0] = 2
+	field.Transform.(*iceberg.BucketTransform).NumBuckets = 32
+
+	assert.True(t, spec.Equals(clone))
+	assert.Equal(t, []int{1}, spec.Field(0).SourceIDs)
+	assert.Equal(t, 16, spec.Field(0).Transform.(*iceberg.BucketTransform).NumBuckets)
+	assert.Len(t, clone.FieldsBySourceID(1), 1)
+}
+
 func TestPartitionSpecCompatibleWithUsesTransformEquals(t *testing.T) {
 	spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
 		SourceIDs: []int{1}, FieldID: 1001, Name: "id",

table/metadata.go

@@ -2404,7 +2404,7 @@ func cloneSchema(schema *iceberg.Schema) *iceberg.Schema {
 	return iceberg.NewSchemaWithIdentifiers(
 		schema.ID,
 		slices.Clone(schema.IdentifierFieldIDs),
-		cloneNestedFields(schema.Fields())...,
+		schema.Fields()...,
 	)
 }
 
@@ -2421,47 +2421,8 @@ func cloneSchemas(schemas []*iceberg.Schema) []*iceberg.Schema {
 	return clones
 }
 
-func cloneNestedFields(fields []iceberg.NestedField) []iceberg.NestedField {
-	clones := slices.Clone(fields)
-	for i := range clones {
-		clones[i].Type = cloneSchemaType(clones[i].Type)
-		clones[i].InitialDefault = iceberg.CloneDefaultValue(clones[i].InitialDefault)
-		clones[i].WriteDefault = iceberg.CloneDefaultValue(clones[i].WriteDefault)
-	}
-
-	return clones
-}
-
-func cloneSchemaType(typ iceberg.Type) iceberg.Type {
-	switch typ := typ.(type) {
-	case *iceberg.StructType:
-		return &iceberg.StructType{FieldList: cloneNestedFields(typ.FieldList)}
-	case *iceberg.ListType:
-		return &iceberg.ListType{
-			ElementID:       typ.ElementID,
-			Element:         cloneSchemaType(typ.Element),
-			ElementRequired: typ.ElementRequired,
-		}
-	case *iceberg.MapType:
-		return &iceberg.MapType{
-			KeyID:         typ.KeyID,
-			KeyType:       cloneSchemaType(typ.KeyType),
-			ValueID:       typ.ValueID,
-			ValueType:     cloneSchemaType(typ.ValueType),
-			ValueRequired: typ.ValueRequired,
-		}
-	default:
-		return typ
-	}
-}
-
 func clonePartitionSpec(spec iceberg.PartitionSpec) iceberg.PartitionSpec {
-	fields := make([]iceberg.PartitionField, spec.NumFields())
-	for i := range fields {
-		fields[i] = spec.Field(i)
-	}
-
-	return iceberg.NewPartitionSpecID(spec.ID(), fields...)
+	return spec.Clone()
 }
 
 func clonePartitionSpecs(specs []iceberg.PartitionSpec) []iceberg.PartitionSpec {
@@ -2575,8 +2536,7 @@ func cloneSortOrder(order SortOrder) SortOrder {
 	clone := order
 	clone.fields = make([]SortField, len(order.fields))
 	for i, field := range order.fields {
-		clone.fields[i] = field
-		clone.fields[i].SourceIDs = slices.Clone(field.SourceIDs)
+		clone.fields[i] = cloneSortField(field)
 	}
 
 	return clone

table/metadata_getters_test.go

@@ -23,6 +23,7 @@ import (
 	"testing"
 
 	"github.com/apache/iceberg-go"
+	iceinternal "github.com/apache/iceberg-go/internal"
 	"github.com/stretchr/testify/require"
 )
 
@@ -42,7 +43,7 @@ func TestMetadataGettersReturnDefensiveCopies(t *testing.T) {
 			WriteDefault:   iceberg.FixedLiteral{3, 4},
 		})},
 		Specs: []iceberg.PartitionSpec{iceberg.NewPartitionSpecID(1, iceberg.PartitionField{
-			SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{},
+			SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: &iceberg.BucketTransform{NumBuckets: 16},
 		})},
 		SnapshotList: []Snapshot{{
 			SnapshotID:       2,
@@ -87,7 +88,7 @@ func TestMetadataGettersReturnDefensiveCopies(t *testing.T) {
 			orderID: 1,
 			fields: []SortField{{
 				SourceIDs: []int{10},
-				Transform: iceberg.IdentityTransform{},
+				Transform: &iceberg.BucketTransform{NumBuckets: 16},
 			}},
 		}},
 	}
@@ -111,7 +112,9 @@ func TestMetadataGettersReturnDefensiveCopies(t *testing.T) {
 	partitionField := partitionSpecs[0].Field(0)
 	partitionField.SourceIDs[0] = 99
 	partitionField.Name = "mutated"
+	partitionField.Transform.(*iceberg.BucketTransform).NumBuckets = 32
 	require.Equal(t, []int{1}, metadata.Specs[0].Field(0).SourceIDs)
+	require.Equal(t, 16, metadata.Specs[0].Field(0).Transform.(*iceberg.BucketTransform).NumBuckets)
 
 	currentSchema := metadata.CurrentSchema()
 	currentSchema.ID = 100
@@ -158,10 +161,73 @@ func TestMetadataGettersReturnDefensiveCopies(t *testing.T) {
 	fields := sortOrders[0].Fields()
 	for _, field := range fields {
 		field.SourceIDs[0] = 99
+		field.Transform.(*iceberg.BucketTransform).NumBuckets = 32
 	}
 	require.Equal(t, []int{10}, metadata.SortOrderList[0].fields[0].SourceIDs)
+	require.Equal(t, 16, metadata.SortOrderList[0].fields[0].Transform.(*iceberg.BucketTransform).NumBuckets)
 
 	got, err := json.Marshal(metadata)
 	require.NoError(t, err)
 	require.JSONEq(t, string(original), string(got))
 }
+
+func TestMetadataSchemaGetterCopiesNestedValues(t *testing.T) {
+	schema := nestedSchemaWithMutableDefaults()
+	metadata := commonMetadata{
+		CurrentSchemaID: schema.ID,
+		SchemaList:      []*iceberg.Schema{schema},
+	}
+	originalFields := schema.Fields()
+
+	cloned := metadata.CurrentSchema()
+	cloned.IdentifierFieldIDs[0] = 99
+	fields := cloned.FieldsRef(iceinternal.SchemaRef{})
+	payload := fields[0].Type.(*iceberg.StructType)
+	payload.FieldList[0].InitialDefault.([]byte)[0] = 99
+	payload.FieldList[0].WriteDefault.(iceberg.BinaryLiteral)[0] = 99
+	payload.FieldList[1].Name = "changed"
+	payload.FieldList[1].Type.(*iceberg.ListType).Element.(*iceberg.StructType).FieldList[0].InitialDefault.([]any)[0].(map[string]any)["bytes"].([]byte)[0] = 99
+	payload.FieldList[2].Type.(*iceberg.MapType).ValueType.(*iceberg.StructType).FieldList[0].WriteDefault.(map[string]any)["values"].([]any)[0] = "changed"
+
+	require.Equal(t, []int{1}, schema.IdentifierFieldIDs)
+	require.Equal(t, originalFields, schema.Fields())
+}
+
+var cloneSchemaBenchmarkSink *iceberg.Schema
+
+func BenchmarkCloneSchemaWithNestedDefaults(b *testing.B) {
+	schema := nestedSchemaWithMutableDefaults()
+	b.ReportAllocs()
+	b.ResetTimer()
+	for range b.N {
+		cloneSchemaBenchmarkSink = cloneSchema(schema)
+	}
+}
+
+func nestedSchemaWithMutableDefaults() *iceberg.Schema {
+	listElement := &iceberg.StructType{FieldList: []iceberg.NestedField{{
+		ID: 5, Name: "element", Type: iceberg.PrimitiveTypes.String,
+		InitialDefault: []any{map[string]any{"bytes": []byte{7, 8}}},
+	}}}
+	mapValue := &iceberg.StructType{FieldList: []iceberg.NestedField{{
+		ID: 9, Name: "value", Type: iceberg.PrimitiveTypes.String,
+		WriteDefault: map[string]any{"values": []any{iceberg.FixedLiteral{10, 11}}},
+	}}}
+	payload := &iceberg.StructType{FieldList: []iceberg.NestedField{
+		{
+			ID: 2, Name: "binary", Type: iceberg.PrimitiveTypes.Binary,
+			InitialDefault: []byte{1, 2, 3}, WriteDefault: iceberg.BinaryLiteral{4, 5, 6},
+		},
+		{ID: 3, Name: "list", Type: &iceberg.ListType{
+			ElementID: 4, Element: listElement, ElementRequired: true,
+		}},
+		{ID: 6, Name: "map", Type: &iceberg.MapType{
+			KeyID: 7, KeyType: iceberg.PrimitiveTypes.String,
+			ValueID: 8, ValueType: mapValue, ValueRequired: false,
+		}},
+	}}
+
+	return iceberg.NewSchemaWithIdentifiers(1, []int{1}, iceberg.NestedField{
+		ID: 1, Name: "payload", Type: payload,
+	})
+}

table/metadata_partition_bench_test.go

@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+)
+
+var clonePartitionSpecsBenchmarkSink []iceberg.PartitionSpec
+
+func BenchmarkClonePartitionSpecs(b *testing.B) {
+	for _, fieldCount := range []int{1, 8, 32} {
+		b.Run("fields="+strconv.Itoa(fieldCount), func(b *testing.B) {
+			specs := []iceberg.PartitionSpec{
+				iceberg.NewPartitionSpecID(1, partitionSpecCloneBenchmarkFields(fieldCount)...),
+			}
+			b.ReportAllocs()
+			b.ResetTimer()
+			for range b.N {
+				clonePartitionSpecsBenchmarkSink = clonePartitionSpecs(specs)
+			}
+		})
+	}
+}
+
+func partitionSpecCloneBenchmarkFields(count int) []iceberg.PartitionField {
+	fields := make([]iceberg.PartitionField, count)
+	for i := range fields {
+		fields[i] = iceberg.PartitionField{
+			SourceIDs: []int{i + 1}, FieldID: i + 1000,
+			Name: "field", Transform: iceberg.IdentityTransform{},
+		}
+	}
+
+	return fields
+}

table/scanner.go

@@ -758,12 +758,12 @@ func splitLineageMetadataFields(selectedFields []string, caseSensitive bool) (us
 // appended only if no field with that ID is already present. Idempotent so
 // callers can pass schemas that already declare the reserved fields.
 func appendMissingLineageFields(s *iceberg.Schema, lineageFields []iceberg.NestedField) *iceberg.Schema {
-	existing := make(map[int]struct{}, len(s.Fields()))
-	for _, f := range s.Fields() {
+	fields := s.Fields()
+	existing := make(map[int]struct{}, len(fields))
+	for _, f := range fields {
 		existing[f.ID] = struct{}{}
 	}
 
-	fields := slices.Clone(s.Fields())
 	for _, f := range lineageFields {
 		if _, ok := existing[f.ID]; ok {
 			continue

table/scanner_internal_test.go

@@ -998,6 +998,19 @@ func TestSplitLineageMetadataFields(t *testing.T) {
 	}
 }
 
+func TestAppendMissingLineageFields(t *testing.T) {
+	schema := iceberg.NewSchemaWithIdentifiers(1, []int{1}, iceberg.NestedField{
+		ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64,
+	})
+
+	withLineage := appendMissingLineageFields(schema, []iceberg.NestedField{iceberg.RowID()})
+
+	assert.Equal(t, []int{1}, schema.IdentifierFieldIDs)
+	assert.Len(t, schema.Fields(), 1)
+	assert.Len(t, withLineage.Fields(), 2)
+	assert.Equal(t, iceberg.RowIDColumnName, withLineage.Fields()[1].Name)
+}
+
 func TestKeyDefaultMapRaceCondition(t *testing.T) {
 	var factoryCallCount atomic.Int64
 	factory := func(key string) int {

view/metadata.go

@@ -422,7 +422,7 @@ func cloneSchema(schema *iceberg.Schema) *iceberg.Schema {
 	return iceberg.NewSchemaWithIdentifiers(
 		schema.ID,
 		slices.Clone(schema.IdentifierFieldIDs),
-		cloneNestedFields(schema.Fields())...,
+		schema.Fields()...,
 	)
 }
 
@@ -439,40 +439,6 @@ func cloneSchemas(schemas []*iceberg.Schema) []*iceberg.Schema {
 	return clones
 }
 
-func cloneNestedFields(fields []iceberg.NestedField) []iceberg.NestedField {
-	clones := slices.Clone(fields)
-	for i := range clones {
-		clones[i].Type = cloneSchemaType(clones[i].Type)
-		clones[i].InitialDefault = iceberg.CloneDefaultValue(clones[i].InitialDefault)
-		clones[i].WriteDefault = iceberg.CloneDefaultValue(clones[i].WriteDefault)
-	}
-
-	return clones
-}
-
-func cloneSchemaType(typ iceberg.Type) iceberg.Type {
-	switch typ := typ.(type) {
-	case *iceberg.StructType:
-		return &iceberg.StructType{FieldList: cloneNestedFields(typ.FieldList)}
-	case *iceberg.ListType:
-		return &iceberg.ListType{
-			ElementID:       typ.ElementID,
-			Element:         cloneSchemaType(typ.Element),
-			ElementRequired: typ.ElementRequired,
-		}
-	case *iceberg.MapType:
-		return &iceberg.MapType{
-			KeyID:         typ.KeyID,
-			KeyType:       cloneSchemaType(typ.KeyType),
-			ValueID:       typ.ValueID,
-			ValueType:     cloneSchemaType(typ.ValueType),
-			ValueRequired: typ.ValueRequired,
-		}
-	default:
-		return typ
-	}
-}
-
 func (m *metadata) validate() error {
 	if m.Loc == "" {
 		return fmt.Errorf("%w: location is required", ErrInvalidViewMetadata)