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 c9b8a7e3f perf(table): reuse partition formatting plans in snapshot
summaries (#2200)
c9b8a7e3f is described below
commit c9b8a7e3fb1713a12de78db375ce297018875d11
Author: Minh Vu <[email protected]>
AuthorDate: Mon Oct 19 19:34:00 2026 +0200
perf(table): reuse partition formatting plans in snapshot summaries (#2200)
* perf(table): reuse partition formatting plans in snapshot summaries
* perf(table): guard snapshot partition plan cache against spec ID reuse
Signed-off-by: Minh Vu <[email protected]>
* test(table): benchmark escaped partition path formatting
* style(table): simplify escaped path benchmark cases
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/snapshots.go | 56 ++++++++++++++++++++--
table/snapshots_bench_test.go | 100 ++++++++++++++++++++++++++++++++++++++-
table/snapshots_internal_test.go | 62 ++++++++++++++++++++++++
3 files changed, 213 insertions(+), 5 deletions(-)
diff --git a/table/snapshots.go b/table/snapshots.go
index a66acce83..6e75bbce5 100644
--- a/table/snapshots.go
+++ b/table/snapshots.go
@@ -580,9 +580,59 @@ func SnapshotLogEntryAsOf(entries
iter.Seq[SnapshotLogEntry], timestampMs int64,
type SnapshotSummaryCollector struct {
metrics updateMetrics
partitionMetrics map[string]updateMetrics
+ partitionPathPlans
map[snapshotPartitionPathKey]snapshotPartitionPathPlan
maxChangedPartitionsForSummaries int
}
+// Spec IDs are unique only within a table. Include schema and spec identity
+// so an independently constructed spec with the same ID cannot reuse a stale
+// formatting plan if the collector is used across table metadata.
+type snapshotPartitionPathKey struct {
+ schema *iceberg.Schema
+ specID int
+ firstField *iceberg.PartitionField
+}
+
+type snapshotPartitionPathPlan struct {
+ partitionType *iceberg.StructType
+ pathPlan partitionPathPlan
+}
+
+// partitionPath reuses schema-dependent formatting state across files in the
+// same snapshot summary. The partition field slice is immutable for the
+// lifetime of its spec, so its first element identifies the spec's layout.
+func (s *SnapshotSummaryCollector) partitionPath(partition map[int]any, sc
*iceberg.Schema, spec iceberg.PartitionSpec) string {
+ if s.partitionPathPlans == nil {
+ s.partitionPathPlans =
make(map[snapshotPartitionPathKey]snapshotPartitionPathPlan)
+ }
+
+ fields := spec.FieldsRef(internal.PartitionSpecRef{})
+ key := snapshotPartitionPathKey{schema: sc, specID: spec.ID()}
+ if len(fields) > 0 {
+ key.firstField = &fields[0]
+ }
+
+ plan, ok := s.partitionPathPlans[key]
+ if !ok {
+ partitionType := spec.PartitionType(sc)
+ pathPlan := partitionPathPlan{fields:
make([]partitionPathField, 0, len(partitionType.FieldList))}
+ for i, field := range spec.Fields() {
+ pathField := partitionPathField{
+ escapedName: field.EscapedName(),
+ transform: field.Transform,
+ resultType: partitionType.FieldList[i].Type,
+ }
+ pathPlan.fields = append(pathPlan.fields, pathField)
+ pathPlan.estimatedSize += len(pathField.escapedName) +
20
+ }
+
+ plan = snapshotPartitionPathPlan{partitionType: partitionType,
pathPlan: pathPlan}
+ s.partitionPathPlans[key] = plan
+ }
+
+ return plan.pathPlan.format(borrowedPartitionRecord{partition:
partition, partitionType: plan.partitionType})
+}
+
func (s *SnapshotSummaryCollector) setPartitionSummaryLimit(limit int) {
s.maxChangedPartitionsForSummaries = limit
}
@@ -614,8 +664,7 @@ func (s *SnapshotSummaryCollector) addFile(df
iceberg.DataFile, sc *iceberg.Sche
partition := dataFilePartition(df)
if len(partition) > 0 {
- partitionPath := spec.PartitionToPath(
- newPartitionRecord(partition, spec.PartitionType(sc)),
sc)
+ partitionPath := s.partitionPath(partition, sc, spec)
return s.updatePartitionMetrics(partitionPath, df, true)
}
@@ -630,8 +679,7 @@ func (s *SnapshotSummaryCollector) removeFile(df
iceberg.DataFile, sc *iceberg.S
partition := dataFilePartition(df)
if len(partition) > 0 {
- partitionPath := spec.PartitionToPath(
- newPartitionRecord(partition, spec.PartitionType(sc)),
sc)
+ partitionPath := s.partitionPath(partition, sc, spec)
return s.updatePartitionMetrics(partitionPath, df, false)
}
diff --git a/table/snapshots_bench_test.go b/table/snapshots_bench_test.go
index d8af4d272..42f8dd887 100644
--- a/table/snapshots_bench_test.go
+++ b/table/snapshots_bench_test.go
@@ -17,7 +17,13 @@
package table
-import "testing"
+import (
+ "fmt"
+ "strconv"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+)
var partitionSummaryBenchmarkResult string
@@ -54,3 +60,95 @@ func BenchmarkPartitionSummary(b *testing.B) {
})
}
}
+
+var snapshotSummaryPathBenchmarkResult string
+
+func BenchmarkSnapshotSummaryPartitionPath(b *testing.B) {
+ for _, fieldCount := range []int{1, 3, 16, 64} {
+ fields := make([]iceberg.NestedField, fieldCount)
+ partitionFields := make([]iceberg.PartitionField, fieldCount)
+ partition := make(map[int]any, fieldCount)
+ for i := range fields {
+ name := "field_" + strconv.Itoa(i)
+ fields[i] = iceberg.NestedField{ID: i + 1, Name: name,
Type: iceberg.PrimitiveTypes.Int32}
+ partitionFields[i] = iceberg.PartitionField{
+ SourceIDs: []int{i + 1}, FieldID: 1000 + i,
Name: name,
+ Transform: iceberg.IdentityTransform{},
+ }
+ partition[1000+i] = int32(i)
+ }
+ schema := iceberg.NewSchema(0, fields...)
+ spec := iceberg.NewPartitionSpec(partitionFields...)
+
+ b.Run(fmt.Sprintf("fields=%d", fieldCount), func(b *testing.B) {
+ b.Run("baseline", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ snapshotSummaryPathBenchmarkResult =
spec.PartitionToPath(
+ newPartitionRecord(partition,
spec.PartitionType(schema)), schema)
+ }
+ })
+
+ b.Run("cold", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ var collector SnapshotSummaryCollector
+ snapshotSummaryPathBenchmarkResult =
collector.partitionPath(partition, schema, spec)
+ }
+ })
+
+ b.Run("cached", func(b *testing.B) {
+ var collector SnapshotSummaryCollector
+ snapshotSummaryPathBenchmarkResult =
collector.partitionPath(partition, schema, spec)
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ snapshotSummaryPathBenchmarkResult =
collector.partitionPath(partition, schema, spec)
+ }
+ })
+ })
+ }
+}
+
+func BenchmarkSnapshotSummaryPartitionPathEscaped(b *testing.B) {
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "region", Type: iceberg.PrimitiveTypes.String,
+ })
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "region",
+ Transform: iceberg.IdentityTransform{},
+ })
+
+ for _, tc := range []struct {
+ name, value string
+ }{
+ {name: "plain", value: "berlin"},
+ {name: "escaped", value: "berlin/east +%?#"},
+ {name: "unicode", value: "München/東京"},
+ } {
+ b.Run(tc.name, func(b *testing.B) {
+ partition := map[int]any{1000: tc.value}
+ want :=
spec.PartitionToPath(newPartitionRecord(partition, spec.PartitionType(schema)),
schema)
+ var collector SnapshotSummaryCollector
+ if got := collector.partitionPath(partition, schema,
spec); got != want {
+ b.Fatalf("partition path changed: got %q, want
%q", got, want)
+ }
+
+ b.Run("baseline", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ snapshotSummaryPathBenchmarkResult =
spec.PartitionToPath(
+ newPartitionRecord(partition,
spec.PartitionType(schema)), schema)
+ }
+ })
+
+ b.Run("cached", func(b *testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ snapshotSummaryPathBenchmarkResult =
collector.partitionPath(partition, schema, spec)
+ }
+ })
+ })
+ }
+}
diff --git a/table/snapshots_internal_test.go b/table/snapshots_internal_test.go
index edefaead6..b35d5f70a 100644
--- a/table/snapshots_internal_test.go
+++ b/table/snapshots_internal_test.go
@@ -73,6 +73,7 @@ func TestSnapshotSummaryCollectorWithPartition(t *testing.T) {
ssc.addFile(dataFile1, sc, spec)
ssc.removeFile(dataFile1, sc, spec)
ssc.removeFile(dataFile2, sc, spec)
+ require.Len(t, ssc.partitionPathPlans, 1)
assert.Equal(t, iceberg.Properties{
"added-files-size": "1234",
@@ -98,6 +99,67 @@ func TestSnapshotSummaryCollectorWithPartition(t *testing.T)
{
}, ssc.build())
}
+func TestSnapshotSummaryCollectorPartitionPathParity(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "ts", Type:
iceberg.PrimitiveTypes.TimestampTz},
+ iceberg.NestedField{ID: 2, Name: "text", Type:
iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 3, Name: "integer", Type:
iceberg.PrimitiveTypes.Int32},
+ iceberg.NestedField{ID: 4, Name: "bytes", Type:
iceberg.PrimitiveTypes.Binary},
+ )
+ unknown, err := iceberg.ParseTransform("custom_transform[42]")
+ require.NoError(t, err)
+ spec := iceberg.NewPartitionSpecID(7,
+ iceberg.PartitionField{SourceIDs: []int{1}, FieldID: 1000,
Name: "ts /tz", Transform: iceberg.IdentityTransform{}},
+ iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1001,
Name: "text#part", Transform: iceberg.TruncateTransform{Width: 8}},
+ iceberg.PartitionField{SourceIDs: []int{3}, FieldID: 1002,
Name: "bucket", Transform: iceberg.BucketTransform{NumBuckets: 16}},
+ iceberg.PartitionField{SourceIDs: []int{9}, FieldID: 1003,
Name: "missing", Transform: iceberg.IdentityTransform{}},
+ iceberg.PartitionField{SourceIDs: []int{3}, FieldID: 1004,
Name: "custom", Transform: unknown},
+ iceberg.PartitionField{SourceIDs: []int{4}, FieldID: 1005,
Name: "binary", Transform: iceberg.IdentityTransform{}},
+ )
+
+ partitions := []map[int]any{
+ {1000: iceberg.Timestamp(1705314600000000), 1001: "a/b c",
1002: int32(7), 1003: nil, 1004: int32(42), 1005: []byte{0, 1, 255}},
+ {1000: iceberg.Timestamp(0), 1001: "", 1002: int32(0), 1003:
nil, 1004: nil, 1005: []byte{}},
+ }
+
+ var collector SnapshotSummaryCollector
+ for _, partition := range partitions {
+ want := spec.PartitionToPath(newPartitionRecord(partition,
spec.PartitionType(schema)), schema)
+ assert.Equal(t, want, collector.partitionPath(partition,
schema, spec))
+ }
+ require.Len(t, collector.partitionPathPlans, 1)
+
+ otherSpec := iceberg.NewPartitionSpecID(8,
+ iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1100,
Name: "other#text", Transform: iceberg.IdentityTransform{}},
+ )
+ otherPartition := map[int]any{1100: "a/b c"}
+ want := otherSpec.PartitionToPath(newPartitionRecord(otherPartition,
otherSpec.PartitionType(schema)), schema)
+ assert.Equal(t, want, collector.partitionPath(otherPartition, schema,
otherSpec))
+ require.Len(t, collector.partitionPathPlans, 2)
+
+ // Partition spec IDs are table-local. Reusing an ID must not reuse the
+ // cached plan when a different spec has been supplied.
+ sameIDSpec := iceberg.NewPartitionSpecID(spec.ID(),
+ iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1101,
Name: "same-id#text", Transform: iceberg.IdentityTransform{}},
+ )
+ sameIDPartition := map[int]any{1101: "a/b c"}
+ want = sameIDSpec.PartitionToPath(
+ newPartitionRecord(sameIDPartition,
sameIDSpec.PartitionType(schema)), schema)
+ assert.Equal(t, want, collector.partitionPath(sameIDPartition, schema,
sameIDSpec))
+ require.Len(t, collector.partitionPathPlans, 3)
+
+ // A schema change can also change how an existing spec formats values.
+ alternateSchema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "ts", Type:
iceberg.PrimitiveTypes.Timestamp},
+ iceberg.NestedField{ID: 2, Name: "text", Type:
iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 3, Name: "integer", Type:
iceberg.PrimitiveTypes.Int32},
+ iceberg.NestedField{ID: 4, Name: "bytes", Type:
iceberg.PrimitiveTypes.Binary},
+ )
+ want = spec.PartitionToPath(newPartitionRecord(partitions[0],
spec.PartitionType(alternateSchema)), alternateSchema)
+ assert.Equal(t, want, collector.partitionPath(partitions[0],
alternateSchema, spec))
+ require.Len(t, collector.partitionPathPlans, 4)
+}
+
func TestMergeSnapshotSummaries(t *testing.T) {
tests := []struct {
sum Summary
