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 d28a6b9c2 feat(table): Add Parquet logical type Info for
`GEOMETRY`/`GEOGRAPHY` type and bump geoarrow-go version (#2125)
d28a6b9c2 is described below
commit d28a6b9c2b199a39984b02a24ef097e2917026de
Author: Colton Loftus <[email protected]>
AuthorDate: Mon Oct 19 11:17:04 2026 -0400
feat(table): Add Parquet logical type Info for `GEOMETRY`/`GEOGRAPHY` type
and bump geoarrow-go version (#2125)
* Bump Geoarrow Version
* Respond to feedback
* fix todo issue number
* add test without arrow schema info
* Respond to feedback
* Respond to last feedback / nits
* add issue into comment
---
go.mod | 2 +-
go.sum | 4 +-
table/arrow_utils_test.go | 198 ++++++++++++++++++++++++--
table/geo_write_test.go | 115 ++++++++++++++-
table/internal/geo_codec.go | 5 +
table/internal/parquet_files.go | 56 ++++++--
table/internal/parquet_files_internal_test.go | 129 +++++++++++++++++
table/internal/parquet_files_test.go | 72 +++++++++-
8 files changed, 552 insertions(+), 29 deletions(-)
diff --git a/go.mod b/go.mod
index 07ffe504f..04605d447 100644
--- a/go.mod
+++ b/go.mod
@@ -38,7 +38,7 @@ require (
github.com/beltran/gohive v1.8.1
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
github.com/docker/docker v28.5.2+incompatible
- github.com/geoarrow/geoarrow-go v0.0.0-20260403143023-f54751c3e3a1
+ github.com/geoarrow/geoarrow-go v0.0.0-20261005150217-fc2b33c3141d
github.com/go-sql-driver/mysql v1.10.1
github.com/google/go-cmp v0.7.0
github.com/google/uuid v1.6.0
diff --git a/go.sum b/go.sum
index 8125a833a..02a51af57 100644
--- a/go.sum
+++ b/go.sum
@@ -277,8 +277,8 @@ github.com/fsnotify/fsevents v0.2.0
h1:BRlvlqjvNTfogHfeBOFvSC9N0Ddy+wzQCQukyoD7o
github.com/fsnotify/fsevents v0.2.0/go.mod
h1:B3eEk39i4hz8y1zaWS/wPrAP4O6wkIl7HQwKBr1qH/w=
github.com/fvbommel/sortorder v1.1.0
h1:fUmoe+HLsBTctBDoaBwpQo5N+nrCp8g/BjKb/6ZQmYw=
github.com/fvbommel/sortorder v1.1.0/go.mod
h1:uk88iVf1ovNn1iLfgUVU2F9o5eO30ui720w+kxuqRs0=
-github.com/geoarrow/geoarrow-go v0.0.0-20260403143023-f54751c3e3a1
h1:8VcjuP5pKZ717K4zOLS9Lm4Zsn8JwD9X+CQGFCiRT2E=
-github.com/geoarrow/geoarrow-go v0.0.0-20260403143023-f54751c3e3a1/go.mod
h1:XMOIOA5J96jEzAOpBROPyz0n14sHsooeF+y2FtGdrCY=
+github.com/geoarrow/geoarrow-go v0.0.0-20261005150217-fc2b33c3141d
h1:QQOSvqmfmuD5LddsaabCTswIGGaRfACq8TafAf3IVJ4=
+github.com/geoarrow/geoarrow-go v0.0.0-20261005150217-fc2b33c3141d/go.mod
h1:axOiUIw8QTsT/Y72+ITKRIXh2UYKjfIQKncOcKsAh54=
github.com/go-jose/go-jose/v4 v4.1.4
h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
github.com/go-jose/go-jose/v4 v4.1.4/go.mod
h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
github.com/go-logr/logr v1.2.2/go.mod
h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
diff --git a/table/arrow_utils_test.go b/table/arrow_utils_test.go
index 8eabb9e08..7b34a366b 100644
--- a/table/arrow_utils_test.go
+++ b/table/arrow_utils_test.go
@@ -22,6 +22,8 @@ import (
"bytes"
"context"
"encoding/json"
+ "os"
+ "path/filepath"
"strconv"
"strings"
"testing"
@@ -38,6 +40,7 @@ import (
"github.com/apache/arrow-go/v18/parquet/schema"
"github.com/apache/arrow-go/v18/parquet/variant"
"github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table"
"github.com/geoarrow/geoarrow-go"
"github.com/google/uuid"
@@ -3339,8 +3342,17 @@ func
TestToRequestedSchemaGeoAbsentCRSAgainstSRID0SchemaFails(t *testing.T) {
}
// TestGeoTypeParquetRoundTrip pins the CRS a geo column carries on the Parquet
-// wire. arrow-go writes no GEOMETRY or GEOGRAPHY logical type, so the CRS
-// travels in the GeoArrow metadata of the stored Arrow schema.
+// wire. geoarrow-go's WKB type maps to the Parquet GEOMETRY or GEOGRAPHY
logical
+// type, and the CRS also travels in the GeoArrow metadata of the stored Arrow
+// schema.
+// Per the Parquet spec, SRIDs are written as "srid:<id>" and the default
+// OGC:CRS84 is omitted.
+//
+// Two cases differ from what Java writes and are pinned so a geoarrow-go bump
+// can't change them silently: EPSG:4326 collapses to OGC:CRS84 (so no Parquet
+// crs, and it reads back as the default CRS) where Java writes "EPSG:4326"
+// verbatim, and a projjson:<property> reference is written as the inline
+// PROJJSON document where Java writes the reference itself.
func TestGeoTypeParquetRoundTrip(t *testing.T) {
geomSRID0, err := iceberg.GeometryTypeOf("srid:0")
require.NoError(t, err)
@@ -3350,36 +3362,71 @@ func TestGeoTypeParquetRoundTrip(t *testing.T) {
require.NoError(t, err)
geomEPSG3857, err := iceberg.GeometryTypeOf("EPSG:3857")
require.NoError(t, err)
+ geomEPSG4326, err := iceberg.GeometryTypeOf("EPSG:4326")
+ require.NoError(t, err)
+ geomPROJJSON, err := iceberg.GeometryTypeOf("projjson:geo.crs")
+ require.NoError(t, err)
+
+ const projjson = `{"type":"GeographicCRS","name":"Custom WGS 84"}`
tests := []struct {
name string
icebergType iceberg.Type
+ tableProps iceberg.Properties
geoarrowMetaJSON string
+ parquetLogical schema.LogicalType
+ // readType is the type read back from the file; nil means
icebergType.
+ readType iceberg.Type
}{
{
name: "geometry_default_crs",
icebergType: iceberg.GeometryType{},
geoarrowMetaJSON:
`{"crs":"OGC:CRS84","crs_type":"authority_code"}`,
+ parquetLogical: schema.GeometryLogicalType{},
},
{
name: "geometry_srid_0",
icebergType: geomSRID0,
geoarrowMetaJSON: `{"crs":"0","crs_type":"srid"}`,
+ parquetLogical: schema.GeometryLogicalType{Crs:
"srid:0"},
+ },
+ {
+ name: "geography_default",
+ icebergType: iceberg.GeographyType{},
+ geoarrowMetaJSON:
`{"crs":"OGC:CRS84","crs_type":"authority_code","edges":"spherical"}`,
+ parquetLogical:
schema.GeographyLogicalType{Algorithm: schema.GeographyEdgeSpherical},
},
{
name: "geography_srid_0",
icebergType: geogSRID0,
geoarrowMetaJSON:
`{"crs":"0","crs_type":"srid","edges":"spherical"}`,
+ parquetLogical: schema.GeographyLogicalType{Crs:
"srid:0", Algorithm: schema.GeographyEdgeSpherical},
},
{
name: "geometry_srid_4326",
icebergType: geomSRID,
geoarrowMetaJSON: `{"crs":"4326","crs_type":"srid"}`,
+ parquetLogical: schema.GeometryLogicalType{Crs:
"srid:4326"},
},
{
name: "geometry_authority_code",
icebergType: geomEPSG3857,
geoarrowMetaJSON:
`{"crs":"EPSG:3857","crs_type":"authority_code"}`,
+ parquetLogical: schema.GeometryLogicalType{Crs:
"EPSG:3857"},
+ },
+ {
+ name: "geometry_epsg_4326",
+ icebergType: geomEPSG4326,
+ geoarrowMetaJSON:
`{"crs":"OGC:CRS84","crs_type":"authority_code"}`,
+ parquetLogical: schema.GeometryLogicalType{},
+ readType: iceberg.GeometryType{},
+ },
+ {
+ name: "geometry_projjson",
+ icebergType: geomPROJJSON,
+ tableProps: iceberg.Properties{"geo.crs":
projjson},
+ geoarrowMetaJSON: `{"crs":` + projjson +
`,"crs_type":"projjson"}`,
+ parquetLogical: schema.GeometryLogicalType{Crs:
projjson},
},
}
@@ -3387,7 +3434,10 @@ func TestGeoTypeParquetRoundTrip(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
mem := memory.NewGoAllocator()
- dt, err := table.TypeToArrowType(tt.icebergType, true,
false)
+ dt, err :=
table.TypeToArrowTypeWithOptions(tt.icebergType, table.ArrowSchemaOptions{
+ IncludeFieldIDs: true,
+ TableProperties: tt.tableProps,
+ })
require.NoError(t, err)
wkb, ok := dt.(*geoarrow.WKBType)
@@ -3414,10 +3464,8 @@ func TestGeoTypeParquetRoundTrip(t *testing.T) {
require.NoError(t, err)
defer rdr.Close()
- // Revisit the CRS spelling once arrow-go emits the geo
logical types:
- // the Parquet CRS field takes the prefixed srid:<id>
form.
logical := rdr.MetaData().Schema.Column(0).LogicalType()
- assert.True(t, logical.Equals(schema.NoLogicalType{}),
"unexpected logical type %s", logical)
+ assert.True(t, logical.Equals(tt.parquetLogical),
"expected logical type %s, got %s", tt.parquetLogical, logical)
arrRdr, err := pqarrow.NewFileReader(rdr,
pqarrow.ArrowReadProperties{}, mem)
require.NoError(t, err)
@@ -3425,13 +3473,145 @@ func TestGeoTypeParquetRoundTrip(t *testing.T) {
readSchema, err := arrRdr.Schema()
require.NoError(t, err)
- readIceberg, err :=
table.ArrowSchemaToIceberg(readSchema, false, nil)
+ readIceberg, err :=
table.ArrowSchemaToIcebergWithOptions(readSchema, table.ArrowToIcebergOptions{
+ TableSchema: iceberg.NewSchema(0,
iceberg.NestedField{
+ ID: 1, Name: "geom", Type:
tt.icebergType, Required: false,
+ }),
+ TableProperties: tt.tableProps,
+ })
require.NoError(t, err)
+ wantType := tt.readType
+ if wantType == nil {
+ wantType = tt.icebergType
+ }
field, ok := readIceberg.FindFieldByID(1)
require.True(t, ok)
- assert.Equal(t, tt.icebergType.String(),
field.Type.String())
- assert.True(t, tt.icebergType.Equals(field.Type),
"expected %s, got %s", tt.icebergType, field.Type)
+ assert.Equal(t, wantType.String(), field.Type.String())
+ assert.True(t, wantType.Equals(field.Type), "expected
%s, got %s", wantType, field.Type)
})
}
}
+
+// TestGeoParquetReadWithoutStoredSchema reads a GEOMETRY file that has no
+// stored ARROW:schema, as Java/Spark writers produce. geoarrow-go maps the
+// Parquet GEOMETRY logical type to geoarrow.wkb on read, so the file projects
+// onto a matching geometry column. A binary table column currently rejects it,
+// both on read and in the FileToDataFile (AddFiles) schema check. Whether
+// geometry should project onto a binary column is still an open decision
+// (https://github.com/apache/iceberg-go/issues/2193); flip the binary cases if
+// that promotion is allowed.
+func TestGeoParquetReadWithoutStoredSchema(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+ defer mem.AssertSize(t, 0)
+
+ geomType, err := iceberg.GeometryTypeOf("srid:4326")
+ require.NoError(t, err)
+
+ dt, err := table.TypeToArrowType(geomType, true, false)
+ require.NoError(t, err)
+ wkb, ok := dt.(*geoarrow.WKBType)
+ require.True(t, ok, "expected geoarrow.wkb, got %T", dt)
+
+ source := newExtensionArrayOverBinary(t, mem, wkb, wkbPointRows())
+ defer source.Release()
+ rec := singleColumnRecord(arrow.Field{
+ Name: "geom", Type: wkb, Nullable: true, Metadata:
fieldIDMeta("1"),
+ }, source)
+ defer rec.Release()
+
+ // No pqarrow.WithStoreSchema: the reader must derive types from the
+ // Parquet schema alone.
+ path := filepath.Join(t.TempDir(), "geo-no-arrow-schema.parquet")
+ f, err := os.Create(path)
+ require.NoError(t, err)
+ writer, err := pqarrow.NewFileWriter(rec.Schema(), f,
+ parquet.NewWriterProperties(parquet.WithAllocator(mem)),
+ pqarrow.NewArrowWriterProperties(pqarrow.WithAllocator(mem)))
+ require.NoError(t, err)
+ require.NoError(t, writer.Write(rec))
+ require.NoError(t, writer.Close())
+
+ readFile := func(t *testing.T) arrow.RecordBatch {
+ t.Helper()
+
+ rdr, err := file.OpenParquetFile(path, false,
file.WithReadProps(parquet.NewReaderProperties(mem)))
+ require.NoError(t, err)
+ defer rdr.Close()
+ require.Nil(t,
rdr.MetaData().KeyValueMetadata().FindValue("ARROW:schema"), "file must not
carry a stored Arrow schema")
+
+ arrRdr, err := pqarrow.NewFileReader(rdr,
pqarrow.ArrowReadProperties{}, mem)
+ require.NoError(t, err)
+ tbl, err := arrRdr.ReadTable(context.Background())
+ require.NoError(t, err)
+ defer tbl.Release()
+
+ trdr := array.NewTableReader(tbl, -1)
+ defer trdr.Release()
+ require.True(t, trdr.Next())
+ out := trdr.RecordBatch()
+ out.Retain()
+
+ return out
+ }
+
+ t.Run("reads as geoarrow.wkb", func(t *testing.T) {
+ got := readFile(t)
+ defer got.Release()
+
+ readType, ok := got.Schema().Field(0).Type.(*geoarrow.WKBType)
+ require.True(t, ok, "expected geoarrow.wkb, got %s",
got.Schema().Field(0).Type)
+
+ fileSchema, err := table.ArrowSchemaToIceberg(got.Schema(),
false, nil)
+ require.NoError(t, err)
+ field, ok := fileSchema.FindFieldByID(1)
+ require.True(t, ok)
+ assert.True(t, geomType.Equals(field.Type), "expected %s, got
%s (%s)", geomType, field.Type, readType.Serialize())
+ })
+
+ t.Run("projects onto geometry column", func(t *testing.T) {
+ got := readFile(t)
+ defer got.Release()
+
+ fileSchema, err := table.ArrowSchemaToIceberg(got.Schema(),
false, nil)
+ require.NoError(t, err)
+ requested := iceberg.NewSchema(0, iceberg.NestedField{ID: 1,
Name: "geom", Type: geomType, Required: false})
+
+ ctx := compute.WithAllocator(context.Background(), mem)
+ projected, err := table.ToRequestedSchema(ctx, requested,
fileSchema, got, table.SchemaOptions{IncludeFieldIDs: true})
+ require.NoError(t, err)
+ defer projected.Release()
+
+ require.EqualValues(t, 3, projected.NumRows())
+ storage := projected.Column(0).(array.ExtensionArray).Storage()
+ assert.Equal(t, wkbPointRows()[0], binaryValueAt(t, storage, 0))
+ assert.True(t, storage.IsNull(1))
+ assert.Equal(t, wkbPointRows()[2], binaryValueAt(t, storage, 2))
+ })
+
+ t.Run("binary column rejects it", func(t *testing.T) {
+ got := readFile(t)
+ defer got.Release()
+
+ fileSchema, err := table.ArrowSchemaToIceberg(got.Schema(),
false, nil)
+ require.NoError(t, err)
+ requested := iceberg.NewSchema(0, iceberg.NestedField{ID: 1,
Name: "geom", Type: iceberg.PrimitiveTypes.Binary, Required: false})
+
+ ctx := compute.WithAllocator(context.Background(), mem)
+ _, err = table.ToRequestedSchema(ctx, requested, fileSchema,
got, table.SchemaOptions{IncludeFieldIDs: true})
+ require.Error(t, err)
+ assert.ErrorContains(t, err, "cannot promote
geometry(srid:4326) to binary")
+ })
+
+ t.Run("AddFiles schema check rejects binary column", func(t *testing.T)
{
+ binarySchema := iceberg.NewSchema(0, iceberg.NestedField{ID: 1,
Name: "geom", Type: iceberg.PrimitiveTypes.Binary, Required: false})
+ _, err := table.FileToDataFile(context.Background(),
iceio.LocalFS{}, path, binarySchema, *iceberg.UnpartitionedSpec, 0, nil)
+ require.Error(t, err)
+ assert.ErrorContains(t, err, "mismatch in fields")
+ assert.ErrorContains(t, err, "geometry(srid:4326)")
+
+ geomSchema := iceberg.NewSchema(0, iceberg.NestedField{ID: 1,
Name: "geom", Type: geomType, Required: false})
+ _, err = table.FileToDataFile(context.Background(),
iceio.LocalFS{}, path, geomSchema, *iceberg.UnpartitionedSpec, 0, nil)
+ require.NoError(t, err)
+ })
+}
diff --git a/table/geo_write_test.go b/table/geo_write_test.go
index 409d0bf41..804ff6738 100644
--- a/table/geo_write_test.go
+++ b/table/geo_write_test.go
@@ -25,6 +25,7 @@ import (
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/compute"
"github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet/file"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/io"
tblutils "github.com/apache/iceberg-go/table/internal"
@@ -61,7 +62,7 @@ func wktToWKB(t *testing.T, s string) geoarrow.WKBBytes {
// exercise the stats-plan dispatch that decides whether geo bounds are
recorded.
//
// The geo columns are all top-level; nested geo columns are still unhandled in
-// the writer (see the TODO(#992) in parquet_files.go), so nothing here covers
+// the writer (see the TODO(#2158) in parquet_files.go), so nothing here covers
// them.
func newGeoTestWriter(t *testing.T, dir string, props iceberg.Properties)
(*defaultDataFileWriter, *iceberg.Schema, *arrow.Schema) {
t.Helper()
@@ -296,3 +297,115 @@ func TestWriteGeometryColumnCheckedAllocator(t
*testing.T) {
require.Contains(t, df.LowerBoundValues(), geoTestGeomFieldID,
"geometry column must record a lower bound")
require.Contains(t, df.UpperBoundValues(), geoTestGeomFieldID,
"geometry column must record an upper bound")
}
+
+// TestWriteGeoColumnMultiRowGroupStats writes geo columns across several row
+// groups and batches. Parquet GEOMETRY/GEOGRAPHY column chunks omit the
+// standard Statistics block (min, max, null count), so DataFileStatsFromMeta
+// invalidates the column in the first row group; value counts and column sizes
+// (which live outside that block) must still sum over every row group, and the
+// null count (tallied from the Arrow data) must sum over every batch. A value
+// count from row group 0 only, paired with the whole-file null count, would
make
+// the file look all-null and let NotNull/IsNull evaluators drop live rows.
+func TestWriteGeoColumnMultiRowGroupStats(t *testing.T) {
+ t.Parallel()
+
+ writer, schema, arrowSchema := newGeoTestWriter(t, t.TempDir(),
iceberg.Properties{
+ tblutils.ParquetRowGroupLimitKey: "2",
+ })
+
+ pt := wktToWKB(t, "POINT (1 2)").String()
+ // Nulls fill row group 0 only. The second batch has none, so a null
tally
+ // that overwrote rather than summed across batches would record 0.
+ first, _, err := array.RecordFromJSON(memory.DefaultAllocator,
arrowSchema, strings.NewReader(`[
+ {"id": 1, "geom": null, "geog": null},
+ {"id": 2, "geom": null, "geog": null},
+ {"id": 3, "geom": "`+pt+`", "geog": "`+pt+`"}
+ ]`))
+ require.NoError(t, err)
+ defer first.Release()
+ second, _, err := array.RecordFromJSON(memory.DefaultAllocator,
arrowSchema, strings.NewReader(`[
+ {"id": 4, "geom": "`+pt+`", "geog": "`+pt+`"},
+ {"id": 5, "geom": "`+pt+`", "geog": "`+pt+`"},
+ {"id": 6, "geom": "`+pt+`", "geog": "`+pt+`"}
+ ]`))
+ require.NoError(t, err)
+ defer second.Release()
+
+ df, err := writer.writeFile(t.Context(), nil, WriteTask{
+ Uuid: uuid.New(),
+ ID: 0,
+ FileCount: 1,
+ Schema: schema,
+ Batches: []arrow.RecordBatch{first, second},
+ })
+ require.NoError(t, err)
+ require.EqualValues(t, 6, df.Count())
+ require.Len(t, df.SplitOffsets(), 3, "row group limit of 2 must produce
3 row groups")
+
+ for _, tt := range []struct {
+ name string
+ fieldID int
+ }{
+ {"geom", geoTestGeomFieldID},
+ {"geog", geoTestGeogFieldID},
+ } {
+ require.Contains(t, df.ValueCounts(), tt.fieldID)
+ assert.EqualValues(t, 6, df.ValueCounts()[tt.fieldID], "value
count must cover every row group")
+ require.Contains(t, df.NullValueCounts(), tt.fieldID)
+ assert.EqualValues(t, 2, df.NullValueCounts()[tt.fieldID],
"null count must cover every batch")
+
+ // Recompute the column's compressed size directly from the row
groups
+ // on disk, independently of DataFileStatsFromMeta, and require
an exact
+ // match. A regression that only summed row group 0 would still
produce
+ // a non-zero, Contains-passing size, so the independent total
is the
+ // only thing that actually pins the multi-row-group sum.
+ require.Contains(t, df.ColumnSizes(), tt.fieldID)
+ wantSize := sumColumnChunkCompressedSize(t, df.FilePath(),
tt.name)
+ assert.Greater(t, wantSize, int64(0), "sanity: the column must
occupy some space across 3 row groups")
+ assert.EqualValues(t, wantSize, df.ColumnSizes()[tt.fieldID],
"column size must sum TotalCompressedSize over every row group")
+ }
+
+ for _, name := range []string{"geom", "geog"} {
+ notNull, err := newInclusiveMetricsEvaluator(schema,
iceberg.NotNull(iceberg.Reference(name)), true, true)
+ require.NoError(t, err)
+ mightMatch, err := notNull(df)
+ require.NoError(t, err)
+ assert.True(t, mightMatch, "NotNull(%s) must not prune a file
with non-null rows", name)
+
+ isNull, err := newStrictMetricsEvaluator(schema,
iceberg.IsNull(iceberg.Reference(name)), true, true)
+ require.NoError(t, err)
+ mustMatch, err := isNull(df)
+ require.NoError(t, err)
+ assert.False(t, mustMatch, "IsNull(%s) must not match every row
of a file with non-null rows", name)
+ }
+}
+
+// sumColumnChunkCompressedSize opens the Parquet file at path and sums
+// TotalCompressedSize for colName's column chunk across every row group, the
+// same quantity DataFileStatsFromMeta accumulates into DataFile.ColumnSizes.
+// Computing it independently here, straight from the file's row groups,
+// catches a regression that only tallies row group 0.
+func sumColumnChunkCompressedSize(t *testing.T, path, colName string) int64 {
+ t.Helper()
+
+ f, err := (io.LocalFS{}).Open(path)
+ require.NoError(t, err)
+ defer f.Close()
+
+ rdr, err := file.NewParquetReader(f)
+ require.NoError(t, err)
+ defer rdr.Close()
+
+ meta := rdr.MetaData()
+ pos := meta.Schema.ColumnIndexByName(colName)
+ require.GreaterOrEqual(t, pos, 0, "column %q not found in Parquet
schema", colName)
+
+ var total int64
+ for rg := range meta.NumRowGroups() {
+ colChunk, err := meta.RowGroup(rg).ColumnChunk(pos)
+ require.NoError(t, err)
+ total += colChunk.TotalCompressedSize()
+ }
+
+ return total
+}
diff --git a/table/internal/geo_codec.go b/table/internal/geo_codec.go
index 48de8c119..6484690c1 100644
--- a/table/internal/geo_codec.go
+++ b/table/internal/geo_codec.go
@@ -72,6 +72,11 @@ type geoBoundsAccumulator struct {
// isGeography marks a column whose edges are geodesics on a sphere, for
// which raw vertex min/max is not a safe bounding box (see Bounds).
isGeography bool
+
+ // nulls counts the null rows seen across every batch written to the
file.
+ // Parquet omits the Statistics block for GEOMETRY/GEOGRAPHY column
chunks,
+ // so the footer cannot supply this count.
+ nulls int64
}
func newGeoBoundsAccumulator(isGeography bool) *geoBoundsAccumulator {
diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index 7a8b36f37..f8ee6b4f7 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -531,7 +531,15 @@ type geoColumn struct {
// 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.
-// TODO(#992): compute geo bounds for geo columns nested in structs/lists/maps.
+//
+// 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.
func collectGeoColumns(sc *arrow.Schema, colMapping map[string]int)
[]geoColumn {
var result []geoColumn
for i, f := range sc.Fields() {
@@ -1188,25 +1196,34 @@ func normalizeWKBArrayReachable(ext
array.ExtensionArray, active []bool, mem mem
}
// accumulateGeoBounds extends the per-field bounding boxes with the WKB values
-// in this batch. Null rows are skipped; a malformed WKB value fails the write.
+// 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
+// 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
{
for _, gc := range w.geoCols {
if gc.colIdx >= int(batch.NumCols()) {
- continue
+ 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 {
- continue
+ 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 {
- continue
+ return fmt.Errorf("geo field %d: unsupported WKB
storage type %s",
+ gc.fieldID, ext.Storage().DataType())
}
acc, ok := w.geoAccs[gc.fieldID]
if !ok {
- continue
+ 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) {
continue
@@ -1268,17 +1285,30 @@ func (w *ParquetFileWriter) Abort() error {
return errors.Join(closeErr, removeErr)
}
-// applyGeoBounds injects the WKB single-point bounds accumulated during the
-// write into the file statistics, so they flow through ToDataFile into the
-// manifest entry like any other typed bound.
+// applyGeoBounds injects the WKB single-point bounds and null counts
+// accumulated during the write into the file statistics, so they flow through
+// ToDataFile into the manifest entry like any other typed bound.
+//
+// Parquet GEOMETRY/GEOGRAPHY columns have an undefined sort order, so the
+// Parquet writer omits the standard Statistics block (min, max, null count)
+// from their column chunks and DataFileStatsFromMeta cannot recover null
+// counts for them.
func (w *ParquetFileWriter) applyGeoBounds(stats *DataFileStatistics) error {
for fieldID, acc := range w.geoAccs {
// Honor the column's metrics mode: a column the caller never
registered
- // (missing key) or one set to counts/none does not record
bounds. Check
- // presence first — a missing key yields a zero-value mode of
"" that would
- // otherwise fall through and write bounds the caller asked to
skip.
+ // (missing key) or one set to none records nothing, and counts
records
+ // no bounds. Check presence first — a missing key yields a
zero-value
+ // mode of "" that would otherwise fall through and write
bounds the
+ // caller asked to skip.
sc, ok := w.info.StatsCols[fieldID]
- if !ok || sc.Mode.Typ == MetricModeNone || sc.Mode.Typ ==
MetricModeCounts {
+ if !ok || sc.Mode.Typ == MetricModeNone {
+ continue
+ }
+ if stats.NullValueCounts == nil {
+ stats.NullValueCounts = make(map[int]int64)
+ }
+ stats.NullValueCounts[fieldID] = acc.nulls
+ if sc.Mode.Typ == MetricModeCounts {
continue
}
diff --git a/table/internal/parquet_files_internal_test.go
b/table/internal/parquet_files_internal_test.go
index a665f8c50..b7f4d6f6f 100644
--- a/table/internal/parquet_files_internal_test.go
+++ b/table/internal/parquet_files_internal_test.go
@@ -23,6 +23,7 @@ import (
"testing"
"github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/extensions"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/parquet"
@@ -96,6 +97,134 @@ func TestWriteDataFileBatchesAbortsOnWriteError(t
*testing.T) {
}
}
+// TestWriteRejectsMismatchedGeoBatch drives the public Write method with a
+// batch that doesn't match the geo columns collected from the writer's Arrow
+// schema, pinning accumulateGeoBounds's error returns against a regression
+// back to silently skipping
+func TestWriteRejectsMismatchedGeoBatch(t *testing.T) {
+ mem := memory.DefaultAllocator
+
+ typeDef := geoarrow.NewWKBType(geoarrow.WKBWithBinaryStorage())
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: false},
+ {Name: "geom", Type: typeDef, Nullable: true},
+ }, nil)
+ fileSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "geom", Type:
iceberg.GeometryType{}, Required: false},
+ )
+ format := parquetFormat{}
+
+ newWriter := func(t *testing.T) FileWriter {
+ t.Helper()
+
+ writer, err := format.NewFileWriter(context.Background(),
iceio.NewMemFS(), nil, WriteFileInfo{
+ FileSchema: fileSchema,
+ Spec: *iceberg.UnpartitionedSpec,
+ FileName: "mismatched-geo.parquet",
+ WriteProps:
format.GetWriteProperties(iceberg.Properties{}),
+ }, arrowSchema)
+ require.NoError(t, err)
+
+ return writer
+ }
+
+ idField := arrow.Field{Name: "id", Type: arrow.PrimitiveTypes.Int32}
+ idBuilder := array.NewInt32Builder(mem)
+ idBuilder.AppendValues([]int32{1, 2}, nil)
+ idArr := idBuilder.NewArray()
+ idBuilder.Release()
+ defer idArr.Release()
+
+ t.Run("column missing from batch", func(t *testing.T) {
+ writer := newWriter(t)
+ defer func() { require.NoError(t, writer.Abort()) }()
+
+ sc := arrow.NewSchema([]arrow.Field{idField}, nil)
+ batch := array.NewRecordBatch(sc, []arrow.Array{idArr},
int64(idArr.Len()))
+ defer batch.Release()
+
+ err := writer.Write(batch)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "missing from batch")
+ })
+
+ t.Run("column is not a WKB extension array", func(t *testing.T) {
+ writer := newWriter(t)
+ defer func() { require.NoError(t, writer.Abort()) }()
+
+ geomBuilder := array.NewInt32Builder(mem)
+ geomBuilder.AppendValues([]int32{10, 20}, nil)
+ geomArr := geomBuilder.NewArray()
+ geomBuilder.Release()
+ defer geomArr.Release()
+
+ sc := arrow.NewSchema([]arrow.Field{idField, {Name: "geom",
Type: arrow.PrimitiveTypes.Int32}}, nil)
+ batch := array.NewRecordBatch(sc, []arrow.Array{idArr,
geomArr}, int64(idArr.Len()))
+ defer batch.Release()
+
+ err := writer.Write(batch)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "expected a WKB extension
array")
+ })
+
+ t.Run("extension storage is not a WKB-compatible array", func(t
*testing.T) {
+ writer := newWriter(t)
+ defer func() { require.NoError(t, writer.Abort()) }()
+
+ // OpaqueType lets the test pick an arbitrary storage type
(here Int32,
+ // which has no Value(int) []byte method) without implementing
a whole
+ // fake extension type by hand. normalizeGeoBatch leaves it
untouched
+ // (it only rewrites *geoarrow.WKBType extensions), so it
reaches
+ // accumulateGeoBounds exactly as built here.
+ opaqueType :=
extensions.NewOpaqueType(arrow.PrimitiveTypes.Int32, "fake-wkb", "test")
+ storageBuilder := array.NewInt32Builder(mem)
+ storageBuilder.AppendValues([]int32{1, 2}, nil)
+ storage := storageBuilder.NewArray()
+ storageBuilder.Release()
+ defer storage.Release()
+
+ geomArr := array.NewExtensionArrayWithStorage(opaqueType,
storage)
+ defer geomArr.Release()
+
+ sc := arrow.NewSchema([]arrow.Field{idField, {Name: "geom",
Type: opaqueType}}, nil)
+ batch := array.NewRecordBatch(sc, []arrow.Array{idArr,
geomArr}, int64(idArr.Len()))
+ defer batch.Release()
+
+ err := writer.Write(batch)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "unsupported WKB storage type")
+ })
+
+ t.Run("geo column has no bounds accumulator", func(t *testing.T) {
+ writer := newWriter(t)
+ defer func() { require.NoError(t, writer.Abort()) }()
+
+ // geoAccs is built from the same geoCols today; drop the entry
to
+ // simulate the two drifting apart.
+ pw, ok := writer.(*ParquetFileWriter)
+ require.True(t, ok, "expected *ParquetFileWriter, got %T",
writer)
+ delete(pw.geoAccs, 2)
+
+ storageBuilder := array.NewBinaryBuilder(mem,
arrow.BinaryTypes.Binary)
+ storageBuilder.AppendNull()
+ storageBuilder.AppendNull()
+ storage := storageBuilder.NewArray()
+ storageBuilder.Release()
+ defer storage.Release()
+
+ geomArr := array.NewExtensionArrayWithStorage(typeDef, storage)
+ defer geomArr.Release()
+
+ batch := array.NewRecordBatch(arrowSchema, []arrow.Array{idArr,
geomArr}, int64(idArr.Len()))
+ defer batch.Release()
+
+ err := writer.Write(batch)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "no bounds accumulator")
+ })
+}
+
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 7eefaed81..2b477cdaa 100644
--- a/table/internal/parquet_files_test.go
+++ b/table/internal/parquet_files_test.go
@@ -3210,7 +3210,8 @@ func TestWriteDataFileGeoBounds(t *testing.T) {
rec, _, err := array.RecordFromJSON(memory.DefaultAllocator,
arrowSchema, strings.NewReader(`[
{"id": 1, "geom": "`+wkbStr("POINT (30 10)")+`", "geog":
"`+wkbStr("POINT (20 5)")+`"},
- {"id": 2, "geom": "`+wkbStr("POINT (5 40)")+`", "geog": null}
+ {"id": 2, "geom": "`+wkbStr("POINT (5 40)")+`", "geog": null},
+ {"id": 3, "geom": null, "geog": "`+wkbStr("POINT (20 5)")+`"}
]`))
require.NoError(t, err)
defer rec.Release()
@@ -3245,7 +3246,7 @@ func TestWriteDataFileGeoBounds(t *testing.T) {
df := writeWithGeomMode(t, internal.MetricsMode{Typ:
internal.MetricModeFull})
lower, upper := df.LowerBoundValues(), df.UpperBoundValues()
- // Geometry bounds span both rows: lower (5, 10), upper (30,
40), encoded as
+ // Geometry bounds span both non-null rows: lower (5, 10),
upper (30, 40), encoded as
// two little-endian float64 coordinates (16 bytes), not WKB.
require.Len(t, lower[2], 16)
require.Len(t, upper[2], 16)
@@ -3263,7 +3264,13 @@ func TestWriteDataFileGeoBounds(t *testing.T) {
assert.True(t, geomType.Equals(lit.Type()), "want geo type, got
%s", lit.Type())
assert.Equal(t, geoBoundBytes(5, 10),
lit.(iceberg.GeoLiteral).Value())
- // Null counts are still recorded for geo columns.
+ // Null counts are still recorded for geo columns. The Parquet
writer
+ // omits the Statistics block (min, max, null count) for
GEOMETRY/
+ // GEOGRAPHY column chunks, so these come from the Arrow data
rather
+ // than the footer.
+ require.Contains(t, df.NullValueCounts(), 2)
+ assert.Equal(t, int64(1), df.NullValueCounts()[2])
+ require.Contains(t, df.NullValueCounts(), 3)
assert.Equal(t, int64(1), df.NullValueCounts()[3])
})
@@ -3282,10 +3289,69 @@ func TestWriteDataFileGeoBounds(t *testing.T) {
df := writeWithGeomMode(t, tt.mode)
assert.NotContains(t, df.LowerBoundValues(), 2,
"geometry lower bound must be omitted for %s", tt.mode.Typ)
assert.NotContains(t, df.UpperBoundValues(), 2,
"geometry upper bound must be omitted for %s", tt.mode.Typ)
+
+ // Counts mode still records the geometry null count;
none records nothing.
+ if tt.mode.Typ == internal.MetricModeCounts {
+ require.Contains(t, df.NullValueCounts(), 2,
"counts mode must record the geometry null count")
+ assert.Equal(t, int64(1),
df.NullValueCounts()[2])
+ } else {
+ assert.NotContains(t, df.NullValueCounts(), 2)
+ }
})
}
}
+// 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)).
+func TestWriteDataFileNestedGeoNullCount(t *testing.T) {
+ iceSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int32, Required: false},
+ iceberg.NestedField{ID: 2, Name: "s", Required: false, Type:
&iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 3, Name: "g", Type:
iceberg.GeometryType{}, Required: false},
+ },
+ }},
+ )
+
+ arrowSchema, err := table.SchemaToArrowSchema(iceSchema, nil, true,
false)
+ require.NoError(t, err)
+
+ pt, err := wktToWKB("POINT (1 2)")
+ require.NoError(t, err)
+
+ 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()+`"}}
+ ]`))
+ require.NoError(t, err)
+ defer rec.Release()
+
+ 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}},
+ },
+ 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)
+}
+
// geoBoundBytes builds the Iceberg geospatial single-value serialization of a
// bound point: little-endian float64 coordinates in X, Y[, Z][, M] order.
func geoBoundBytes(coords ...float64) []byte {
