[GH-3502] Read a NULL GeoPackage geometry as null

spark/common/src/main/scala/org/apache/sedona/sql/datasources/geopackage/transform/GeometryReader.scala

@@ -26,7 +26,15 @@ import java.nio.{ByteBuffer, ByteOrder}
 
 object GeometryReader {
 
+  /**
+   * Convert a GeoPackage geometry blob to Sedona's serialized geometry. A SQL NULL in a geometry
+   * column comes back from the driver as a null array, which maps to a null geometry.
+   */
   def extractWKB(gpkgGeom: Array[Byte]): Array[Byte] = {
+    if (gpkgGeom == null) {
+      return null
+    }
+
     val reader = ByteBuffer.wrap(gpkgGeom)
 
     val magic = new Array[Byte](2)

spark/common/src/test/scala/org/apache/sedona/sql/GeoPackageReaderTest.scala

@@ -340,6 +340,24 @@ class GeoPackageReaderTest extends TestBaseScala with Matchers {
         df.filter(s"$name IS NULL").count() shouldEqual 1
       }
     }
+
+    it("should read a NULL geometry as null") {
+      // A NULL in a geometry column used to surface as a NullPointerException from
+      // GeometryReader.extractWKB and abort the whole scan.
+      val df = sparkSession.read
+        .format("geopackage")
+        .option("tableName", "test_features")
+        .load(resourceFolder + "geopackage/test_null_geometry.gpkg")
+
+      val rows = df.select("fid", "geom", "name").collect().sortBy(_.getInt(0))
+
+      rows.map(row => (row.getInt(0), row.isNullAt(1), row.getString(2))) shouldEqual Array(
+        (1, false, "point"),
+        (2, true, "null geometry"),
+        (3, false, "point"))
+
+      df.filter("geom IS NULL").count() shouldEqual 1
+    }
   }
 
   describe("GeoPackage Raster Data Test") {

spark/common/src/test/scala/org/apache/sedona/sql/datasources/geopackage/transform/GeometryReaderTest.scala

@@ -27,6 +27,10 @@ import java.nio.{ByteBuffer, ByteOrder}
 import java.util.EnumSet
 
 class GeometryReaderTest extends AnyFunSuite {
+  test("extractWKB returns null for a NULL geometry") {
+    assert(GeometryReader.extractWKB(null) == null)
+  }
+
   test("extractWKB retains an empty geometry's declared layout and GeoPackage SRID") {
     val factory = new GeometryFactory()
     Seq((3, 0), (3, 1), (4, 1)).foreach { case (dimension, measures) =>