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 89c5e6415 perf(table): append inspect partition values directly (#2116)
89c5e6415 is described below
commit 89c5e6415f704590c6cc566e822419bb14ef0768
Author: Minh Vu <[email protected]>
AuthorDate: Tue Oct 6 10:26:30 2026 +0200
perf(table): append inspect partition values directly (#2116)
* perf(table): append inspect partition values directly
* test(table): cover mixed and special partition values
---
table/inspect_files.go | 99 +++++++++++--
table/inspect_files_bench_test.go | 52 +++++++
table/inspect_partition_append_test.go | 259 +++++++++++++++++++++++++++++++++
3 files changed, 395 insertions(+), 15 deletions(-)
diff --git a/table/inspect_files.go b/table/inspect_files.go
index f10575216..4705f785f 100644
--- a/table/inspect_files.go
+++ b/table/inspect_files.go
@@ -555,11 +555,12 @@ type inspectPartitionBuilder struct {
}
type inspectPartitionFieldBuilder struct {
- id int
- name string
- typ iceberg.Type
- arrowType arrow.DataType
- builder array.Builder
+ id int
+ name string
+ typ iceberg.Type
+ arrowType arrow.DataType
+ builder array.Builder
+ appendValue func(any) error
}
func newInspectContentFileBuilder(bldr *array.RecordBuilder, partitionType
*iceberg.StructType) (inspectContentFileBuilder, error) {
@@ -737,13 +738,15 @@ func newInspectPartitionBuilder(
fields := make([]inspectPartitionFieldBuilder, 0,
len(partitionType.FieldList))
for _, field := range partitionType.FieldList {
fieldBuilder := lookup[field.ID]
- fields = append(fields, inspectPartitionFieldBuilder{
+ partitionField := inspectPartitionFieldBuilder{
id: field.ID,
name: field.Name,
typ: field.Type,
arrowType: fieldBuilder.Type(),
builder: fieldBuilder,
- })
+ }
+ partitionField.appendValue =
newInspectPartitionFieldAppender(partitionField)
+ fields = append(fields, partitionField)
}
return &inspectPartitionBuilder{builder: builder, fields: fields}, nil
@@ -752,22 +755,88 @@ func newInspectPartitionBuilder(
func (b *inspectPartitionBuilder) append(values map[int]any) error {
b.builder.Append(true)
for _, field := range b.fields {
- value := values[field.id]
- if value == nil {
- field.builder.AppendNull()
-
- continue
+ if err := field.appendValue(values[field.id]); err != nil {
+ return err
}
+ }
+
+ return nil
+}
+
+func newInspectPartitionFieldAppender(field inspectPartitionFieldBuilder)
func(any) error {
+ appendScalar := func(value any) error {
sc, err := inspectValueScalar(value, field.typ, field.arrowType)
if err != nil {
return fmt.Errorf("partition field %q: %w", field.name,
err)
}
- if err := scalar.Append(field.builder, sc); err != nil {
- return err
+
+ return scalar.Append(field.builder, sc)
+ }
+ appendNull := func() {
+ field.builder.AppendNull()
+ }
+
+ switch field.typ.(type) {
+ case iceberg.BooleanType:
+ if builder, ok := field.builder.(*array.BooleanBuilder); ok {
+ return newInspectTypedPartitionAppender(builder.Append,
builder.AppendNull, appendScalar)
+ }
+ case iceberg.Int32Type:
+ if builder, ok := field.builder.(*array.Int32Builder); ok {
+ return newInspectTypedPartitionAppender(builder.Append,
builder.AppendNull, appendScalar)
+ }
+ case iceberg.Int64Type:
+ if builder, ok := field.builder.(*array.Int64Builder); ok {
+ return newInspectTypedPartitionAppender(builder.Append,
builder.AppendNull, appendScalar)
+ }
+ case iceberg.Float32Type:
+ if builder, ok := field.builder.(*array.Float32Builder); ok {
+ return newInspectTypedPartitionAppender(builder.Append,
builder.AppendNull, appendScalar)
+ }
+ case iceberg.Float64Type:
+ if builder, ok := field.builder.(*array.Float64Builder); ok {
+ return newInspectTypedPartitionAppender(builder.Append,
builder.AppendNull, appendScalar)
+ }
+ case iceberg.StringType:
+ if builder, ok := field.builder.(*array.StringBuilder); ok {
+ return newInspectTypedPartitionAppender(builder.Append,
builder.AppendNull, appendScalar)
+ }
+ case iceberg.BinaryType:
+ if builder, ok := field.builder.(*array.BinaryBuilder); ok {
+ return newInspectTypedPartitionAppender(builder.Append,
builder.AppendNull, appendScalar)
}
}
- return nil
+ return func(value any) error {
+ if value == nil {
+ appendNull()
+
+ return nil
+ }
+
+ return appendScalar(value)
+ }
+}
+
+func newInspectTypedPartitionAppender[T any](
+ appendValue func(T),
+ appendNull func(),
+ appendScalar func(any) error,
+) func(any) error {
+ return func(value any) error {
+ if value == nil {
+ appendNull()
+
+ return nil
+ }
+ typed, ok := value.(T)
+ if !ok {
+ return appendScalar(value)
+ }
+ appendValue(typed)
+
+ return nil
+ }
}
func inspectValueScalar(value any, typ iceberg.Type, arrowType arrow.DataType)
(scalar.Scalar, error) {
diff --git a/table/inspect_files_bench_test.go
b/table/inspect_files_bench_test.go
index ac5604a44..7e980aaf3 100644
--- a/table/inspect_files_bench_test.go
+++ b/table/inspect_files_bench_test.go
@@ -97,3 +97,55 @@ func benchmarkInspectContentFiles(
return partitionType, files
}
+
+func BenchmarkInspectContentFileAppenderMixedPartitions(b *testing.B) {
+ types := []iceberg.Type{iceberg.PrimitiveTypes.Bool,
iceberg.PrimitiveTypes.Int32, iceberg.PrimitiveTypes.Int64,
iceberg.PrimitiveTypes.Float32, iceberg.PrimitiveTypes.Float64,
iceberg.PrimitiveTypes.String, iceberg.PrimitiveTypes.Binary,
iceberg.PrimitiveTypes.Date}
+ raw := []any{true, int32(34), int64(9007199254740993), float32(1.25),
float64(2.5), "partition", []byte("binary"), iceberg.Date(12)}
+ fields := make([]iceberg.NestedField, len(types))
+ partitions := make([]iceberg.PartitionField, len(types))
+ for i, typ := range types {
+ fields[i] = iceberg.NestedField{ID: i + 1, Name:
fmt.Sprintf("field_%d", i), Type: typ}
+ partitions[i] = iceberg.PartitionField{SourceIDs: []int{i + 1},
FieldID: 1000 + i, Name: fields[i].Name, Transform: iceberg.IdentityTransform{}}
+ }
+ spec := iceberg.NewPartitionSpec(partitions...)
+ partitionType := spec.PartitionType(iceberg.NewSchema(0, fields...))
+ for _, nullEvery := range []int{0, 2} {
+ b.Run(fmt.Sprintf("fields=8/files=4096/null_every=%d",
nullEvery), func(b *testing.B) {
+ files := make([]iceberg.DataFile, 4096)
+ for i := range files {
+ values := make(map[int]any, len(types))
+ for j, value := range raw {
+ if nullEvery > 0 && (i+j)%nullEvery ==
0 {
+ value = nil
+ }
+ values[1000+j] = value
+ }
+ file, err := iceberg.NewDataFileBuilder(spec,
iceberg.EntryContentData, fmt.Sprintf("file-%d.parquet", i),
iceberg.ParquetFile, values, map[int]string{1007: "date"}, nil, 1, 1)
+ if err != nil {
+ b.Fatal(err)
+ }
+ files[i] = file.Build()
+ }
+ arrowSchema, err :=
SchemaToArrowSchema(DataFilesSchema(partitionType), nil, true, false)
+ if err != nil {
+ b.Fatal(err)
+ }
+ builder :=
array.NewRecordBuilder(memory.DefaultAllocator, arrowSchema)
+ defer builder.Release()
+ appendFile :=
newInspectContentFileAppender(partitionType)
+ b.ReportAllocs()
+ for b.Loop() {
+ for _, file := range files {
+ if err := appendFile(builder, file);
err != nil {
+ b.Fatal(err)
+ }
+ }
+ record := builder.NewRecordBatch()
+ if record.NumRows() != int64(len(files)) {
+ b.Fatalf("got %d rows, want %d",
record.NumRows(), len(files))
+ }
+ record.Release()
+ }
+ })
+ }
+}
diff --git a/table/inspect_partition_append_test.go
b/table/inspect_partition_append_test.go
new file mode 100644
index 000000000..1706f3431
--- /dev/null
+++ b/table/inspect_partition_append_test.go
@@ -0,0 +1,259 @@
+// 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 (
+ "math"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/decimal128"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/require"
+)
+
+func TestInspectPartitionBuilderAppendsTypedAndFallbackValues(t *testing.T) {
+ tests := []struct {
+ name string
+ typ iceberg.Type
+ value any
+ check func(*testing.T, arrow.Array)
+ }{
+ {
+ name: "boolean",
+ typ: iceberg.PrimitiveTypes.Bool,
+ value: true,
+ check: func(t *testing.T, value arrow.Array) {
+ require.True(t, value.(*array.Boolean).Value(0))
+ },
+ },
+ {
+ name: "int32",
+ typ: iceberg.PrimitiveTypes.Int32,
+ value: int32(32),
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, int32(32),
value.(*array.Int32).Value(0))
+ },
+ },
+ {
+ name: "int64",
+ typ: iceberg.PrimitiveTypes.Int64,
+ value: int64(64),
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, int64(64),
value.(*array.Int64).Value(0))
+ },
+ },
+ {
+ name: "float32",
+ typ: iceberg.PrimitiveTypes.Float32,
+ value: float32(1.25),
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, float32(1.25),
value.(*array.Float32).Value(0))
+ },
+ },
+ {
+ name: "float64",
+ typ: iceberg.PrimitiveTypes.Float64,
+ value: float64(2.5),
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, float64(2.5),
value.(*array.Float64).Value(0))
+ },
+ },
+ {
+ name: "string",
+ typ: iceberg.PrimitiveTypes.String,
+ value: "partition",
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, "partition",
value.(*array.String).Value(0))
+ },
+ },
+ {
+ name: "binary",
+ typ: iceberg.PrimitiveTypes.Binary,
+ value: []byte{0, 1, 255},
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, []byte{0, 1, 255},
value.(*array.Binary).Value(0))
+ },
+ },
+ {
+ name: "date scalar fallback",
+ typ: iceberg.PrimitiveTypes.Date,
+ value: iceberg.Date(12),
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, arrow.Date32(12),
value.(*array.Date32).Value(0))
+ },
+ },
+ {
+ name: "time scalar fallback",
+ typ: iceberg.PrimitiveTypes.Time,
+ value: iceberg.Time(1234),
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, arrow.Time64(1234),
value.(*array.Time64).Value(0))
+ },
+ },
+ {
+ name: "timestamp scalar fallback",
+ typ: iceberg.PrimitiveTypes.Timestamp,
+ value: iceberg.Timestamp(1234),
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, arrow.Timestamp(1234),
value.(*array.Timestamp).Value(0))
+ },
+ },
+ {
+ name: "nanosecond timestamp scalar fallback",
+ typ: iceberg.PrimitiveTypes.TimestampNs,
+ value: iceberg.TimestampNano(1234),
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, arrow.Timestamp(1234),
value.(*array.Timestamp).Value(0))
+ },
+ },
+ {
+ name: "decimal scalar fallback",
+ typ: iceberg.DecimalTypeOf(10, 2),
+ value: iceberg.DecimalLiteral{Val:
decimal128.FromI64(123), Scale: 2},
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, decimal128.FromI64(123),
value.(*array.Decimal128).Value(0))
+ },
+ },
+ {
+ name: "fixed scalar fallback",
+ typ: iceberg.FixedTypeOf(3),
+ value: []byte("abc"),
+ check: func(t *testing.T, value arrow.Array) {
+ require.Equal(t, []byte("abc"),
value.(*array.FixedSizeBinary).Value(0))
+ },
+ },
+ }
+
+ fields := make([]iceberg.NestedField, len(tests))
+ values := make(map[int]any, len(tests))
+ for i, tt := range tests {
+ fields[i] = iceberg.NestedField{ID: i + 2, Name: tt.name, Type:
tt.typ}
+ values[i+2] = tt.value
+ }
+ builder, partitionBuilder := newInspectPartitionBuilderForTest(t,
fields)
+ require.NoError(t, partitionBuilder.append(values))
+
+ partition := builder.NewArray().(*array.Struct)
+ defer partition.Release()
+ for i, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tt.check(t, partition.Field(i))
+ })
+ }
+}
+
+func TestInspectPartitionBuilderAppendsNullValues(t *testing.T) {
+ builder, partitionBuilder := newInspectPartitionBuilderForTest(t,
[]iceberg.NestedField{
+ {ID: 2, Name: "value", Type: iceberg.PrimitiveTypes.Int32},
+ })
+ require.NoError(t, partitionBuilder.append(nil))
+
+ partition := builder.NewArray().(*array.Struct)
+ defer partition.Release()
+ require.True(t, partition.Field(0).IsNull(0))
+}
+
+func TestInspectPartitionBuilderScalarFallbackPreservesValidation(t
*testing.T) {
+ tests := []struct {
+ name string
+ typ iceberg.Type
+ value any
+ wantErr string
+ }{
+ {name: "compatible integer string", typ:
iceberg.PrimitiveTypes.Int32, value: "42"},
+ {name: "invalid integer string", typ:
iceberg.PrimitiveTypes.Int32, value: "not-an-int", wantErr: "partition field
\"value\": strconv.ParseInt"},
+ {name: "wrong date value", typ: iceberg.PrimitiveTypes.Date,
value: int64(1), wantErr: "partition field \"value\": unsupported date"},
+ {name: "wrong fixed width", typ: iceberg.FixedTypeOf(2), value:
[]byte("x"), wantErr: "invalid scalar value of len"},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ builder, partitionBuilder :=
newInspectPartitionBuilderForTest(t, []iceberg.NestedField{
+ {ID: 2, Name: "value", Type: tt.typ},
+ })
+ err := partitionBuilder.append(map[int]any{2: tt.value})
+ if tt.wantErr != "" {
+ require.ErrorContains(t, err, tt.wantErr)
+
+ return
+ }
+ require.NoError(t, err)
+ partition := builder.NewArray().(*array.Struct)
+ defer partition.Release()
+ require.Equal(t, int32(42),
partition.Field(0).(*array.Int32).Value(0))
+ })
+ }
+}
+
+func newInspectPartitionBuilderForTest(
+ t *testing.T,
+ fields []iceberg.NestedField,
+) (*array.StructBuilder, *inspectPartitionBuilder) {
+ t.Helper()
+ partitionType := &iceberg.StructType{FieldList: fields}
+ schema, err := SchemaToArrowSchema(iceberg.NewSchema(0,
iceberg.NestedField{
+ ID: 1, Name: "partition", Type: partitionType, Required: true,
+ }), nil, true, false)
+ require.NoError(t, err)
+ structType, ok := schema.Field(0).Type.(*arrow.StructType)
+ require.True(t, ok)
+ builder := array.NewStructBuilder(memory.DefaultAllocator, structType)
+ t.Cleanup(builder.Release)
+ partitionBuilder, err := newInspectPartitionBuilder(builder,
partitionType)
+ require.NoError(t, err)
+
+ return builder, partitionBuilder
+}
+
+func TestInspectPartitionBuilderPreservesEmptyAndSpecialValues(t *testing.T) {
+ for _, tt := range []struct {
+ name string
+ typ iceberg.Type
+ value any
+ check func(*testing.T, arrow.Array)
+ }{
+ {"empty string", iceberg.PrimitiveTypes.String, "", func(t
*testing.T, a arrow.Array) { require.Empty(t, a.(*array.String).Value(0)) }},
+ {"empty binary", iceberg.PrimitiveTypes.Binary, []byte{},
func(t *testing.T, a arrow.Array) { require.Empty(t,
a.(*array.Binary).Value(0)) }},
+ {"float32 negative zero", iceberg.PrimitiveTypes.Float32,
float32(math.Copysign(0, -1)), func(t *testing.T, a arrow.Array) {
require.True(t, math.Signbit(float64(a.(*array.Float32).Value(0)))) }},
+ {"float64 NaN", iceberg.PrimitiveTypes.Float64, math.NaN(),
func(t *testing.T, a arrow.Array) { require.True(t,
math.IsNaN(a.(*array.Float64).Value(0))) }},
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ builder, partitionBuilder :=
newInspectPartitionBuilderForTest(t, []iceberg.NestedField{{ID: 2, Name:
"value", Type: tt.typ}})
+ require.NoError(t,
partitionBuilder.append(map[int]any{2: tt.value}))
+ require.NoError(t, partitionBuilder.append(nil))
+ partition := builder.NewArray().(*array.Struct)
+ defer partition.Release()
+ require.True(t, partition.Field(0).IsValid(0))
+ require.True(t, partition.Field(0).IsNull(1))
+ tt.check(t, partition.Field(0))
+ })
+ }
+}
+
+func TestInspectPartitionBuilderCopiesBinaryValues(t *testing.T) {
+ builder, partitionBuilder := newInspectPartitionBuilderForTest(t,
[]iceberg.NestedField{{ID: 2, Name: "value", Type:
iceberg.PrimitiveTypes.Binary}})
+ value := []byte("abc")
+ require.NoError(t, partitionBuilder.append(map[int]any{2: value}))
+ value[0] = 'x'
+ partition := builder.NewArray().(*array.Struct)
+ defer partition.Release()
+ require.Equal(t, []byte("abc"),
partition.Field(0).(*array.Binary).Value(0))
+}
