From 178fc767fe4a2ad62c6e39e51ef0a7b5483a3f78 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 29 Aug 2026 13:04:25 +0200 Subject: [PATCH 1/3] perf(table): index schemas by ID Signed-off-by: Minh Vu --- table/metadata.go | 143 +++++++++++++- table/metadata_builder_internal_test.go | 1 + table/schema_index_bench_test.go | 156 +++++++++++++++ table/schema_index_test.go | 253 ++++++++++++++++++++++++ 4 files changed, 545 insertions(+), 8 deletions(-) create mode 100644 table/schema_index_bench_test.go create mode 100644 table/schema_index_test.go diff --git a/table/metadata.go b/table/metadata.go index 1c71860da..efcf5baf3 100644 --- a/table/metadata.go +++ b/table/metadata.go @@ -142,6 +142,92 @@ func snapshotIndexPosition(index *snapshotIndexData, snapshots []Snapshot, id in return 0, false } +type schemaIndexData struct { + schemas map[int]*iceberg.Schema + // sourceCount is kept separately because duplicate IDs make the map + // smaller than the schema slice in invalid or in-package fixture state. + sourceCount int + // firstSchemaSlot identifies the slice used to build the index. + firstSchemaSlot **iceberg.Schema + // shared means the map is owned by more than one builder or metadata value. + shared bool +} + +func schemaListFirst(schemas []*iceberg.Schema) **iceberg.Schema { + if len(schemas) == 0 { + return nil + } + + return &schemas[0] +} + +func buildSchemaIndex(schemas []*iceberg.Schema) *schemaIndexData { + byID := make(map[int]*iceberg.Schema, len(schemas)) + for _, schema := range schemas { + if schema == nil { + continue + } + if _, exists := byID[schema.ID]; !exists { + byID[schema.ID] = schema + } + } + + return &schemaIndexData{ + schemas: byID, + sourceCount: len(schemas), + firstSchemaSlot: schemaListFirst(schemas), + } +} + +func cloneSchemaIndex(index *schemaIndexData) *schemaIndexData { + if index == nil { + return nil + } + + return &schemaIndexData{ + schemas: maps.Clone(index.schemas), + sourceCount: index.sourceCount, + firstSchemaSlot: index.firstSchemaSlot, + } +} + +func schemaIndexNeedsRebuild(index *schemaIndexData, schemas []*iceberg.Schema) bool { + if index == nil || index.sourceCount != len(schemas) { + return true + } + + return index.firstSchemaSlot != schemaListFirst(schemas) +} + +func schemaIndexLookup(index *schemaIndexData, schemas []*iceberg.Schema, id int) (*iceberg.Schema, bool) { + if index != nil { + if schema, ok := index.schemas[id]; ok { + if schema != nil && schema.ID == id { + return schema, true + } + + index = buildSchemaIndex(schemas) + if schema, ok := index.schemas[id]; ok && schema != nil && schema.ID == id { + return schema, true + } + + return nil, false + } + + if !schemaIndexNeedsRebuild(index, schemas) { + return nil, false + } + } + + for _, schema := range schemas { + if schema != nil && schema.ID == id { + return schema, true + } + } + + return nil, false +} + // Metadata for an iceberg table as specified in the Iceberg spec // // https://iceberg.apache.org/spec/#iceberg-table-spec @@ -257,6 +343,7 @@ type MetadataBuilder struct { lastUpdatedMS int64 lastColumnId int schemaList []*iceberg.Schema + schemaIndex *schemaIndexData // Derived from schemaList; not serialized. currentSchemaID int specs []iceberg.PartitionSpec defaultSpecID int @@ -293,6 +380,7 @@ func NewMetadataBuilder(formatVersion int) (*MetadataBuilder, error) { return &MetadataBuilder{ updates: make([]Update, 0), schemaList: make([]*iceberg.Schema, 0), + schemaIndex: buildSchemaIndex(nil), specs: make([]iceberg.PartitionSpec, 0), props: make(iceberg.Properties), snapshotList: make([]Snapshot, 0), @@ -405,6 +493,7 @@ func MetadataBuilderFromBase(metadata Metadata, currentFileLocation string) (*Me b.partitionStatsList = slices.Collect(metadata.PartitionStatistics()) b.encryptionKeyList = slices.Collect(metadata.EncryptionKeys()) } + b.schemaIndex = buildSchemaIndex(b.schemaList) if currentFileLocation != "" { b.previousFileEntry = &MetadataLogEntry{ @@ -466,6 +555,15 @@ func (b *MetadataBuilder) clone() *MetadataBuilder { lastAddedPartitionID: clonePtr(b.lastAddedPartitionID), lastAddedSortOrderID: clonePtr(b.lastAddedSortOrderID), } + if b.schemaIndex != nil { + cloned.schemaIndex = &schemaIndexData{ + schemas: b.schemaIndex.schemas, + sourceCount: b.schemaIndex.sourceCount, + firstSchemaSlot: schemaListFirst(cloned.schemaList), + shared: true, + } + b.schemaIndex.shared = true + } if b.snapshotIndex != nil { cloned.snapshotIndex = &snapshotIndexData{ positions: b.snapshotIndex.positions, @@ -521,6 +619,19 @@ func (b *MetadataBuilder) newSnapshotID() int64 { } } +func (b *MetadataBuilder) ensureSchemaIndex() { + if schemaIndexNeedsRebuild(b.schemaIndex, b.schemaList) { + b.schemaIndex = buildSchemaIndex(b.schemaList) + } +} + +func (b *MetadataBuilder) ensureSchemaIndexMutable() { + b.ensureSchemaIndex() + if b.schemaIndex.shared { + b.schemaIndex = cloneSchemaIndex(b.schemaIndex) + } +} + func (b *MetadataBuilder) ensureSnapshotIndex() { if snapshotIndexNeedsRebuild(b.snapshotIndex, b.snapshotList) { b.snapshotIndex = buildSnapshotIndex(b.snapshotList) @@ -590,7 +701,11 @@ func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema) error { schema.ID = newSchemaID + b.ensureSchemaIndexMutable() b.schemaList = append(b.schemaList, schema) + b.schemaIndex.schemas[newSchemaID] = schema + b.schemaIndex.sourceCount = len(b.schemaList) + b.schemaIndex.firstSchemaSlot = schemaListFirst(b.schemaList) b.updates = append(b.updates, NewAddSchemaUpdate(schema)) b.lastAddedSchemaID = &newSchemaID @@ -1250,6 +1365,10 @@ func (b *MetadataBuilder) SetLastUpdatedMS() *MetadataBuilder { } func (b *MetadataBuilder) buildCommonMetadata() (*commonMetadata, error) { + b.ensureSchemaIndex() + if b.schemaIndex != nil { + b.schemaIndex.shared = true + } b.ensureSnapshotIndex() if b.snapshotIndex != nil { b.snapshotIndex.shared = true @@ -1283,6 +1402,7 @@ func (b *MetadataBuilder) buildCommonMetadata() (*commonMetadata, error) { LastUpdatedMS: b.lastUpdatedMS, LastColumnId: b.lastColumnId, SchemaList: b.schemaList, + schemaIndex: b.schemaIndex, CurrentSchemaID: b.currentSchemaID, Specs: b.specs, DefaultSpecID: defaultSpecID, @@ -1350,10 +1470,12 @@ func (b *MetadataBuilder) updateSnapshotLog() error { } func (b *MetadataBuilder) GetSchemaByID(id int) (*iceberg.Schema, error) { - for _, s := range b.schemaList { - if s.ID == id { - return s, nil - } + index := b.schemaIndex + if schemaIndexNeedsRebuild(index, b.schemaList) { + index = buildSchemaIndex(b.schemaList) + } + if schema, ok := schemaIndexLookup(index, b.schemaList, id); ok { + return schema, nil } return nil, fmt.Errorf("%w: schema with id %d not found", iceberg.ErrInvalidArgument, id) @@ -1598,6 +1720,7 @@ func (b *MetadataBuilder) RemoveSchemas(ints []int) error { }) if len(removed) != 0 { + b.schemaIndex = buildSchemaIndex(b.schemaList) b.updates = append(b.updates, NewRemoveSchemasUpdate(removed)) } @@ -1957,6 +2080,7 @@ type commonMetadata struct { // V3+ fields NextRowID *int64 `json:"next-row-id,omitempty"` // V3: Next available row ID + schemaIndex *schemaIndexData snapshotIndex *snapshotIndexData } @@ -2121,10 +2245,12 @@ func (c *commonMetadata) CurrentSchema() *iceberg.Schema { // versions through commonMetadata and is intentionally not part of Metadata, // so custom Metadata implementations keep the public Schemas fallback. func (c *commonMetadata) schemaByID(id int) *iceberg.Schema { - for _, schema := range c.SchemaList { - if schema.ID == id { - return cloneSchema(schema) - } + index := c.schemaIndex + if schemaIndexNeedsRebuild(index, c.SchemaList) { + index = buildSchemaIndex(c.SchemaList) + } + if schema, ok := schemaIndexLookup(index, c.SchemaList, id); ok { + return cloneSchema(schema) } return nil @@ -2520,6 +2646,7 @@ func (c *commonMetadata) preValidate() { c.SnapshotLog = []SnapshotLogEntry{} } + c.schemaIndex = buildSchemaIndex(c.SchemaList) c.snapshotIndex = buildSnapshotIndex(c.SnapshotList) } diff --git a/table/metadata_builder_internal_test.go b/table/metadata_builder_internal_test.go index b0a3ffe55..053378ece 100644 --- a/table/metadata_builder_internal_test.go +++ b/table/metadata_builder_internal_test.go @@ -3446,6 +3446,7 @@ func TestSetFormatVersionV2ToV3FromDeserializedMetadata(t *testing.T) { // both the drift guard and its filler consult it. var sharedCloneFields = map[string]struct{}{ "base": {}, // immutable snapshot, shared by design. + "schemaIndex": {}, // immutable schema references, copied on schema mutation. "snapshotIndex": {}, // immutable index positions, copied on snapshot mutation. } diff --git a/table/schema_index_bench_test.go b/table/schema_index_bench_test.go new file mode 100644 index 000000000..3499376fd --- /dev/null +++ b/table/schema_index_bench_test.go @@ -0,0 +1,156 @@ +// 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 ( + "fmt" + "strconv" + "testing" + + "github.com/apache/iceberg-go" +) + +var schemaLookupBenchmarkSink int + +func BenchmarkCommonMetadataSchemaByID(b *testing.B) { + for _, schemaCount := range []int{1, 16, 128, 1_024, 8_192} { + schemas := schemaIndexBenchmarkSchemas(schemaCount) + metadata := commonMetadata{ + SchemaList: schemas, + schemaIndex: buildSchemaIndex(schemas), + } + + for _, tt := range []struct { + name string + id int + }{ + {name: "first", id: schemas[0].ID}, + {name: "middle", id: schemas[schemaCount/2].ID}, + {name: "last", id: schemas[schemaCount-1].ID}, + {name: "miss", id: -1}, + } { + b.Run(schemaCountName(schemaCount)+"/"+tt.name+"/indexed", func(b *testing.B) { + benchmarkSchemaLookup(b, metadata.schemaByID, tt.id, schemaCount) + }) + b.Run(schemaCountName(schemaCount)+"/"+tt.name+"/linear", func(b *testing.B) { + benchmarkSchemaLookup(b, func(id int) *iceberg.Schema { + return linearMetadataSchemaByID(schemas, id) + }, tt.id, schemaCount) + }) + } + } +} + +func BenchmarkMetadataBuilderGetSchemaByID(b *testing.B) { + for _, schemaCount := range []int{1, 16, 128, 1_024, 8_192} { + schemas := schemaIndexBenchmarkSchemas(schemaCount) + builder := MetadataBuilder{ + schemaList: schemas, + schemaIndex: buildSchemaIndex(schemas), + } + + for _, tt := range []struct { + name string + id int + }{ + {name: "first", id: schemas[0].ID}, + {name: "middle", id: schemas[schemaCount/2].ID}, + {name: "last", id: schemas[schemaCount-1].ID}, + {name: "miss", id: -1}, + } { + b.Run(schemaCountName(schemaCount)+"/"+tt.name+"/indexed", func(b *testing.B) { + benchmarkBuilderSchemaLookup(b, builder.GetSchemaByID, tt.id, schemaCount) + }) + b.Run(schemaCountName(schemaCount)+"/"+tt.name+"/linear", func(b *testing.B) { + benchmarkBuilderSchemaLookup(b, func(id int) (*iceberg.Schema, error) { + return linearBuilderSchemaByID(schemas, id) + }, tt.id, schemaCount) + }) + } + } +} + +func benchmarkSchemaLookup(b *testing.B, lookup func(int) *iceberg.Schema, id, schemaCount int) { + b.Helper() + b.ReportAllocs() + b.ReportMetric(float64(schemaCount), "schemas") + b.ResetTimer() + for range b.N { + schema := lookup(id) + if schema == nil { + schemaLookupBenchmarkSink = -1 + } else { + schemaLookupBenchmarkSink = schema.ID + } + } +} + +func benchmarkBuilderSchemaLookup(b *testing.B, lookup func(int) (*iceberg.Schema, error), id, schemaCount int) { + b.Helper() + b.ReportAllocs() + b.ReportMetric(float64(schemaCount), "schemas") + b.ResetTimer() + for range b.N { + schema, err := lookup(id) + if err != nil { + schemaLookupBenchmarkSink = -1 + } else { + schemaLookupBenchmarkSink = schema.ID + } + } +} + +func linearMetadataSchemaByID(schemas []*iceberg.Schema, id int) *iceberg.Schema { + schema := linearSchemaByID(schemas, id) + if schema == nil { + return nil + } + + return cloneSchema(schema) +} + +func linearSchemaByID(schemas []*iceberg.Schema, id int) *iceberg.Schema { + for _, schema := range schemas { + if schema != nil && schema.ID == id { + return schema + } + } + + return nil +} + +func linearBuilderSchemaByID(schemas []*iceberg.Schema, id int) (*iceberg.Schema, error) { + if schema := linearSchemaByID(schemas, id); schema != nil { + return schema, nil + } + + return nil, fmt.Errorf("%w: schema with id %d not found", iceberg.ErrInvalidArgument, id) +} + +func schemaIndexBenchmarkSchemas(count int) []*iceberg.Schema { + schemas := make([]*iceberg.Schema, count) + for i := range count { + schemas[i] = iceberg.NewSchema(i) + } + + return schemas +} + +func schemaCountName(count int) string { + return "schemas=" + strconv.Itoa(count) +} diff --git a/table/schema_index_test.go b/table/schema_index_test.go new file mode 100644 index 000000000..30e926ea9 --- /dev/null +++ b/table/schema_index_test.go @@ -0,0 +1,253 @@ +// 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 ( + "sync" + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCommonMetadataSchemaIndexLookups(t *testing.T) { + schemas := schemaIndexTestSchemas(10, 20, 42) + metadata := commonMetadata{ + SchemaList: schemas, + CurrentSchemaID: 42, + schemaIndex: buildSchemaIndex(schemas), + } + + got := metadata.schemaByID(20) + require.NotNil(t, got) + assert.Equal(t, 20, got.ID) + + current := metadata.CurrentSchema() + require.NotNil(t, current) + assert.Equal(t, 42, current.ID) + assert.Nil(t, metadata.schemaByID(99)) +} + +func TestParsedMetadataBuildsSchemaIndex(t *testing.T) { + metadata, err := ParseMetadataBytes([]byte(ExampleTableMetadataV2)) + require.NoError(t, err) + + common := metadataCommon(metadata) + require.Len(t, common.schemaIndex.schemas, len(common.SchemaList)) + for _, schema := range common.SchemaList { + assert.Same(t, schema, common.schemaIndex.schemas[schema.ID]) + } +} + +func TestMetadataBuilderFromBaseBuildsSchemaIndex(t *testing.T) { + metadata, err := ParseMetadataBytes([]byte(ExampleTableMetadataV2)) + require.NoError(t, err) + + builder, err := MetadataBuilderFromBase(metadata, "") + require.NoError(t, err) + require.Len(t, builder.schemaIndex.schemas, len(builder.schemaList)) + for _, schema := range builder.schemaList { + assert.Same(t, schema, builder.schemaIndex.schemas[schema.ID]) + } +} + +func TestCommonMetadataSchemaIndexFallsBackAfterSliceReplacement(t *testing.T) { + schemas := schemaIndexTestSchemas(1, 2) + metadata := commonMetadata{ + SchemaList: schemas, + schemaIndex: buildSchemaIndex(schemas), + } + originalIndex := metadata.schemaIndex + + replacement := schemaIndexTestSchemas(3, 4) + replacement[0] = schemas[0] + metadata.SchemaList = replacement + got := metadata.schemaByID(4) + require.NotNil(t, got) + assert.Equal(t, 4, got.ID) + assert.Nil(t, metadata.schemaByID(2)) + assert.Same(t, originalIndex, metadata.schemaIndex) + assert.Same(t, schemas[1], originalIndex.schemas[2]) +} + +func TestCommonMetadataSchemaIndexFallsBackAfterElementMutation(t *testing.T) { + schemas := schemaIndexTestSchemas(1, 2) + metadata := commonMetadata{ + SchemaList: schemas, + schemaIndex: buildSchemaIndex(schemas), + } + + schemas[1].ID = 3 + assert.Nil(t, metadata.schemaByID(2)) +} + +func TestMetadataBuilderSchemaIndexFallsBackAfterSliceReplacement(t *testing.T) { + schemas := schemaIndexTestSchemas(1, 2) + builder := MetadataBuilder{ + schemaList: schemas, + schemaIndex: buildSchemaIndex(schemas), + } + originalIndex := builder.schemaIndex + + replacement := schemaIndexTestSchemas(3, 4) + replacement[0] = schemas[0] + builder.schemaList = replacement + got, err := builder.GetSchemaByID(4) + require.NoError(t, err) + assert.Equal(t, 4, got.ID) + _, err = builder.GetSchemaByID(2) + assert.Error(t, err) + assert.Same(t, originalIndex, builder.schemaIndex) + assert.Same(t, schemas[1], originalIndex.schemas[2]) +} + +func TestMetadataBuilderSchemaIndexFollowsUpdates(t *testing.T) { + builder := builderWithoutChanges(2) + assert.Contains(t, builder.schemaIndex.schemas, builder.currentSchemaID) + + added := iceberg.NewSchema(99, iceberg.NestedField{ + ID: 4, Name: "new", Type: iceberg.PrimitiveTypes.Int64, Required: true, + }) + require.NoError(t, builder.AddSchema(added)) + assert.Same(t, added, builder.schemaIndex.schemas[99]) + got, err := builder.GetSchemaByID(99) + require.NoError(t, err) + assert.Same(t, added, got) + + clone := builder.clone() + cloneAdded := iceberg.NewSchema(100, iceberg.NestedField{ + ID: 5, Name: "clone", Type: iceberg.PrimitiveTypes.Int64, Required: true, + }) + require.NoError(t, clone.AddSchema(cloneAdded)) + assert.NotContains(t, builder.schemaIndex.schemas, 100) + assert.Same(t, cloneAdded, clone.schemaIndex.schemas[100]) + + require.NoError(t, builder.RemoveSchemas([]int{99})) + assert.NotContains(t, builder.schemaIndex.schemas, 99) + _, err = builder.GetSchemaByID(99) + assert.Error(t, err) + got, err = clone.GetSchemaByID(99) + require.NoError(t, err) + assert.Same(t, added, got) +} + +func TestMetadataBuilderSchemaIndexIsolatedFromBuiltMetadata(t *testing.T) { + builder := builderWithoutChanges(2) + metadata, err := builder.Build() + require.NoError(t, err) + common := metadataCommon(metadata) + originalIndex := common.schemaIndex + + added := iceberg.NewSchema(99, iceberg.NestedField{ + ID: 4, Name: "new", Type: iceberg.PrimitiveTypes.Int64, Required: true, + }) + require.NoError(t, builder.AddSchema(added)) + + assert.NotSame(t, originalIndex, builder.schemaIndex) + assert.NotContains(t, originalIndex.schemas, 99) + assert.Contains(t, builder.schemaIndex.schemas, 99) + assert.Nil(t, common.schemaByID(99)) +} + +func TestCommonMetadataSchemaLookupsConcurrent(t *testing.T) { + schemas := schemaIndexTestSchemas(1, 2, 3) + metadata := &commonMetadata{ + SchemaList: schemas, + CurrentSchemaID: 3, + schemaIndex: buildSchemaIndex(schemas), + } + + const ( + goroutineCount = 8 + lookupCount = 100 + ) + + var wg sync.WaitGroup + wg.Add(goroutineCount) + for i := range goroutineCount { + go func(i int) { + defer wg.Done() + for range lookupCount { + var schema *iceberg.Schema + if i%2 == 0 { + schema = metadata.schemaByID(2) + } else { + schema = metadata.CurrentSchema() + } + if schema == nil { + t.Errorf("expected schema, got nil") + + return + } + } + }(i) + } + wg.Wait() +} + +func TestMetadataBuilderSchemaIndexCopyOnWriteConcurrent(t *testing.T) { + builderValue := builderWithoutChanges(2) + metadata, err := builderValue.Build() + require.NoError(t, err) + common := metadataCommon(metadata) + builder := &builderValue + + readerStarted := make(chan struct{}) + writerErr := make(chan error, 1) + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + for range 1_000 { + schema := common.CurrentSchema() + if schema == nil || schema.ID != builder.currentSchemaID { + t.Errorf("expected current schema %d, got %v", builder.currentSchemaID, schema) + + return + } + select { + case <-readerStarted: + default: + close(readerStarted) + } + } + }() + go func() { + defer wg.Done() + <-readerStarted + added := iceberg.NewSchema(99, iceberg.NestedField{ + ID: 4, Name: "copy-on-write", Type: iceberg.PrimitiveTypes.Int64, Required: true, + }) + writerErr <- builder.AddSchema(added) + }() + wg.Wait() + require.NoError(t, <-writerErr) + assert.NotSame(t, common.schemaIndex, builder.schemaIndex) + assert.NotContains(t, common.schemaIndex.schemas, 99) + assert.Contains(t, builder.schemaIndex.schemas, 99) +} + +func schemaIndexTestSchemas(ids ...int) []*iceberg.Schema { + schemas := make([]*iceberg.Schema, len(ids)) + for i, id := range ids { + schemas[i] = iceberg.NewSchema(id) + } + + return schemas +} From cb3e742eb3d5102fbc22217adf2eacc527a64818 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sun, 30 Aug 2026 22:36:16 +0200 Subject: [PATCH 2/3] fix(table): preserve schema lookup after ID changes Signed-off-by: Minh Vu --- table/metadata.go | 19 ++++--------------- table/schema_index_test.go | 17 +++++++++++++++++ 2 files changed, 21 insertions(+), 15 deletions(-) diff --git a/table/metadata.go b/table/metadata.go index efcf5baf3..504cc1d49 100644 --- a/table/metadata.go +++ b/table/metadata.go @@ -201,24 +201,13 @@ func schemaIndexNeedsRebuild(index *schemaIndexData, schemas []*iceberg.Schema) func schemaIndexLookup(index *schemaIndexData, schemas []*iceberg.Schema, id int) (*iceberg.Schema, bool) { if index != nil { - if schema, ok := index.schemas[id]; ok { - if schema != nil && schema.ID == id { - return schema, true - } - - index = buildSchemaIndex(schemas) - if schema, ok := index.schemas[id]; ok && schema != nil && schema.ID == id { - return schema, true - } - - return nil, false - } - - if !schemaIndexNeedsRebuild(index, schemas) { - return nil, false + if schema, ok := index.schemas[id]; ok && schema != nil && schema.ID == id { + return schema, true } } + // Builder lookups return mutable schemas, so a changed ID may not have + // an entry in the index even when the slice itself is unchanged. for _, schema := range schemas { if schema != nil && schema.ID == id { return schema, true diff --git a/table/schema_index_test.go b/table/schema_index_test.go index 30e926ea9..f35e172c1 100644 --- a/table/schema_index_test.go +++ b/table/schema_index_test.go @@ -95,6 +95,23 @@ func TestCommonMetadataSchemaIndexFallsBackAfterElementMutation(t *testing.T) { schemas[1].ID = 3 assert.Nil(t, metadata.schemaByID(2)) + got := metadata.schemaByID(3) + require.NotNil(t, got) + assert.Equal(t, 3, got.ID) +} + +func TestMetadataBuilderSchemaIndexFindsRenamedSchema(t *testing.T) { + builder := builderWithoutChanges(2) + schema, err := builder.GetSchemaByID(builder.currentSchemaID) + require.NoError(t, err) + originalID := schema.ID + schema.ID = 42 + + got, err := builder.GetSchemaByID(42) + require.NoError(t, err) + assert.Same(t, schema, got) + _, err = builder.GetSchemaByID(originalID) + assert.ErrorIs(t, err, iceberg.ErrInvalidArgument) } func TestMetadataBuilderSchemaIndexFallsBackAfterSliceReplacement(t *testing.T) { From 76c4cb3bd7a54b55ab95996329ae53e6a7404fc8 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sun, 30 Aug 2026 23:28:52 +0200 Subject: [PATCH 3/3] fix(schema): avoid copying lazy caches during JSON encoding Signed-off-by: Minh Vu --- schema.go | 3 +-- schema_test.go | 46 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/schema.go b/schema.go index 24b0ff4b1..4330103dc 100644 --- a/schema.go +++ b/schema.go @@ -346,8 +346,7 @@ func (s *Schema) MarshalJSON() ([]byte, error) { type Alias Schema - aliasCopy := *(*Alias)(s) - aliasCopy.IdentifierFieldIDs = ids + aliasCopy := Alias{ID: s.ID, IdentifierFieldIDs: ids} return json.Marshal(struct { Type string `json:"type"` diff --git a/schema_test.go b/schema_test.go index 7fa16e345..6d794ab71 100644 --- a/schema_test.go +++ b/schema_test.go @@ -24,6 +24,7 @@ import ( "path/filepath" "runtime" "strings" + "sync" "testing" "github.com/apache/iceberg-go" @@ -2293,3 +2294,48 @@ func TestVisitGeoSchemaWithSchemaVisitorPerPrimitiveType(t *testing.T) { assert.Equal(t, 1, v.geometryCalls) assert.Equal(t, 1, v.geographyCalls) } + +func TestSchemaMarshalJSONConcurrentLazyLookups(t *testing.T) { + for range 32 { + schema := iceberg.NewSchemaWithIdentifiers(17, nil, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String}, + ) + start := make(chan struct{}) + var wg sync.WaitGroup + for range 8 { + wg.Go(func() { + <-start + for range 8 { + _, err := json.Marshal(schema) + assert.NoError(t, err) + } + }) + wg.Go(func() { + <-start + _, found := schema.FindFieldByID(1) + assert.True(t, found) + _, found = schema.FindFieldByName("data") + assert.True(t, found) + _, found = schema.FindFieldByNameCaseInsensitive("DATA") + assert.True(t, found) + name, found := schema.FindColumnName(2) + assert.True(t, found) + assert.Equal(t, "data", name) + }) + } + close(start) + wg.Wait() + + data, err := json.Marshal(schema) + require.NoError(t, err) + assert.JSONEq(t, `{ + "type": "struct", "schema-id": 17, "identifier-field-ids": [], + "fields": [ + {"id": 1, "name": "id", "type": "long", "required": true}, + {"id": 2, "name": "data", "type": "string", "required": false} + ] + }`, string(data)) + assert.Nil(t, schema.IdentifierFieldIDs) + } +}