(iceberg-go) branch main updated: perf(table): borrow DataFile stats in metadata tables (#1896)

[at...@apache.org]

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 8748cbd89 perf(table): borrow DataFile stats in metadata tables (#1896)
8748cbd89 is described below

commit 8748cbd896551145610ff9d59ec952948b010e4b
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Thu Aug 27 19:14:51 2026 +0200

    perf(table): borrow DataFile stats in metadata tables (#1896)
---
 table/inspect_files.go                  |  13 +--
 table/inspect_files_stats_bench_test.go | 166 ++++++++++++++++++++++++++++++++
 table/inspect_files_stats_test.go       | 117 ++++++++++++++++++++++
 3 files changed, 290 insertions(+), 6 deletions(-)

diff --git a/table/inspect_files.go b/table/inspect_files.go
index fe80b4699..d1fc4c7b2 100644
--- a/table/inspect_files.go
+++ b/table/inspect_files.go
@@ -655,7 +655,7 @@ func (b inspectContentFileBuilder) append(file iceberg.DataFile) error {
 	b.specID.Append(file.SpecID())
 
 	if b.partition != nil {
-		if err := b.partition.append(file.Partition()); err != nil {
+		if err := b.partition.append(dataFilePartition(file)); err != nil {
 			return err
 		}
 	}
@@ -665,19 +665,20 @@ func (b inspectContentFileBuilder) append(file iceberg.DataFile) error {
 	if err := appendInspectInt64Map(b.columnSizes, file.ColumnSizes()); err != nil {
 		return err
 	}
-	if err := appendInspectInt64Map(b.valueCounts, file.ValueCounts()); err != nil {
+	valueCounts, nullCounts, nanCounts, lowerBounds, upperBounds := dataFileStats(file)
+	if err := appendInspectInt64Map(b.valueCounts, valueCounts); err != nil {
 		return err
 	}
-	if err := appendInspectInt64Map(b.nullValueCounts, file.NullValueCounts()); err != nil {
+	if err := appendInspectInt64Map(b.nullValueCounts, nullCounts); err != nil {
 		return err
 	}
-	if err := appendInspectInt64Map(b.nanValueCounts, file.NaNValueCounts()); err != nil {
+	if err := appendInspectInt64Map(b.nanValueCounts, nanCounts); err != nil {
 		return err
 	}
-	if err := appendInspectBinaryMap(b.lowerBounds, file.LowerBoundValues()); err != nil {
+	if err := appendInspectBinaryMap(b.lowerBounds, lowerBounds); err != nil {
 		return err
 	}
-	if err := appendInspectBinaryMap(b.upperBounds, file.UpperBoundValues()); err != nil {
+	if err := appendInspectBinaryMap(b.upperBounds, upperBounds); err != nil {
 		return err
 	}
 	appendInspectBytes(b.keyMetadata, file.KeyMetadata())
diff --git a/table/inspect_files_stats_bench_test.go b/table/inspect_files_stats_bench_test.go
new file mode 100644
index 000000000..429954f6d
--- /dev/null
+++ b/table/inspect_files_stats_bench_test.go
@@ -0,0 +1,166 @@
+// 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 (
+	"fmt"
+	"testing"
+
+	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/iceberg-go"
+)
+
+type inspectPublicStatsDataFile struct {
+	iceberg.DataFile
+}
+
+func (f *inspectPublicStatsDataFile) Partition() map[int]any {
+	return f.DataFile.Partition()
+}
+
+func (f *inspectPublicStatsDataFile) ValueCounts() map[int]int64 {
+	return f.DataFile.ValueCounts()
+}
+
+func (f *inspectPublicStatsDataFile) NullValueCounts() map[int]int64 {
+	return f.DataFile.NullValueCounts()
+}
+
+func (f *inspectPublicStatsDataFile) NaNValueCounts() map[int]int64 {
+	return f.DataFile.NaNValueCounts()
+}
+
+func (f *inspectPublicStatsDataFile) LowerBoundValues() map[int][]byte {
+	return f.DataFile.LowerBoundValues()
+}
+
+func (f *inspectPublicStatsDataFile) UpperBoundValues() map[int][]byte {
+	return f.DataFile.UpperBoundValues()
+}
+
+func BenchmarkInspectContentFileAppenderDataFileStats(b *testing.B) {
+	for _, fileCount := range []int{4096, 16384} {
+		for _, statsWidth := range []int{0, 8, 32, 128} {
+			b.Run(fmt.Sprintf("files=%d/stats=%d", fileCount, statsWidth), func(b *testing.B) {
+				partitionType, files := benchmarkInspectContentFilesWithStats(b, statsWidth, fileCount)
+				publicFiles := make([]iceberg.DataFile, len(files))
+				for i, file := range files {
+					publicFiles[i] = &inspectPublicStatsDataFile{DataFile: file}
+				}
+
+				b.Run("borrowed", func(b *testing.B) {
+					benchmarkAppendInspectContentFiles(b, partitionType, files)
+				})
+				b.Run("public", func(b *testing.B) {
+					benchmarkAppendInspectContentFiles(b, partitionType, publicFiles)
+				})
+			})
+		}
+	}
+}
+
+func benchmarkAppendInspectContentFiles(
+	b *testing.B,
+	partitionType *iceberg.StructType,
+	files []iceberg.DataFile,
+) {
+	b.Helper()
+
+	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()
+	b.ResetTimer()
+	b.ReportMetric(float64(len(files)), "files/op")
+	for range b.N {
+		for _, file := range files {
+			if err := appendFile(builder, file); err != nil {
+				b.Fatal(err)
+			}
+		}
+		record := builder.NewRecordBatch()
+		record.Release()
+	}
+}
+
+func benchmarkInspectContentFilesWithStats(
+	b *testing.B,
+	statsWidth int,
+	fileCount int,
+) (*iceberg.StructType, []iceberg.DataFile) {
+	b.Helper()
+
+	schemaWidth := max(statsWidth, 1)
+	schemaFields := make([]iceberg.NestedField, schemaWidth)
+	for i := range schemaFields {
+		fieldID := i + 1
+		schemaFields[i] = iceberg.NestedField{
+			ID: fieldID, Name: fmt.Sprintf("field_%d", fieldID),
+			Type: iceberg.PrimitiveTypes.Int32, Required: true,
+		}
+	}
+	schema := iceberg.NewSchema(0, schemaFields...)
+	spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+		SourceIDs: []int{1}, FieldID: 1000, Name: "partition", Transform: iceberg.IdentityTransform{},
+	})
+	partitionType := spec.PartitionType(schema)
+
+	files := make([]iceberg.DataFile, fileCount)
+	for i := range files {
+		builder, err := iceberg.NewDataFileBuilder(
+			spec, iceberg.EntryContentData, fmt.Sprintf("file-%d.parquet", i),
+			iceberg.ParquetFile, map[int]any{1000: int32(i)}, nil, nil, 1, 1,
+		)
+		if err != nil {
+			b.Fatal(err)
+		}
+		if statsWidth > 0 {
+			columnSizes := make(map[int]int64, statsWidth)
+			valueCounts := make(map[int]int64, statsWidth)
+			nullCounts := make(map[int]int64, statsWidth)
+			nanCounts := make(map[int]int64, statsWidth)
+			lowerBounds := make(map[int][]byte, statsWidth)
+			upperBounds := make(map[int][]byte, statsWidth)
+			for field := range statsWidth {
+				fieldID := field + 1
+				columnSizes[fieldID] = int64(field + 1)
+				valueCounts[fieldID] = int64(i + 1)
+				nullCounts[fieldID] = int64(field)
+				nanCounts[fieldID] = 0
+				lowerBounds[fieldID] = []byte{byte(field), byte(i)}
+				upperBounds[fieldID] = []byte{byte(field + 1), byte(i)}
+			}
+			builder = builder.
+				ColumnSizes(columnSizes).
+				ValueCounts(valueCounts).
+				NullValueCounts(nullCounts).
+				NaNValueCounts(nanCounts).
+				LowerBoundValues(lowerBounds).
+				UpperBoundValues(upperBounds)
+		}
+		files[i] = builder.Build()
+	}
+
+	return partitionType, files
+}
diff --git a/table/inspect_files_stats_test.go b/table/inspect_files_stats_test.go
new file mode 100644
index 000000000..9fa224e1c
--- /dev/null
+++ b/table/inspect_files_stats_test.go
@@ -0,0 +1,117 @@
+// 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/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	iceberg "github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/internal"
+	"github.com/stretchr/testify/require"
+)
+
+type borrowedInspectDataFile struct {
+	iceberg.DataFile
+}
+
+func (f *borrowedInspectDataFile) Partition() map[int]any {
+	panic("metadata appender called the public Partition getter")
+}
+
+func (f *borrowedInspectDataFile) ValueCounts() map[int]int64 {
+	panic("metadata appender called the public ValueCounts getter")
+}
+
+func (f *borrowedInspectDataFile) NullValueCounts() map[int]int64 {
+	panic("metadata appender called the public NullValueCounts getter")
+}
+
+func (f *borrowedInspectDataFile) NaNValueCounts() map[int]int64 {
+	panic("metadata appender called the public NaNValueCounts getter")
+}
+
+func (f *borrowedInspectDataFile) LowerBoundValues() map[int][]byte {
+	panic("metadata appender called the public LowerBoundValues getter")
+}
+
+func (f *borrowedInspectDataFile) UpperBoundValues() map[int][]byte {
+	panic("metadata appender called the public UpperBoundValues getter")
+}
+
+func (f *borrowedInspectDataFile) DataFileStatsRef(_ internal.DataFileRef) (
+	map[int]int64, map[int]int64, map[int]int64, map[int][]byte, map[int][]byte,
+) {
+	return internal.BorrowedDataFileStats(f.DataFile)
+}
+
+func (f *borrowedInspectDataFile) DataFilePartitionRef(_ internal.DataFileRef) map[int]any {
+	return internal.BorrowedDataFilePartition(f.DataFile)
+}
+
+func TestInspectContentFileBuilderUsesBorrowedDataFileMetadata(t *testing.T) {
+	file := &borrowedInspectDataFile{DataFile: testDataFileWithStats(t)}
+	partitionType := &iceberg.StructType{FieldList: []iceberg.NestedField{
+		{ID: 1000, Name: "part", Type: iceberg.PrimitiveTypes.String},
+	}}
+
+	withInspectContentFileRecord(t, partitionType, file, func(record arrow.RecordBatch) {
+		require.EqualValues(t, 1, record.NumRows())
+		partition := record.Column(4).(*array.Struct)
+		require.Equal(t, "partition", partition.Field(0).(*array.String).Value(0))
+
+		valueCounts := record.Column(8).(*array.Map)
+		require.False(t, valueCounts.IsNull(0))
+		start, end := valueCounts.ValueOffsets(0)
+		require.EqualValues(t, 1, end-start)
+		require.EqualValues(t, 2, valueCounts.Items().(*array.Int64).Value(int(start)))
+	})
+}
+
+func TestInspectContentFileBuilderFallsBackToPublicDataFileMetadata(t *testing.T) {
+	file := &publicStatsDataFile{DataFile: testDataFileWithStats(t)}
+	partitionType := &iceberg.StructType{FieldList: []iceberg.NestedField{
+		{ID: 1000, Name: "part", Type: iceberg.PrimitiveTypes.String},
+	}}
+
+	withInspectContentFileRecord(t, partitionType, file, func(record arrow.RecordBatch) {
+		require.EqualValues(t, 1, record.NumRows())
+	})
+	require.Equal(t, 5, file.getterCalls)
+}
+
+func withInspectContentFileRecord(
+	t *testing.T,
+	partitionType *iceberg.StructType,
+	file iceberg.DataFile,
+	check func(arrow.RecordBatch),
+) {
+	t.Helper()
+
+	arrowSchema, err := SchemaToArrowSchema(DataFilesSchema(partitionType), nil, true, false)
+	require.NoError(t, err)
+	builder := array.NewRecordBuilder(memory.DefaultAllocator, arrowSchema)
+	defer builder.Release()
+	require.NoError(t, appendContentFileRecord(builder, partitionType, file))
+
+	record := builder.NewRecordBatch()
+	defer record.Release()
+	check(record)
+}