✨ feat(api): Add WKB geometry bounds

- Accumulate two-dimensional bounds across ISO WKB geometry values
- Reject unsupported dimensions and malformed structures

Generated-by: Codex

Author: Xin Huang

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

@@ -0,0 +1,197 @@
+/*
+ * 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;
+
+/**
+ * Accumulates geometry bounds from values encoded as Well-Known Binary (WKB).
+ *
+ * <p>Coordinates are tracked independently for the X and Y dimensions. Null or {@code NaN} values
+ * do not contribute to a dimension, and no bounds are produced unless both dimensions are present.
+ * WKB values with Z or M dimensions are not currently supported.
+ */
+public final class WKBGeometryBounds {
+
+  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;
+  private static final int ANY_GEOMETRY = 0;
+
+  private static final int MAX_DEPTH = 100;
+
+  private final DimensionBounds xBounds = new DimensionBounds();
+  private final DimensionBounds yBounds = new DimensionBounds();
+
+  /**
+   * Adds the coordinates from one WKB geometry to these bounds.
+   *
+   * <p>The input is read through a duplicate, so its position and limit are left unchanged.
+   *
+   * @param wkb a buffer containing exactly one WKB geometry
+   * @throws IllegalArgumentException if the WKB is invalid or unsupported
+   */
+  public void add(ByteBuffer wkb) {
+    Preconditions.checkArgument(wkb != null, "Invalid WKB buffer: null");
+    ByteBuffer buffer = wkb.duplicate();
+    parseGeometry(buffer, 0, ANY_GEOMETRY);
+    Preconditions.checkArgument(!buffer.hasRemaining(), "Invalid WKB: trailing data");
+  }
+
+  /** Returns whether both X and Y have accumulated a non-NaN value. */
+  public boolean hasBounds() {
+    return xBounds.hasValue() && yBounds.hasValue();
+  }
+
+  /** Returns the lower bound, or {@code null} if either X or Y has no value. */
+  public GeospatialBound lowerBound() {
+    return hasBounds() ? GeospatialBound.createXY(xBounds.lower(), yBounds.lower()) : null;
+  }
+
+  /** Returns the upper bound, or {@code null} if either X or Y has no value. */
+  public GeospatialBound upperBound() {
+    return hasBounds() ? GeospatialBound.createXY(xBounds.upper(), yBounds.upper()) : null;
+  }
+
+  private void parseGeometry(ByteBuffer buffer, int depth, int expectedType) {
+    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);
+    Preconditions.checkArgument(
+        typeCode / 1000 == 0, "Unsupported WKB: only 2D geometries are supported");
+    Preconditions.checkArgument(
+        expectedType == ANY_GEOMETRY || geometryType == expectedType,
+        "Invalid WKB: expected geometry type %s but found %s",
+        expectedType,
+        geometryType);
+
+    switch (geometryType) {
+      case TYPE_POINT -> readCoordinate(buffer);
+      case TYPE_LINE_STRING -> readCoordinateSequence(buffer, true);
+      case TYPE_POLYGON -> {
+        int numRings = readCount(buffer);
+        if (numRings > 0) {
+          readCoordinateSequence(buffer, true);
+        }
+
+        for (int i = 1; i < numRings; i += 1) {
+          readCoordinateSequence(buffer, false);
+        }
+      }
+      case TYPE_MULTI_POINT -> readCollection(buffer, depth, TYPE_POINT);
+      case TYPE_MULTI_LINE_STRING -> readCollection(buffer, depth, TYPE_LINE_STRING);
+      case TYPE_MULTI_POLYGON -> readCollection(buffer, depth, TYPE_POLYGON);
+      case TYPE_GEOMETRY_COLLECTION -> readCollection(buffer, depth, ANY_GEOMETRY);
+      default ->
+          throw new IllegalArgumentException(
+              "Invalid or unsupported WKB geometry type: " + typeCode);
+    }
+  }
+
+  private void readCollection(ByteBuffer buffer, int depth, int expectedChildType) {
+    int numElements = readCount(buffer);
+    for (int i = 0; i < numElements; i += 1) {
+      parseGeometry(buffer, depth + 1, expectedChildType);
+    }
+  }
+
+  private void readCoordinateSequence(ByteBuffer buffer, boolean updateBounds) {
+    int numPoints = readCount(buffer);
+    long numBytes = (long) numPoints * 2 * Double.BYTES;
+    checkRemaining(buffer, numBytes);
+    if (!updateBounds) {
+      buffer.position(buffer.position() + (int) numBytes);
+      return;
+    }
+
+    for (int i = 0; i < numPoints; i += 1) {
+      readCoordinate(buffer);
+    }
+  }
+
+  private void readCoordinate(ByteBuffer buffer) {
+    checkRemaining(buffer, 2 * Double.BYTES);
+    double x = buffer.getDouble();
+    double y = buffer.getDouble();
+    xBounds.add(x);
+    yBounds.add(y);
+  }
+
+  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");
+  }
+
+  private static class DimensionBounds {
+    private double lower;
+    private double upper;
+    private boolean hasValue = false;
+
+    private void add(double value) {
+      if (Double.isNaN(value)) {
+        return;
+      }
+
+      if (hasValue) {
+        lower = Math.min(lower, value);
+        upper = Math.max(upper, value);
+      } else {
+        lower = value;
+        upper = value;
+        hasValue = true;
+      }
+    }
+
+    private boolean hasValue() {
+      return hasValue;
+    }
+
+    private double lower() {
+      return lower;
+    }
+
+    private double upper() {
+      return upper;
+    }
+  }
+}

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

@@ -0,0 +1,167 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class TestWKBGeometryBounds {
+
+  @ParameterizedTest(name = "{0}")
+  @MethodSource("boundingBoxCases")
+  public void testBoundingBox(
+      String wkt, String hexWkb, GeospatialBound expectedLower, GeospatialBound expectedUpper) {
+    WKBGeometryBounds bounds = new WKBGeometryBounds();
+    ByteBuffer wkb = decode(hexWkb);
+    int position = wkb.position();
+    int limit = wkb.limit();
+
+    bounds.add(wkb);
+
+    assertThat(wkb.position()).as(wkt).isEqualTo(position);
+    assertThat(wkb.limit()).as(wkt).isEqualTo(limit);
+    assertThat(bounds.lowerBound()).as(wkt).isEqualTo(expectedLower);
+    assertThat(bounds.upperBound()).as(wkt).isEqualTo(expectedUpper);
+  }
+
+  @Test
+  public void testBoundsAcrossValuesWithMissingCoordinates() {
+    WKBGeometryBounds bounds = new WKBGeometryBounds();
+    bounds.add(decode("0101000000000000000000f03f000000000000f87f"));
+    bounds.add(decode("0101000000000000000000f87f0000000000000040"));
+
+    assertThat(bounds.lowerBound()).isEqualTo(xy(1, 2));
+    assertThat(bounds.upperBound()).isEqualTo(xy(1, 2));
+  }
+
+  @ParameterizedTest(name = "{0}")
+  @MethodSource("invalidWkbCases")
+  public void testInvalidWkb(String description, String hexWkb, String expectedMessage) {
+    WKBGeometryBounds bounds = new WKBGeometryBounds();
+    assertThatThrownBy(() -> bounds.add(decode(hexWkb)))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining(expectedMessage);
+  }
+
+  private static Stream<Arguments> boundingBoxCases() {
+    return Stream.of(
+        Arguments.of("POINT EMPTY", "0101000000000000000000f87f000000000000f87f", null, null),
+        Arguments.of(
+            "POINT(1 2)", "0101000000000000000000f03f0000000000000040", xy(1, 2), xy(1, 2)),
+        Arguments.of(
+            "POINT(1 2) big endian",
+            "00000000013ff00000000000004000000000000000",
+            xy(1, 2),
+            xy(1, 2)),
+        Arguments.of(
+            "LINESTRING(0 1,1 0,2 -1,-1 -2,0 1)",
+            "0102000000050000000000000000000000000000000000f03f000000000000f03f"
+                + "00000000000000000000000000000040000000000000f0bf000000000000f0bf"
+                + "00000000000000c00000000000000000000000000000f03f",
+            xy(-1, -2),
+            xy(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",
+            xy(0, 0),
+            xy(10, 10)),
+        Arguments.of(
+            "MULTIPOINT((1 2),EMPTY,EMPTY,(3 4))",
+            "0104000000040000000101000000000000000000f03f000000000000004001010000"
+                + "00000000000000f87f000000000000f87f0101000000000000000000f87f00000000"
+                + "0000f87f010100000000000000000008400000000000001040",
+            xy(1, 2),
+            xy(3, 4)),
+        Arguments.of(
+            "MULTILINESTRING((1 2,3 4),(5 6,7 8))",
+            "010500000002000000010200000002000000000000000000f03f0000000000000040"
+                + "0000000000000840000000000000104001020000000200000000000000000014400000"
+                + "0000000018400000000000001c400000000000002040",
+            xy(1, 2),
+            xy(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",
+            xy(0, 0),
+            xy(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",
+            xy(1, 2),
+            xy(1, 2)));
+  }
+
+  private static Stream<Arguments> invalidWkbCases() {
+    return Stream.of(
+        Arguments.of(
+            "trailing data", "0101000000000000000000f03f000000000000004000", "trailing data"),
+        Arguments.of(
+            "invalid multi-point child",
+            "010400000001000000010200000000000000",
+            "expected geometry type"),
+        Arguments.of(
+            "unsupported Z geometry",
+            "01e9030000000000000000f03f00000000000000400000000000000840",
+            "only 2D geometries are supported"),
+        Arguments.of(
+            "unsupported M geometry",
+            "01d1070000000000000000f03f00000000000000400000000000000840",
+            "only 2D geometries are supported"),
+        Arguments.of(
+            "unsupported ZM geometry",
+            "01b90b0000000000000000f03f000000000000004000000000000008400000000000001040",
+            "only 2D geometries are supported"),
+        Arguments.of("truncated point", "0101000000", "unexpected end of buffer"));
+  }
+
+  private static GeospatialBound xy(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);
+  }
+}