From 4a1d8e80924c752c15357bf4396f03f99d7c4f00 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Fri, 10 Jul 2026 21:09:32 -0700 Subject: [PATCH] Parquet: Compute geometry bounding box metrics The Parquet footer's lexicographic min/max over WKB bytes is not meaningful for geometry, so geometry columns previously wrote counts only and could not be data skipped. Scan each value's coordinates as it is written and accumulate a 2D (XY) bounding box, emitting it as a FieldMetrics through the writer-side metrics channel (the same path float and double use for NaN counts). The box's lower and upper corners serialize into the existing lower_bounds and upper_bounds maps via the geometry conversion. A new pure-Java WKB coordinate scanner (WKBBoundingBox, in api/geospatial, no JTS dependency) walks all OGC geometry types, skips Z and M, ignores NaN coordinates, and validates the buffer defensively. Geography, higher dimensions, and the content_stats geo struct bounds are left as follow-ups. --- .../iceberg/geospatial/WKBBoundingBox.java | 217 +++++++++ .../geospatial/TestWKBBoundingBox.java | 429 ++++++++++++++++++ .../apache/iceberg/GeometryFieldMetrics.java | 64 +++ .../iceberg/TestGeometryFieldMetrics.java | 95 ++++ .../java/org/apache/iceberg/TestMetrics.java | 18 +- .../data/parquet/BaseParquetWriter.java | 7 +- .../iceberg/parquet/ParquetValueWriters.java | 30 ++ 7 files changed, 853 insertions(+), 7 deletions(-) create mode 100644 api/src/main/java/org/apache/iceberg/geospatial/WKBBoundingBox.java create mode 100644 api/src/test/java/org/apache/iceberg/geospatial/TestWKBBoundingBox.java create mode 100644 core/src/main/java/org/apache/iceberg/GeometryFieldMetrics.java create mode 100644 core/src/test/java/org/apache/iceberg/TestGeometryFieldMetrics.java diff --git a/api/src/main/java/org/apache/iceberg/geospatial/WKBBoundingBox.java b/api/src/main/java/org/apache/iceberg/geospatial/WKBBoundingBox.java new file mode 100644 index 000000000000..46497a71a6f5 --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/geospatial/WKBBoundingBox.java @@ -0,0 +1,217 @@ +/* + * 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 org.apache.iceberg.geospatial; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Computes a two-dimensional (XY) bounding box from geometry values encoded as Well-Known Binary + * (WKB). + * + *

The bounding box is the minimum axis-aligned rectangle that contains every coordinate of the + * geometries. Only the X and Y dimensions are considered; any Z or M values present in the WKB are + * read past and ignored. Coordinates whose X or Y is {@code NaN} are skipped, matching the spec + * rule that null or NaN values in a coordinate dimension do not contribute to bounds (an empty + * geometry such as {@code POINT EMPTY} therefore contributes nothing). + * + *

Parsing follows the OGC Simple Feature Access WKB layout and validates the input defensively: + * a malformed or truncated buffer results in an {@link IllegalArgumentException} rather than an + * out-of-bounds read. + */ +public class WKBBoundingBox { + + // OGC WKB base geometry type codes (after stripping the dimension offset). + private static final int TYPE_POINT = 1; + private static final int TYPE_LINE_STRING = 2; + private static final int TYPE_POLYGON = 3; + private static final int TYPE_MULTI_POINT = 4; + private static final int TYPE_MULTI_LINE_STRING = 5; + private static final int TYPE_MULTI_POLYGON = 6; + private static final int TYPE_GEOMETRY_COLLECTION = 7; + + // Bounds the recursion depth for nested collections to reject pathological or malicious input. + private static final int MAX_DEPTH = 100; + + private WKBBoundingBox() {} + + /** + * Parses one WKB geometry and folds all of its XY coordinates into the given accumulator. + * + *

The buffer is read via a duplicate, so the caller's position and limit are left unchanged. + * + * @param wkb a buffer containing a single WKB geometry + * @param accumulator the accumulator to update with the geometry's coordinates + * @throws IllegalArgumentException if the buffer is not a well-formed WKB geometry + */ + public static void accumulate(ByteBuffer wkb, XYAccumulator accumulator) { + Preconditions.checkArgument(wkb != null, "Invalid WKB buffer: null"); + Preconditions.checkArgument(accumulator != null, "Invalid accumulator: null"); + ByteBuffer buffer = wkb.duplicate(); + parseGeometry(buffer, accumulator, 0); + } + + private static void parseGeometry(ByteBuffer buffer, XYAccumulator accumulator, int depth) { + Preconditions.checkArgument(depth <= MAX_DEPTH, "Invalid WKB: nesting too deep"); + checkRemaining(buffer, 5); + + byte order = buffer.get(); + if (order == 0) { + buffer.order(ByteOrder.BIG_ENDIAN); + } else if (order == 1) { + buffer.order(ByteOrder.LITTLE_ENDIAN); + } else { + throw new IllegalArgumentException("Invalid WKB byte order: " + order); + } + + long typeCode = buffer.getInt() & 0xFFFFFFFFL; + int geometryType = (int) (typeCode % 1000); + int dimensionGroup = (int) (typeCode / 1000); + int numDimensions = numDimensions(dimensionGroup, typeCode); + + switch (geometryType) { + case TYPE_POINT: + readCoordinate(buffer, accumulator, numDimensions); + break; + case TYPE_LINE_STRING: + readCoordinateSequence(buffer, accumulator, numDimensions); + break; + case TYPE_POLYGON: + int numRings = readCount(buffer); + for (int i = 0; i < numRings; i += 1) { + readCoordinateSequence(buffer, accumulator, numDimensions); + } + break; + case TYPE_MULTI_POINT: + case TYPE_MULTI_LINE_STRING: + case TYPE_MULTI_POLYGON: + case TYPE_GEOMETRY_COLLECTION: + int numElements = readCount(buffer); + for (int i = 0; i < numElements; i += 1) { + // Each child carries its own byte-order and type header. + parseGeometry(buffer, accumulator, depth + 1); + } + break; + default: + throw new IllegalArgumentException("Invalid or unsupported WKB geometry type: " + typeCode); + } + } + + private static int numDimensions(int dimensionGroup, long typeCode) { + switch (dimensionGroup) { + case 0: // XY + return 2; + case 1: // XYZ + case 2: // XYM + return 3; + case 3: // XYZM + return 4; + default: + throw new IllegalArgumentException("Invalid or unsupported WKB geometry type: " + typeCode); + } + } + + private static void readCoordinateSequence( + ByteBuffer buffer, XYAccumulator accumulator, int numDimensions) { + int numPoints = readCount(buffer); + // Validate the full extent up front so a bogus point count cannot drive a long read loop, but + // never pre-allocate from the count itself. + checkRemaining(buffer, (long) numPoints * numDimensions * Double.BYTES); + for (int i = 0; i < numPoints; i += 1) { + readCoordinate(buffer, accumulator, numDimensions); + } + } + + private static void readCoordinate( + ByteBuffer buffer, XYAccumulator accumulator, int numDimensions) { + checkRemaining(buffer, (long) numDimensions * Double.BYTES); + double xCoord = buffer.getDouble(); + double yCoord = buffer.getDouble(); + // Skip any Z and/or M values; only X and Y contribute to the box. + for (int i = 2; i < numDimensions; i += 1) { + buffer.getDouble(); + } + accumulator.addXY(xCoord, yCoord); + } + + private static int readCount(ByteBuffer buffer) { + checkRemaining(buffer, Integer.BYTES); + long count = buffer.getInt() & 0xFFFFFFFFL; + Preconditions.checkArgument(count <= Integer.MAX_VALUE, "Invalid WKB element count: %s", count); + return (int) count; + } + + private static void checkRemaining(ByteBuffer buffer, long bytes) { + Preconditions.checkArgument( + buffer.remaining() >= bytes, "Invalid WKB: unexpected end of buffer"); + } + + /** + * A mutable accumulator of the minimum and maximum X and Y coordinates seen so far. + * + *

Coordinates whose X or Y is {@code NaN} are ignored. An accumulator that has seen no non-NaN + * coordinate reports {@link #hasBounds()} as {@code false} and returns {@code null} bounds. + */ + public static class XYAccumulator { + private double minX = Double.POSITIVE_INFINITY; + private double minY = Double.POSITIVE_INFINITY; + private double maxX = Double.NEGATIVE_INFINITY; + private double maxY = Double.NEGATIVE_INFINITY; + private boolean hasBounds = false; + + /** + * Folds a single coordinate into the box, ignoring it if either dimension is {@code NaN}. + * + * @param xCoord the X coordinate + * @param yCoord the Y coordinate + */ + public void addXY(double xCoord, double yCoord) { + if (Double.isNaN(xCoord) || Double.isNaN(yCoord)) { + return; + } + minX = Math.min(minX, xCoord); + minY = Math.min(minY, yCoord); + maxX = Math.max(maxX, xCoord); + maxY = Math.max(maxY, yCoord); + hasBounds = true; + } + + /** Returns whether any non-NaN coordinate has been accumulated. */ + public boolean hasBounds() { + return hasBounds; + } + + /** + * Returns the lower corner (minimum X and Y) of the box, or {@code null} if no coordinate has + * been accumulated. + */ + public GeospatialBound minBound() { + return hasBounds ? GeospatialBound.createXY(minX, minY) : null; + } + + /** + * Returns the upper corner (maximum X and Y) of the box, or {@code null} if no coordinate has + * been accumulated. + */ + public GeospatialBound maxBound() { + return hasBounds ? GeospatialBound.createXY(maxX, maxY) : null; + } + } +} diff --git a/api/src/test/java/org/apache/iceberg/geospatial/TestWKBBoundingBox.java b/api/src/test/java/org/apache/iceberg/geospatial/TestWKBBoundingBox.java new file mode 100644 index 000000000000..e423a369f0bf --- /dev/null +++ b/api/src/test/java/org/apache/iceberg/geospatial/TestWKBBoundingBox.java @@ -0,0 +1,429 @@ +/* + * 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 org.apache.iceberg.geospatial; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.ByteArrayOutputStream; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.Arrays; +import org.apache.iceberg.geospatial.WKBBoundingBox.XYAccumulator; +import org.junit.jupiter.api.Test; + +public class TestWKBBoundingBox { + + // OGC WKB geometry type codes; the dimension offset (+1000 Z, +2000 M, +3000 ZM) is added + // separately by the builders below. + private static final int POINT = 1; + private static final int LINE_STRING = 2; + private static final int POLYGON = 3; + private static final int MULTI_POINT = 4; + private static final int MULTI_LINE_STRING = 5; + private static final int MULTI_POLYGON = 6; + private static final int GEOMETRY_COLLECTION = 7; + + @Test + public void testPoint() { + XYAccumulator acc = accumulate(point(ByteOrder.LITTLE_ENDIAN, 30, 10)); + assertThat(acc.hasBounds()).isTrue(); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(30, 10)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(30, 10)); + } + + @Test + public void testPointBigEndian() { + XYAccumulator acc = accumulate(point(ByteOrder.BIG_ENDIAN, -71, 42)); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(-71, 42)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(-71, 42)); + } + + @Test + public void testLineString() { + XYAccumulator acc = + accumulate(lineString(ByteOrder.LITTLE_ENDIAN, new double[][] {{0, 0}, {5, 3}, {2, -4}})); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(0, -4)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(5, 3)); + } + + @Test + public void testPolygon() { + // A single ring; the box spans all ring vertices. + double[][] ring = {{1, 1}, {1, 9}, {8, 9}, {8, 1}, {1, 1}}; + XYAccumulator acc = accumulate(polygon(ByteOrder.LITTLE_ENDIAN, ring)); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(1, 1)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(8, 9)); + } + + @Test + public void testMultiPoint() { + byte[] wkb = + collection( + ByteOrder.LITTLE_ENDIAN, + MULTI_POINT, + point(ByteOrder.LITTLE_ENDIAN, 30, 10), + point(ByteOrder.LITTLE_ENDIAN, -5, 40)); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(-5, 10)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(30, 40)); + } + + @Test + public void testMultiLineString() { + byte[] wkb = + collection( + ByteOrder.LITTLE_ENDIAN, + MULTI_LINE_STRING, + lineString(ByteOrder.LITTLE_ENDIAN, new double[][] {{0, 0}, {1, 1}}), + lineString(ByteOrder.LITTLE_ENDIAN, new double[][] {{-3, 7}, {2, 2}})); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(-3, 0)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(2, 7)); + } + + @Test + public void testMultiPolygon() { + byte[] wkb = + collection( + ByteOrder.LITTLE_ENDIAN, + MULTI_POLYGON, + polygon(ByteOrder.LITTLE_ENDIAN, new double[][] {{0, 0}, {0, 2}, {2, 2}, {0, 0}}), + polygon(ByteOrder.LITTLE_ENDIAN, new double[][] {{5, 5}, {5, 9}, {9, 9}, {5, 5}})); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(0, 0)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(9, 9)); + } + + @Test + public void testGeometryCollection() { + byte[] wkb = + collection( + ByteOrder.LITTLE_ENDIAN, + GEOMETRY_COLLECTION, + point(ByteOrder.LITTLE_ENDIAN, 30, 10), + lineString(ByteOrder.LITTLE_ENDIAN, new double[][] {{-8, 3}, {12, 25}})); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(-8, 3)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(30, 25)); + } + + @Test + public void testMixedEndianChildren() { + // A collection whose children use different byte orders; each child re-reads its own order. + byte[] wkb = + collection( + ByteOrder.BIG_ENDIAN, + MULTI_POINT, + point(ByteOrder.LITTLE_ENDIAN, 30, 10), + point(ByteOrder.BIG_ENDIAN, -5, 40)); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(-5, 10)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(30, 40)); + } + + @Test + public void testNestedGeometryCollection() { + byte[] inner = + collection( + ByteOrder.LITTLE_ENDIAN, GEOMETRY_COLLECTION, point(ByteOrder.LITTLE_ENDIAN, 100, 50)); + byte[] wkb = + collection( + ByteOrder.LITTLE_ENDIAN, + GEOMETRY_COLLECTION, + point(ByteOrder.LITTLE_ENDIAN, 1, 2), + inner); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(1, 2)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(100, 50)); + } + + @Test + public void testXYZSkipsZ() { + // A 3D point (type code 1001). Only X and Y should contribute; Z is read past. + byte[] wkb = pointWithDims(ByteOrder.LITTLE_ENDIAN, 1000, 30, 10, 999.0); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(30, 10)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(30, 10)); + } + + @Test + public void testXYMSkipsM() { + // A measured point (type code 2001). + byte[] wkb = pointWithDims(ByteOrder.LITTLE_ENDIAN, 2000, -5, 40, 7.5); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(-5, 40)); + } + + @Test + public void testXYZMSkipsZAndM() { + // A 4D point (type code 3001). + byte[] wkb = pointWithDims(ByteOrder.LITTLE_ENDIAN, 3000, 12, 34, 5.0, 6.0); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(12, 34)); + } + + @Test + public void testEmptyPointContributesNothing() { + // POINT EMPTY is encoded as a point with NaN coordinates. + XYAccumulator acc = accumulate(point(ByteOrder.LITTLE_ENDIAN, Double.NaN, Double.NaN)); + assertThat(acc.hasBounds()).isFalse(); + assertThat(acc.minBound()).isNull(); + assertThat(acc.maxBound()).isNull(); + } + + @Test + public void testEmptyLineStringContributesNothing() { + XYAccumulator acc = accumulate(lineString(ByteOrder.LITTLE_ENDIAN, new double[0][])); + assertThat(acc.hasBounds()).isFalse(); + } + + @Test + public void testNaNCoordinateSkipped() { + // The valid point still sets the box; the NaN point contributes nothing. + byte[] wkb = + collection( + ByteOrder.LITTLE_ENDIAN, + MULTI_POINT, + point(ByteOrder.LITTLE_ENDIAN, 30, 10), + point(ByteOrder.LITTLE_ENDIAN, Double.NaN, Double.NaN)); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(30, 10)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(30, 10)); + } + + @Test + public void testInfinityWidensBounds() { + // Only NaN is skipped; +/-Infinity is a real value that widens the box (per spec). + byte[] wkb = + collection( + ByteOrder.LITTLE_ENDIAN, + MULTI_POINT, + point(ByteOrder.LITTLE_ENDIAN, 0, 0), + point(ByteOrder.LITTLE_ENDIAN, Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY)); + XYAccumulator acc = accumulate(wkb); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(0, Double.NEGATIVE_INFINITY)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(Double.POSITIVE_INFINITY, 0)); + } + + @Test + public void testCallerBufferPositionUnchanged() { + ByteBuffer buffer = ByteBuffer.wrap(point(ByteOrder.LITTLE_ENDIAN, 30, 10)); + int position = buffer.position(); + int limit = buffer.limit(); + XYAccumulator acc = new XYAccumulator(); + WKBBoundingBox.accumulate(buffer, acc); + assertThat(buffer.position()).isEqualTo(position); + assertThat(buffer.limit()).isEqualTo(limit); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(30, 10)); + } + + @Test + public void testAccumulateAcrossMultipleGeometries() { + XYAccumulator acc = new XYAccumulator(); + WKBBoundingBox.accumulate(ByteBuffer.wrap(point(ByteOrder.LITTLE_ENDIAN, 30, 10)), acc); + WKBBoundingBox.accumulate(ByteBuffer.wrap(point(ByteOrder.LITTLE_ENDIAN, -5, 40)), acc); + assertThat(acc.minBound()).isEqualTo(GeospatialBound.createXY(-5, 10)); + assertThat(acc.maxBound()).isEqualTo(GeospatialBound.createXY(30, 40)); + } + + @Test + public void testRejectsBadByteOrder() { + byte[] wkb = point(ByteOrder.LITTLE_ENDIAN, 30, 10); + wkb[0] = 2; // neither 0 (BE) nor 1 (LE) + assertThatThrownBy(() -> accumulate(wkb)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("byte order"); + } + + @Test + public void testRejectsUnknownGeometryType() { + byte[] wkb = + ByteBuffer.allocate(21) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) + .putInt(99) // not a valid geometry type + .putDouble(0) + .putDouble(0) + .array(); + assertThatThrownBy(() -> accumulate(wkb)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("geometry type"); + } + + @Test + public void testRejectsUnknownDimension() { + // Dimension group 4 (type code 4001) is not a valid OGC dimension offset. + byte[] wkb = + ByteBuffer.allocate(21) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) + .putInt(4001) + .putDouble(0) + .putDouble(0) + .array(); + assertThatThrownBy(() -> accumulate(wkb)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("geometry type"); + } + + @Test + public void testRejectsTruncatedHeader() { + byte[] wkb = Arrays.copyOf(point(ByteOrder.LITTLE_ENDIAN, 30, 10), 3); + assertThatThrownBy(() -> accumulate(wkb)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("end of buffer"); + } + + @Test + public void testRejectsTruncatedCoordinates() { + // A point header that promises two doubles but only supplies one. + byte[] wkb = + ByteBuffer.allocate(13) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) + .putInt(POINT) + .putDouble(30) + .array(); + assertThatThrownBy(() -> accumulate(wkb)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("end of buffer"); + } + + @Test + public void testRejectsHugeElementCount() { + // A multipoint that claims 0xFFFFFFFF elements but supplies none. + byte[] wkb = + ByteBuffer.allocate(9) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) + .putInt(MULTI_POINT) + .putInt(0xFFFFFFFF) // -1 as signed, ~4 billion unsigned + .array(); + assertThatThrownBy(() -> accumulate(wkb)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("WKB"); + } + + @Test + public void testRejectsHugePointCount() { + // A linestring that claims a massive point count but supplies no coordinates. + byte[] wkb = + ByteBuffer.allocate(9) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) + .putInt(LINE_STRING) + .putInt(Integer.MAX_VALUE) + .array(); + assertThatThrownBy(() -> accumulate(wkb)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("end of buffer"); + } + + @Test + public void testRejectsDeeplyNestedCollection() { + // Build 200 nested geometry collections, each containing the next. + byte[] wkb = point(ByteOrder.LITTLE_ENDIAN, 0, 0); + for (int i = 0; i < 200; i += 1) { + wkb = collection(ByteOrder.LITTLE_ENDIAN, GEOMETRY_COLLECTION, wkb); + } + byte[] deep = wkb; + assertThatThrownBy(() -> accumulate(deep)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("nesting too deep"); + } + + private static XYAccumulator accumulate(byte[] wkb) { + XYAccumulator acc = new XYAccumulator(); + WKBBoundingBox.accumulate(ByteBuffer.wrap(wkb), acc); + return acc; + } + + private static byte[] point(ByteOrder order, double xCoord, double yCoord) { + return ByteBuffer.allocate(21) + .order(order) + .put(orderFlag(order)) + .putInt(POINT) + .putDouble(xCoord) + .putDouble(yCoord) + .array(); + } + + private static byte[] pointWithDims( + ByteOrder order, int dimOffset, double xCoord, double yCoord, double... extra) { + ByteBuffer buffer = + ByteBuffer.allocate(5 + (2 + extra.length) * Double.BYTES) + .order(order) + .put(orderFlag(order)) + .putInt(POINT + dimOffset) + .putDouble(xCoord) + .putDouble(yCoord); + for (double value : extra) { + buffer.putDouble(value); + } + return buffer.array(); + } + + private static byte[] lineString(ByteOrder order, double[][] coords) { + ByteBuffer buffer = + ByteBuffer.allocate(5 + Integer.BYTES + coords.length * 2 * Double.BYTES) + .order(order) + .put(orderFlag(order)) + .putInt(LINE_STRING) + .putInt(coords.length); + for (double[] coord : coords) { + buffer.putDouble(coord[0]).putDouble(coord[1]); + } + return buffer.array(); + } + + private static byte[] polygon(ByteOrder order, double[]... ring) { + ByteBuffer buffer = + ByteBuffer.allocate(5 + Integer.BYTES + Integer.BYTES + ring.length * 2 * Double.BYTES) + .order(order) + .put(orderFlag(order)) + .putInt(POLYGON) + .putInt(1) // one ring + .putInt(ring.length); + for (double[] coord : ring) { + buffer.putDouble(coord[0]).putDouble(coord[1]); + } + return buffer.array(); + } + + private static byte[] collection(ByteOrder order, int type, byte[]... children) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] header = + ByteBuffer.allocate(5 + Integer.BYTES) + .order(order) + .put(orderFlag(order)) + .putInt(type) + .putInt(children.length) + .array(); + out.writeBytes(header); + for (byte[] child : children) { + out.writeBytes(child); + } + return out.toByteArray(); + } + + private static byte orderFlag(ByteOrder order) { + return (byte) (order == ByteOrder.LITTLE_ENDIAN ? 1 : 0); + } +} diff --git a/core/src/main/java/org/apache/iceberg/GeometryFieldMetrics.java b/core/src/main/java/org/apache/iceberg/GeometryFieldMetrics.java new file mode 100644 index 000000000000..59c76cc0298e --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/GeometryFieldMetrics.java @@ -0,0 +1,64 @@ +/* + * 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 org.apache.iceberg; + +import java.nio.ByteBuffer; +import org.apache.iceberg.geospatial.GeospatialBound; +import org.apache.iceberg.geospatial.WKBBoundingBox; +import org.apache.iceberg.types.Type; + +/** + * Field-level metrics for a geometry column, tracked by the Parquet writer. + * + *

The Parquet footer's lexicographic min/max over the WKB bytes is not meaningful for spatial + * values, so the writer instead scans each value's coordinates and accumulates a two-dimensional + * bounding box. The resulting {@link FieldMetrics} carries the box's lower and upper corners as + * {@link GeospatialBound} points, which serialize through the existing geometry conversion path. + * The null value count is left at zero here; it is reconciled by the optional-field writer that + * owns null accounting. + */ +public class GeometryFieldMetrics extends FieldMetrics { + + private GeometryFieldMetrics( + int id, long valueCount, GeospatialBound lowerBound, GeospatialBound upperBound, Type type) { + super(id, valueCount, 0L, lowerBound, upperBound, type); + } + + public static class Builder { + private final int id; + private final Type geometryType; + private final WKBBoundingBox.XYAccumulator accumulator = new WKBBoundingBox.XYAccumulator(); + private long valueCount = 0; + + public Builder(int id, Type geometryType) { + this.id = id; + this.geometryType = geometryType; + } + + public void addValue(ByteBuffer wkb) { + this.valueCount += 1; + WKBBoundingBox.accumulate(wkb, accumulator); + } + + public GeometryFieldMetrics build() { + return new GeometryFieldMetrics( + id, valueCount, accumulator.minBound(), accumulator.maxBound(), geometryType); + } + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestGeometryFieldMetrics.java b/core/src/test/java/org/apache/iceberg/TestGeometryFieldMetrics.java new file mode 100644 index 000000000000..6e600b269ca8 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestGeometryFieldMetrics.java @@ -0,0 +1,95 @@ +/* + * 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 org.apache.iceberg; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import org.apache.iceberg.geospatial.GeospatialBound; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; + +public class TestGeometryFieldMetrics { + + private static final Types.GeometryType GEOMETRY = Types.GeometryType.crs84(); + + @Test + public void testBoundsAcrossMultiplePoints() { + GeometryFieldMetrics.Builder builder = new GeometryFieldMetrics.Builder(2, GEOMETRY); + builder.addValue(wkbPoint(30, 10)); + builder.addValue(wkbPoint(-5, 40)); + + FieldMetrics metrics = builder.build(); + + assertThat(metrics.id()).isEqualTo(2); + assertThat(metrics.valueCount()).isEqualTo(2L); + // null accounting is reconciled by the optional-field writer, not here + assertThat(metrics.nullValueCount()).isEqualTo(0L); + // geometry has no NaN metric; -1 suppresses the entry downstream + assertThat(metrics.nanValueCount()).isEqualTo(-1L); + assertThat(metrics.originalType()).isEqualTo(GEOMETRY); + assertThat(metrics.lowerBound()).isEqualTo(GeospatialBound.createXY(-5, 10)); + assertThat(metrics.upperBound()).isEqualTo(GeospatialBound.createXY(30, 40)); + } + + @Test + public void testSinglePointBoundsAreThatPoint() { + GeometryFieldMetrics.Builder builder = new GeometryFieldMetrics.Builder(2, GEOMETRY); + builder.addValue(wkbPoint(12, 34)); + + FieldMetrics metrics = builder.build(); + + assertThat(metrics.valueCount()).isEqualTo(1L); + assertThat(metrics.lowerBound()).isEqualTo(GeospatialBound.createXY(12, 34)); + assertThat(metrics.upperBound()).isEqualTo(GeospatialBound.createXY(12, 34)); + } + + @Test + public void testEmptyGeometryProducesNoBounds() { + // POINT EMPTY (NaN coordinates) still counts as a value but contributes no coordinate. + GeometryFieldMetrics.Builder builder = new GeometryFieldMetrics.Builder(2, GEOMETRY); + builder.addValue(wkbPoint(Double.NaN, Double.NaN)); + + FieldMetrics metrics = builder.build(); + + assertThat(metrics.valueCount()).isEqualTo(1L); + assertThat(metrics.lowerBound()).isNull(); + assertThat(metrics.upperBound()).isNull(); + } + + @Test + public void testNoValuesProducesNoBounds() { + FieldMetrics metrics = new GeometryFieldMetrics.Builder(2, GEOMETRY).build(); + + assertThat(metrics.valueCount()).isEqualTo(0L); + assertThat(metrics.lowerBound()).isNull(); + assertThat(metrics.upperBound()).isNull(); + } + + private static ByteBuffer wkbPoint(double xCoord, double yCoord) { + return ByteBuffer.allocate(21) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) // little-endian + .putInt(1) // WKB geometry type: Point + .putDouble(xCoord) + .putDouble(yCoord) + .flip(); + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestMetrics.java b/core/src/test/java/org/apache/iceberg/TestMetrics.java index 095feb29382f..0bb334a775d8 100644 --- a/core/src/test/java/org/apache/iceberg/TestMetrics.java +++ b/core/src/test/java/org/apache/iceberg/TestMetrics.java @@ -37,6 +37,7 @@ import java.util.Map; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.geospatial.GeospatialBound; import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; @@ -276,15 +277,22 @@ public void testMetricsForGeospatialTypes() throws IOException { first.setField("geog", wkbPoint(-5, 40)); Record second = GenericRecord.create(schema); second.setField("id", 2L); - // both geo columns are left null + second.setField("geom", wkbPoint(-5, 40)); + // geog on the second row is left null Metrics metrics = getMetrics(schema, first, second); assertThat(metrics.recordCount()).isEqualTo(2L); - // geometry and geography keep value/null counts but no bounds: lexicographic WKB min/max is not - // meaningful, so bounds are intentionally skipped (spatial bounds are a separate follow-up). - assertCounts(2, 2L, 1L, metrics); - assertBounds(2, Types.GeometryType.crs84(), null, null, metrics); + // geometry bounds are the XY bounding box of all values: min corner (-5, 10), max corner + // (30, 40). The null geography row leaves geography with value/null counts but no bounds + // (geography bounds are a separate follow-up). + assertCounts(2, 2L, 0L, metrics); + assertBounds( + 2, + Types.GeometryType.crs84(), + GeospatialBound.createXY(-5, 10), + GeospatialBound.createXY(30, 40), + metrics); assertCounts(3, 2L, 1L, metrics); assertBounds(3, Types.GeographyType.crs84(), null, null, metrics); } diff --git a/parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java b/parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java index dcc93f939d8e..3eb3fe5d1f1f 100644 --- a/parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java +++ b/parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java @@ -268,8 +268,11 @@ public Optional> visit( @Override public Optional> visit( LogicalTypeAnnotation.GeometryLogicalTypeAnnotation geometryType) { - // geometry values are pure WKB stored in a BINARY column - return Optional.of(ParquetValueWriters.byteBuffers(desc)); + // geometry values are pure WKB stored in a BINARY column; the writer also scans the + // coordinates to produce a bounding box (the Parquet footer cannot). The concrete CRS is + // immaterial here: ParquetMetrics rewrites the bound's type from the table schema before + // serialization, and the serialization is keyed on the type id, which ignores the CRS. + return Optional.of(ParquetValueWriters.geometry(desc, Types.GeometryType.crs84())); } @Override diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java index 5554c86b4671..89f37dc4d55b 100644 --- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java +++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java @@ -36,6 +36,7 @@ import org.apache.iceberg.DoubleFieldMetrics; import org.apache.iceberg.FieldMetrics; import org.apache.iceberg.FloatFieldMetrics; +import org.apache.iceberg.GeometryFieldMetrics; import org.apache.iceberg.StructLike; import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; @@ -120,6 +121,11 @@ public static PrimitiveWriter byteBuffers(ColumnDescriptor desc) { return new BytesWriter(desc); } + public static PrimitiveWriter geometry( + ColumnDescriptor desc, org.apache.iceberg.types.Type geometryType) { + return new GeometryWriter(desc, geometryType); + } + public static PrimitiveWriter fixedBuffers(ColumnDescriptor desc) { return new FixedBufferWriter(desc); } @@ -336,6 +342,30 @@ public void write(int repetitionLevel, ByteBuffer buffer) { } } + private static class GeometryWriter extends PrimitiveWriter { + private final GeometryFieldMetrics.Builder metricsBuilder; + + private GeometryWriter(ColumnDescriptor desc, org.apache.iceberg.types.Type geometryType) { + super(desc); + this.metricsBuilder = + new GeometryFieldMetrics.Builder( + desc.getPrimitiveType().getId().intValue(), geometryType); + } + + @Override + public void write(int repetitionLevel, ByteBuffer buffer) { + // Accumulate the bounding box before writing, so it reads the buffer's coordinates while the + // position is intact (the scanner reads a duplicate and leaves this buffer untouched). + metricsBuilder.addValue(buffer); + column.writeBinary(repetitionLevel, Binary.fromReusedByteBuffer(buffer)); + } + + @Override + public Stream> metrics() { + return Stream.of(metricsBuilder.build()); + } + } + private static class FixedBufferWriter extends PrimitiveWriter { private final int length;