diff --git a/weed/worker/tasks/iceberg/compact_order.go b/weed/worker/tasks/iceberg/compact_order.go index 56b6176fbb..ffda8acc73 100644 --- a/weed/worker/tasks/iceberg/compact_order.go +++ b/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 { diff --git a/weed/worker/tasks/iceberg/compact_order_test.go b/weed/worker/tasks/iceberg/compact_order_test.go index 8a4b1fdf78..1a36f7618f 100644 --- a/weed/worker/tasks/iceberg/compact_order_test.go +++ b/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) + } +} diff --git a/weed/worker/tasks/iceberg/compact_stats.go b/weed/worker/tasks/iceberg/compact_stats.go index 5ee092eeca..277752c976 100644 --- a/weed/worker/tasks/iceberg/compact_stats.go +++ b/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 } diff --git a/weed/worker/tasks/iceberg/geometry_bounds.go b/weed/worker/tasks/iceberg/geometry_bounds.go new file mode 100644 index 0000000000..34789c7d2d --- /dev/null +++ b/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 +} diff --git a/weed/worker/tasks/iceberg/geometry_bounds_test.go b/weed/worker/tasks/iceberg/geometry_bounds_test.go new file mode 100644 index 0000000000..dea73f4ce8 --- /dev/null +++ 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) + } + } +}