iceberg: record geometry bounds on the data files compaction writes (#11898)

* iceberg: record geometry bounds on the data files compaction writes

Compaction collected column statistics for every primitive column except
geometry: the bound converter has no geometry case, so a compacted file's
manifest entry carried value and null counts for a geometry column but no
lower or upper bound. Readers that skip files by these bounds for a spatial
filter (for example DuckDB for `geom && <box>`) had to open every compacted
file, where the files before compaction could be skipped.

The bound follows the spec's single-value serialization for geometry: the
lower bound is the point (xmin, ymin), the upper bound (xmax, ymax), each an
8-byte little-endian double; z and m are not written. It is the union of the
written file's row-group geospatial statistics, which parquet-go computes
from every non-null value.

A bound that is too small makes a reader skip rows, so any doubt gives none:
a row group with non-null values but no geospatial statistics (parquet-go
omits them when a value does not parse as WKB), a box that is not finite, an
inverted box, or a chunk whose statistics cannot be read. A row group of
nulls only constrains nothing. Geography gets no bound, as iceberg-go writes
none for it (geodesic edges, boxes across the antimeridian).

Tests: points and polygons far apart through a bin-packed compaction (the
bound equals the union of every geometry), a sorted compaction (any bound
covers every geometry), a column of nulls only, an unparseable value in a
second row group (no bound; without the rule the file got the bound of the
first row group only), geography, and the aggregation rules.

* iceberg: do not order compaction bins by a geometry column

When a table declares no usable sort order, compaction orders a bin's files
by the first column every file has bounds for. A v3 table whose first column
is a geometry qualifies as soon as its files carry geometry bounds (DuckDB
writes them, and compacted files now do too). Those bounds decode into geo
literals that compareLiterals cannot compare, so the order sorted nothing,
yet an oversized bin was still split into "ordered" runs by splitOrderedBin.

Geometry and geography columns are now skipped in both the declared sort
order and the inferred one, so such a table is ordered by its next column
with bounds.

Test: TestResolveCompactionOrderSkipsGeometry, a geometry column first and
an id column second; on master the order is the geometry column.

---------

Co-authored-by: Javier de la Torre <jatorre@lakehousebox.com>

weed/worker/tasks/iceberg/compact_order.go

@@ -37,7 +37,7 @@ func resolveCompactionOrder(meta table.Metadata, entries []iceberg.ManifestEntry
 				continue
 			}
 			typ, ok := field.Type.(iceberg.PrimitiveType)
-			if !ok {
+			if !ok || !orderableType(typ) {
 				continue
 			}
 			candidate := &compactionOrder{fieldID: field.ID, typ: typ, descending: sortField.Direction == table.SortDESC}
@@ -57,7 +57,7 @@ func resolveCompactionOrder(meta table.Metadata, entries []iceberg.ManifestEntry
 	// different format, or another partition) cannot disable ordering here.
 	for _, field := range schema.Fields() {
 		typ, ok := field.Type.(iceberg.PrimitiveType)
-		if !ok {
+		if !ok || !orderableType(typ) {
 			continue
 		}
 		candidate := &compactionOrder{fieldID: field.ID, typ: typ, bestEffort: true}
@@ -68,6 +68,18 @@ func resolveCompactionOrder(meta table.Metadata, entries []iceberg.ManifestEntry
 	return nil
 }
 
+// orderableType reports whether bounds of the type order files. Geometry and
+// geography bounds are corners of a box: they decode into literals that do
+// not compare, so ordering by them would sort nothing while still splitting
+// oversized bins as if the files were in order.
+func orderableType(t iceberg.PrimitiveType) bool {
+	switch t.(type) {
+	case iceberg.GeometryType, iceberg.GeographyType:
+		return false
+	}
+	return true
+}
+
 func (o *compactionOrder) coversEntries(entries []iceberg.ManifestEntry) bool {
 	for _, entry := range entries {
 		if _, ok := o.key(entry); !ok {

weed/worker/tasks/iceberg/compact_order_test.go

@@ -386,3 +386,36 @@ func TestCompactDataFilesKeepsRunsWhenRepackFails(t *testing.T) {
 		t.Fatalf("expected the ordered run to survive the failed repack, got %q", result)
 	}
 }
+
+// Geometry bounds have no order, so they never order a bin's files: a table
+// whose first column is a geometry is ordered by the next column with bounds.
+func TestResolveCompactionOrderSkipsGeometry(t *testing.T) {
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Type: iceberg.GeometryType{}, Name: "geom"},
+		iceberg.NestedField{ID: 2, Type: iceberg.PrimitiveTypes.Int64, Name: "id", Required: true})
+	meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, "s3://tb/ns/geo", iceberg.Properties{"format-version": "3"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	geomField, _ := meta.CurrentSchema().FindFieldByName("geom")
+	idField, _ := meta.CurrentSchema().FindFieldByName("id")
+	snap := int64(1)
+	var entries []iceberg.ManifestEntry
+	for i, name := range []string{"a.parquet", "b.parquet", "c.parquet"} {
+		b, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentData, "s3://tb/ns/geo/data/"+name, iceberg.ParquetFile, map[int]any{}, nil, nil, 1, 100)
+		if err != nil {
+			t.Fatal(err)
+		}
+		b.LowerBoundValues(map[int][]byte{geomField.ID: xyBound(float64(i), 0), idField.ID: geoLE64(int64(10 - i))})
+		b.UpperBoundValues(map[int][]byte{geomField.ID: xyBound(float64(i)+1, 1), idField.ID: geoLE64(int64(10 - i))})
+		entries = append(entries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snap, nil, nil, b.Build()))
+	}
+	order := resolveCompactionOrder(meta, entries)
+	if order == nil || order.fieldID != idField.ID {
+		t.Fatalf("order %+v, want the id column (field %d), not the geometry (field %d)", order, idField.ID, geomField.ID)
+	}
+	order.sortEntries(entries)
+	if first := entries[0].DataFile().FilePath(); !strings.HasSuffix(first, "c.parquet") {
+		t.Errorf("first file %s, want c.parquet (the lowest id)", first)
+	}
+}

weed/worker/tasks/iceberg/compact_stats.go

@@ -64,6 +64,7 @@ func collectColumnStats(data []byte, schema *iceberg.Schema) (*columnStats, erro
 	stats := &columnStats{sizes: map[int]int64{}, values: map[int]int64{}, nulls: map[int]int64{},
 		lower: map[int][]byte{}, upper: map[int][]byte{}}
 	agg := map[int]*bounds{}
+	geoAgg := map[int]*geometryBounds{}
 	meta := file.Metadata()
 	for rgIdx, rg := range file.RowGroups() {
 		stats.splitOffsets = append(stats.splitOffsets, rowGroupOffset(&meta.RowGroups[rgIdx]))
@@ -82,9 +83,24 @@ func collectColumnStats(data []byte, schema *iceberg.Schema) (*columnStats, erro
 			chunk, ok := cc.(*parquet.FileColumnChunk)
 			if !ok {
 				b.invalid = true
+				geoAgg[field.ID] = &geometryBounds{invalid: true}
 				continue
 			}
 			stats.nulls[field.ID] += chunk.NullCount()
+			switch field.Type.(type) {
+			case iceberg.GeometryType:
+				// the bound comes from the row group's geospatial statistics
+				g := geoAgg[field.ID]
+				if g == nil {
+					g = &geometryBounds{}
+					geoAgg[field.ID] = g
+				}
+				g.add(meta.RowGroups[rgIdx].Columns[col].MetaData.GeospatialStatistics, chunk.NumValues(), chunk.NullCount())
+				continue
+			case iceberg.GeographyType:
+				// no bound: edges are geodesic and a box may cross the antimeridian
+				continue
+			}
 			if chunk.NullCount() == chunk.NumValues() {
 				continue // an all-null chunk constrains nothing
 			}
@@ -122,6 +138,11 @@ func collectColumnStats(data []byte, schema *iceberg.Schema) (*columnStats, erro
 			stats.upper[id] = raw
 		}
 	}
+	for id, g := range geoAgg {
+		if lower, upper, ok := g.encoded(); ok {
+			stats.lower[id], stats.upper[id] = lower, upper
+		}
+	}
 	sort.Slice(stats.splitOffsets, func(i, j int) bool { return stats.splitOffsets[i] < stats.splitOffsets[j] })
 	return stats, nil
 }

weed/worker/tasks/iceberg/geometry_bounds.go

@@ -0,0 +1,71 @@
+package iceberg
+
+import (
+	"encoding/binary"
+	"math"
+
+	"github.com/parquet-go/parquet-go/format"
+)
+
+// geometryBounds aggregates the bound of a geometry column over the row groups
+// of a written file. The spec serializes a geometry bound as a point: the
+// lower bound is (xmin, ymin), the upper bound (xmax, ymax), each coordinate
+// an 8-byte little-endian double; z and m are optional and not written here.
+//
+// Readers skip a file whose bound does not intersect a spatial filter, so a
+// box smaller than the file's geometries returns wrong results. Any doubt
+// gives no bound:
+//   - the box is the union of the row groups' geospatial statistics, which
+//     the Parquet writer computes from every non-null value;
+//   - a row group with non-null values but no statistics (the writer omits
+//     them when a value does not parse as WKB) gives the column no bound, as
+//     does a box that is not finite or is inverted;
+//   - a row group of nulls only constrains nothing.
+type geometryBounds struct {
+	xmin, ymin, xmax, ymax float64
+	seen                   bool
+	invalid                bool
+}
+
+func (g *geometryBounds) add(gs format.GeospatialStatistics, numValues, nullCount int64) {
+	if g.invalid || numValues <= 0 || nullCount >= numValues {
+		return
+	}
+	if len(gs.GeoSpatialTypes) == 0 {
+		g.invalid = true
+		return
+	}
+	bb := gs.BBox
+	for _, v := range []float64{bb.XMin, bb.XMax, bb.YMin, bb.YMax} {
+		if math.IsNaN(v) || math.IsInf(v, 0) {
+			g.invalid = true
+			return
+		}
+	}
+	if bb.XMin > bb.XMax || bb.YMin > bb.YMax {
+		g.invalid = true
+		return
+	}
+	if !g.seen {
+		g.xmin, g.ymin, g.xmax, g.ymax, g.seen = bb.XMin, bb.YMin, bb.XMax, bb.YMax, true
+		return
+	}
+	g.xmin, g.ymin = math.Min(g.xmin, bb.XMin), math.Min(g.ymin, bb.YMin)
+	g.xmax, g.ymax = math.Max(g.xmax, bb.XMax), math.Max(g.ymax, bb.YMax)
+}
+
+// encoded returns the lower and upper bound, or false when the column gets
+// none.
+func (g *geometryBounds) encoded() (lower, upper []byte, ok bool) {
+	if g == nil || g.invalid || !g.seen {
+		return nil, nil, false
+	}
+	return encodeGeometryPoint(g.xmin, g.ymin), encodeGeometryPoint(g.xmax, g.ymax), true
+}
+
+func encodeGeometryPoint(x, y float64) []byte {
+	b := make([]byte, 16)
+	binary.LittleEndian.PutUint64(b, math.Float64bits(x))
+	binary.LittleEndian.PutUint64(b[8:], math.Float64bits(y))
+	return b
+}

weed/worker/tasks/iceberg/geometry_bounds_test.go

@@ -0,0 +1,215 @@
+package iceberg
+
+import (
+	"fmt"
+	"math"
+	"strings"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+	"github.com/parquet-go/parquet-go/format"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+type geoBox struct{ xmin, ymin, xmax, ymax float64 }
+
+func (b geoBox) String() string {
+	return fmt.Sprintf("(%g %g, %g %g)", b.xmin, b.ymin, b.xmax, b.ymax)
+}
+
+func (b geoBox) covers(o geoBox) bool {
+	return b.xmin <= o.xmin && b.ymin <= o.ymin && b.xmax >= o.xmax && b.ymax >= o.ymax
+}
+
+// compactedGeoBounds is the geometry bound of each compacted file in the
+// current manifests (nil: none), and whether its id column has a bound, the
+// control that statistics were collected at all.
+func compactedGeoBounds(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup) (map[string]*geoBox, map[string]bool) {
+	t.Helper()
+	geo, idBounded := map[string]*geoBox{}, map[string]bool{}
+	for _, e := range currentDataEntries(t, client, setup) {
+		df := e.DataFile()
+		if !strings.Contains(df.FilePath(), "/compact-") {
+			continue
+		}
+		idBounded[df.FilePath()] = df.LowerBoundValues()[geoTestIDField] != nil
+		lo, hi := df.LowerBoundValues()[geoTestGeomField], df.UpperBoundValues()[geoTestGeomField]
+		if lo == nil && hi == nil {
+			geo[df.FilePath()] = nil
+			continue
+		}
+		x0, y0 := decodeXYBound(t, lo)
+		x1, y1 := decodeXYBound(t, hi)
+		geo[df.FilePath()] = &geoBox{x0, y0, x1, y1}
+	}
+	if len(geo) == 0 {
+		t.Fatal("no compacted file in the manifests")
+	}
+	return geo, idBounded
+}
+
+// mixedGeoFiles holds points and polygons far apart, so a bound that misses
+// any of them shows.
+func mixedGeoFiles() ([]geoFile, geoBox) {
+	files := []geoFile{
+		{Name: "pts.parquet", Rows: []geoRow{{1, wkbPoint(-3.70, 40.41)}, {2, wkbPoint(2.17, 41.38)}, {3, wkbPoint(-5.98, 37.38)}}},
+		{Name: "boxes.parquet", Rows: []geoRow{{4, wkbBox(-9.3, 42.1, -8.5, 43.4)}, {5, wkbBox(1.0, 36.0, 4.3, 39.9)}}},
+		{Name: "nulls.parquet", Rows: []geoRow{{6, nil}, {7, nil}}},
+		{Name: "mixed.parquet", Rows: []geoRow{{8, nil}, {9, wkbPoint(-1.10, 40.34)}, {10, wkbBox(-0.5, 38.9, -0.3, 39.5)}}},
+	}
+	return files, geoBox{-9.3, 36.0, 4.3, 43.4}
+}
+
+func TestCompactRecordsGeometryBounds(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+	files, want := mixedGeoFiles()
+	populateGeoTable(t, fs, setup, files, geoTableOptions{})
+	if result := compactGeo(t, client, setup, Config{}); !strings.Contains(result, "compacted 4 files") {
+		t.Fatalf("result %q", result)
+	}
+	geo, idBounded := compactedGeoBounds(t, client, setup)
+	for p, got := range geo {
+		if !idBounded[p] {
+			t.Errorf("%s: no id bound: statistics were not collected", p)
+		}
+		if got == nil {
+			t.Fatalf("%s: no geometry bound", p)
+		}
+		if *got != want {
+			t.Errorf("%s: geometry bound %s, want the union of every geometry %s", p, got, want)
+		}
+	}
+}
+
+// A sorted rewrite may write its output without row-group geospatial
+// statistics; the file then gets no geometry bound, never one that misses
+// a row.
+func TestCompactSortedGeometryBoundCoversEveryRow(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+	so, err := table.NewSortOrder(1, []table.SortField{{SourceIDs: []int{geoTestIDField}, Transform: iceberg.IdentityTransform{}, Direction: table.SortDESC, NullOrder: table.NullsFirst}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	files, want := mixedGeoFiles()
+	populateGeoTable(t, fs, setup, files, geoTableOptions{sortOrder: &so})
+	compactGeo(t, client, setup, Config{RewriteStrategy: "sort", SortBufferRows: minSortBufferRows})
+	geo, _ := compactedGeoBounds(t, client, setup)
+	for p, got := range geo {
+		if got != nil && !got.covers(want) {
+			t.Errorf("%s: geometry bound %s does not cover every geometry %s", p, got, want)
+		}
+	}
+}
+
+func TestCompactAllNullGeometryGetsNoBound(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+	populateGeoTable(t, fs, setup, []geoFile{
+		{Name: "a.parquet", Rows: []geoRow{{1, nil}, {2, nil}}},
+		{Name: "b.parquet", Rows: []geoRow{{3, nil}}},
+	}, geoTableOptions{})
+	compactGeo(t, client, setup, Config{})
+	geo, idBounded := compactedGeoBounds(t, client, setup)
+	for p, got := range geo {
+		if got != nil {
+			t.Errorf("%s: geometry bound %s on a column of nulls", p, got)
+		}
+		if !idBounded[p] {
+			t.Errorf("%s: no id bound: statistics were not collected", p)
+		}
+	}
+}
+
+// A value that does not parse as WKB leaves its row group without
+// geospatial statistics; the file then gets no geometry bound rather than
+// one from the other row groups that does not cover that row group.
+func TestCompactUnparseableGeometryGivesNoBound(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+	populateGeoTable(t, fs, setup, []geoFile{
+		{Name: "ok.parquet", Rows: []geoRow{{1, wkbPoint(10, 10)}, {2, wkbPoint(11, 11)}}},
+		{Name: "bad.parquet", Rows: []geoRow{{3, []byte("not wkb at all")}, {4, wkbPoint(-50, -50)}}},
+	}, geoTableOptions{})
+	// two rows a row group: the valid file's rows, then the other file's
+	compactGeo(t, client, setup, Config{RowGroupRowLimit: 2})
+	geo, idBounded := compactedGeoBounds(t, client, setup)
+	for p, got := range geo {
+		if got != nil {
+			t.Errorf("%s: geometry bound %s although a value did not parse", p, got)
+		}
+		if !idBounded[p] {
+			t.Errorf("%s: no id bound: statistics were not collected", p)
+		}
+	}
+}
+
+func TestCollectColumnStatsGivesGeographyNoBound(t *testing.T) {
+	data := writeGeoParquet(t, []geoRow{{1, wkbPoint(170, 10)}, {2, wkbPoint(-170, 12)}}, "")
+	geog, err := iceberg.GeographyTypeOf(iceberg.DefaultGeoCRS, "spherical")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, tc := range []struct {
+		typ       iceberg.Type
+		wantBound bool
+	}{{iceberg.GeometryType{}, true}, {geog, false}} {
+		stats, err := collectColumnStats(data, geoTestSchema(tc.typ))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, has := stats.lower[geoTestGeomField]; has != tc.wantBound {
+			t.Errorf("%s: lower bound present %v, want %v", tc.typ, has, tc.wantBound)
+		}
+		if stats.values[geoTestGeomField] != 2 {
+			t.Errorf("%s: value count %d, want 2", tc.typ, stats.values[geoTestGeomField])
+		}
+	}
+}
+
+func TestGeometryBoundsAggregation(t *testing.T) {
+	stats := func(xmin, ymin, xmax, ymax float64) format.GeospatialStatistics {
+		return format.GeospatialStatistics{BBox: format.BoundingBox{XMin: xmin, XMax: xmax, YMin: ymin, YMax: ymax}, GeoSpatialTypes: []int32{1}}
+	}
+	type chunk struct {
+		gs            format.GeospatialStatistics
+		values, nulls int64
+	}
+	for _, tc := range []struct {
+		name   string
+		chunks []chunk
+		want   *geoBox
+	}{
+		{"one", []chunk{{stats(1, 2, 3, 4), 5, 0}}, &geoBox{1, 2, 3, 4}},
+		{"union", []chunk{{stats(1, 2, 3, 4), 5, 0}, {stats(-1, 3, 2, 9), 5, 1}}, &geoBox{-1, 2, 3, 9}},
+		{"nulls only constrain nothing", []chunk{{stats(1, 2, 3, 4), 5, 0}, {format.GeospatialStatistics{}, 4, 4}}, &geoBox{1, 2, 3, 4}},
+		{"values without statistics", []chunk{{stats(1, 2, 3, 4), 5, 0}, {format.GeospatialStatistics{}, 4, 1}}, nil},
+		{"NaN", []chunk{{stats(1, math.NaN(), 3, 4), 5, 0}}, nil},
+		{"infinity", []chunk{{stats(1, 2, math.Inf(1), 4), 5, 0}}, nil},
+		{"inverted", []chunk{{stats(3, 2, 1, 4), 5, 0}}, nil},
+		{"nothing", nil, nil},
+	} {
+		var g geometryBounds
+		for _, c := range tc.chunks {
+			g.add(c.gs, c.values, c.nulls)
+		}
+		lo, hi, ok := g.encoded()
+		if tc.want == nil {
+			if ok {
+				t.Errorf("%s: bound given, want none", tc.name)
+			}
+			continue
+		}
+		if !ok {
+			t.Errorf("%s: no bound, want %s", tc.name, tc.want)
+			continue
+		}
+		x0, y0 := decodeXYBound(t, lo)
+		x1, y1 := decodeXYBound(t, hi)
+		if got := (geoBox{x0, y0, x1, y1}); got != *tc.want {
+			t.Errorf("%s: bound %s, want %s", tc.name, got, tc.want)
+		}
+	}
+}