You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/18 18:12:26 UTC
(iceberg-go) branch main updated: perf(table): pre-resolve Parquet stats columns (#1775)
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 6a226bd8f perf(table): pre-resolve Parquet stats columns (#1775)
6a226bd8f is described below
commit 6a226bd8f0e3333dfeb40089b86460a333b76fbb
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Tue Aug 18 20:12:19 2026 +0200
perf(table): pre-resolve Parquet stats columns (#1775)
perf(table): pre-resolve parquet stats columns
---
table/internal/parquet_files.go | 121 ++++++---
table/internal/parquet_stats_columns_bench_test.go | 281 +++++++++++++++++++++
table/internal/parquet_stats_columns_test.go | 164 ++++++++++++
3 files changed, 528 insertions(+), 38 deletions(-)
diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index a0332d9fd..7173d683a 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -1378,8 +1378,72 @@ func (w wrappedDecByteArrayStats) Max() iceberg.Decimal {
return iceberg.Decimal{Val: dec, Scale: w.scale}
}
+// parquetStatsColumn contains the immutable per-file resolution needed to
+// interpret a physical Parquet column's statistics. The row-group metadata
+// still carries the values that are accumulated, but the path-to-field lookup
+// is independent of the row group and is done once from the resolved schema.
+type parquetStatsColumn struct {
+ path string
+ fieldID int
+ statsCol StatisticsCollector
+ variantChild bool
+ skipStats bool
+ resolveErr error
+}
+
+func resolveParquetStatsColumns(meta *metadata.FileMetaData, statsCols map[int]StatisticsCollector, colMapping map[string]int, variantFieldIDs map[int]struct{}) []parquetStatsColumn {
+ columns := make([]parquetStatsColumn, meta.Schema.NumColumns())
+ for pos := range meta.Schema.NumColumns() {
+ column := meta.Schema.Column(pos)
+ columnPath := column.ColumnPath()
+ descriptor := parquetStatsColumn{path: column.Path()}
+
+ fieldID, ok := colMapping[descriptor.path]
+ if !ok {
+ // Variant sub-columns (metadata, value, typed_value children) have
+ // no Iceberg field ID (spec: "must not be assigned field IDs").
+ // Walk up the resolved path once to find a variant ancestor.
+ for depth := len(columnPath) - 1; depth >= 1; depth-- {
+ ancestorPath := strings.Join(columnPath[:depth], ".")
+ if ancestorID, hasAncestor := colMapping[ancestorPath]; hasAncestor {
+ _, descriptor.variantChild = variantFieldIDs[ancestorID]
+
+ break
+ }
+ }
+ if !descriptor.variantChild {
+ descriptor.resolveErr = fmt.Errorf("column chunk %q not found in column mapping", columnPath)
+ }
+
+ columns[pos] = descriptor
+
+ continue
+ }
+
+ descriptor.fieldID = fieldID
+ statsCol, ok := statsCols[fieldID]
+ if !ok {
+ // Only the reserved row-lineage metadata columns (_row_id,
+ // _last_updated_sequence_number) may legitimately appear in a file
+ // without a metrics-plan entry. Writers materialize them without
+ // adding them to the table schema, so skip them entirely.
+ if iceberg.IsMetadataColumn(fieldID) {
+ descriptor.skipStats = true
+ } else {
+ descriptor.resolveErr = fmt.Errorf("field id %d (column %q) not found in the metrics plan", fieldID, descriptor.path)
+ }
+ } else {
+ descriptor.statsCol = statsCol
+ }
+ columns[pos] = descriptor
+ }
+
+ return columns
+}
+
func (p parquetFormat) DataFileStatsFromMeta(meta Metadata, statsCols map[int]StatisticsCollector, colMapping map[string]int, variantFieldIDs map[int]struct{}, arrowSchema *arrow.Schema) *DataFileStatistics {
pqmeta := meta.(*metadata.FileMetaData)
+ columns := resolveParquetStatsColumns(pqmeta, statsCols, colMapping, variantFieldIDs)
var (
colSizes = make(map[int]int64)
valueCounts = make(map[int]int64)
@@ -1411,44 +1475,15 @@ func (p parquetFormat) DataFileStatsFromMeta(meta Metadata, statsCols map[int]St
panic(err)
}
- fieldID, ok := colMapping[colChunk.PathInSchema().String()]
- if !ok {
- // Variant sub-columns (metadata, value, typed_value children) have no
- // Iceberg field ID (spec: "must not be assigned field IDs"). Walk up
- // the path hierarchy to find a variant ancestor.
- path := colChunk.PathInSchema()
- found := false
- for depth := len(path) - 1; depth >= 1; depth-- {
- ancestorPath := strings.Join(path[:depth], ".")
- if ancestorID, hasAncestor := colMapping[ancestorPath]; hasAncestor {
- if _, isVariant := variantFieldIDs[ancestorID]; isVariant {
- found = true
- }
-
- break
- }
- }
- if found {
- continue
- }
-
- panic(fmt.Errorf("column chunk %q not found in column mapping", colChunk.PathInSchema()))
+ column := columns[pos]
+ if column.resolveErr != nil {
+ panic(column.resolveErr)
}
- statsCol, ok := statsCols[fieldID]
- if !ok {
- // Only the reserved row-lineage metadata columns (_row_id,
- // _last_updated_sequence_number) may legitimately appear in
- // a file without a metrics-plan entry: writers materialize
- // them without adding them to the table schema. Any other
- // field id missing from the plan means the plan and the
- // file disagree — fail loudly rather than silently dropping
- // that column's stats.
- if iceberg.IsMetadataColumn(fieldID) {
- continue
- }
-
- panic(fmt.Errorf("field id %d (column %q) not found in the metrics plan", fieldID, colChunk.PathInSchema()))
+ if column.variantChild || column.skipStats {
+ continue
}
+
+ fieldID, statsCol := column.fieldID, column.statsCol
if statsCol.Mode.Typ == MetricModeNone {
continue
}
@@ -1528,7 +1563,7 @@ func (p parquetFormat) DataFileStatsFromMeta(meta Metadata, statsCols map[int]St
return ok
})
- vlo, vup := p.collectVariantBounds(pqmeta, arrowSchema, colMapping, statsCols)
+ vlo, vup := p.collectVariantBoundsWithColumns(pqmeta, arrowSchema, colMapping, statsCols, columns)
return &DataFileStatistics{
RecordCount: pqmeta.GetNumRows(),
@@ -1594,6 +1629,16 @@ func (p parquetFormat) collectVariantBounds(meta *metadata.FileMetaData, arrowSc
return nil, nil
}
+ columns := resolveParquetStatsColumns(meta, statsCols, colMapping, nil)
+
+ return p.collectVariantBoundsWithColumns(meta, arrowSchema, colMapping, statsCols, columns)
+}
+
+func (p parquetFormat) collectVariantBoundsWithColumns(meta *metadata.FileMetaData, arrowSchema *arrow.Schema, colMapping map[string]int, statsCols map[int]StatisticsCollector, columns []parquetStatsColumn) (lower, upper map[int][]byte) {
+ if arrowSchema == nil {
+ return nil, nil
+ }
+
type leafAgg struct {
leaf variantLeaf
parentID int
@@ -1636,7 +1681,7 @@ func (p parquetFormat) collectVariantBounds(meta *metadata.FileMetaData, arrowSc
if err != nil {
continue
}
- path := cc.PathInSchema().String()
+ path := columns[pos].path
if _, isResidual := residualParent[path]; isResidual {
// Non-null residual entries invalidate the bound unless they are all variant-null.
diff --git a/table/internal/parquet_stats_columns_bench_test.go b/table/internal/parquet_stats_columns_bench_test.go
new file mode 100644
index 000000000..7e832fd05
--- /dev/null
+++ b/table/internal/parquet_stats_columns_bench_test.go
@@ -0,0 +1,281 @@
+// 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 internal
+
+import (
+ "encoding/binary"
+ "math"
+ "strings"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/extensions"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/metadata"
+ "github.com/apache/iceberg-go"
+)
+
+// benchmarkLegacyStatsColumnResolution models the path and field lookup that
+// ran once for every physical column in every row group before descriptors
+// were introduced. It deliberately stops after resolution so the paired
+// benchmark isolates the work changed by this optimization.
+func benchmarkLegacyStatsColumnResolution(meta *metadata.FileMetaData, statsCols map[int]StatisticsCollector, colMapping map[string]int, variantFieldIDs map[int]struct{}) int {
+ checksum := 0
+ for rg := range meta.NumRowGroups() {
+ rowGroup := meta.RowGroup(rg)
+ for pos := range rowGroup.NumColumns() {
+ colChunk, err := rowGroup.ColumnChunk(pos)
+ if err != nil {
+ panic(err)
+ }
+
+ pathString := colChunk.PathInSchema().String()
+ fieldID, ok := colMapping[pathString]
+ checksum += len(pathString)
+ if !ok {
+ path := colChunk.PathInSchema()
+ for depth := len(path) - 1; depth >= 1; depth-- {
+ ancestorPath := strings.Join(path[:depth], ".")
+ if ancestorID, hasAncestor := colMapping[ancestorPath]; hasAncestor {
+ if _, isVariant := variantFieldIDs[ancestorID]; !isVariant {
+ panic("unexpected non-variant column mapping in benchmark")
+ }
+
+ break
+ }
+ }
+
+ continue
+ }
+
+ checksum += fieldID + len(statsCols[fieldID].Mode.Typ)
+ }
+ }
+
+ return checksum
+}
+
+func benchmarkResolvedStatsColumnResolution(meta *metadata.FileMetaData, statsCols map[int]StatisticsCollector, colMapping map[string]int, variantFieldIDs map[int]struct{}) int {
+ columns := resolveParquetStatsColumns(meta, statsCols, colMapping, variantFieldIDs)
+ checksum := 0
+ for rg := range meta.NumRowGroups() {
+ rowGroup := meta.RowGroup(rg)
+ for pos := range rowGroup.NumColumns() {
+ if _, err := rowGroup.ColumnChunk(pos); err != nil {
+ panic(err)
+ }
+
+ column := columns[pos]
+ if column.resolveErr != nil {
+ panic(column.resolveErr)
+ }
+ checksum += len(column.path)
+ if column.variantChild || column.skipStats {
+ continue
+ }
+
+ checksum += column.fieldID + len(column.statsCol.Mode.Typ)
+ }
+ }
+
+ return checksum
+}
+
+func benchmarkStats(min, max []byte, nullCount int64) metadata.EncodedStatistics {
+ var stats metadata.EncodedStatistics
+ stats.SetMin(min)
+ stats.SetMax(max)
+ stats.SetNullCount(nullCount)
+
+ return stats
+}
+
+func benchmarkAllNullStats(nullCount int64) metadata.EncodedStatistics {
+ var stats metadata.EncodedStatistics
+ stats.SetNullCount(nullCount)
+
+ return stats
+}
+
+func buildStatsColumnsBenchmarkMetadata(b *testing.B, rowGroups, rows int) *metadata.FileMetaData {
+ b.Helper()
+
+ meta := buildStatsColumnsMetadata(b)
+ builder := metadata.NewFileMetadataBuilder(meta.Schema, parquet.NewWriterProperties(), nil)
+ rowGroupBytes := int64(9 * rows * 16)
+
+ for rg := range rowGroups {
+ rowGroup := builder.AppendRowGroup()
+ rowGroup.SetNumRows(rows)
+
+ idMin := make([]byte, 4)
+ idMax := make([]byte, 4)
+ binary.LittleEndian.PutUint32(idMin, uint32(rg*rows))
+ binary.LittleEndian.PutUint32(idMax, uint32((rg+1)*rows-1))
+
+ scoreMin := make([]byte, 8)
+ scoreMax := make([]byte, 8)
+ binary.LittleEndian.PutUint64(scoreMin, math.Float64bits(float64(rg)))
+ binary.LittleEndian.PutUint64(scoreMax, math.Float64bits(float64(rg+1)))
+
+ stats := []metadata.EncodedStatistics{
+ benchmarkStats(idMin, idMax, 0),
+ benchmarkStats(scoreMin, scoreMax, 0),
+ benchmarkStats([]byte("name-a"), []byte("name-z"), 0),
+ benchmarkAllNullStats(int64(rows)),
+ benchmarkAllNullStats(int64(rows)),
+ benchmarkAllNullStats(int64(rows)),
+ benchmarkStats(scoreMin, scoreMax, 0),
+ benchmarkAllNullStats(int64(rows)),
+ benchmarkStats([]byte("city-a"), []byte("city-z"), 0),
+ }
+
+ for pos, stat := range stats {
+ chunk := rowGroup.NextColumnChunk()
+ chunk.SetStats(stat)
+ info := metadata.ChunkMetaInfo{
+ NumValues: int64(rows),
+ DataPageOffset: int64(100 + rg*1000 + pos*100),
+ IndexPageOffset: -1,
+ CompressedSize: 16,
+ UncompressedSize: 32,
+ }
+ if err := chunk.Finish(info, false, false, metadata.EncodingStats{}); err != nil {
+ b.Fatal(err)
+ }
+ }
+
+ if err := rowGroup.Finish(rowGroupBytes, int16(rg)); err != nil {
+ b.Fatal(err)
+ }
+ }
+
+ result, err := builder.Finish()
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ return result
+}
+
+type statsColumnsBenchmarkFixture struct {
+ meta *metadata.FileMetaData
+ arrowSchema *arrow.Schema
+ statsCols map[int]StatisticsCollector
+ colMapping map[string]int
+ variantFieldIDs map[int]struct{}
+}
+
+func newStatsColumnsBenchmarkFixture(b *testing.B, rowGroups, rows int) statsColumnsBenchmarkFixture {
+ b.Helper()
+
+ return statsColumnsBenchmarkFixture{
+ meta: buildStatsColumnsBenchmarkMetadata(b, rowGroups, rows),
+ arrowSchema: arrow.NewSchema([]arrow.Field{{
+ Name: "payload",
+ Type: extensions.NewShreddedVariantType(arrow.StructOf(
+ arrow.Field{Name: "score", Type: arrow.PrimitiveTypes.Float64},
+ arrow.Field{Name: "city", Type: arrow.BinaryTypes.String},
+ )),
+ Nullable: true,
+ }}, nil),
+ statsCols: map[int]StatisticsCollector{
+ 1: {FieldID: 1, Mode: MetricsMode{Typ: MetricModeFull}, IcebergTyp: iceberg.PrimitiveTypes.Int32},
+ 2: {FieldID: 2, Mode: MetricsMode{Typ: MetricModeCounts}, IcebergTyp: iceberg.PrimitiveTypes.Float64},
+ 3: {FieldID: 3, Mode: MetricsMode{Typ: MetricModeCounts}, IcebergTyp: iceberg.PrimitiveTypes.String},
+ 4: {FieldID: 4, Mode: MetricsMode{Typ: MetricModeFull}, ColName: "payload"},
+ },
+ colMapping: map[string]int{
+ "id": 1,
+ "details.score": 2,
+ "details.name": 3,
+ "payload": 4,
+ },
+ variantFieldIDs: map[int]struct{}{4: {}},
+ }
+}
+
+// BenchmarkParquetStatsColumnResolution compares the repeated per-row-group
+// lookup with the one-time descriptor resolution. The end-to-end benchmark
+// below measures the same change with statistics aggregation included.
+func BenchmarkParquetStatsColumnResolution(b *testing.B) {
+ const (
+ rowGroups = 32
+ rows = 10_000
+ )
+
+ fixture := newStatsColumnsBenchmarkFixture(b, rowGroups, rows)
+
+ want := benchmarkLegacyStatsColumnResolution(fixture.meta, fixture.statsCols, fixture.colMapping, fixture.variantFieldIDs)
+ if got := benchmarkResolvedStatsColumnResolution(fixture.meta, fixture.statsCols, fixture.colMapping, fixture.variantFieldIDs); got != want {
+ b.Fatalf("resolution checksum mismatch: before=%d after=%d", want, got)
+ }
+
+ for _, tc := range []struct {
+ name string
+ fn func() int
+ }{
+ {
+ name: "before",
+ fn: func() int {
+ return benchmarkLegacyStatsColumnResolution(fixture.meta, fixture.statsCols, fixture.colMapping, fixture.variantFieldIDs)
+ },
+ },
+ {
+ name: "after",
+ fn: func() int {
+ return benchmarkResolvedStatsColumnResolution(fixture.meta, fixture.statsCols, fixture.colMapping, fixture.variantFieldIDs)
+ },
+ },
+ } {
+ b.Run(tc.name, func(b *testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for range b.N {
+ if got := tc.fn(); got != want {
+ b.Fatalf("unexpected resolution checksum: got=%d want=%d", got, want)
+ }
+ }
+ b.ReportMetric(float64(rowGroups*9), "physical-columns/op")
+ })
+ }
+}
+
+func BenchmarkDataFileStatsFromMetaNestedVariant(b *testing.B) {
+ const (
+ rowGroups = 32
+ rows = 10_000
+ )
+
+ fixture := newStatsColumnsBenchmarkFixture(b, rowGroups, rows)
+
+ format := parquetFormat{}
+ b.ReportAllocs()
+ b.ResetTimer()
+ for range b.N {
+ stats := format.DataFileStatsFromMeta(fixture.meta, fixture.statsCols, fixture.colMapping, fixture.variantFieldIDs, fixture.arrowSchema)
+ if stats.RecordCount != int64(rowGroups*rows) {
+ b.Fatalf("unexpected record count: %d", stats.RecordCount)
+ }
+ if len(stats.VariantLowerBounds) != 1 || len(stats.VariantUpperBounds) != 1 {
+ b.Fatalf("expected one variant bound object, got lower=%d upper=%d", len(stats.VariantLowerBounds), len(stats.VariantUpperBounds))
+ }
+ }
+ b.ReportMetric(float64(rowGroups*rows), "rows/op")
+ b.ReportMetric(float64(rowGroups*9), "columns/op")
+}
diff --git a/table/internal/parquet_stats_columns_test.go b/table/internal/parquet_stats_columns_test.go
new file mode 100644
index 000000000..3cbc70ce6
--- /dev/null
+++ b/table/internal/parquet_stats_columns_test.go
@@ -0,0 +1,164 @@
+// 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 internal
+
+import (
+ "testing"
+
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/metadata"
+ "github.com/apache/arrow-go/v18/parquet/schema"
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func buildStatsColumnsMetadata(t testing.TB) *metadata.FileMetaData {
+ t.Helper()
+
+ details, err := schema.NewGroupNode("details", parquet.Repetitions.Optional, schema.FieldList{
+ schema.NewFloat64Node("score", parquet.Repetitions.Optional, -1),
+ schema.NewByteArrayNode("name", parquet.Repetitions.Optional, -1),
+ }, -1)
+ require.NoError(t, err)
+
+ variantScore, err := schema.NewGroupNode("score", parquet.Repetitions.Optional, schema.FieldList{
+ schema.NewByteArrayNode("value", parquet.Repetitions.Optional, -1),
+ schema.NewFloat64Node("typed_value", parquet.Repetitions.Optional, -1),
+ }, -1)
+ require.NoError(t, err)
+
+ variantCity, err := schema.NewGroupNode("city", parquet.Repetitions.Optional, schema.FieldList{
+ schema.NewByteArrayNode("value", parquet.Repetitions.Optional, -1),
+ schema.NewByteArrayNode("typed_value", parquet.Repetitions.Optional, -1),
+ }, -1)
+ require.NoError(t, err)
+
+ variantTypedValue, err := schema.NewGroupNode("typed_value", parquet.Repetitions.Optional, schema.FieldList{
+ variantScore,
+ variantCity,
+ }, -1)
+ require.NoError(t, err)
+
+ payload, err := schema.NewGroupNode("payload", parquet.Repetitions.Optional, schema.FieldList{
+ schema.NewByteArrayNode("metadata", parquet.Repetitions.Optional, -1),
+ schema.NewByteArrayNode("value", parquet.Repetitions.Optional, -1),
+ variantTypedValue,
+ }, -1)
+ require.NoError(t, err)
+
+ root, err := schema.NewGroupNode("schema", parquet.Repetitions.Required, schema.FieldList{
+ schema.NewInt32Node("id", parquet.Repetitions.Required, -1),
+ details,
+ payload,
+ }, -1)
+ require.NoError(t, err)
+
+ builder := metadata.NewFileMetadataBuilder(schema.NewSchema(root), parquet.NewWriterProperties(), nil)
+ meta, err := builder.Finish()
+ require.NoError(t, err)
+
+ return meta
+}
+
+func TestResolveParquetStatsColumnsPreservesPathResolution(t *testing.T) {
+ meta := buildStatsColumnsMetadata(t)
+ statsCols := map[int]StatisticsCollector{
+ 1: {FieldID: 1, Mode: MetricsMode{Typ: MetricModeFull}, IcebergTyp: iceberg.PrimitiveTypes.Int32},
+ 2: {FieldID: 2, Mode: MetricsMode{Typ: MetricModeCounts}, IcebergTyp: iceberg.PrimitiveTypes.Float64},
+ 3: {FieldID: 3, Mode: MetricsMode{Typ: MetricModeTruncate, Len: 8}, IcebergTyp: iceberg.PrimitiveTypes.String},
+ 4: {FieldID: 4, Mode: MetricsMode{Typ: MetricModeFull}},
+ }
+ colMapping := map[string]int{
+ "id": 1,
+ "details.score": 2,
+ "details.name": 3,
+ "payload": 4,
+ }
+ variantFieldIDs := map[int]struct{}{4: {}}
+
+ columns := resolveParquetStatsColumns(meta, statsCols, colMapping, variantFieldIDs)
+ require.Len(t, columns, 9)
+
+ wantPaths := []string{
+ "id",
+ "details.score",
+ "details.name",
+ "payload.metadata",
+ "payload.value",
+ "payload.typed_value.score.value",
+ "payload.typed_value.score.typed_value",
+ "payload.typed_value.city.value",
+ "payload.typed_value.city.typed_value",
+ }
+ for pos, wantPath := range wantPaths {
+ assert.Equal(t, wantPath, columns[pos].path, "physical column %d path", pos)
+ assert.Nil(t, columns[pos].resolveErr, "physical column %d should resolve", pos)
+ }
+
+ assert.Equal(t, 1, columns[0].fieldID)
+ assert.Equal(t, 2, columns[1].fieldID)
+ assert.Equal(t, 3, columns[2].fieldID)
+ assert.Equal(t, MetricModeFull, columns[0].statsCol.Mode.Typ)
+ assert.Equal(t, MetricModeCounts, columns[1].statsCol.Mode.Typ)
+ assert.Equal(t, MetricModeTruncate, columns[2].statsCol.Mode.Typ)
+ for _, column := range columns[3:] {
+ assert.True(t, column.variantChild, "variant child %q should not require an Iceberg field ID", column.path)
+ }
+}
+
+func TestResolveParquetStatsColumnsRetainsMissingPathError(t *testing.T) {
+ meta := buildStatsColumnsMetadata(t)
+ columns := resolveParquetStatsColumns(meta, nil, map[string]int{
+ "id": 1,
+ "payload": 4,
+ }, map[int]struct{}{4: {}})
+
+ require.Len(t, columns, 9)
+ assert.ErrorContains(t, columns[1].resolveErr, `column chunk "details.score" not found in column mapping`)
+ assert.ErrorContains(t, columns[2].resolveErr, `column chunk "details.name" not found in column mapping`)
+ for _, column := range columns[3:] {
+ assert.True(t, column.variantChild)
+ assert.NoError(t, column.resolveErr)
+ }
+}
+
+func TestResolveParquetStatsColumnsHandlesMissingMetricsPlan(t *testing.T) {
+ t.Run("reserved row-lineage column is skipped", func(t *testing.T) {
+ meta := buildStatsColumnsMetadata(t)
+ columns := resolveParquetStatsColumns(meta, nil, map[string]int{
+ "id": iceberg.RowIDFieldID,
+ }, nil)
+
+ require.Len(t, columns, 9)
+ assert.Equal(t, iceberg.RowIDFieldID, columns[0].fieldID)
+ assert.True(t, columns[0].skipStats)
+ assert.NoError(t, columns[0].resolveErr)
+ })
+
+ t.Run("non-reserved column retains metrics plan error", func(t *testing.T) {
+ meta := buildStatsColumnsMetadata(t)
+ columns := resolveParquetStatsColumns(meta, nil, map[string]int{
+ "id": 1,
+ }, nil)
+
+ require.Len(t, columns, 9)
+ assert.False(t, columns[0].skipStats)
+ assert.ErrorContains(t, columns[0].resolveErr, `field id 1 (column "id") not found in the metrics plan`)
+ })
+}