✨ feat(api): Add WKB geometry bounds

- Accumulate X, Y, Z, and M bounds across ISO WKB geometry values
- Validate structure, collection child types, and trailing data

Generated-by: Codex

Author: Xin Huang

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

@@ -0,0 +1,244 @@
+/*
+ * 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 java.util.Arrays;
+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, Y, Z, and M dimensions. Null or {@code NaN}
+ * values do not contribute to a dimension. Z and M are omitted from the resulting bounds when no
+ * value contributes to that dimension, and no bounds are produced unless both X and Y are present.
+ */
+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 X = 0;
+  private static final int Y = 1;
+  private static final int Z = 2;
+  private static final int M = 3;
+  private static final int NUM_DIMENSIONS = 4;
+
+  private static final int MAX_DEPTH = 100;
+
+  private final double[] lower = new double[NUM_DIMENSIONS];
+  private final double[] upper = new double[NUM_DIMENSIONS];
+  private final boolean[] hasValue = new boolean[NUM_DIMENSIONS];
+
+  public WKBGeometryBounds() {
+    Arrays.fill(lower, Double.POSITIVE_INFINITY);
+    Arrays.fill(upper, Double.NEGATIVE_INFINITY);
+  }
+
+  /**
+   * 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 hasValue[X] && hasValue[Y];
+  }
+
+  /** Returns the lower bound, or {@code null} if either X or Y has no value. */
+  public GeospatialBound lowerBound() {
+    return hasBounds() ? bound(lower) : null;
+  }
+
+  /** Returns the upper bound, or {@code null} if either X or Y has no value. */
+  public GeospatialBound upperBound() {
+    return hasBounds() ? bound(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);
+    CoordinateLayout layout = coordinateLayout((int) (typeCode / 1000), typeCode);
+    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, layout);
+        break;
+      case TYPE_LINE_STRING:
+        readCoordinateSequence(buffer, layout);
+        break;
+      case TYPE_POLYGON:
+        int numRings = readCount(buffer);
+        for (int i = 0; i < numRings; i += 1) {
+          readCoordinateSequence(buffer, layout);
+        }
+        break;
+      case TYPE_MULTI_POINT:
+        readCollection(buffer, depth, TYPE_POINT);
+        break;
+      case TYPE_MULTI_LINE_STRING:
+        readCollection(buffer, depth, TYPE_LINE_STRING);
+        break;
+      case TYPE_MULTI_POLYGON:
+        readCollection(buffer, depth, TYPE_POLYGON);
+        break;
+      case TYPE_GEOMETRY_COLLECTION:
+        readCollection(buffer, depth, ANY_GEOMETRY);
+        break;
+      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, CoordinateLayout layout) {
+    int numPoints = readCount(buffer);
+    checkRemaining(buffer, (long) numPoints * layout.size() * Double.BYTES);
+    for (int i = 0; i < numPoints; i += 1) {
+      readCoordinate(buffer, layout);
+    }
+  }
+
+  private void readCoordinate(ByteBuffer buffer, CoordinateLayout layout) {
+    checkRemaining(buffer, (long) layout.size() * Double.BYTES);
+    update(X, buffer.getDouble());
+    update(Y, buffer.getDouble());
+    if (layout.hasZ()) {
+      update(Z, buffer.getDouble());
+    }
+
+    if (layout.hasM()) {
+      update(M, buffer.getDouble());
+    }
+  }
+
+  private void update(int dimension, double value) {
+    if (!Double.isNaN(value)) {
+      lower[dimension] = Math.min(lower[dimension], value);
+      upper[dimension] = Math.max(upper[dimension], value);
+      hasValue[dimension] = true;
+    }
+  }
+
+  private GeospatialBound bound(double[] values) {
+    if (hasValue[Z] && hasValue[M]) {
+      return GeospatialBound.createXYZM(values[X], values[Y], values[Z], values[M]);
+    } else if (hasValue[Z]) {
+      return GeospatialBound.createXYZ(values[X], values[Y], values[Z]);
+    } else if (hasValue[M]) {
+      return GeospatialBound.createXYM(values[X], values[Y], values[M]);
+    }
+
+    return GeospatialBound.createXY(values[X], values[Y]);
+  }
+
+  private static CoordinateLayout coordinateLayout(int dimensionGroup, long typeCode) {
+    switch (dimensionGroup) {
+      case 0:
+        return CoordinateLayout.XY;
+      case 1:
+        return CoordinateLayout.XYZ;
+      case 2:
+        return CoordinateLayout.XYM;
+      case 3:
+        return CoordinateLayout.XYZM;
+      default:
+        throw new IllegalArgumentException("Invalid or unsupported WKB geometry type: " + typeCode);
+    }
+  }
+
+  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 enum CoordinateLayout {
+    XY(false, false),
+    XYZ(true, false),
+    XYM(false, true),
+    XYZM(true, true);
+
+    private final boolean hasZ;
+    private final boolean hasM;
+
+    CoordinateLayout(boolean hasZ, boolean hasM) {
+      this.hasZ = hasZ;
+      this.hasM = hasM;
+    }
+
+    private boolean hasZ() {
+      return hasZ;
+    }
+
+    private boolean hasM() {
+      return hasM;
+    }
+
+    private int size() {
+      return 2 + (hasZ ? 1 : 0) + (hasM ? 1 : 0);
+    }
+  }
+}