zeroshade commented on code in PR #2517:
URL: https://github.com/apache/iceberg-go/pull/2517#discussion_r4541129667
##########
table/rewrite_data_files.go:
##########
@@ -363,6 +364,43 @@ func WithCompactionSortOrder(order SortOrder)
CompactionGroupOption {
}
}
+type hilbertSpec struct {
+ column string
+ opts []HilbertOption
+}
+
+// WithCompactionHilbertOrder switches the group rewrite from bin-pack to
+// spatial clustering: the rows of the group are ordered by the Hilbert
+// curve index of a top-level geometry or geography column, so
+// rows close in space land tend to more often share the same file and row
group.
+//
+// Each geometry is represented by the center of its planar XY bounding
+// box and snapped to a grid of 2^bits cells per axis ([WithHilbertBits])
+// over the extent of the data ([WithHilbertBounds] fixes the extent).
+// Geometries are decoded as the writer's statistics decode them: ISO
+// WKB and EWKB (with or without an SRID) are accepted, nested geometry
+// collections are walked, and empty members are ignored. Rows with a
+// null or empty geometry, or one made only of empty members, have no
+// location and go last. Coordinates are treated as planar x/y, so
+// geography across the antimeridian is not wrapped.
+//
+// Rows are split by the table's current partition spec first and each
+// partition is ordered on its own. The whole group is held in memory
+// decoded until its files are written, so plan groups with
+// [compaction.Config.MaxGroupSizeBytes] large enough to cover a
+// partition and small enough to fit in memory. Rows are ordered only
+// within a group.
+//
+// Hilbert order is not an Iceberg sort order: output files record no
+// sort_order_id. The option cannot be combined with
+// [WithCompactionSortOrder]; [ExecuteCompactionGroup] rejects that, and
+// a column that is not a geometry or geography column.
+func WithCompactionHilbertOrder(column string, opts ...HilbertOption)
CompactionGroupOption {
Review Comment:
Java's Hilbert strategy (apache/iceberg#16827) works differently. It's a
general multi-column sort, `sort_order => 'hilbert(c1, c2)'`, over primitive
columns encoded with `ZOrderByteUtils`, and it doesn't support geometry
columns. This option orders one geometry/geography column by its bounding-box
centre, which is a different feature. If it takes the generic
`WithCompactionHilbertOrder` name, a Java-compatible multi-column option (which
#2043 asks for) has nowhere to go.
Could this get a geometry-specific name, such as
`WithCompactionSpatialOrder` or `WithCompactionGeometryHilbertOrder`? The
alternative is a signature that can grow to several columns later, with
geometry columns handled as a special case.
##########
table/sort_compaction.go:
##########
@@ -542,25 +542,56 @@ func yieldSortedPartition(
return true
}
+ total, err := checkCompactionBatches(batches)
+ if err != nil {
+ yield(nil, err)
+
+ return false
+ }
+
+ refs := naturalRowRefs(batches, total)
+ slices.SortStableFunc(refs, rows.bind(batches))
+
+ return yieldOrderedRows(ctx, batches, refs, batchRows, yield)
+}
+
+// checkCompactionBatches verifies batches share one schema and returns
+// their total row count.
+func checkCompactionBatches(batches []arrow.RecordBatch) (int64, error) {
sc := batches[0].Schema()
var total int64
for _, b := range batches {
if !b.Schema().Equal(sc) {
- yield(nil, errors.New("sort compaction: scan produced
batches with differing schemas"))
-
- return false
+ return 0, errors.New("sort compaction: scan produced
batches with differing schemas")
Review Comment:
`checkCompactionBatches` and `yieldOrderedRows` are now shared with the
Hilbert strategy, but their errors still begin `sort compaction:` (here and in
`gather sorted rows` below). A Hilbert rewrite that hits them reports a sort
error. Please use a strategy-neutral `compaction:` prefix, or let the caller
add its own prefix.
##########
table/rewrite_data_files.go:
##########
@@ -363,6 +364,43 @@ func WithCompactionSortOrder(order SortOrder)
CompactionGroupOption {
}
}
+type hilbertSpec struct {
+ column string
+ opts []HilbertOption
+}
+
+// WithCompactionHilbertOrder switches the group rewrite from bin-pack to
+// spatial clustering: the rows of the group are ordered by the Hilbert
+// curve index of a top-level geometry or geography column, so
+// rows close in space land tend to more often share the same file and row
group.
Review Comment:
Typo: "rows close in space land tend to more often share" should read "rows
close in space tend to share".
##########
table/hilbert_compaction_internal_test.go:
##########
@@ -0,0 +1,710 @@
+// 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 table
+
+import (
+ "context"
+ "encoding/binary"
+ "math"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "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/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "github.com/twpayne/go-geom"
+ "github.com/twpayne/go-geom/encoding/ewkb"
+ "github.com/twpayne/go-geom/encoding/wkb"
+ "github.com/twpayne/go-geom/encoding/wkbcommon"
+ "github.com/twpayne/go-geom/encoding/wkt"
+)
+
+func mustHilbertGrid(t *testing.T, bits int) hilbertGrid {
+ t.Helper()
+
+ g, err := newHilbertGrid(bits)
+ require.NoError(t, err)
+
+ return g
+}
+
+func hilbertTestWKB(t *testing.T, s string) []byte {
+ t.Helper()
+
+ g, err := wkt.Unmarshal(s)
+ require.NoError(t, err)
+ // ISO WKB has no empty-point encoding; the convention, which go-geom
+ // follows with this option, is a point of NaN coordinates. Without it
+ // go-geom refuses to encode an empty point, even inside a collection.
+ b, err := wkb.Marshal(g, wkb.NDR,
wkbcommon.WKBOptionEmptyPointHandling(wkbcommon.EmptyPointHandlingNaN))
+ require.NoError(t, err)
+
+ return b
+}
+
+// hilbertTestBatch builds a one-column batch of binary values; a nil
+// value is a null row.
+func hilbertTestBatch(t *testing.T, dt arrow.BinaryDataType, values [][]byte)
arrow.RecordBatch {
+ t.Helper()
+
+ bldr := array.NewBinaryBuilder(memory.DefaultAllocator, dt)
+ defer bldr.Release()
+ for _, v := range values {
+ if v == nil {
+ bldr.AppendNull()
+ } else {
+ bldr.Append(v)
+ }
+ }
+ col := bldr.NewArray()
+ defer col.Release()
+
+ sc := arrow.NewSchema([]arrow.Field{{Name: "geom", Type: dt, Nullable:
true}}, nil)
+
+ return array.NewRecordBatch(sc, []arrow.Array{col}, int64(len(values)))
+}
+
+func TestNewHilbertGrid(t *testing.T) {
+ for _, bits := range []int{0, -1, 33, 100} {
+ _, err := newHilbertGrid(bits)
+ require.ErrorIs(t, err, ErrInvalidOperation, "bits=%d", bits)
+ }
+ for bits, want := range map[int]uint64{1: 2, 4: 16, 16: 65536, 32: 1 <<
32} {
+ g, err := newHilbertGrid(bits)
+ require.NoError(t, err)
+ assert.Equal(t, bits, g.bits)
+ assert.Equal(t, want, g.cellsPerAxis)
+ }
+}
+
+func TestHilbertGridCurveDistanceForCellKnownOrder(t *testing.T) {
+ // The first-order curve visits (0,0), (0,1), (1,1), (1,0).
+ assert.Equal(t, uint64(0), mustHilbertGrid(t,
1).curveDistanceForCell(0, 0))
+ assert.Equal(t, uint64(1), mustHilbertGrid(t,
1).curveDistanceForCell(0, 1))
+ assert.Equal(t, uint64(2), mustHilbertGrid(t,
1).curveDistanceForCell(1, 1))
+ assert.Equal(t, uint64(3), mustHilbertGrid(t,
1).curveDistanceForCell(1, 0))
+}
+
+func TestHilbertGridCurveDistanceForCellIsContinuousBijection(t *testing.T) {
+ for _, bits := range []int{2, 3, 5} {
+ grid := mustHilbertGrid(t, bits)
+ side := grid.cellsPerAxis
+ cells := make([][2]uint64, side*side)
+ seen := make([]bool, side*side)
+ for x := range side {
+ for y := range side {
+ d := grid.curveDistanceForCell(x, y)
+ require.Less(t, d, side*side)
+ require.False(t, seen[d], "bits=%d: index %d
assigned twice", bits, d)
+ seen[d] = true
+ cells[d] = [2]uint64{x, y}
+ }
+ }
+ for d := 1; d < len(cells); d++ {
+ dx := int64(cells[d][0]) - int64(cells[d-1][0])
+ dy := int64(cells[d][1]) - int64(cells[d-1][1])
+ require.Equal(t, int64(1), dx*dx+dy*dy, "bits=%d: steps
%d and %d not adjacent", bits, d-1, d)
+ }
+ }
+}
+
+func TestHilbertGridCurveDistanceForCellMaxBits(t *testing.T) {
+ // The last cell of the full 32-bit grid must not overflow.
+ assert.Equal(t, uint64(math.MaxUint32)<<32|uint64(math.MaxUint32),
mustHilbertGrid(t, 32).curveDistanceForCell(math.MaxUint32, 0))
+}
+
+func TestHilbertGridCellForCoord(t *testing.T) {
+ assert.Equal(t, uint64(0), mustHilbertGrid(t, 4).cellForCoord(-5, 0,
10))
+ assert.Equal(t, uint64(15), mustHilbertGrid(t, 4).cellForCoord(50, 0,
10))
+ assert.Equal(t, uint64(0), mustHilbertGrid(t, 4).cellForCoord(0, 0, 10))
+ assert.Equal(t, uint64(15), mustHilbertGrid(t, 4).cellForCoord(10, 0,
10))
+ assert.Equal(t, uint64(0), mustHilbertGrid(t, 4).cellForCoord(3, 3, 3),
"degenerate range")
+ assert.Equal(t, uint64(math.MaxUint32), mustHilbertGrid(t,
32).cellForCoord(1, 0, 1))
+}
+
+func TestHilbertGridCurveDistanceForCellKeepsNeighborsClose(t *testing.T) {
+ // Cells of one quadrant occupy one contiguous quarter of the curve,
+ // which is what makes a sorted file spatially compact.
+ grid := mustHilbertGrid(t, 4)
+ side := grid.cellsPerAxis
+ half := side / 2
+ for qx := range uint64(2) {
+ for qy := range uint64(2) {
+ lo, hi := uint64(math.MaxUint64), uint64(0)
+ for x := qx * half; x < (qx+1)*half; x++ {
+ for y := qy * half; y < (qy+1)*half; y++ {
+ d := grid.curveDistanceForCell(x, y)
+ lo, hi = min(lo, d), max(hi, d)
+ }
+ }
+ assert.Equal(t, half*half-1, hi-lo, "quadrant (%d,%d)
is not a contiguous run", qx, qy)
+ assert.Zero(t, lo%(half*half), "quadrant (%d,%d) does
not start on a quarter boundary", qx, qy)
+ }
+ }
+}
+
+func TestHilbertGridCellForCoordEdgeCases(t *testing.T) {
+ assert.Equal(t, uint64(0), mustHilbertGrid(t,
4).cellForCoord(math.NaN(), 0, 10), "NaN clamps to the first cell")
+ assert.Equal(t, uint64(15), mustHilbertGrid(t,
4).cellForCoord(math.Inf(1), 0, 10))
+ assert.Equal(t, uint64(0), mustHilbertGrid(t,
4).cellForCoord(math.Inf(-1), 0, 10))
+ assert.Equal(t, uint64(0), mustHilbertGrid(t, 4).cellForCoord(5, 10,
0), "inverted range is degenerate")
+ assert.Equal(t, uint64(0), mustHilbertGrid(t, 1).cellForCoord(4.9, 0,
10))
+ assert.Equal(t, uint64(1), mustHilbertGrid(t, 1).cellForCoord(5, 0,
10), "midpoint starts the upper half")
+ assert.Equal(t, uint64(8), mustHilbertGrid(t, 4).cellForCoord(5, 0,
10), "midpoint of a 4-bit grid")
+ assert.Equal(t, uint64(14), mustHilbertGrid(t, 4).cellForCoord(9.3, 0,
10), "the last cell is a full-width bucket, not just hi")
+ assert.Equal(t, uint64(15), mustHilbertGrid(t, 4).cellForCoord(9.4, 0,
10))
+ // Monotonic: a larger value never lands in an earlier cell.
+ grid := mustHilbertGrid(t, 6)
+ prev := uint64(0)
+ for i := range 101 {
+ cell := grid.cellForCoord(float64(i), 0, 100)
+ assert.GreaterOrEqual(t, cell, prev)
+ prev = cell
+ }
+}
+
+func TestWKBCenter(t *testing.T) {
Review Comment:
`TestWKBCenter`, `TestWKBCenterExtremeAndNonFinite`,
`TestWKBCenterByteOrder` and `TestWKBCenterAcceptsEWKB` repeat the
`TestWKBCenterXY*` cases in `table/internal/geo_center_test.go`, through
`wkbCenter`, which is a thin wrapper. Please keep only the cases specific to
the wrapper (the nil and zero-length errors and the NaN mapping) and drop the
rest.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
