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 7536627ab perf(table): reuse partition path formatting plan (#1958)
7536627ab is described below
commit 7536627ab9a538b64bd0f86d7a9bdf5ec74180b1
Author: Minh Vu <[email protected]>
AuthorDate: Fri Sep 25 10:59:30 2026 +0200
perf(table): reuse partition path formatting plan (#1958)
Signed-off-by: Minh Vu <[email protected]>
---
table/clustered_writer.go | 4 +-
table/partition_extraction_bench_test.go | 44 +++++++++++++++++
table/partition_path.go | 60 +++++++++++++++++++++++
table/partition_path_test.go | 81 ++++++++++++++++++++++++++++++++
table/partitioned_fanout_writer.go | 12 ++++-
5 files changed, 198 insertions(+), 3 deletions(-)
diff --git a/table/clustered_writer.go b/table/clustered_writer.go
index 9f5b7e97d..b14f6689c 100644
--- a/table/clustered_writer.go
+++ b/table/clustered_writer.go
@@ -119,12 +119,12 @@ func clusteredPartitionedWrite(
return err
}
if
completedPartitions.contains(part.partitionRec) {
- partitionPath :=
spec.PartitionToPath(part.partitionRec, schema)
+ partitionPath :=
extractionPlan.pathPlan.format(part.partitionRec)
return fmt.Errorf("clustered write:
incoming records violate the clustering assumption; "+
"partition %q has records
arriving after its writer was already closed", partitionPath)
}
- partitionPath :=
spec.PartitionToPath(part.partitionRec, schema)
+ partitionPath :=
extractionPlan.pathPlan.format(part.partitionRec)
currentWriter = factory.newRollingDataWriter(
ctx, partitionPath,
part.partitionValues, outputCh)
currentRec = part.partitionRec
diff --git a/table/partition_extraction_bench_test.go
b/table/partition_extraction_bench_test.go
index d88bd78fd..ff0f02a6c 100644
--- a/table/partition_extraction_bench_test.go
+++ b/table/partition_extraction_bench_test.go
@@ -258,3 +258,47 @@ func BenchmarkPartitionTransforms(b *testing.B) {
})
}
}
+
+func BenchmarkPartitionPathFormatting(b *testing.B) {
+ for _, fieldCount := range []int{1, 4, 16, 64} {
+ b.Run(fmt.Sprintf("fields_%d", fieldCount), func(b *testing.B) {
+ arrowFields := make([]arrow.Field, fieldCount)
+ icebergFields := make([]iceberg.NestedField, fieldCount)
+ specFields := make([]iceberg.PartitionField, fieldCount)
+ values := make(partitionRecord, fieldCount)
+ for i := range fieldCount {
+ name := fmt.Sprintf("part #%d", i)
+ arrowFields[i] = arrow.Field{Name: name, Type:
arrow.BinaryTypes.String}
+ icebergFields[i] = iceberg.NestedField{ID: i +
1, Name: name, Type: iceberg.PrimitiveTypes.String}
+ specFields[i] = iceberg.PartitionField{
+ SourceIDs: []int{i + 1}, FieldID: 1000
+ i,
+ Transform: iceberg.IdentityTransform{},
Name: name,
+ }
+ values[i] = fmt.Sprintf("value/%d", i)
+ }
+
+ schema := iceberg.NewSchema(0, icebergFields...)
+ spec := iceberg.NewPartitionSpec(specFields...)
+ arrowSchema := arrow.NewSchema(arrowFields, nil)
+ plan, err := newPartitionExtractionPlan(spec, schema,
arrowSchema)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ b.Run("partition_to_path", func(b *testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ _ = spec.PartitionToPath(values, schema)
+ }
+ })
+ b.Run("reused_plan", func(b *testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ _ = plan.pathPlan.format(values)
+ }
+ })
+ })
+ }
+}
diff --git a/table/partition_path.go b/table/partition_path.go
new file mode 100644
index 000000000..5c360d1e2
--- /dev/null
+++ b/table/partition_path.go
@@ -0,0 +1,60 @@
+// 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 (
+ "net/url"
+ "strings"
+
+ "github.com/apache/iceberg-go"
+)
+
+type partitionPathField struct {
+ escapedName string
+ transform iceberg.Transform
+ resultType iceberg.Type
+}
+
+// partitionPathPlan contains the spec and schema state needed to format a
+// partition path. The values are immutable after construction, so a plan can
+// be shared by all workers handling the same table write.
+type partitionPathPlan struct {
+ fields []partitionPathField
+ estimatedSize int
+}
+
+func (p partitionPathPlan) format(data iceberg.StructLike) string {
+ if len(p.fields) == 0 {
+ return ""
+ }
+
+ var sb strings.Builder
+ sb.Grow(p.estimatedSize)
+ for i, field := range p.fields {
+ if i > 0 {
+ sb.WriteByte('/')
+ }
+
+ sb.WriteString(field.escapedName)
+ sb.WriteByte('=')
+ value := field.transform.ToHumanStrType(field.resultType,
data.Get(i))
+ sb.WriteString(url.QueryEscape(value))
+ }
+
+ return sb.String()
+}
diff --git a/table/partition_path_test.go b/table/partition_path_test.go
new file mode 100644
index 000000000..358cecd36
--- /dev/null
+++ b/table/partition_path_test.go
@@ -0,0 +1,81 @@
+// 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/arrow-go/v18/arrow"
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestPartitionPathPlanMatchesPartitionToPath(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "created_ts_tz", Type:
iceberg.PrimitiveTypes.TimestampTz, Required: true},
+ iceberg.NestedField{ID: 2, Name: "str", Type:
iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 3, Name: "id", Type:
iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+
+ unknown, err := iceberg.ParseTransform("custom_transform[42]")
+ require.NoError(t, err)
+ spec := iceberg.NewPartitionSpec(
+ iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000,
+ Transform: iceberg.IdentityTransform{}, Name: "created
ts/tz",
+ },
+ iceberg.PartitionField{
+ SourceIDs: []int{2}, FieldID: 1001,
+ Transform: iceberg.TruncateTransform{Width: 8}, Name:
"str#part",
+ },
+ iceberg.PartitionField{
+ SourceIDs: []int{3}, FieldID: 1002,
+ Transform: iceberg.BucketTransform{NumBuckets: 16},
Name: "id_bucket",
+ },
+ iceberg.PartitionField{
+ SourceIDs: []int{4}, FieldID: 1003,
+ Transform: iceberg.IdentityTransform{}, Name: "missing",
+ },
+ iceberg.PartitionField{
+ SourceIDs: []int{3}, FieldID: 1004,
+ Transform: unknown, Name: "unknown",
+ },
+ )
+
+ recordSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "created_ts_tz", Type:
arrow.FixedWidthTypes.Timestamp_us, Nullable: false},
+ {Name: "str", Type: arrow.BinaryTypes.String},
+ {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: false},
+ }, nil)
+ plan, err := newPartitionExtractionPlan(spec, schema, recordSchema)
+ require.NoError(t, err)
+
+ record := partitionRecord{
+ iceberg.Timestamp(1705314600000000),
+ "partition/value",
+ int32(7),
+ nil,
+ int32(42),
+ }
+
+ assert.Equal(t, spec.PartitionToPath(record, schema),
plan.pathPlan.format(record))
+ assert.Equal(t,
+
"created+ts%2Ftz=2024-01-15T10%3A30%3A00%2B00%3A00/str%23part=partition%2Fvalue/id_bucket=7/missing=null/unknown=42",
+ plan.pathPlan.format(record))
+}
diff --git a/table/partitioned_fanout_writer.go
b/table/partitioned_fanout_writer.go
index b70d144cd..50c5f7f9a 100644
--- a/table/partitioned_fanout_writer.go
+++ b/table/partitioned_fanout_writer.go
@@ -76,6 +76,7 @@ type partitionExtractionPlan struct {
schema *iceberg.Schema
recordSchema *arrow.Schema
fields []partitionFieldInfo
+ pathPlan partitionPathPlan
}
type (
@@ -155,7 +156,7 @@ func newPartitionedFanoutWriter(partitionSpec
iceberg.PartitionSpec, schema *ice
}
func (p *partitionedFanoutWriter) partitionPath(data partitionRecord) string {
- return p.partitionSpec.PartitionToPath(data, p.schema)
+ return p.plan.pathPlan.format(data)
}
// Write writes the Arrow records to the specified location using a fanout
pattern with
@@ -356,6 +357,7 @@ func getRecordPartitions(spec iceberg.PartitionSpec, schema
*iceberg.Schema, rec
func newPartitionExtractionPlan(spec iceberg.PartitionSpec, schema
*iceberg.Schema, recordSchema *arrow.Schema) (*partitionExtractionPlan, error) {
partitionFields := spec.PartitionType(schema).FieldList
partitionFieldsInfo := make([]partitionFieldInfo, len(partitionFields))
+ pathPlan := partitionPathPlan{fields: make([]partitionPathField,
len(partitionFields))}
specFieldsByID := make(map[int]iceberg.PartitionField, spec.NumFields())
for _, field := range spec.Fields() {
specFieldsByID[field.FieldID] = field
@@ -366,6 +368,13 @@ func newPartitionExtractionPlan(spec
iceberg.PartitionSpec, schema *iceberg.Sche
if !ok {
return nil, fmt.Errorf("failed to find partition field
ID %d in spec", partitionField.ID)
}
+ pathField := partitionPathField{
+ escapedName: sourceField.EscapedName(),
+ transform: sourceField.Transform,
+ resultType: partitionField.Type,
+ }
+ pathPlan.fields[i] = pathField
+ pathPlan.estimatedSize += len(pathField.escapedName) + 20
partitionFieldsInfo[i].fieldID = partitionField.ID
colName, ok := schema.FindColumnName(sourceField.SourceID())
if !ok {
@@ -394,6 +403,7 @@ func newPartitionExtractionPlan(spec iceberg.PartitionSpec,
schema *iceberg.Sche
schema: schema,
recordSchema: recordSchema,
fields: partitionFieldsInfo,
+ pathPlan: pathPlan,
}, nil
}
