You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/28 21:44:33 UTC
(iceberg-go) branch main updated: perf(table): avoid redundant MetadataBuilderFromBase copies (#1932)
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 94cedde67 perf(table): avoid redundant MetadataBuilderFromBase copies (#1932)
94cedde67 is described below
commit 94cedde6741f026eccd1178d921a4a2f18807d4b
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Fri Aug 28 23:44:27 2026 +0200
perf(table): avoid redundant MetadataBuilderFromBase copies (#1932)
* perf(table): avoid redundant metadata builder copies
* perf(table): preserve nil snapshot refs
---
table/metadata.go | 136 ++++++++++++++++++------
table/metadata_builder_bench_test.go | 113 ++++++++++++++++++++
table/metadata_builder_from_base_test.go | 172 +++++++++++++++++++++++++++++++
3 files changed, 388 insertions(+), 33 deletions(-)
diff --git a/table/metadata.go b/table/metadata.go
index 133c6227c..1c71860da 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -310,6 +310,10 @@ func NewMetadataBuilder(formatVersion int) (*MetadataBuilder, error) {
}, nil
}
+type metadataBuilderSource interface {
+ metadataBuilderCommon() *commonMetadata
+}
+
// MetadataBuilderFromBase creates a MetadataBuilder from an existing Metadata object.
// currentFileLocation is the location where the current version of the metadata
// file is stored. This is used to update the metadata log. If currentFileLocation is
@@ -323,42 +327,85 @@ func MetadataBuilderFromBase(metadata Metadata, currentFileLocation string) (*Me
b.loc = metadata.Location()
b.lastUpdatedMS = 0
b.lastColumnId = metadata.LastColumnID()
- b.schemaList = slices.Clone(metadata.Schemas())
- currentSchema := metadata.CurrentSchema()
- if currentSchema == nil {
- return nil, fmt.Errorf("%w: current schema is missing", ErrInvalidMetadata)
- }
- b.currentSchemaID = currentSchema.ID
- b.specs = slices.Clone(metadata.PartitionSpecs())
- defaultSpecID := metadata.DefaultPartitionSpec()
- b.defaultSpecID = defaultSpecID
- b.lastPartitionID = metadata.LastPartitionSpecID()
- b.props = maps.Clone(metadata.Properties())
- b.snapshotList = slices.Clone(metadata.Snapshots())
- b.snapshotIndex = buildSnapshotIndex(b.snapshotList)
- b.sortOrderList = slices.Clone(metadata.SortOrders())
- b.defaultSortOrderID = metadata.DefaultSortOrder()
- if metadata.Version() > 1 {
- seq := metadata.LastSequenceNumber()
- b.lastSequenceNumber = &seq
- }
-
- if metadata.Version() >= 3 {
- nextRowID := metadata.NextRowID()
- b.nextRowID = &nextRowID
- }
- if metadata.CurrentSnapshot() != nil {
- b.currentSnapshotID = &metadata.CurrentSnapshot().SnapshotID
+ if source, ok := metadata.(metadataBuilderSource); ok {
+ common := source.metadataBuilderCommon()
+ if common == nil || !slices.ContainsFunc(common.SchemaList, func(schema *iceberg.Schema) bool {
+ return schema != nil && schema.ID == common.CurrentSchemaID
+ }) {
+ return nil, fmt.Errorf("%w: current schema is missing", ErrInvalidMetadata)
+ }
+
+ b.currentSchemaID = common.CurrentSchemaID
+ b.defaultSpecID = metadata.DefaultPartitionSpec()
+ b.defaultSortOrderID = metadata.DefaultSortOrder()
+ b.schemaList = cloneSchemas(common.SchemaList)
+ b.specs = clonePartitionSpecs(common.Specs)
+ b.lastPartitionID = clonePtr(common.LastPartitionID)
+ b.props = maps.Clone(common.Props)
+ if b.props == nil {
+ b.props = iceberg.Properties{}
+ }
+ b.snapshotList = cloneSnapshots(common.SnapshotList)
+ b.snapshotLog = cloneCollected(common.SnapshotLog)
+ b.metadataLog = cloneCollected(common.MetadataLog)
+ b.sortOrderList = cloneSortOrders(common.SortOrderList)
+ b.refs = cloneSnapshotRefs(common.SnapshotRefs)
+ if len(common.StatisticsList) > 0 {
+ b.statisticsList = cloneStatisticsFiles(common.StatisticsList)
+ }
+ b.partitionStatsList = cloneCollected(common.PartitionStatsList)
+ if len(common.EncryptionKeyList) > 0 {
+ b.encryptionKeyList = cloneEncryptionKeys(common.EncryptionKeyList)
+ }
+ b.snapshotIndex = buildSnapshotIndex(b.snapshotList)
+ if metadata.Version() > 1 {
+ seq := metadata.LastSequenceNumber()
+ b.lastSequenceNumber = &seq
+ }
+ if metadata.Version() >= 3 {
+ nextRowID := metadata.NextRowID()
+ b.nextRowID = &nextRowID
+ }
+ if common.CurrentSnapshotID != nil {
+ if _, ok := snapshotIndexPosition(common.snapshotIndex, common.SnapshotList, *common.CurrentSnapshotID); ok {
+ b.currentSnapshotID = clonePtr(common.CurrentSnapshotID)
+ }
+ }
+ } else {
+ b.schemaList = slices.Clone(metadata.Schemas())
+ currentSchema := metadata.CurrentSchema()
+ if currentSchema == nil {
+ return nil, fmt.Errorf("%w: current schema is missing", ErrInvalidMetadata)
+ }
+ b.currentSchemaID = currentSchema.ID
+ b.specs = slices.Clone(metadata.PartitionSpecs())
+ b.defaultSpecID = metadata.DefaultPartitionSpec()
+ b.lastPartitionID = metadata.LastPartitionSpecID()
+ b.props = maps.Clone(metadata.Properties())
+ b.snapshotList = slices.Clone(metadata.Snapshots())
+ b.snapshotIndex = buildSnapshotIndex(b.snapshotList)
+ b.sortOrderList = slices.Clone(metadata.SortOrders())
+ b.defaultSortOrderID = metadata.DefaultSortOrder()
+ if metadata.Version() > 1 {
+ seq := metadata.LastSequenceNumber()
+ b.lastSequenceNumber = &seq
+ }
+ if metadata.Version() >= 3 {
+ nextRowID := metadata.NextRowID()
+ b.nextRowID = &nextRowID
+ }
+ if metadata.CurrentSnapshot() != nil {
+ b.currentSnapshotID = &metadata.CurrentSnapshot().SnapshotID
+ }
+ b.refs = maps.Collect(metadata.Refs())
+ b.snapshotLog = slices.Collect(metadata.SnapshotLogs())
+ b.metadataLog = slices.Collect(metadata.PreviousFiles())
+ b.statisticsList = slices.Collect(metadata.Statistics())
+ b.partitionStatsList = slices.Collect(metadata.PartitionStatistics())
+ b.encryptionKeyList = slices.Collect(metadata.EncryptionKeys())
}
- b.refs = maps.Collect(metadata.Refs())
- b.snapshotLog = slices.Collect(metadata.SnapshotLogs())
- b.metadataLog = slices.Collect(metadata.PreviousFiles())
- b.statisticsList = slices.Collect(metadata.Statistics())
- b.partitionStatsList = slices.Collect(metadata.PartitionStatistics())
- b.encryptionKeyList = slices.Collect(metadata.EncryptionKeys())
-
if currentFileLocation != "" {
b.previousFileEntry = &MetadataLogEntry{
MetadataFile: currentFileLocation,
@@ -1913,6 +1960,8 @@ type commonMetadata struct {
snapshotIndex *snapshotIndexData
}
+func (c *commonMetadata) metadataBuilderCommon() *commonMetadata { return c }
+
func initCommonMetadataForDeserialization() commonMetadata {
return commonMetadata{
// These fields use negative sentinels so validation can distinguish omitted
@@ -2292,6 +2341,27 @@ func cloneSnapshotRef(ref SnapshotRef) SnapshotRef {
return clone
}
+func cloneSnapshotRefs(refs map[string]SnapshotRef) map[string]SnapshotRef {
+ if refs == nil {
+ return nil
+ }
+
+ clones := make(map[string]SnapshotRef, len(refs))
+ for name, ref := range refs {
+ clones[name] = cloneSnapshotRef(ref)
+ }
+
+ return clones
+}
+
+func cloneCollected[T any](values []T) []T {
+ if len(values) == 0 {
+ return nil
+ }
+
+ return slices.Clone(values)
+}
+
func cloneSnapshot(snapshot Snapshot) Snapshot {
clone := snapshot
if snapshot.ManifestLocations != nil {
diff --git a/table/metadata_builder_bench_test.go b/table/metadata_builder_bench_test.go
index 3caa164c2..93c732f68 100644
--- a/table/metadata_builder_bench_test.go
+++ b/table/metadata_builder_bench_test.go
@@ -21,6 +21,9 @@ import (
"fmt"
"slices"
"testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/google/uuid"
)
var (
@@ -149,6 +152,116 @@ func BenchmarkMetadataBuilderFromBase(b *testing.B) {
}
}
+func BenchmarkMetadataBuilderFromBaseCollections(b *testing.B) {
+ cases := []struct {
+ name string
+ schemaCount int
+ specCount int
+ snapshotCount int
+ sortOrderCount int
+ propertyCount int
+ }{
+ {name: "schemas=1", schemaCount: 1, specCount: 1, snapshotCount: 1, sortOrderCount: 1},
+ {name: "schemas=16", schemaCount: 16, specCount: 1, snapshotCount: 1, sortOrderCount: 1},
+ {name: "schemas=128", schemaCount: 128, specCount: 1, snapshotCount: 1, sortOrderCount: 1},
+ {name: "schemas=1024", schemaCount: 1024, specCount: 1, snapshotCount: 1, sortOrderCount: 1},
+ {name: "partition-specs=16", schemaCount: 1, specCount: 16, snapshotCount: 1, sortOrderCount: 1},
+ {name: "partition-specs=128", schemaCount: 1, specCount: 128, snapshotCount: 1, sortOrderCount: 1},
+ {name: "snapshots=1000", schemaCount: 1, specCount: 1, snapshotCount: 1000, sortOrderCount: 1},
+ {name: "sort-orders=32", schemaCount: 1, specCount: 1, snapshotCount: 1, sortOrderCount: 32},
+ {name: "properties=32", schemaCount: 1, specCount: 1, snapshotCount: 1, sortOrderCount: 1, propertyCount: 32},
+ }
+
+ for _, tc := range cases {
+ b.Run(tc.name, func(b *testing.B) {
+ metadata := benchmarkMetadataBuilderBase(tc.schemaCount, tc.specCount, tc.snapshotCount, tc.sortOrderCount, tc.propertyCount)
+
+ b.ReportAllocs()
+ b.ReportMetric(float64(tc.schemaCount), "schemas")
+ b.ReportMetric(float64(tc.specCount), "partition_specs")
+ b.ReportMetric(float64(tc.snapshotCount), "snapshots")
+ b.ReportMetric(float64(tc.sortOrderCount), "sort_orders")
+ b.ReportMetric(float64(tc.propertyCount), "properties")
+ b.ResetTimer()
+
+ for range b.N {
+ builder, err := MetadataBuilderFromBase(metadata, "")
+ if err != nil {
+ b.Fatal(err)
+ }
+ metadataBuilderBenchmarkSink = len(builder.schemaList) + len(builder.specs) +
+ len(builder.snapshotList) + len(builder.sortOrderList) + len(builder.props)
+ }
+ })
+ }
+}
+
+func benchmarkMetadataBuilderBase(schemaCount, specCount, snapshotCount, sortOrderCount, propertyCount int) Metadata {
+ schemas := make([]*iceberg.Schema, schemaCount)
+ schemas[0] = iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 1,
+ Name: "id",
+ Type: iceberg.PrimitiveTypes.Int64,
+ Required: true,
+ })
+ for i := 1; i < schemaCount; i++ {
+ schemas[i] = iceberg.NewSchema(i)
+ }
+
+ specs := make([]iceberg.PartitionSpec, specCount)
+ for i := range specs {
+ specs[i] = iceberg.NewPartitionSpecID(i)
+ }
+
+ sortOrders := make([]SortOrder, sortOrderCount)
+ sortOrders[0] = UnsortedSortOrder
+ for i := 1; i < sortOrderCount; i++ {
+ order, err := NewSortOrder(i, []SortField{{
+ SourceIDs: []int{1},
+ Transform: iceberg.IdentityTransform{},
+ Direction: SortASC,
+ NullOrder: NullsFirst,
+ }})
+ if err != nil {
+ panic(err)
+ }
+ sortOrders[i] = order
+ }
+
+ snapshots := benchmarkSnapshots(snapshotCount)
+ currentSnapshotID := int64(snapshotCount - 1)
+ lastPartitionID := 0
+ properties := make(iceberg.Properties, propertyCount)
+ for i := range propertyCount {
+ properties[fmt.Sprintf("property-%d", i)] = "value"
+ }
+
+ return &metadataV2{
+ LastSeqNum: int64(snapshotCount),
+ commonMetadata: commonMetadata{
+ FormatVersion: 2,
+ UUID: uuid.New(),
+ Loc: "s3://bucket/test/location",
+ LastColumnId: 1,
+ SchemaList: schemas,
+ CurrentSchemaID: 0,
+ Specs: specs,
+ DefaultSpecID: 0,
+ LastPartitionID: &lastPartitionID,
+ Props: properties,
+ SnapshotList: snapshots,
+ snapshotIndex: buildSnapshotIndex(snapshots),
+ CurrentSnapshotID: ¤tSnapshotID,
+ SnapshotLog: benchmarkSnapshotLog(snapshotCount),
+ SortOrderList: sortOrders,
+ DefaultSortOrderID: 0,
+ SnapshotRefs: map[string]SnapshotRef{
+ MainBranch: {SnapshotID: currentSnapshotID, SnapshotRefType: BranchRef},
+ },
+ },
+ }
+}
+
func benchmarkSnapshotLog(snapshotCount int) []SnapshotLogEntry {
log := make([]SnapshotLogEntry, snapshotCount)
for i := range log {
diff --git a/table/metadata_builder_from_base_test.go b/table/metadata_builder_from_base_test.go
new file mode 100644
index 000000000..8f1ddf8b7
--- /dev/null
+++ b/table/metadata_builder_from_base_test.go
@@ -0,0 +1,172 @@
+// 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 (
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/google/uuid"
+ "github.com/stretchr/testify/require"
+)
+
+func TestMetadataBuilderFromBaseCopiesBuiltinMetadata(t *testing.T) {
+ tableSchema := iceberg.NewSchemaWithIdentifiers(
+ 0,
+ []int{1},
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ partitionSpec := iceberg.NewPartitionSpecID(0, iceberg.PartitionField{
+ SourceIDs: []int{1},
+ Name: "id",
+ Transform: iceberg.IdentityTransform{},
+ })
+ sortOrder, err := NewSortOrder(1, []SortField{{
+ SourceIDs: []int{1},
+ Transform: iceberg.IdentityTransform{},
+ Direction: SortASC,
+ NullOrder: NullsFirst,
+ }})
+ require.NoError(t, err)
+
+ currentSnapshotID := int64(2)
+ parentSnapshotID := int64(1)
+ schemaID := 0
+ lastPartitionID := 1
+ minSnapshotsToKeep := 2
+ maxSnapshotAgeMs := int64(3)
+ maxRefAgeMs := int64(4)
+ statisticsKeyMetadata := "statistics-key"
+ encryptedByID := "kms-key"
+
+ metadata := &metadataV3{
+ LastSeqNum: 2,
+ NextRowIDValue: 3,
+ commonMetadata: commonMetadata{
+ FormatVersion: 3,
+ UUID: uuid.New(),
+ Loc: "s3://test/table",
+ LastUpdatedMS: 1000,
+ LastColumnId: 1,
+ SchemaList: []*iceberg.Schema{tableSchema},
+ CurrentSchemaID: 0,
+ Specs: []iceberg.PartitionSpec{partitionSpec},
+ DefaultSpecID: 0,
+ LastPartitionID: &lastPartitionID,
+ Props: iceberg.Properties{"property": "value"},
+ SnapshotList: []Snapshot{{
+ SnapshotID: currentSnapshotID,
+ ParentSnapshotID: &parentSnapshotID,
+ ManifestLocations: []string{"manifest.avro"},
+ Summary: &Summary{Operation: OpAppend, Properties: map[string]string{"summary": "value"}},
+ SchemaID: &schemaID,
+ }},
+ CurrentSnapshotID: ¤tSnapshotID,
+ SnapshotLog: []SnapshotLogEntry{{SnapshotID: currentSnapshotID, TimestampMs: 1000}},
+ MetadataLog: []MetadataLogEntry{{MetadataFile: "metadata.json", TimestampMs: 900}},
+ SortOrderList: []SortOrder{sortOrder},
+ DefaultSortOrderID: 1,
+ SnapshotRefs: map[string]SnapshotRef{
+ MainBranch: {
+ SnapshotID: currentSnapshotID,
+ SnapshotRefType: BranchRef,
+ MinSnapshotsToKeep: &minSnapshotsToKeep,
+ MaxSnapshotAgeMs: &maxSnapshotAgeMs,
+ MaxRefAgeMs: &maxRefAgeMs,
+ },
+ },
+ StatisticsList: []StatisticsFile{{
+ SnapshotID: currentSnapshotID,
+ KeyMetadata: &statisticsKeyMetadata,
+ BlobMetadata: []BlobMetadata{{Fields: []int32{1}, Properties: map[string]string{"blob": "value"}}},
+ }},
+ PartitionStatsList: []PartitionStatisticsFile{{
+ SnapshotID: currentSnapshotID,
+ StatisticsPath: "partition-stats.parquet",
+ }},
+ EncryptionKeyList: []EncryptionKey{{
+ KeyID: "key",
+ EncryptedKeyMetadata: "metadata",
+ EncryptedByID: &encryptedByID,
+ Properties: map[string]string{"encryption": "value"},
+ }},
+ },
+ }
+
+ builder, err := MetadataBuilderFromBase(metadata, "")
+ require.NoError(t, err)
+
+ require.NotSame(t, &metadata.SchemaList[0], &builder.schemaList[0])
+ require.NotSame(t, metadata.SchemaList[0], builder.schemaList[0])
+ builder.schemaList[0].IdentifierFieldIDs[0] = 99
+ require.Equal(t, []int{1}, metadata.SchemaList[0].IdentifierFieldIDs)
+
+ require.NotSame(t, &metadata.Specs[0], &builder.specs[0])
+ require.NotSame(t, &metadata.SnapshotList[0], &builder.snapshotList[0])
+ builder.snapshotList[0].ManifestLocations[0] = "changed.avro"
+ *builder.snapshotList[0].ParentSnapshotID = 99
+ builder.snapshotList[0].Summary.Properties["summary"] = "changed"
+ *builder.snapshotList[0].SchemaID = 99
+ require.Equal(t, []string{"manifest.avro"}, metadata.SnapshotList[0].ManifestLocations)
+ require.Equal(t, int64(1), *metadata.SnapshotList[0].ParentSnapshotID)
+ require.Equal(t, "value", metadata.SnapshotList[0].Summary.Properties["summary"])
+ require.Equal(t, 0, *metadata.SnapshotList[0].SchemaID)
+
+ require.NotSame(t, &metadata.SortOrderList[0], &builder.sortOrderList[0])
+ builder.sortOrderList[0].fields[0].SourceIDs[0] = 99
+ require.Equal(t, []int{1}, metadata.SortOrderList[0].fields[0].SourceIDs)
+
+ builder.props["property"] = "changed"
+ require.Equal(t, "value", metadata.Props["property"])
+ *builder.lastPartitionID = 99
+ require.Equal(t, 1, *metadata.LastPartitionID)
+ *builder.currentSnapshotID = 99
+ require.Equal(t, currentSnapshotID, *metadata.CurrentSnapshotID)
+
+ ref := builder.refs[MainBranch]
+ ref.MinSnapshotsToKeep = clonePtr(ref.MinSnapshotsToKeep)
+ *ref.MinSnapshotsToKeep = 99
+ builder.refs[MainBranch] = ref
+ require.Equal(t, 2, *metadata.SnapshotRefs[MainBranch].MinSnapshotsToKeep)
+ delete(builder.refs, MainBranch)
+ require.Contains(t, metadata.SnapshotRefs, MainBranch)
+
+ *builder.statisticsList[0].KeyMetadata = "changed"
+ builder.statisticsList[0].BlobMetadata[0].Fields[0] = 99
+ builder.statisticsList[0].BlobMetadata[0].Properties["blob"] = "changed"
+ require.Equal(t, "statistics-key", *metadata.StatisticsList[0].KeyMetadata)
+ require.Equal(t, int32(1), metadata.StatisticsList[0].BlobMetadata[0].Fields[0])
+ require.Equal(t, "value", metadata.StatisticsList[0].BlobMetadata[0].Properties["blob"])
+
+ *builder.encryptionKeyList[0].EncryptedByID = "changed"
+ builder.encryptionKeyList[0].Properties["encryption"] = "changed"
+ require.Equal(t, "kms-key", *metadata.EncryptionKeyList[0].EncryptedByID)
+ require.Equal(t, "value", metadata.EncryptionKeyList[0].Properties["encryption"])
+}
+
+func TestMetadataBuilderFromBaseKeepsNilSnapshotRefs(t *testing.T) {
+ metadata := &metadataV2{commonMetadata: commonMetadata{
+ SchemaList: []*iceberg.Schema{iceberg.NewSchema(0)},
+ CurrentSchemaID: 0,
+ SnapshotRefs: nil,
+ }}
+
+ builder, err := MetadataBuilderFromBase(metadata, "")
+ require.NoError(t, err)
+ require.Nil(t, builder.refs)
+}