[SPARK-55640][GEO][SQL] Propagate WKB parsing errors for Geometry and Geography

### What changes were proposed in this pull request?
WKB reader was implemented for Geometry and Geography, but only using internal exception handling. This PR addresses this by introducing proper user-facing error classes for WKB parsing.

### Why are the changes needed?
Propagate the WKB parsing errors properly to the user.

### Does this PR introduce _any_ user-facing change?
Yes, users now get proper errors for invalid WKB parsing.

### How was this patch tested?
Added new unit tests and end-to-end SQL tests for WKB parsing.

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

Closes #54424 from uros-db/geo-wkb-parse-exceptions.

Authored-by: Uros Bojanic <uros.bojanic@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: uros-db

common/utils/src/main/resources/error/error-conditions.json

@@ -7754,6 +7754,12 @@
     ],
     "sqlState" : "42601"
   },
+  "WKB_PARSE_ERROR" : {
+    "message" : [
+      "Error parsing WKB: <parseError> at position <pos>"
+    ],
+    "sqlState" : "22023"
+  },
   "WRITE_STREAM_NOT_ALLOWED" : {
     "message" : [
       "`writeStream` can be called only on streaming Dataset/DataFrame."

python/pyspark/errors/error-conditions.json

@@ -1488,6 +1488,12 @@
       "Value for `<arg_name>` must be between <lower_bound> and <upper_bound> (inclusive), got <actual>"
     ]
   },
+  "WKB_PARSE_ERROR" : {
+    "message" : [
+      "Error parsing WKB: <parseError> at position <pos>"
+    ],
+    "sqlState" : "22023"
+  },
   "WRONG_NUM_ARGS_FOR_HIGHER_ORDER_FUNCTION": {
     "message": [
       "Function `<func_name>` should take between 1 and 3 arguments, but the provided function takes <num_args>."

python/pyspark/sql/tests/test_functions.py

@@ -26,6 +26,7 @@
 import unittest
 
 from pyspark.errors import PySparkTypeError, PySparkValueError, SparkRuntimeException
+from pyspark.errors.exceptions.base import IllegalArgumentException
 from pyspark.sql import Row, Window, functions as F, types
 from pyspark.sql.avro.functions import from_avro, to_avro
 from pyspark.sql.column import Column
@@ -3798,6 +3799,25 @@ def test_st_asbinary(self):
         )
         self.assertEqual(results, [expected])
 
+    def test_st_geogfromwkb(self):
+        df = self.spark.createDataFrame(
+            [(bytes.fromhex("0101000000000000000000F03F0000000000000040"),)],
+            ["wkb"],
+        )
+        results = df.select(
+            F.hex(F.st_asbinary(F.st_geogfromwkb("wkb"))),
+        ).collect()
+        expected = Row(
+            "0101000000000000000000F03F0000000000000040",
+        )
+        self.assertEqual(results, [expected])
+        # ST_GeogFromWKB with invalid WKB.
+        df = self.spark.createDataFrame([(bytearray(b"\x6f"),)], ["wkb"])
+        with self.assertRaises(IllegalArgumentException) as error_context:
+            df.select(F.st_geogfromwkb("wkb")).collect()
+        self.assertIn("[WKB_PARSE_ERROR]", str(error_context.exception))
+        self.assertIn("Unexpected end of WKB buffer", str(error_context.exception))
+
     def test_st_geomfromwkb(self):
         df = self.spark.createDataFrame(
             [(bytes.fromhex("0101000000000000000000F03F0000000000000040"), 4326)],
@@ -3814,6 +3834,12 @@ def test_st_geomfromwkb(self):
             "0101000000000000000000F03F0000000000000040",
         )
         self.assertEqual(results, [expected])
+        # ST_GeomFromWKB with invalid WKB.
+        df = self.spark.createDataFrame([(bytearray(b"\x6f"),)], ["wkb"])
+        with self.assertRaises(IllegalArgumentException) as error_context:
+            df.select(F.st_geomfromwkb("wkb")).collect()
+        self.assertIn("[WKB_PARSE_ERROR]", str(error_context.exception))
+        self.assertIn("Unexpected end of WKB buffer", str(error_context.exception))
 
     def test_st_setsrid(self):
         df = self.spark.createDataFrame(

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/util/Geography.java

@@ -17,8 +17,10 @@
 package org.apache.spark.sql.catalyst.util;
 
 import org.apache.spark.sql.catalyst.util.geo.GeometryModel;
+import org.apache.spark.sql.catalyst.util.geo.WkbParseException;
 import org.apache.spark.sql.catalyst.util.geo.WkbReader;
 import org.apache.spark.sql.catalyst.util.geo.WkbWriter;
+import org.apache.spark.sql.errors.QueryExecutionErrors;
 import org.apache.spark.unsafe.types.GeographyVal;
 
 import java.nio.ByteBuffer;
@@ -81,13 +83,17 @@ public Geography copy() {
 
   // Returns a Geography object with the specified SRID value by parsing the input WKB.
   public static Geography fromWkb(byte[] wkb, int srid) {
-    WkbReader reader = new WkbReader(true);
-    reader.read(wkb); // Validate WKB with geography coordinate bounds.
-
-    byte[] bytes = new byte[HEADER_SIZE + wkb.length];
-    ByteBuffer.wrap(bytes).order(DEFAULT_ENDIANNESS).putInt(srid);
-    System.arraycopy(wkb, 0, bytes, WKB_OFFSET, wkb.length);
-    return fromBytes(bytes);
+    try {
+      WkbReader reader = new WkbReader(true);
+      reader.read(wkb); // Validate WKB with geography coordinate bounds.
+
+      byte[] bytes = new byte[HEADER_SIZE + wkb.length];
+      ByteBuffer.wrap(bytes).order(DEFAULT_ENDIANNESS).putInt(srid);
+      System.arraycopy(wkb, 0, bytes, WKB_OFFSET, wkb.length);
+      return fromBytes(bytes);
+    } catch (WkbParseException e) {
+      throw QueryExecutionErrors.wkbParseError(e.getParseError(), e.getPosition());
+    }
   }
 
   // Overload for the WKB reader where we use the default SRID for Geography.

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/util/Geometry.java

@@ -17,8 +17,10 @@
 package org.apache.spark.sql.catalyst.util;
 
 import org.apache.spark.sql.catalyst.util.geo.GeometryModel;
+import org.apache.spark.sql.catalyst.util.geo.WkbParseException;
 import org.apache.spark.sql.catalyst.util.geo.WkbReader;
 import org.apache.spark.sql.catalyst.util.geo.WkbWriter;
+import org.apache.spark.sql.errors.QueryExecutionErrors;
 import org.apache.spark.unsafe.types.GeometryVal;
 
 import java.nio.ByteBuffer;
@@ -81,13 +83,17 @@ public Geometry copy() {
 
   // Returns a Geometry object with the specified SRID value by parsing the input WKB.
   public static Geometry fromWkb(byte[] wkb, int srid) {
-    WkbReader reader = new WkbReader();
-    reader.read(wkb); // Validate WKB
-
-    byte[] bytes = new byte[HEADER_SIZE + wkb.length];
-    ByteBuffer.wrap(bytes).order(DEFAULT_ENDIANNESS).putInt(srid);
-    System.arraycopy(wkb, 0, bytes, WKB_OFFSET, wkb.length);
-    return fromBytes(bytes);
+    try {
+      WkbReader reader = new WkbReader();
+      reader.read(wkb); // Validate WKB
+
+      byte[] bytes = new byte[HEADER_SIZE + wkb.length];
+      ByteBuffer.wrap(bytes).order(DEFAULT_ENDIANNESS).putInt(srid);
+      System.arraycopy(wkb, 0, bytes, WKB_OFFSET, wkb.length);
+      return fromBytes(bytes);
+    } catch (WkbParseException e) {
+      throw QueryExecutionErrors.wkbParseError(e.getParseError(), e.getPosition());
+    }
   }
 
   // Overload for the WKB reader where we use the default SRID for Geometry.

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

@@ -19,40 +19,27 @@
 /**
  * Exception thrown when parsing WKB data fails.
  */
-class WkbParseException extends RuntimeException {
+public class WkbParseException extends RuntimeException {
+  private final String parseError;
   private final long position;
-  private final String wkbString;
+  private final byte[] wkb;
 
-  WkbParseException(String message, long position, byte[] wkb) {
-    super(formatMessage(message, position, wkb));
+  WkbParseException(String parseError, long position, byte[] wkb) {
+    super();
+    this.parseError = parseError;
     this.position = position;
-    this.wkbString = wkb != null ? bytesToHex(wkb) : "";
+    this.wkb = wkb;
   }
 
-  private static String formatMessage(String message, long position, byte[] wkb) {
-    String baseMessage = message + " at position " + position;
-    if (wkb != null && wkb.length > 0) {
-      baseMessage += " in WKB: " + bytesToHex(wkb);
-    }
-    return baseMessage;
+  public String getParseError() {
+    return parseError;
   }
 
-  private static String bytesToHex(byte[] bytes) {
-    if (bytes == null || bytes.length == 0) {
-      return "";
-    }
-    StringBuilder sb = new StringBuilder(bytes.length * 2);
-    for (byte b : bytes) {
-      sb.append(String.format("%02X", b));
-    }
-    return sb.toString();
-  }
-
-  long getPosition() {
+  public long getPosition() {
     return position;
   }
 
-  String getWkbString() {
-    return wkbString;
+  public byte[] getWkb() {
+    return wkb;
   }
 }

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

@@ -222,7 +222,7 @@ private GeometryModel readGeometry(int defaultSrid) {
 
     // Check that we have enough bytes for header (endianness byte + 4-byte type)
     if (currentWkb.length < WkbUtil.BYTE_SIZE + WkbUtil.TYPE_SIZE) {
-      throw new WkbParseException("WKB data too short", 0, currentWkb);
+      throw new WkbParseException("Unexpected end of WKB buffer", 0, currentWkb);
     }
 
     // Create buffer wrapping the entire byte array

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -676,6 +676,15 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
     stInvalidSridValueError(srid.toString)
   }
 
+  def wkbParseError(msg: String, pos: String): SparkIllegalArgumentException = {
+    new SparkIllegalArgumentException(errorClass = "WKB_PARSE_ERROR",
+      messageParameters = Map("parseError" -> msg, "pos" -> pos))
+  }
+
+  def wkbParseError(msg: String, pos: Long): SparkIllegalArgumentException = {
+    wkbParseError(msg, pos.toString)
+  }
+
   def withSuggestionIntervalArithmeticOverflowError(
       suggestedFunc: String,
       context: QueryContext): ArithmeticException = {

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

@@ -46,13 +46,12 @@ private void assertParseError(String hex, String expectedMessagePart, int valida
     WkbParseException ex = Assertions.assertThrows(
         WkbParseException.class, () -> reader.read(wkb),
         "Should throw WkbParseException for WKB: " + hex);
-    Assertions.assertTrue(ex.getMessage().toUpperCase().contains(hex.toUpperCase()),
-        "Exception message should contain the WKB hex: " + hex + ", actual: " + ex.getMessage());
+    Assertions.assertSame(wkb, ex.getWkb());
     if (expectedMessagePart != null && !expectedMessagePart.isEmpty()) {
       Assertions.assertTrue(
-          ex.getMessage().toLowerCase().contains(expectedMessagePart.toLowerCase()),
+          ex.getParseError().toLowerCase().contains(expectedMessagePart.toLowerCase()),
           "Exception message should contain '" + expectedMessagePart + "', actual: " +
-              ex.getMessage());
+              ex.getParseError());
     }
   }
 
@@ -63,69 +62,60 @@ public void testEmptyWkb() {
     WkbParseException ex = Assertions.assertThrows(
       WkbParseException.class, () -> reader.read(emptyWkb));
     // Empty WKB produces empty hex string, so just verify exception was thrown
-    Assertions.assertNotNull(ex.getMessage());
+    Assertions.assertNotNull(ex.getParseError());
   }
 
   @Test
   public void testTooShortWkb() {
     // Only endianness byte
-    String hex = "01";
-    byte[] tooShort = hexToBytes(hex);
+    byte[] tooShort = hexToBytes("01");
     WkbReader reader = new WkbReader();
     WkbParseException ex = Assertions.assertThrows(
       WkbParseException.class, () -> reader.read(tooShort));
-    Assertions.assertTrue(ex.getMessage().toUpperCase().contains(hex.toUpperCase()),
-      "Exception message should contain the WKB hex: " + hex);
+    Assertions.assertSame(tooShort, ex.getWkb());
   }
 
   @Test
   public void testInvalidGeometryTypeZero() {
     // Type = 0 (invalid, should be 1-7)
-    String hex = "0100000000000000000000F03F0000000000000040";
-    byte[] invalidType = hexToBytes(hex);
+    byte[] invalidType = hexToBytes("0100000000000000000000F03F0000000000000040");
     WkbReader reader = new WkbReader();
     WkbParseException ex = Assertions.assertThrows(
       WkbParseException.class, () -> reader.read(invalidType));
-    Assertions.assertTrue(ex.getMessage().toUpperCase().contains(hex.toUpperCase()),
-      "Exception message should contain the WKB hex: " + hex);
+    Assertions.assertSame(invalidType, ex.getWkb());
   }
 
   @Test
   public void testTruncatedPointCoordinates() {
     // Point WKB with truncated coordinates (missing Y coordinate)
-    String hex = "0101000000000000000000F03F";
-    byte[] truncated = hexToBytes(hex);
+    byte[] truncated = hexToBytes("0101000000000000000000F03F");
     WkbReader reader = new WkbReader();
     WkbParseException ex = Assertions.assertThrows(
       WkbParseException.class, () -> reader.read(truncated));
-    Assertions.assertTrue(ex.getMessage().toUpperCase().contains(hex.toUpperCase()),
-      "Exception message should contain the WKB hex: " + hex);
+    Assertions.assertSame(truncated, ex.getWkb());
   }
 
   @Test
   public void testTruncatedByte() {
     // Only one byte (FF) of the 4-byte INT field.
-    String hex = "0102000000ff";
-    byte[] truncated = hexToBytes(hex);
+    byte[] truncated = hexToBytes("0102000000ff");
     WkbReader reader = new WkbReader();
     WkbParseException ex = Assertions.assertThrows(
       WkbParseException.class, () -> reader.read(truncated));
-    Assertions.assertTrue(ex.getMessage().toUpperCase().contains(hex.toUpperCase()),
-      "Exception message should contain the WKB hex: " + hex);
+    Assertions.assertSame(truncated, ex.getWkb());
   }
 
   @Test
   public void testTruncatedLineString() {
     // LineString with declared 2 points but only 1 provided
-    String hex = "010200000002000000" +  // LineString with 2 points
-      "0000000000000000" +              // X of first point
-      "0000000000000000";               // Y of first point (missing second point)
-    byte[] truncated = hexToBytes(hex);
+    byte[] truncated = hexToBytes(
+      "010200000002000000" +  // LineString with 2 points
+      "0000000000000000" +    // X of first point
+      "0000000000000000");    // Y of first point (missing second point)
     WkbReader reader = new WkbReader();
     WkbParseException ex = Assertions.assertThrows(
       WkbParseException.class, () -> reader.read(truncated));
-    Assertions.assertTrue(ex.getMessage().toUpperCase().contains(hex.toUpperCase()),
-      "Exception message should contain the WKB hex: " + hex);
+    Assertions.assertSame(truncated, ex.getWkb());
   }
 
   @Test
@@ -164,8 +154,7 @@ public void testRingWithTooFewPoints() {
 
     WkbParseException ex = Assertions.assertThrows(
       WkbParseException.class, () -> reader.read(invalidPolygon));
-    Assertions.assertTrue(ex.getMessage().toUpperCase().contains(hex.toUpperCase()),
-      "Exception message should contain the WKB hex: " + hex);
+    Assertions.assertSame(invalidPolygon, ex.getWkb());
   }
 
   @Test
@@ -174,20 +163,19 @@ public void testNonClosedRing() {
     WkbReader reader = new WkbReader(1);
 
     // Polygon with ring where first and last points don't match
-    String hex = "01" +                                      // Little endian
+    byte[] nonClosedRing = hexToBytes(
+      "01" +                                               // Little endian
       "03000000" +                                         // Polygon type
       "01000000" +                                         // 1 ring
       "04000000" +                                         // 4 points
       "0000000000000000" + "0000000000000000" +            // (0, 0)
       "000000000000F03F" + "0000000000000000" +            // (1, 0)
       "000000000000F03F" + "000000000000F03F" +            // (1, 1)
-      "0000000000000040" + "0000000000000040";             // (2, 2) - doesn't match first point!
-    byte[] nonClosedRing = hexToBytes(hex);
+      "0000000000000040" + "0000000000000040");            // (2, 2) - doesn't match first point!
 
     WkbParseException ex = Assertions.assertThrows(
       WkbParseException.class, () -> reader.read(nonClosedRing));
-    Assertions.assertTrue(ex.getMessage().toUpperCase().contains(hex.toUpperCase()),
-      "Exception message should contain the WKB hex: " + hex);
+    Assertions.assertSame(nonClosedRing, ex.getWkb());
   }
 
   @Test
@@ -197,7 +185,7 @@ public void testNullByteArray() {
       WkbParseException.class, () -> reader.read(null),
       "Should throw WKBParseException for null byte array");
     // Null WKB cannot produce hex string, just verify exception was thrown
-    Assertions.assertNotNull(ex.getMessage());
+    Assertions.assertNotNull(ex.getParseError());
   }
 
   // ========== Invalid Byte Order Tests ==========

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

@@ -629,7 +629,7 @@ public void testGeographyErrorMessageContainsBoundsInfo() {
     byte[] wkb = makePointWkb2D(200.0, 0.0);
     WkbParseException ex = Assertions.assertThrows(
         WkbParseException.class, () -> geographyReader1().read(wkb));
-    String msg = ex.getMessage();
+    String msg = ex.getParseError();
     Assertions.assertTrue(msg.contains("Invalid coordinate value"));
   }
 }