[SEDONA-468] Add useSpheroid provision in ST_DWithin

binder/Sedona_OvertureMaps_GeoParquet.ipynb

@@ -23,19 +23,19 @@
     "tags": []
    },
    "source": [
-    "# Overture Maps data"
+    "# Overture Map data"
    ]
   },
   {
    "cell_type": "markdown",
    "id": "3cfd484a-e6ac-4cdb-a8ac-2ed77fbd23db",
    "metadata": {},
    "source": [
-    "## Wherobots distribution\n",
+    "## Wherobots version\n",
     "\n",
-    "At this time, Wherobots only maintains a distribution of the Overture Maps dataset for `overturemaps-us-west-2/release/2023-07-26-alpha.0/`.\n",
+    "Wherobots only releases a version for overturemaps-us-west-2/release/2023-07-26-alpha.0/.\n",
     "\n",
-    "This data is in GeoParquet format and is clustered by spatial promity to ensure efficient filter pushdown performance."
+    "This data is in GeoParquet format and data is clusterd by their spatial promity to ensure efficient filter pushdown performance"
    ]
   },
   {
@@ -47,17 +47,17 @@
    },
    "outputs": [],
    "source": [
-    "DATA_LINK = \"s3a://wherobots-examples/data/overturemaps-us-west-2/release/2023-07-26-alpha.0/\""
+    "DATA_LINK = \"s3a://wherobots-public-data/overturemaps-us-west-2/release/2023-07-26-alpha.0/\""
    ]
   },
   {
    "cell_type": "markdown",
    "id": "b5fcf043-a97d-4321-ac16-fafcdbbcf3aa",
    "metadata": {},
    "source": [
-    "## Overture Maps Foundation distributions\n",
+    "## OMF versions\n",
     "\n",
-    "The following files are from the official Overture Maps Foundation releases. They are GeoParquet files generated by Apache Sedona.\n",
+    "The following files are official OMF releases. They are GeoParquet files generated by Apache Sedona.\n",
     "\n",
     "However, unlike the Wherobots version, these data might not be in the same file structures and hence the spatial query might be a bit slower.\n"
    ]

common/src/main/java/org/apache/sedona/common/Predicates.java

@@ -14,6 +14,7 @@
 package org.apache.sedona.common;
 
 import org.locationtech.jts.geom.Geometry;
+import org.apache.sedona.common.sphere.Spheroid;
 
 public class Predicates {
     public static boolean contains(Geometry leftGeometry, Geometry rightGeometry) {
@@ -50,6 +51,15 @@ public static boolean orderingEquals(Geometry leftGeometry, Geometry rightGeomet
         return leftGeometry.equalsExact(rightGeometry);
     }
     public static boolean dWithin(Geometry leftGeometry, Geometry rightGeometry, double distance) {
-        return leftGeometry.isWithinDistance(rightGeometry, distance);
+       return dWithin(leftGeometry, rightGeometry, distance, false);
+    }
+
+    public static boolean dWithin(Geometry leftGeometry, Geometry rightGeometry, double distance, boolean useSpheroid) {
+        if (useSpheroid) {
+            double distanceSpheroid = Spheroid.distance(leftGeometry, rightGeometry);
+            return distanceSpheroid <= distance;
+        }else {
+            return leftGeometry.isWithinDistance(rightGeometry, distance);
+        }
     }
 }

common/src/test/java/org/apache/sedona/common/PredicatesTest.java

@@ -62,5 +62,15 @@ public void testDWithinGeomCollection() {
         assertTrue(actual);
     }
 
+    @Test
+    public void testDWithinSpheroid() {
+        Geometry seattlePoint = GEOMETRY_FACTORY.createPoint(new Coordinate(-122.335167, 47.608013));
+        Geometry newYorkPoint = GEOMETRY_FACTORY.createPoint(new Coordinate(-73.935242, 40.730610));
+
+        double distance = 4000 * 1e3; //distance between NY and Seattle is less than 4000 km
+        boolean actual = Predicates.dWithin(newYorkPoint, seattlePoint, distance, true);
+        assertTrue(actual);
+    }
+
 
 }

docs/api/flink/Predicate.md

@@ -60,9 +60,15 @@ true
 
 ## ST_DWithin
 
-Introduction: Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'. This function essentially checks if the shortest distance between the envelope of the two geometries is <= the provided distance.
+Introduction: Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'.
 
-Format: `ST_DWithin (leftGeometry: Geometry, rightGeometry: Geometry, distance: Double)`
+If `useSpheroid` is passed true, ST_DWithin uses Sedona's ST_DistanceSpheroid to check the spheroid distance between the centroids of two geometries. The unit of the distance in this case is meter.
+
+If `useSpheroid` is passed false, ST_DWithin uses Euclidean distance and the unit of the distance is the same as the CRS of the geometries. To obtain the correct result, please consider using ST_Transform to put data in an appropriate CRS.
+
+If useSpheroid is not given, it defaults to false
+
+Format: `ST_DWithin (leftGeometry: Geometry, rightGeometry: Geometry, distance: Double, useSpheroid: Optional(Boolean) = false)`
 
 Since: `v1.5.1`
 
@@ -78,6 +84,19 @@ Output:
 true
 ```
 
+```text
+Check for distance between New York and Seattle (< 4000 km)
+```
+
+```sql
+SELECT ST_DWithin(ST_GeomFromWKT(-122.335167 47.608013), ST_GeomFromWKT(-73.935242 40.730610), 4000000, true)
+```
+
+Output:
+```
+true
+```
+
 ## ST_Equals
 
 Introduction: Return true if A equals to B

docs/api/sql/Predicate.md

@@ -60,9 +60,15 @@ true
 
 ## ST_DWithin
 
-Introduction: Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'. This function essentially checks if the shortest distance between the envelope of the two geometries is <= the provided distance.
+Introduction: Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'.
 
-Format: `ST_DWithin (leftGeometry: Geometry, rightGeometry: Geometry, distance: Double)`
+If `useSpheroid` is passed true, ST_DWithin uses Sedona's ST_DistanceSpheroid to check the spheroid distance between the centroids of two geometries. The unit of the distance in this case is meter.
+
+If `useSpheroid` is passed false, ST_DWithin uses Euclidean distance and the unit of the distance is the same as the CRS of the geometries. To obtain the correct result, please consider using ST_Transform to put data in an appropriate CRS.
+
+If useSpheroid is not given, it defaults to false
+
+Format: `ST_DWithin (leftGeometry: Geometry, rightGeometry: Geometry, distance: Double, useSpheroid: Optional(Boolean) = false)`
 
 Since: `v1.5.1`
 
@@ -78,6 +84,19 @@ Output:
 true
 ```
 
+```text
+Check for distance between New York and Seattle (< 4000 km)
+```
+
+```sql
+SELECT ST_DWithin(ST_GeomFromWKT(-122.335167 47.608013), ST_GeomFromWKT(-73.935242 40.730610), 4000000, true)
+```
+
+Output:
+```
+true
+```
+
 ## ST_Equals
 
 Introduction: Return true if A equals to B

flink/src/main/java/org/apache/sedona/flink/expressions/Predicates.java

@@ -248,5 +248,13 @@ public Boolean eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jt
             Geometry geom2 = (Geometry) o2;
             return org.apache.sedona.common.Predicates.dWithin(geom1, geom2, distance);
         }
+
+        @DataTypeHint("Boolean")
+        public Boolean eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o1, @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o2, @DataTypeHint("Double") Double distance, @DataTypeHint("Boolean") Boolean useSphere)
+        {
+            Geometry geom1 = (Geometry) o1;
+            Geometry geom2 = (Geometry) o2;
+            return org.apache.sedona.common.Predicates.dWithin(geom1, geom2, distance, useSphere);
+        }
     }
 }

flink/src/test/java/org/apache/sedona/flink/PredicateTest.java

@@ -145,4 +145,12 @@ public void testDWithinFailure() {
         Boolean actual = (Boolean) first(table).getField(0);
         assertEquals(false, actual);
     }
+
+    @Test
+    public void testDWithinSphere() {
+        Table table = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('POINT (-122.335167 47.608013)') as seattle, ST_GeomFromWKT('POINT (-73.935242 40.730610)') as ny");
+        table = table.select(call(Predicates.ST_DWithin.class.getSimpleName(), $("seattle"), $("ny"), 3500 * 1e3, true));
+        Boolean actual = (Boolean) first(table).getField(0);
+        assertEquals(false, actual);
+    }
 }

python/sedona/sql/st_predicates.py

@@ -18,7 +18,7 @@
 from functools import partial
 
 from pyspark.sql import Column
-from typing import Union
+from typing import Union, Optional
 
 from sedona.sql.dataframe_api import ColumnOrName, call_sedona_function, validate_argument_types
 
@@ -194,13 +194,14 @@ def ST_CoveredBy(a: ColumnOrName, b: ColumnOrName) -> Column:
     return _call_predicate_function("ST_CoveredBy", (a, b))
 
 @validate_argument_types
-def ST_DWithin(a: ColumnOrName, b: ColumnOrName, distance: Union[ColumnOrName, float]):
+def ST_DWithin(a: ColumnOrName, b: ColumnOrName, distance: Union[ColumnOrName, float], use_sphere: Optional[Union[ColumnOrName, bool]] = None):
     """
     Check if geometry a is within 'distance' units of geometry b
     :param a: Geometry column to check
     :param b: Geometry column to check
     :param distance: distance units to check the within predicate
+    :param use_sphere: whether to use spheroid distance or euclidean distance
     :return: True if a is within distance units of Geometry b
     """
-
-    return _call_predicate_function("ST_DWithin", (a, b, distance))
+    args = (a, b, distance, use_sphere) if use_sphere is not None else (a, b, distance,)
+    return _call_predicate_function("ST_DWithin", args)

python/tests/sql/test_dataframe_api.py

@@ -180,7 +180,8 @@
     (stp.ST_Within, (lambda: f.expr("ST_Point(0.0, 0.0)"), "geom"), "triangle_geom", "", False),
     (stp.ST_Covers, ("geom", lambda: f.expr("ST_Point(0.0, 0.0)")), "triangle_geom", "", True),
     (stp.ST_CoveredBy, (lambda: f.expr("ST_Point(0.0, 0.0)"), "geom"), "triangle_geom", "", True),
-    (stp.ST_DWithin, ("origin", "point", 5.0), "origin_and_point", "", True),
+    (stp.ST_DWithin, ("origin", "point", 5.0,), "origin_and_point", "", True),
+    (stp.ST_DWithin, ("ny", "seattle", 4000000.0, True), "ny_seattle", "", True),
 
     # aggregates
     (sta.ST_Envelope_Aggr, ("geom",), "exploded_points", "", "POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))"),
@@ -436,6 +437,8 @@ def base_df(self, request):
             return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('LINESTRING (0 2, 1 1, 2 0)') AS line, ST_GeomFromWKT('POINT (0 0)') AS point")
         elif request.param == "origin_and_point":
             return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('POINT (0 0)') AS origin, ST_GeomFromWKT('POINT (1 0)') as point")
+        elif request.param == "ny_seattle":
+            return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('POINT (-122.335167 47.608013)') AS seattle, ST_GeomFromWKT('POINT (-73.935242 40.730610)') as ny")
         raise ValueError(f"Invalid base_df name passed: {request.param}")
 
     def _id_test_configuration(val):

python/tests/sql/test_predicate.py

@@ -17,6 +17,7 @@
 
 from tests import csv_point_input_location, csv_point1_input_location, csv_polygon1_input_location
 from tests.test_base import TestBase
+from pyspark.sql.functions import expr
 
 
 class TestPredicate(TestBase):
@@ -222,3 +223,17 @@ def test_st_dwithin(self):
         test_table.createOrReplaceTempView("test_table")
         isWithin = self.spark.sql("select ST_DWithin(origin, point_1, 3) from test_table").head()[0]
         assert isWithin is True
+
+    def test_dwithin_use_sphere(self):
+        test_table = self.spark.sql("select ST_GeomFromWKT('POINT (-122.335167 47.608013)') as seattle, ST_GeomFromWKT('POINT (-73.935242 40.730610)') as ny")
+        test_table.createOrReplaceTempView("test_table")
+        isWithin = self.spark.sql("select ST_DWithin(seattle, ny, 2000000, true) from test_table").head()[0]
+        assert isWithin is False
+
+
+    def test_dwithin_use_sphere_complex_boolean_expression(self):
+        expected = 55
+        df_point = self.spark.range(10).withColumn("pt", expr("ST_Point(id, id)"))
+        df_polygon = self.spark.range(10).withColumn("poly", expr("ST_Point(id, id + 0.01)"))
+        actual = df_point.alias("a").join(df_polygon.alias("b"), expr("ST_DWithin(pt, poly, 10000, a.`id` % 2 = 0)")).count()
+        assert expected == actual

spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Predicates.scala

@@ -254,7 +254,7 @@ case class ST_OrderingEquals(inputExpressions: Seq[Expression])
 }
 
 case class ST_DWithin(inputExpressions: Seq[Expression])
-  extends InferredExpression(Predicates.dWithin _) {
+  extends InferredExpression(inferrableFunction3(Predicates.dWithin), inferrableFunction4(Predicates.dWithin)) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)

spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_predicates.scala

@@ -55,6 +55,7 @@ object st_predicates extends DataFrameAPI {
   def ST_CoveredBy(a: Column, b: Column): Column = wrapExpression[ST_CoveredBy](a, b)
   def ST_CoveredBy(a: String, b: String): Column = wrapExpression[ST_CoveredBy](a, b)
   def ST_DWithin(a: Column, b: Column, distance: Column): Column = wrapExpression[ST_DWithin](a, b, distance)
-
   def ST_DWithin(a: String, b: String, distance: Double): Column = wrapExpression[ST_DWithin](a, b, distance)
+  def ST_DWithin(a: Column, b: Column, distance: Column, useSphere: Column): Column = wrapExpression[ST_DWithin](a, b, distance, useSphere)
+  def ST_DWithin(a: String, b: String, distance: Double, useSphere: Boolean): Column = wrapExpression[ST_DWithin](a, b, distance, useSphere)
 }

spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/strategy/join/JoinQueryDetector.scala

@@ -126,12 +126,6 @@ class JoinQueryDetector(sparkSession: SparkSession) extends Strategy {
           getJoinDetection(left, right, predicate, Some(extraCondition))
         case Some(And(extraCondition, predicate: ST_Predicate)) =>
           getJoinDetection(left, right, predicate, Some(extraCondition))
-        case Some(ST_DWithin(Seq(leftShape, rightShape, distance))) =>
-          Some(JoinQueryDetection(left, right, leftShape, ST_Buffer(Seq(rightShape, distance)), SpatialPredicate.INTERSECTS, isGeography = false, condition, None))
-        case Some(And(ST_DWithin(Seq(leftShape, rightShape, distance)), extraCondition)) =>
-          Some(JoinQueryDetection(left, right, leftShape, ST_Buffer(Seq(rightShape, distance)), SpatialPredicate.INTERSECTS, isGeography = false, condition, Some(extraCondition)))
-        case Some(And(extraCondition, ST_DWithin(Seq(leftShape, rightShape, distance)))) =>
-          Some(JoinQueryDetection(left, right, leftShape, ST_Buffer(Seq(rightShape, distance)), SpatialPredicate.INTERSECTS, isGeography = false, condition, Some(extraCondition)))
           //For raster-vector joins
         case Some(predicate: RS_Predicate) =>
           getRasterJoinDetection(left, right, predicate, None)
@@ -140,6 +134,36 @@ class JoinQueryDetector(sparkSession: SparkSession) extends Strategy {
         case Some(And(extraCondition, predicate: RS_Predicate)) =>
           getRasterJoinDetection(left, right, predicate, Some(extraCondition))
         // For distance joins we execute the actual predicate (condition) and not only extraConditions.
+        case Some(ST_DWithin(Seq(leftShape, rightShape, distance))) =>
+          Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, isGeography = false, condition, Some(distance)))
+        case Some(And(ST_DWithin(Seq(leftShape, rightShape, distance)), _)) =>
+          Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, isGeography = false, condition, Some(distance)))
+        case Some(And(_, ST_DWithin(Seq(leftShape, rightShape, distance)))) =>
+          Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, isGeography = false, condition, Some(distance)))
+        case Some(ST_DWithin(Seq(leftShape, rightShape, distance, useSpheroid))) =>
+          try {
+            val useSpheroidUnwrapped = useSpheroid.eval().asInstanceOf[Boolean]
+            Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, isGeography = useSpheroidUnwrapped, condition, Some(distance)))
+          }catch {
+            case _: Throwable =>
+              Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, isGeography = false, condition, Some(distance)))
+          }
+        case Some(And(ST_DWithin(Seq(leftShape, rightShape, distance, useSpheroid)), _)) =>
+          try {
+            val useSpheroidUnwrapped = useSpheroid.eval().asInstanceOf[Boolean]
+            Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, isGeography = useSpheroidUnwrapped, condition, Some(distance)))
+          }catch {
+            case _: Throwable =>
+              Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, isGeography = false, condition, Some(distance)))
+          }
+        case Some(And(_, ST_DWithin(Seq(leftShape, rightShape, distance, useSpheroid)))) =>
+          try {
+            val useSpheroidUnwrapped = useSpheroid.eval().asInstanceOf[Boolean]
+            Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, isGeography = useSpheroidUnwrapped, condition, Some(distance)))
+          }catch {
+            case _: Throwable =>
+              Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, isGeography = false, condition, Some(distance)))
+          }
         case Some(LessThanOrEqual(ST_Distance(Seq(leftShape, rightShape)), distance)) =>
           Some(JoinQueryDetection(left, right, leftShape, rightShape, SpatialPredicate.INTERSECTS, false, condition, Some(distance)))
         case Some(And(LessThanOrEqual(ST_Distance(Seq(leftShape, rightShape)), distance), _)) =>
@@ -397,13 +421,15 @@ class JoinQueryDetector(sparkSession: SparkSession) extends Strategy {
     val b = children.tail.head
     val isRasterPredicate = a.dataType.isInstanceOf[RasterUDT] || b.dataType.isInstanceOf[RasterUDT]
 
-    val relationship = (distance, spatialPredicate, isGeography, isRasterPredicate) match {
-      case (Some(_), SpatialPredicate.INTERSECTS, false, false) => "ST_Distance <="
-      case (Some(_), _, false, false) => "ST_Distance <"
-      case (Some(_), SpatialPredicate.INTERSECTS, true, false) => "ST_Distance (Geography) <="
-      case (Some(_), _, true, false) => "ST_Distance (Geography) <"
-      case (None, _, false, false) => s"ST_$spatialPredicate"
-      case (None, _, false, true) => s"RS_$spatialPredicate"
+    val relationship = (distance, spatialPredicate, isGeography, extraCondition, isRasterPredicate) match {
+      case (Some(_), SpatialPredicate.INTERSECTS, false, Some(ST_DWithin(Seq(_*))), false) => "ST_DWithin"
+      case (Some(_), SpatialPredicate.INTERSECTS, false, _, false) => "ST_Distance <="
+      case (Some(_), _, false, _, false) => "ST_Distance <"
+      case (Some(_), SpatialPredicate.INTERSECTS, true, Some(ST_DWithin(Seq(_*))), false) => "ST_DWithin(useSpheroid = true)"
+      case (Some(_), SpatialPredicate.INTERSECTS, true, _, false) => "ST_Distance (Geography) <="
+      case (Some(_), _, true, _, false) => "ST_Distance (Geography) <"
+      case (None, _, false, _, false) => s"ST_$spatialPredicate"
+      case (None, _, false,_,  true) => s"RS_$spatialPredicate"
     }
     val (distanceOnIndexSide, distanceOnStreamSide) = distance.map { distanceExpr =>
       matchDistanceExpressionToJoinSide(distanceExpr, left, right) match {