[SPARK-57766][SQL] Validate WKB element counts before allocation

### What changes were proposed in this pull request?

This PR validates WKB collection element counts before using them as `ArrayList` initial capacities in `WkbReader`.

The new `readCount` helper rejects negative counts and counts that cannot fit in the remaining WKB buffer before parsing these structures:

- LineString points
- Polygon ring points
- Polygon rings
- MultiPoint points
- MultiLineString line strings
- MultiPolygon polygons
- GeometryCollection geometries

It also adds regression coverage for negative and oversized counts, including the public `Geometry.fromWkb` path.

### Why are the changes needed?

Malformed WKB can encode invalid collection counts. Before this change, those counts were passed directly to `new ArrayList<>(count)`, which could throw raw allocation-related exceptions such as `IllegalArgumentException` for negative capacities or attempt excessive allocation for very large counts.

Invalid WKB should be rejected consistently as a WKB parse error before allocation.

### Does this PR introduce _any_ user-facing change?

Yes. For malformed WKB with invalid collection counts, parsing now fails with Spark's normal `WKB_PARSE_ERROR` instead of raw Java allocation failures. This affects unreleased WKB parsing behavior.

### How was this patch tested?

Added tests in `WkbErrorHandlingTest` for negative and oversized counts across all count-bearing WKB collection readers, plus a `Geometry.fromWkb` regression test.

Result: 37 tests passed, 0 failed.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex (GPT-5)

Closes #56875 from szehon-ho/SPARK-57766-wkb-count-validation.

Authored-by: Szehon Ho <szehon.apache@gmail.com>
Signed-off-by: Szehon Ho <szehon.apache@gmail.com>
(cherry picked from commit e58192f)
Signed-off-by: Szehon Ho <szehon.apache@gmail.com>

Author: szehon-ho

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/util/geo/WkbReader.java

@@ -178,6 +178,20 @@ private int readInt() {
     return buffer.getInt();
   }
 
+  private int readCount(String countName, int minBytesPerItem) {
+    long countPos = buffer.position();
+    int count = readInt();
+    if (count < 0) {
+      throw new WkbParseException("Invalid count for " + countName + ": " + count, countPos,
+        currentWkb);
+    }
+    if (count > buffer.remaining() / minBytesPerItem) {
+      throw new WkbParseException("Invalid count for " + countName
+        + ": exceeds remaining bytes", countPos, currentWkb);
+    }
+    return count;
+  }
+
   /**
    * Reads a double coordinate value, allowing NaN for empty points.
    */
@@ -386,7 +400,7 @@ private Point readInternalPoint(int srid, int dimensionCount, boolean hasZ,
 
   private LineString readLineString(int srid, int dimensionCount, boolean hasZ, boolean hasM) {
     long numPointsPos = buffer.position();
-    int numPoints = readInt();
+    int numPoints = readCount("LineString points", dimensionCount * WkbUtil.DOUBLE_SIZE);
 
     if (validationLevel > 0 && numPoints == 1) {
       throw new WkbParseException("Too few points in linestring", numPointsPos, currentWkb);
@@ -402,7 +416,7 @@ private LineString readLineString(int srid, int dimensionCount, boolean hasZ, bo
 
   private Ring readRing(int srid, int dimensionCount, boolean hasZ, boolean hasM) {
     long numPointsPos = buffer.position();
-    int numPoints = readInt();
+    int numPoints = readCount("ring points", dimensionCount * WkbUtil.DOUBLE_SIZE);
 
     List<Point> points = new ArrayList<>(numPoints);
 
@@ -425,7 +439,7 @@ private Ring readRing(int srid, int dimensionCount, boolean hasZ, boolean hasM)
   }
 
   private Polygon readPolygon(int srid, int dimensionCount, boolean hasZ, boolean hasM) {
-    int numRings = readInt();
+    int numRings = readCount("polygon rings", WkbUtil.INT_SIZE);
     List<Ring> rings = new ArrayList<>(numRings);
 
     for (int i = 0; i < numRings; i++) {
@@ -436,7 +450,7 @@ private Polygon readPolygon(int srid, int dimensionCount, boolean hasZ, boolean
   }
 
   private MultiPoint readMultiPoint(int srid, boolean hasZ, boolean hasM) {
-    int numPoints = readInt();
+    int numPoints = readCount("MultiPoint points", WkbUtil.BYTE_SIZE + WkbUtil.TYPE_SIZE);
     List<Point> points = new ArrayList<>(numPoints);
 
     for (int i = 0; i < numPoints; i++) {
@@ -452,7 +466,8 @@ private MultiPoint readMultiPoint(int srid, boolean hasZ, boolean hasM) {
   }
 
   private MultiLineString readMultiLineString(int srid, boolean hasZ, boolean hasM) {
-    int numLineStrings = readInt();
+    int numLineStrings =
+        readCount("MultiLineString line strings", WkbUtil.BYTE_SIZE + WkbUtil.TYPE_SIZE);
     List<LineString> lineStrings = new ArrayList<>(numLineStrings);
 
     for (int i = 0; i < numLineStrings; i++) {
@@ -468,7 +483,8 @@ private MultiLineString readMultiLineString(int srid, boolean hasZ, boolean hasM
   }
 
   private MultiPolygon readMultiPolygon(int srid, boolean hasZ, boolean hasM) {
-    int numPolygons = readInt();
+    int numPolygons =
+        readCount("MultiPolygon polygons", WkbUtil.BYTE_SIZE + WkbUtil.TYPE_SIZE);
     List<Polygon> polygons = new ArrayList<>(numPolygons);
 
     for (int i = 0; i < numPolygons; i++) {
@@ -484,7 +500,8 @@ private MultiPolygon readMultiPolygon(int srid, boolean hasZ, boolean hasM) {
   }
 
   private GeometryCollection readGeometryCollection(int srid, boolean hasZ, boolean hasM) {
-    int numGeometries = readInt();
+    int numGeometries =
+        readCount("GeometryCollection geometries", WkbUtil.BYTE_SIZE + WkbUtil.TYPE_SIZE);
     List<GeometryModel> geometries = new ArrayList<>(numGeometries);
 
     for (int i = 0; i < numGeometries; i++) {

sql/catalyst/src/test/java/org/apache/spark/sql/catalyst/util/geo/WkbErrorHandlingTest.java

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.catalyst.util.geo;
 
+import org.apache.spark.SparkIllegalArgumentException;
+import org.apache.spark.sql.catalyst.util.Geometry;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -118,6 +120,50 @@ public void testTruncatedLineString() {
     Assertions.assertSame(truncated, ex.getWkb());
   }
 
+  @Test
+  public void testNegativeElementCounts() {
+    String[] invalidCounts = new String[] {
+      "0102000000ffffffff",          // LineString with -1 points
+      "010300000001000000ffffffff",  // Polygon ring with -1 points
+      "0103000000ffffffff",          // Polygon with -1 rings
+      "0104000000ffffffff",          // MultiPoint with -1 points
+      "0105000000ffffffff",          // MultiLineString with -1 linestrings
+      "0106000000ffffffff",          // MultiPolygon with -1 polygons
+      "0107000000ffffffff"           // GeometryCollection with -1 geometries
+    };
+
+    for (String invalidCount : invalidCounts) {
+      assertParseError(invalidCount, "Invalid count");
+    }
+  }
+
+  @Test
+  public void testElementCountsExceedRemainingBytes() {
+    String[] invalidCounts = new String[] {
+      "0102000000ffffff7f",          // LineString with too many points
+      "010300000001000000ffffff7f",  // Polygon ring with too many points
+      "0103000000ffffff7f",          // Polygon with too many rings
+      "0104000000ffffff7f",          // MultiPoint with too many points
+      "0105000000ffffff7f",          // MultiLineString with too many linestrings
+      "0106000000ffffff7f",          // MultiPolygon with too many polygons
+      "0107000000ffffff7f"           // GeometryCollection with too many geometries
+    };
+
+    for (String invalidCount : invalidCounts) {
+      assertParseError(invalidCount, "Invalid count");
+    }
+  }
+
+  @Test
+  public void testGeometryFromWkbRejectsInvalidCount() {
+    byte[] wkb = hexToBytes("0102000000ffffffff");
+    SparkIllegalArgumentException ex = Assertions.assertThrows(
+      SparkIllegalArgumentException.class, () -> Geometry.fromWkb(wkb));
+
+    Assertions.assertEquals("WKB_PARSE_ERROR", ex.getCondition());
+    Assertions.assertTrue(ex.getMessage().contains("Invalid count"));
+  }
+
   @Test
   public void testValidationLevels() {
     // With validation level 0, invalid geometries might be accepted