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 f593b08b6 fix(table): compute geo metrics for geo fields nested in
structs, lists and maps (#2303)
f593b08b6 is described below
commit f593b08b6f10bcaf99f9009975269510d93d8929
Author: Nirajan Sah <[email protected]>
AuthorDate: Mon Oct 26 22:02:08 2026 +0545
fix(table): compute geo metrics for geo fields nested in structs, lists and
maps (#2303)
Parquet writes no statistics for GEOMETRY/GEOGRAPHY columns, so the
writer computes them from the Arrow data, but collectGeoColumns only
looked at top-level columns. Nested geo fields got a value count but
no null count or bounds.
Walk structs, lists and maps down to each geo field. Count nulls the
way Parquet does for nested fields: a null value, a null parent, and
an empty list or map each count once. Values under a null parent are
not decoded.
Fixes #2158
Signed-off-by: nirajan1111 <[email protected]>
---
table/internal/parquet_files.go | 139 +++++++++++++++++++-------
table/internal/parquet_files_internal_test.go | 75 ++++++++++++++
table/internal/parquet_files_test.go | 70 +++++++++----
website/src/feature-status.md | 2 +-
4 files changed, 232 insertions(+), 54 deletions(-)
diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index 7238b6347..648874036 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -521,36 +521,52 @@ type ParquetFileWriter struct {
rowGroupBytes int64
}
-// geoColumn locates a top-level geometry/geography column: its position in the
-// Arrow record batch and the Iceberg field ID its bounds are recorded under.
+// geoColumn locates a geometry/geography leaf: the position of its top-level
+// column in the Arrow record batch, the struct child index taken at each
+// struct (including a map's key/value entries) on the way down to the leaf,
+// and the Iceberg field ID its bounds are recorded under.
type geoColumn struct {
colIdx int
+ path []int
fieldID int
}
-// collectGeoColumns finds top-level WKB-encoded geo columns in the Arrow
schema
-// and pairs each with its Iceberg field ID. Geo bounds for columns nested
inside
-// structs/lists/maps are not yet computed, which diverges from Java/PyIceberg.
+// collectGeoColumns finds WKB-encoded geo leaves in the Arrow schema,
including
+// those nested in structs/lists/maps, and pairs each with its Iceberg field
ID.
//
-// Nested geo leaves also lose their Parquet Statistics block (min, max, null
-// count), because the GEOMETRY/GEOGRAPHY logical types have an undefined sort
-// order. They are not collected here, so their null counts are not recovered
-// either: the manifest omits them, which costs pruning but never gives a wrong
-// answer. The same applies to
-// geo files imported via DataFileStatsFromMeta alone (e.g. AddFiles).
-// TODO(#2158): compute geo bounds and null counts for geo columns nested in
-// structs/lists/maps.
+// Parquet drops the Statistics block (min, max, null count) for GEOMETRY and
+// GEOGRAPHY columns because their sort order is undefined, so these metrics
are
+// computed from the Arrow data instead. Geo files imported via
+// DataFileStatsFromMeta alone (e.g. AddFiles) get no geo null counts or
bounds,
+// which costs pruning but never gives a wrong answer.
func collectGeoColumns(sc *arrow.Schema, colMapping map[string]int)
[]geoColumn {
var result []geoColumn
for i, f := range sc.Fields() {
- if _, ok := f.Type.(*geoarrow.WKBType); !ok {
- continue
+ result = appendGeoColumns(result, f.Type, f.Name, i, nil,
colMapping)
+ }
+
+ return result
+}
+
+// appendGeoColumns walks dt and appends a geoColumn for each WKB leaf. name is
+// the Parquet column path used as the colMapping key.
+func appendGeoColumns(result []geoColumn, dt arrow.DataType, name string,
colIdx int, path []int, colMapping map[string]int) []geoColumn {
+ switch t := dt.(type) {
+ case *geoarrow.WKBType:
+ if fieldID, ok := colMapping[name]; ok {
+ result = append(result, geoColumn{colIdx: colIdx, path:
slices.Clone(path), fieldID: fieldID})
}
- fieldID, ok := colMapping[f.Name]
- if !ok {
- continue
+ case *arrow.StructType:
+ for i, f := range t.Fields() {
+ result = appendGeoColumns(result, f.Type,
name+"."+f.Name, colIdx, append(path, i), colMapping)
}
- result = append(result, geoColumn{colIdx: i, fieldID: fieldID})
+ case *arrow.MapType:
+ kv := name + ".key_value."
+ result = appendGeoColumns(result, t.KeyType(),
kv+t.KeyField().Name, colIdx, append(path, 0), colMapping)
+ result = appendGeoColumns(result, t.ItemType(),
kv+t.ItemField().Name, colIdx, append(path, 1), colMapping)
+ case *arrow.ListType, *arrow.LargeListType, *arrow.FixedSizeListType:
+ elem := dt.(arrow.ListLikeType).ElemField()
+ result = appendGeoColumns(result, elem.Type,
name+".list."+elem.Name, colIdx, path, colMapping)
}
return result
@@ -1199,7 +1215,7 @@ func normalizeWKBArrayReachable(ext array.ExtensionArray,
active []bool, mem mem
// in this batch and tallies their null counts. Null rows are skipped; a
// malformed WKB value fails the write.
//
-// Every geoCols entry comes from a *geoarrow.WKBType field of the writer's
+// Every geoCols entry comes from a *geoarrow.WKBType leaf of the writer's
// schema, so a batch whose column does not match is an error rather than a
// skip: skipping would leave a partial null count that under-reports nulls.
func (w *ParquetFileWriter) accumulateGeoBounds(batch arrow.RecordBatch) error
{
@@ -1208,30 +1224,83 @@ func (w *ParquetFileWriter) accumulateGeoBounds(batch
arrow.RecordBatch) error {
return fmt.Errorf("geo field %d: column %d missing from
batch with %d columns",
gc.fieldID, gc.colIdx, batch.NumCols())
}
- ext, ok := batch.Column(gc.colIdx).(array.ExtensionArray)
- if !ok {
- return fmt.Errorf("geo field %d: expected a WKB
extension array, got %s",
- gc.fieldID, batch.Column(gc.colIdx).DataType())
- }
- storage, ok := ext.Storage().(wkbStorage)
- if !ok {
- return fmt.Errorf("geo field %d: unsupported WKB
storage type %s",
- gc.fieldID, ext.Storage().DataType())
- }
acc, ok := w.geoAccs[gc.fieldID]
if !ok {
return fmt.Errorf("geo field %d: no bounds
accumulator", gc.fieldID)
}
- acc.nulls += int64(storage.NullN())
- for i := range storage.Len() {
- if storage.IsNull(i) {
+ col := batch.Column(gc.colIdx)
+ if err := accumulateGeoValues(acc, col, gc.path, 0, col.Len());
err != nil {
+ return fmt.Errorf("geo field %d: %w", gc.fieldID, err)
+ }
+ }
+
+ return nil
+}
+
+// accumulateGeoValues feeds the WKB values in arr[lo:hi] to acc, following
path
+// down to the geo leaf. Nulls are counted the way Parquet's leaf null_count
+// does: a null leaf, a null parent, and an empty list or map each count once.
+// Values under a null parent are never decoded.
+func accumulateGeoValues(acc *geoBoundsAccumulator, arr arrow.Array, path
[]int, lo, hi int) error {
+ switch arr := arr.(type) {
+ case *array.Struct:
+ if len(path) == 0 || path[0] >= arr.NumField() {
+ return fmt.Errorf("struct array %s does not match the
geo column path", arr.DataType())
+ }
+ child := arr.Field(path[0])
+ for i := lo; i < hi; i++ {
+ if arr.IsNull(i) {
+ acc.nulls++
+
continue
}
- if err := acc.AddWKB(storage.Value(i)); err != nil {
- return fmt.Errorf("computing geo bounds for
field %d: %w", gc.fieldID, err)
+ if err := accumulateGeoValues(acc, child, path[1:], i,
i+1); err != nil {
+ return err
}
}
+ case array.ListLike:
+ for i := lo; i < hi; i++ {
+ start, end := arr.ValueOffsets(i)
+ if arr.IsNull(i) || start == end {
+ acc.nulls++
+
+ continue
+ }
+ if err := accumulateGeoValues(acc, arr.ListValues(),
path, int(start), int(end)); err != nil {
+ return err
+ }
+ }
+ default:
+ if len(path) != 0 {
+ return fmt.Errorf("%s array does not match the geo
column path", arr.DataType())
+ }
+
+ return accumulateWKBValues(acc, arr, lo, hi)
+ }
+
+ return nil
+}
+
+func accumulateWKBValues(acc *geoBoundsAccumulator, arr arrow.Array, lo, hi
int) error {
+ ext, ok := arr.(array.ExtensionArray)
+ if !ok {
+ return fmt.Errorf("expected a WKB extension array, got %s",
arr.DataType())
+ }
+ storage, ok := ext.Storage().(wkbStorage)
+ if !ok {
+ return fmt.Errorf("unsupported WKB storage type %s",
ext.Storage().DataType())
+ }
+
+ for i := lo; i < hi; i++ {
+ if storage.IsNull(i) {
+ acc.nulls++
+
+ continue
+ }
+ if err := acc.AddWKB(storage.Value(i)); err != nil {
+ return fmt.Errorf("computing geo bounds: %w", err)
+ }
}
return nil
diff --git a/table/internal/parquet_files_internal_test.go
b/table/internal/parquet_files_internal_test.go
index b7f4d6f6f..d95a4ca1b 100644
--- a/table/internal/parquet_files_internal_test.go
+++ b/table/internal/parquet_files_internal_test.go
@@ -20,6 +20,7 @@ package internal
import (
"context"
"errors"
+ "strings"
"testing"
"github.com/apache/arrow-go/v18/arrow"
@@ -225,6 +226,80 @@ func TestWriteRejectsMismatchedGeoBatch(t *testing.T) {
})
}
+func TestAccumulateGeoValuesNested(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ wkbType := geoarrow.NewWKBType(geoarrow.WKBWithBinaryStorage())
+ hex := func(s string) string { return geoarrow.WKBBytes(wkbOf(t,
s)).String() }
+
+ listOfStructs, _, err := array.FromJSON(mem,
+ arrow.ListOf(arrow.StructOf(arrow.Field{Name: "g", Type:
wkbType, Nullable: true})),
+ strings.NewReader(`[
+ [{"g": "`+hex("POINT (1 2)")+`"}, null, {"g": null}],
+ [],
+ null,
+ [{"g": "`+hex("POINT (3 -4)")+`"}]
+ ]`))
+ require.NoError(t, err)
+ defer listOfStructs.Release()
+
+ t.Run("list of structs", func(t *testing.T) {
+ acc := newGeoBoundsAccumulator(false)
+ require.NoError(t, accumulateGeoValues(acc, listOfStructs,
[]int{0}, 0, listOfStructs.Len()))
+
+ // null struct, null leaf, empty list, null list
+ assert.EqualValues(t, 4, acc.nulls)
+ lower, upper := acc.Bounds()
+ assert.Equal(t, []float64{1, -4}, decodeBound(t, lower))
+ assert.Equal(t, []float64{3, 2}, decodeBound(t, upper))
+ })
+
+ t.Run("sliced list", func(t *testing.T) {
+ sliced := array.NewSlice(listOfStructs, 1, 4)
+ defer sliced.Release()
+
+ acc := newGeoBoundsAccumulator(false)
+ require.NoError(t, accumulateGeoValues(acc, sliced, []int{0},
0, sliced.Len()))
+
+ assert.EqualValues(t, 2, acc.nulls)
+ lower, upper := acc.Bounds()
+ assert.Equal(t, []float64{3, -4}, decodeBound(t, lower))
+ assert.Equal(t, []float64{3, -4}, decodeBound(t, upper))
+ })
+
+ t.Run("value under a null parent is not decoded", func(t *testing.T) {
+ storageBuilder := array.NewBinaryBuilder(mem,
arrow.BinaryTypes.Binary)
+ storageBuilder.Append(wkbOf(t, "POINT (1 2)"))
+ storageBuilder.Append([]byte{0xff, 0xff})
+ storage := storageBuilder.NewArray()
+ storageBuilder.Release()
+ defer storage.Release()
+
+ geomArr := array.NewExtensionArrayWithStorage(wkbType, storage)
+ defer geomArr.Release()
+
+ structArr, err :=
array.NewStructArrayWithNulls([]arrow.Array{geomArr}, []string{"g"},
+ memory.NewBufferBytes([]byte{0b01}), 1, 0)
+ require.NoError(t, err)
+ defer structArr.Release()
+
+ acc := newGeoBoundsAccumulator(false)
+ require.NoError(t, accumulateGeoValues(acc, structArr,
[]int{0}, 0, structArr.Len()))
+
+ assert.EqualValues(t, 1, acc.nulls)
+ lower, _ := acc.Bounds()
+ assert.Equal(t, []float64{1, 2}, decodeBound(t, lower))
+ })
+
+ t.Run("array does not match path", func(t *testing.T) {
+ acc := newGeoBoundsAccumulator(false)
+ err := accumulateGeoValues(acc, listOfStructs, nil, 0,
listOfStructs.Len())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "does not match the geo column
path")
+ })
+}
+
func TestNewFileWriterCachesGeoNormalizationColumns(t *testing.T) {
typeDef := geoarrow.NewWKBType(geoarrow.WKBWithBinaryStorage())
arrowSchema := arrow.NewSchema([]arrow.Field{
diff --git a/table/internal/parquet_files_test.go
b/table/internal/parquet_files_test.go
index c4dc92bf3..5effcffdf 100644
--- a/table/internal/parquet_files_test.go
+++ b/table/internal/parquet_files_test.go
@@ -3301,12 +3301,11 @@ func TestWriteDataFileGeoBounds(t *testing.T) {
}
}
-// TestWriteDataFileNestedGeoNullCount pins the current metrics for a geo
column
-// nested in a struct. The Parquet GEOMETRY logical type drops the leaf's
footer
-// statistics, and only top-level geo columns have their null counts tallied
-// from the Arrow data, so the nested leaf records a value count but no null
-// count. A missing null count only costs pruning. Flip this when nested geo
-// metrics land (TODO(#2158)).
+// TestWriteDataFileNestedGeoNullCount checks the metrics for geo columns
nested
+// in a struct, a list and a map. The Parquet GEOMETRY logical type drops the
+// leaf's footer statistics, so null counts and bounds are computed from the
+// Arrow data. Null counts follow Parquet's leaf null_count: a null leaf, a
null
+// parent, and an empty list or map each count once.
func TestWriteDataFileNestedGeoNullCount(t *testing.T) {
iceSchema := iceberg.NewSchema(0,
iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int32, Required: false},
@@ -3315,41 +3314,76 @@ func TestWriteDataFileNestedGeoNullCount(t *testing.T) {
{ID: 3, Name: "g", Type:
iceberg.GeometryType{}, Required: false},
},
}},
+ iceberg.NestedField{ID: 4, Name: "l", Required: false, Type:
&iceberg.ListType{
+ ElementID: 5, Element: iceberg.GeometryType{},
ElementRequired: false,
+ }},
+ iceberg.NestedField{ID: 6, Name: "m", Required: false, Type:
&iceberg.MapType{
+ KeyID: 7, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 8, ValueType: iceberg.GeometryType{},
ValueRequired: false,
+ }},
)
arrowSchema, err := table.SchemaToArrowSchema(iceSchema, nil, true,
false)
require.NoError(t, err)
- pt, err := wktToWKB("POINT (1 2)")
- require.NoError(t, err)
+ wkbHex := func(s string) string {
+ g, err := wktToWKB(s)
+ require.NoError(t, err)
+
+ return g.String()
+ }
rec, _, err := array.RecordFromJSON(memory.DefaultAllocator,
arrowSchema, strings.NewReader(`[
- {"id": 1, "s": {"g": "`+pt.String()+`"}},
- {"id": 2, "s": {"g": null}},
- {"id": 3, "s": {"g": "`+pt.String()+`"}}
+ {"id": 1, "s": {"g": "`+wkbHex("POINT (1 2)")+`"},
+ "l": ["`+wkbHex("POINT (3 4)")+`", null, "`+wkbHex("POINT (-1
7)")+`"],
+ "m": [{"key": "a", "value": "`+wkbHex("POINT (5 6)")+`"},
{"key": "b", "value": null}]},
+ {"id": 2, "s": null, "l": null, "m": null},
+ {"id": 3, "s": {"g": null}, "l": [], "m": []}
]`))
require.NoError(t, err)
defer rec.Release()
+ full := internal.MetricsMode{Typ: internal.MetricModeFull}
fm := internal.GetFileFormat(iceberg.ParquetFile)
df, err := fm.WriteDataFile(context.Background(), iceio.NewMemFS(),
nil, internal.WriteFileInfo{
FileSchema: iceSchema,
Spec: *iceberg.UnpartitionedSpec,
FileName: "nested-geo.parquet",
StatsCols: map[int]internal.StatisticsCollector{
- 1: {FieldID: 1, IcebergTyp:
iceberg.PrimitiveTypes.Int32, ColName: "id", Mode: internal.MetricsMode{Typ:
internal.MetricModeFull}},
- 3: {FieldID: 3, IcebergTyp: iceberg.GeometryType{},
ColName: "s.g", Mode: internal.MetricsMode{Typ: internal.MetricModeFull}},
+ 1: {FieldID: 1, IcebergTyp:
iceberg.PrimitiveTypes.Int32, ColName: "id", Mode: full},
+ 3: {FieldID: 3, IcebergTyp: iceberg.GeometryType{},
ColName: "s.g", Mode: full},
+ 5: {FieldID: 5, IcebergTyp: iceberg.GeometryType{},
ColName: "l.element", Mode: full},
+ 7: {FieldID: 7, IcebergTyp:
iceberg.PrimitiveTypes.String, ColName: "m.key", Mode: full},
+ 8: {FieldID: 8, IcebergTyp: iceberg.GeometryType{},
ColName: "m.value", Mode: full},
},
WriteProps: fm.GetWriteProperties(iceberg.Properties{}),
Content: iceberg.EntryContentData,
}, []arrow.RecordBatch{rec})
require.NoError(t, err)
- require.Contains(t, df.ValueCounts(), 3)
- assert.EqualValues(t, 3, df.ValueCounts()[3])
- assert.NotContains(t, df.NullValueCounts(), 3, "nested geo null counts
are not recovered yet")
- assert.NotContains(t, df.LowerBoundValues(), 3)
- assert.NotContains(t, df.UpperBoundValues(), 3)
+ tests := []struct {
+ name string
+ fieldID int
+ values int64
+ nulls int64
+ lower, upper []byte
+ }{
+ // rows: {g: POINT (1 2)}, null, {g: null}
+ {"struct", 3, 3, 2, geoBoundBytes(1, 2), geoBoundBytes(1, 2)},
+ // rows: [POINT (3 4), null, POINT (-1 7)], null, []
+ {"list", 5, 5, 3, geoBoundBytes(-1, 4), geoBoundBytes(3, 7)},
+ // rows: {a: POINT (5 6), b: null}, null, {}
+ {"map", 8, 4, 3, geoBoundBytes(5, 6), geoBoundBytes(5, 6)},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ assert.Equal(t, tt.values, df.ValueCounts()[tt.fieldID])
+ require.Contains(t, df.NullValueCounts(), tt.fieldID)
+ assert.Equal(t, tt.nulls,
df.NullValueCounts()[tt.fieldID])
+ assert.Equal(t, tt.lower,
df.LowerBoundValues()[tt.fieldID])
+ assert.Equal(t, tt.upper,
df.UpperBoundValues()[tt.fieldID])
+ })
+ }
}
// geoBoundBytes builds the Iceberg geospatial single-value serialization of a
diff --git a/website/src/feature-status.md b/website/src/feature-status.md
index 9021bcbbf..c4c732521 100644
--- a/website/src/feature-status.md
+++ b/website/src/feature-status.md
@@ -58,7 +58,7 @@ All V1 features are supported. V1 is the format-version
baseline.
| Geometry / Geography types (schema) | Supported |
| Geometry: statistics and pruning | Supported |
| Geography: read, write, and schema | Supported (bounds/pruning intentionally
omitted for geodesic safety) |
-| Geometry / Geography: nested-geo bounds | Deferred |
+| Geometry / Geography: nested-geo null counts | Supported (like other nested
fields, no bounds are written) |
| Multi-argument transforms | Infrastructure present; no concrete
implementations exercised yet |
| Statistics files (read / round-trip) | Supported |
| NDV (Theta-sketch) statistics computation | Roadmap
([#1929](https://github.com/apache/iceberg-go/issues/1929)) |
