[SPARK][Geo][SQL] Enable WKB parsing for Geography

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

@@ -16,6 +16,7 @@
  */
 package org.apache.spark.sql.catalyst.util;
 
+import org.apache.spark.sql.catalyst.util.geo.WkbReader;
 import org.apache.spark.unsafe.types.GeographyVal;
 
 import java.nio.ByteBuffer;
@@ -77,6 +78,9 @@ 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);

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

@@ -25,29 +25,61 @@
 import java.util.List;
 
 /**
- * Reader for parsing Well-Known Binary (WKB) format geometries.
+ * Reader for parsing Well-Known Binary (WKB) format geometries and geographies.
  * This class implements the OGC Simple Features specification for WKB parsing.
+ * For geographies, coordinate bounds validation is enforced:
+ *   - X (longitude) must be between -180 and 180 (inclusive),
+ *   - Y (latitude) must be between -90 and 90 (inclusive).
  * This class is not thread-safe. Create a new instance for each thread.
  * This class should be catalyst-internal.
  */
 public class WkbReader {
   private ByteBuffer buffer;
   private final int validationLevel;
+  private final boolean geographyMode;
   private byte[] currentWkb;
 
+  // Geography coordinate bounds.
+  private static final double MIN_LONGITUDE = -180.0;
+  private static final double MAX_LONGITUDE = 180.0;
+  private static final double MIN_LATITUDE = -90.0;
+  private static final double MAX_LATITUDE = 90.0;
+  // Default WKB reader settings.
+  private static final int DEFAULT_VALIDATION_LEVEL = 1; // basic validation
+  private static final boolean DEFAULT_GEOGRAPHY_MODE = false; // no geography bounds checking
+
   /**
-   * Constructor for WkbReader with default validation level (1 = basic validation).
+   * Constructor for WkbReader with default validation level (1 = basic validation)
+   * and geometry mode (no geography coordinate bounds checking).
    */
   public WkbReader() {
-    this(1);
+    this(DEFAULT_VALIDATION_LEVEL, DEFAULT_GEOGRAPHY_MODE);
   }
 
   /**
-   * Constructor for WkbReader with specified validation level.
+   * Constructor for WkbReader with specified validation level and geometry mode.
    * @param validationLevel validation level (0 = no validation, 1 = basic validation)
    */
   public WkbReader(int validationLevel) {
+    this(validationLevel, DEFAULT_GEOGRAPHY_MODE);
+  }
+
+  /**
+   * Constructor for WkbReader with default validation level and geography mode.
+   * @param geographyMode if true, validates geography coordinate bounds for longitude and latitude
+   */
+  public WkbReader(boolean geographyMode) {
+    this(DEFAULT_VALIDATION_LEVEL, geographyMode);
+  }
+
+  /**
+   * Constructor for WkbReader with specified validation level and geography mode.
+   * @param validationLevel validation level (0 = no validation, 1 = basic validation)
+   * @param geographyMode if true, validates geography coordinate bounds for longitude and latitude
+   */
+  public WkbReader(int validationLevel, boolean geographyMode) {
     this.validationLevel = validationLevel;
+    this.geographyMode = geographyMode;
   }
 
   // ========== Coordinate Validation Helpers ==========
@@ -69,6 +101,35 @@ private static boolean isValidCoordinateAllowEmpty(double value) {
     return Double.isFinite(value) || Double.isNaN(value);
   }
 
+  /**
+   * Returns true if the longitude value is within valid geography bounds [-180, 180].
+   */
+  private static boolean isValidLongitude(double value) {
+    return value >= MIN_LONGITUDE && value <= MAX_LONGITUDE;
+  }
+
+  /**
+   * Returns true if the latitude value is within valid geography bounds [-90, 90].
+   */
+  private static boolean isValidLatitude(double value) {
+    return value >= MIN_LATITUDE && value <= MAX_LATITUDE;
+  }
+
+  /**
+   * Validates geography coordinate bounds for a point. In geography mode with validation
+   * level > 0, longitude must be between -180 and 180, and latitude must be between -90 and 90.
+   */
+  private void validateGeographyBounds(Point point, long pos) {
+    if (geographyMode && validationLevel > 0 && !point.isEmpty()) {
+      if (!isValidLongitude(point.getX())) {
+        throw new WkbParseException("Longitude must be between -180 and 180", pos, currentWkb);
+      }
+      if (!isValidLatitude(point.getY())) {
+        throw new WkbParseException("Latitude must be between -90 and 90", pos, currentWkb);
+      }
+    }
+  }
+
   /**
    * Reads a geometry from WKB bytes.
    */
@@ -301,11 +362,14 @@ private GeometryModel readGeometryData(
    * Reads a top-level point geometry (allows empty points with NaN coordinates).
    */
   private Point readPoint(int srid, int dimensionCount, boolean hasZ, boolean hasM) {
+    long coordsStartPos = buffer.position();
     double[] coords = new double[dimensionCount];
     for (int i = 0; i < dimensionCount; i++) {
       coords[i] = readDoubleAllowEmpty();
     }
-    return new Point(coords, srid, hasZ, hasM);
+    Point point = new Point(coords, srid, hasZ, hasM);
+    validateGeographyBounds(point, coordsStartPos);
+    return point;
   }
 
   /**
@@ -314,11 +378,14 @@ private Point readPoint(int srid, int dimensionCount, boolean hasZ, boolean hasM
    */
   private Point readInternalPoint(int srid, int dimensionCount, boolean hasZ,
                                   boolean hasM) {
+    long coordsStartPos = buffer.position();
     double[] coords = new double[dimensionCount];
     for (int i = 0; i < dimensionCount; i++) {
       coords[i] = readDoubleNoEmpty();
     }
-    return new Point(coords, srid, hasZ, hasM);
+    Point point = new Point(coords, srid, hasZ, hasM);
+    validateGeographyBounds(point, coordsStartPos);
+    return point;
   }
 
   private LineString readLineString(int srid, int dimensionCount, boolean hasZ, boolean hasM) {