✨ feat(api): Add WKB bounding box scanner

- Accumulate XY bounds across all OGC WKB geometry types\n- Validate byte order, dimensions, counts, truncation, and nesting limits

Generated-by: Codex

Author: Xin Huang

api/src/main/java/org/apache/iceberg/geospatial/WKBBoundingBox.java

@@ -0,0 +1,222 @@
+/*
+ * 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).
+ *
+ * <p>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. A {@code NaN} value is skipped independently in its coordinate dimension,
+ * matching the spec rule that null or NaN values in a dimension do not contribute to bounds (an
+ * empty geometry such as {@code POINT EMPTY} therefore contributes nothing).
+ *
+ * <p>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.
+   *
+   * <p>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.
+   *
+   * <p>NaN values are ignored independently for each dimension. An accumulator reports {@link
+   * #hasBounds()} as {@code true} only after both dimensions have seen a non-NaN value.
+   */
+  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 hasXBounds = false;
+    private boolean hasYBounds = false;
+
+    /**
+     * Folds a single coordinate into the box, ignoring {@code NaN} values independently in each
+     * dimension.
+     *
+     * @param xCoord the X coordinate
+     * @param yCoord the Y coordinate
+     */
+    public void addXY(double xCoord, double yCoord) {
+      if (!Double.isNaN(xCoord)) {
+        minX = Math.min(minX, xCoord);
+        maxX = Math.max(maxX, xCoord);
+        hasXBounds = true;
+      }
+
+      if (!Double.isNaN(yCoord)) {
+        minY = Math.min(minY, yCoord);
+        maxY = Math.max(maxY, yCoord);
+        hasYBounds = true;
+      }
+    }
+
+    /** Returns whether both dimensions have accumulated a non-NaN value. */
+    public boolean hasBounds() {
+      return hasXBounds && hasYBounds;
+    }
+
+    /**
+     * Returns the lower corner (minimum X and Y) of the box, or {@code null} if either dimension
+     * has no accumulated value.
+     */
+    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 either dimension
+     * has no accumulated value.
+     */
+    public GeospatialBound maxBound() {
+      return hasBounds() ? GeospatialBound.createXY(maxX, maxY) : null;
+    }
+  }
+}

api/src/test/java/org/apache/iceberg/geospatial/TestWKBBoundingBox.java

@@ -0,0 +1,133 @@
+/*
+ * 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 java.nio.ByteBuffer;
+import java.util.stream.Stream;
+import org.apache.iceberg.geospatial.WKBBoundingBox.XYAccumulator;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class TestWKBBoundingBox {
+
+  @ParameterizedTest(name = "{0}")
+  @MethodSource("boundingBoxCases")
+  public void testBoundingBox(
+      String wkt, String hexWkb, GeospatialBound expectedMin, GeospatialBound expectedMax) {
+    XYAccumulator accumulator = new XYAccumulator();
+    WKBBoundingBox.accumulate(decode(hexWkb), accumulator);
+
+    assertThat(accumulator.minBound()).as(wkt).isEqualTo(expectedMin);
+    assertThat(accumulator.maxBound()).as(wkt).isEqualTo(expectedMax);
+  }
+
+  private static Stream<Arguments> boundingBoxCases() {
+    return Stream.of(
+        Arguments.of("POINT EMPTY", "0101000000000000000000f87f000000000000f87f", null, null),
+        Arguments.of(
+            "POINT(1 2)", "0101000000000000000000f03f0000000000000040", bound(1, 2), bound(1, 2)),
+        Arguments.of(
+            "POINT(1 2) big endian",
+            "00000000013ff00000000000004000000000000000",
+            bound(1, 2),
+            bound(1, 2)),
+        Arguments.of(
+            "POINT Z (1 2 3)",
+            "01e9030000000000000000f03f00000000000000400000000000000840",
+            bound(1, 2),
+            bound(1, 2)),
+        Arguments.of(
+            "POINT M (1 2 3)",
+            "01d1070000000000000000f03f00000000000000400000000000000840",
+            bound(1, 2),
+            bound(1, 2)),
+        Arguments.of(
+            "POINT ZM (1 2 3 4)",
+            "01b90b0000000000000000f03f000000000000004000000000000008400000000000001040",
+            bound(1, 2),
+            bound(1, 2)),
+        Arguments.of(
+            "LINESTRING(0 1,1 0,2 -1,-1 -2,0 1)",
+            "0102000000050000000000000000000000000000000000f03f000000000000f03f"
+                + "00000000000000000000000000000040000000000000f0bf000000000000f0bf"
+                + "00000000000000c00000000000000000000000000000f03f",
+            bound(-1, -2),
+            bound(2, 1)),
+        Arguments.of(
+            "POLYGON((0 0,10 0,0 10,0 0),(1 1,1 2,2 1,1 1))",
+            "010300000002000000040000000000000000000000000000000000000000000000"
+                + "000024400000000000000000000000000000000000000000000024400000000000"
+                + "000000000000000000000004000000000000000000f03f000000000000f03f0000"
+                + "00000000f03f00000000000000400000000000000040000000000000f03f000000"
+                + "000000f03f000000000000f03f",
+            bound(0, 0),
+            bound(10, 10)),
+        Arguments.of(
+            "MULTIPOINT((1 2),EMPTY,EMPTY,(3 4))",
+            "0104000000040000000101000000000000000000f03f000000000000004001010000"
+                + "00000000000000f87f000000000000f87f0101000000000000000000f87f00000000"
+                + "0000f87f010100000000000000000008400000000000001040",
+            bound(1, 2),
+            bound(3, 4)),
+        Arguments.of(
+            "MULTILINESTRING((1 2,3 4),(5 6,7 8))",
+            "010500000002000000010200000002000000000000000000f03f0000000000000040"
+                + "0000000000000840000000000000104001020000000200000000000000000014400000"
+                + "0000000018400000000000001c400000000000002040",
+            bound(1, 2),
+            bound(7, 8)),
+        Arguments.of(
+            "MULTIPOLYGON(EMPTY,((0 0,10 0,0 10,0 0),(1 1,1 2,2 1,1 1)))",
+            "01060000000200000001030000000000000001030000000200000004000000000000000000000000000"
+                + "00000000000000000000000244000000000000000000000000000000000000000000000244000000000"
+                + "00000000000000000000000004000000000000000000f03f000000000000f03f000000000000f03f000"
+                + "00000000000400000000000000040000000000000f03f000000000000f03f000000000000f03f",
+            bound(0, 0),
+            bound(10, 10)),
+        Arguments.of(
+            "GEOMETRYCOLLECTION(POINT(1 2),LINESTRING EMPTY,POLYGON EMPTY,"
+                + "MULTIPOINT EMPTY,MULTILINESTRING EMPTY,MULTIPOLYGON EMPTY,"
+                + "GEOMETRYCOLLECTION(POINT EMPTY,LINESTRING EMPTY,POLYGON EMPTY,"
+                + "MULTIPOINT EMPTY,MULTILINESTRING EMPTY,MULTIPOLYGON EMPTY))",
+            "0107000000070000000101000000000000000000f03f000000000000004001020000"
+                + "00000000000103000000000000000104000000000000000105000000000000000106"
+                + "000000000000000107000000060000000101000000000000000000f87f0000000000"
+                + "00f87f01020000000000000001030000000000000001040000000000000001050000"
+                + "0000000000010600000000000000",
+            bound(1, 2),
+            bound(1, 2)));
+  }
+
+  private static GeospatialBound bound(double x, double y) {
+    return GeospatialBound.createXY(x, y);
+  }
+
+  private static ByteBuffer decode(String hex) {
+    byte[] bytes = new byte[hex.length() / 2];
+    for (int i = 0; i < bytes.length; i += 1) {
+      int offset = i * 2;
+      bytes[i] = (byte) Integer.parseInt(hex.substring(offset, offset + 2), 16);
+    }
+
+    return ByteBuffer.wrap(bytes);
+  }
+}