This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 a7f0d585a perf(rest): avoid repeated schema-history clones in
partition decoding (#2362)
a7f0d585a is described below
commit a7f0d585a69adcc5ada3d8da697208f8f3b1fcbd
Author: Minh Vu <[email protected]>
AuthorDate: Sat Oct 31 09:47:41 2026 +0100
perf(rest): avoid repeated schema-history clones in partition decoding
(#2362)
---
catalog/rest/scan_task_decoder.go | 29 +++-
.../rest/scan_task_schema_history_bench_test.go | 86 ++++++++++++
.../rest/scan_task_schema_history_lookup_test.go | 155 +++++++++++++++++++++
3 files changed, 265 insertions(+), 5 deletions(-)
diff --git a/catalog/rest/scan_task_decoder.go
b/catalog/rest/scan_task_decoder.go
index 0d852837d..0d3b2b231 100644
--- a/catalog/rest/scan_task_decoder.go
+++ b/catalog/rest/scan_task_decoder.go
@@ -598,6 +598,7 @@ func newPartitionDecodePlan(spec *iceberg.PartitionSpec,
metadata table.ScanPlan
spec: spec,
fields: make([]partitionDecodeField, spec.NumFields()),
}
+ var sourceTypes partitionSourceTypeLookup
for i, field := range spec.Fields() {
transform := reflect.ValueOf(field.Transform)
if !transform.IsValid() || (transform.Kind() == reflect.Pointer
&& transform.IsNil()) {
@@ -610,7 +611,7 @@ func newPartitionDecodePlan(spec *iceberg.PartitionSpec,
metadata table.ScanPlan
fieldName: field.Name,
sourceID: field.FirstSourceID(),
}
- sourceType, ok := findFieldType(fieldPlan.sourceID, metadata)
+ sourceType, ok := sourceTypes.find(fieldPlan.sourceID, metadata)
fieldPlan.sourceTypeFound = ok
if ok {
fieldPlan.resultType =
field.Transform.ResultType(sourceType)
@@ -662,13 +663,31 @@ func decodePartition(
return partition, logicalTypes, nil
}
-func findFieldType(fieldID int, metadata table.ScanPlanningMetadata)
(iceberg.Type, bool) {
- if current := metadata.CurrentSchema(); current != nil {
- if typ, ok := current.FindTypeByID(fieldID); ok {
+// partitionSourceTypeLookup keeps the defensive copies of table schemas for
+// one decode plan. Metadata.Schemas clones the entire schema history, so
+// calling it for every historical partition field multiplies the cost.
+type partitionSourceTypeLookup struct {
+ current *iceberg.Schema
+ history []*iceberg.Schema
+ currentLoaded bool
+ historyLoaded bool
+}
+
+func (l *partitionSourceTypeLookup) find(fieldID int, metadata
table.ScanPlanningMetadata) (iceberg.Type, bool) {
+ if !l.currentLoaded {
+ l.current = metadata.CurrentSchema()
+ l.currentLoaded = true
+ }
+ if l.current != nil {
+ if typ, ok := l.current.FindTypeByID(fieldID); ok {
return typ, true
}
}
- for _, schema := range metadata.Schemas() {
+ if !l.historyLoaded {
+ l.history = metadata.Schemas()
+ l.historyLoaded = true
+ }
+ for _, schema := range l.history {
if schema == nil {
continue
}
diff --git a/catalog/rest/scan_task_schema_history_bench_test.go
b/catalog/rest/scan_task_schema_history_bench_test.go
new file mode 100644
index 000000000..f00581a4e
--- /dev/null
+++ b/catalog/rest/scan_task_schema_history_bench_test.go
@@ -0,0 +1,86 @@
+// 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 rest
+
+import (
+ "fmt"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+)
+
+func historicalSchemaBenchmarkFixture(schemaCount, fieldCount, sourceCount
int) (*cloningSchemaHistoryMetadata, []int) {
+ history := make([]*iceberg.Schema, schemaCount)
+ for version := range history {
+ fields := make([]iceberg.NestedField, fieldCount)
+ for index := range fields {
+ fields[index] = iceberg.NestedField{
+ ID: 1 + version*fieldCount + index,
+ Name: fmt.Sprintf("field_%d", index),
+ Type: iceberg.PrimitiveTypes.Int64,
+ }
+ }
+ history[version] = iceberg.NewSchema(version, fields...)
+ }
+ sources := make([]int, sourceCount)
+ for index := range sources {
+ version := index * schemaCount / sourceCount
+ sources[index] = 1 + version*fieldCount + index%fieldCount
+ }
+ metadata := &cloningSchemaHistoryMetadata{
+ scanTaskDecoderMetadata: &scanTaskDecoderMetadata{
+ schema: iceberg.NewSchema(schemaCount+1,
iceberg.NestedField{
+ ID: 900000, Name: "current", Type:
iceberg.PrimitiveTypes.String,
+ }),
+ },
+ history: history,
+ }
+
+ return metadata, sources
+}
+
+var benchmarkPartitionSourceTypeSink int
+
+// BenchmarkPartitionSourceTypeSchemaHistory measures source-type lookups
+// with defensive schema copies; it excludes JSON decoding and REST scans.
+func BenchmarkPartitionSourceTypeSchemaHistory(b *testing.B) {
+ for _, tc := range []struct {
+ schemas, fields, sources int
+ }{
+ {schemas: 16, fields: 32, sources: 4},
+ {schemas: 64, fields: 64, sources: 16},
+ {schemas: 128, fields: 64, sources: 32},
+ } {
+ name := fmt.Sprintf("%d_schemas_%d_fields_%d_sources",
tc.schemas, tc.fields, tc.sources)
+ metadata, sources :=
historicalSchemaBenchmarkFixture(tc.schemas, tc.fields, tc.sources)
+ b.Run(name, func(b *testing.B) {
+ b.ReportAllocs()
+ hits := 0
+ for b.Loop() {
+ var lookup partitionSourceTypeLookup
+ for _, id := range sources {
+ if typ, ok := lookup.find(id,
metadata); !ok || typ == nil {
+ b.Fatal("failed historical
source lookup", id)
+ }
+ hits++
+ }
+ }
+ benchmarkPartitionSourceTypeSink = hits
+ })
+ }
+}
diff --git a/catalog/rest/scan_task_schema_history_lookup_test.go
b/catalog/rest/scan_task_schema_history_lookup_test.go
new file mode 100644
index 000000000..79eccc510
--- /dev/null
+++ b/catalog/rest/scan_task_schema_history_lookup_test.go
@@ -0,0 +1,155 @@
+// 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 rest
+
+import (
+ "encoding/json"
+ "fmt"
+ "slices"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// cloningSchemaHistoryMetadata models the defensive schema copies returned by
+// real table metadata and counts how often the decode plan requests them.
+type cloningSchemaHistoryMetadata struct {
+ *scanTaskDecoderMetadata
+ history []*iceberg.Schema
+ currentLookups int
+ historyLookups int
+}
+
+func cloneDecoderSchemaForTest(schema *iceberg.Schema) *iceberg.Schema {
+ if schema == nil {
+ return nil
+ }
+
+ return iceberg.NewSchemaWithIdentifiers(
+ schema.ID, slices.Clone(schema.IdentifierFieldIDs),
schema.Fields()...,
+ )
+}
+
+func (m *cloningSchemaHistoryMetadata) CurrentSchema() *iceberg.Schema {
+ m.currentLookups++
+
+ return cloneDecoderSchemaForTest(m.schema)
+}
+
+func (m *cloningSchemaHistoryMetadata) Schemas() []*iceberg.Schema {
+ m.historyLookups++
+ copies := make([]*iceberg.Schema, len(m.history))
+ for index, schema := range m.history {
+ copies[index] = cloneDecoderSchemaForTest(schema)
+ }
+
+ return copies
+}
+
+func TestPartitionDecodePlanReusesHistoricalSchemaCopies(t *testing.T) {
+ t.Parallel()
+
+ current := iceberg.NewSchema(30,
+ iceberg.NestedField{ID: 1, Name: "active", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ firstHistory := iceberg.NewSchema(20,
+ iceberg.NestedField{ID: 2, Name: "old_date", Type:
iceberg.PrimitiveTypes.Date},
+ iceberg.NestedField{ID: 3, Name: "old_long", Type:
iceberg.PrimitiveTypes.Int64},
+ )
+ // An ID appearing in multiple historical schemas must keep the first
match.
+ secondHistory := iceberg.NewSchema(10,
+ iceberg.NestedField{ID: 2, Name: "old_string", Type:
iceberg.PrimitiveTypes.String},
+ )
+ metadata := &cloningSchemaHistoryMetadata{
+ scanTaskDecoderMetadata: &scanTaskDecoderMetadata{schema:
current},
+ history: []*iceberg.Schema{nil, firstHistory,
secondHistory},
+ }
+ spec := iceberg.NewPartitionSpecID(7,
+ iceberg.PartitionField{SourceIDs: []int{1}, FieldID: 1000,
Name: "active", Transform: iceberg.IdentityTransform{}},
+ iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1001,
Name: "old_date", Transform: iceberg.IdentityTransform{}},
+ iceberg.PartitionField{SourceIDs: []int{3}, FieldID: 1002,
Name: "old_long", Transform: iceberg.IdentityTransform{}},
+ iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1003,
Name: "old_date_again", Transform: iceberg.IdentityTransform{}},
+ iceberg.PartitionField{SourceIDs: []int{404}, FieldID: 1004,
Name: "missing", Transform: iceberg.IdentityTransform{}},
+ )
+
+ plan, err := newPartitionDecodePlan(&spec, metadata)
+ require.NoError(t, err)
+ assert.Equal(t, 1, metadata.currentLookups)
+ assert.Equal(t, 1, metadata.historyLookups)
+ require.Len(t, plan.fields, 5)
+ assert.Equal(t, iceberg.PrimitiveTypes.Int64, plan.fields[0].resultType)
+ assert.Equal(t, iceberg.PrimitiveTypes.Date, plan.fields[1].resultType)
+ assert.Equal(t, iceberg.PrimitiveTypes.Int64, plan.fields[2].resultType)
+ assert.Equal(t, iceberg.PrimitiveTypes.Date, plan.fields[3].resultType)
+ assert.False(t, plan.fields[4].sourceTypeFound)
+
+ // Old, dropped partition fields remain readable even after schema
evolution.
+ partition, _, err := decodePartition([]json.RawMessage{
+ json.RawMessage("12"), json.RawMessage("\"2026-07-17\""),
+ json.RawMessage("45"), json.RawMessage("\"2026-07-18\""),
+ json.RawMessage("null"),
+ }, plan)
+ require.NoError(t, err)
+ assert.Equal(t, int64(12), partition[1000])
+ assert.Equal(t, iceberg.Date(20651), partition[1001])
+ assert.Equal(t, int64(45), partition[1002])
+ assert.Equal(t, iceberg.Date(20652), partition[1003])
+ assert.Nil(t, partition[1004])
+}
+
+func TestPartitionDecodePlanSkipsUnneededSchemaCopies(t *testing.T) {
+ t.Parallel()
+
+ current := iceberg.NewSchema(10,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64},
+ iceberg.NestedField{ID: 2, Name: "category", Type:
iceberg.PrimitiveTypes.String},
+ )
+ for _, tc := range []struct {
+ name string
+ sourceIDs []int
+ currentCalls int
+ historyCalls int
+ }{
+ {name: "all current", sourceIDs: []int{1, 2, 1}, currentCalls:
1},
+ {name: "unpartitioned"},
+ {name: "missing source", sourceIDs: []int{101, 102},
currentCalls: 1, historyCalls: 1},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ metadata := &cloningSchemaHistoryMetadata{
+ scanTaskDecoderMetadata:
&scanTaskDecoderMetadata{schema: current},
+ }
+ fields := make([]iceberg.PartitionField,
len(tc.sourceIDs))
+ for index, id := range tc.sourceIDs {
+ fields[index] = iceberg.PartitionField{
+ SourceIDs: []int{id},
+ FieldID: 1000 + index,
+ Name: fmt.Sprintf("part_%d",
index),
+ Transform: iceberg.IdentityTransform{},
+ }
+ }
+ spec := iceberg.NewPartitionSpecID(7, fields...)
+ plan, err := newPartitionDecodePlan(&spec, metadata)
+ require.NoError(t, err)
+ assert.Len(t, plan.fields, len(fields))
+ assert.Equal(t, tc.currentCalls,
metadata.currentLookups)
+ assert.Equal(t, tc.historyCalls,
metadata.historyLookups)
+ })
+ }
+}
