[GH-2830] Adds Geography dual-dispatch to ST_Intersects

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

@@ -29,6 +29,22 @@ public boolean S2_intersects(
     return S2BooleanOperation.intersects(geo1.shapeIndex, geo2.shapeIndex, options);
   }
 
+  /**
+   * Fast intersects between a single point and a ShapeIndex. Avoids building a ShapeIndex for the
+   * point side — only the complex geometry needs an index. Uses S2ClosestEdgeQuery with
+   * includeInteriors=true (default) so a point in a polygon interior returns distance 0.
+   */
+  public static boolean S2_intersectsPointWithIndex(S2Point point, ShapeIndexGeography geo) {
+    S2ClosestEdgeQuery query = S2ClosestEdgeQuery.builder().build(geo.shapeIndex);
+    S2ClosestEdgeQuery.PointTarget<S1ChordAngle> target =
+        new S2ClosestEdgeQuery.PointTarget<>(point);
+    Optional<S2BestEdgesQueryBase.Result> result = query.findClosestEdge(target);
+    if (!result.isPresent()) {
+      return false;
+    }
+    return ((S1ChordAngle) result.get().distance()).getLength2() == 0.0;
+  }
+
   public boolean S2_equals(
       ShapeIndexGeography geo1, ShapeIndexGeography geo2, S2BooleanOperation.Options options) {
     return S2BooleanOperation.equals(geo1.shapeIndex, geo2.shapeIndex, options);

common/src/main/java/org/apache/sedona/common/geography/Functions.java

@@ -312,6 +312,33 @@ public static boolean equals(Geography g1, Geography g2) {
     return pred.S2_equals(toShapeIndex(g1), toShapeIndex(g2), s2Options());
   }
 
+  /**
+   * Spherical intersection test using S2 boolean operations. Takes fast paths for point-to-point
+   * and point-to-complex inputs backed by WKBGeography, avoiding ShapeIndex construction on the
+   * point side.
+   */
+  public static boolean intersects(Geography g1, Geography g2) {
+    if (g1 == null || g2 == null) return false;
+    if (g1 instanceof WKBGeography && g2 instanceof WKBGeography) {
+      WKBGeography w1 = (WKBGeography) g1;
+      WKBGeography w2 = (WKBGeography) g2;
+      // Fast path: point-to-point intersects iff the points are equal
+      if (w1.isPoint() && w2.isPoint()) {
+        return w1.extractPoint().equalsPoint(w2.extractPoint());
+      }
+      // Fast path: point-to-complex uses PointTarget (avoids building ShapeIndex for point side)
+      if (w1.isPoint()) {
+        return Predicates.S2_intersectsPointWithIndex(w1.extractPoint(), toShapeIndex(w2));
+      }
+      if (w2.isPoint()) {
+        return Predicates.S2_intersectsPointWithIndex(w2.extractPoint(), toShapeIndex(w1));
+      }
+    }
+    // General path via ShapeIndex
+    Predicates pred = new Predicates();
+    return pred.S2_intersects(toShapeIndex(g1), toShapeIndex(g2), s2Options());
+  }
+
   /**
    * Spherical "distance within" test. Returns true iff the minimum geodesic distance between g1 and
    * g2 (in meters) is less than or equal to {@code distanceMeters}.

common/src/test/java/org/apache/sedona/common/Geography/FunctionTest.java

@@ -489,6 +489,67 @@ public void equals_nullHandling() throws ParseException {
     assertFalse(Functions.equals(null, null));
   }
 
+  @Test
+  public void intersects_overlappingPolygons() throws ParseException {
+    Geography g1 = Constructors.geogFromWKT("POLYGON ((0 0, 2 0, 2 2, 0 2, 0 0))", 4326);
+    Geography g2 = Constructors.geogFromWKT("POLYGON ((1 1, 3 1, 3 3, 1 3, 1 1))", 4326);
+    assertTrue(Functions.intersects(g1, g2));
+  }
+
+  @Test
+  public void intersects_disjointPolygons() throws ParseException {
+    Geography g1 = Constructors.geogFromWKT("POLYGON ((0 0, 1 0, 1 1, 0 1, 0 0))", 4326);
+    Geography g2 = Constructors.geogFromWKT("POLYGON ((10 10, 11 10, 11 11, 10 11, 10 10))", 4326);
+    assertFalse(Functions.intersects(g1, g2));
+  }
+
+  @Test
+  public void intersects_pointInPolygon() throws ParseException {
+    Geography g1 = Constructors.geogFromWKT("POLYGON ((0 0, 2 0, 2 2, 0 2, 0 0))", 4326);
+    Geography g2 = Constructors.geogFromWKT("POINT (1 1)", 4326);
+    assertTrue(Functions.intersects(g1, g2));
+  }
+
+  @Test
+  public void intersects_pointToPoint_samePoint() throws ParseException {
+    // Exercises the point-to-point fast path (no ShapeIndex built on either side)
+    Geography g1 = Constructors.geogFromWKT("POINT (1 2)", 4326);
+    Geography g2 = Constructors.geogFromWKT("POINT (1 2)", 4326);
+    assertTrue(Functions.intersects(g1, g2));
+  }
+
+  @Test
+  public void intersects_pointToPoint_differentPoints() throws ParseException {
+    Geography g1 = Constructors.geogFromWKT("POINT (1 2)", 4326);
+    Geography g2 = Constructors.geogFromWKT("POINT (3 4)", 4326);
+    assertFalse(Functions.intersects(g1, g2));
+  }
+
+  @Test
+  public void intersects_pointOnLinestring() throws ParseException {
+    // Exercises the point-to-complex fast path
+    Geography line = Constructors.geogFromWKT("LINESTRING (0 0, 2 0)", 4326);
+    Geography pt = Constructors.geogFromWKT("POINT (1 0)", 4326);
+    assertTrue(Functions.intersects(line, pt));
+    assertTrue(Functions.intersects(pt, line));
+  }
+
+  @Test
+  public void intersects_pointOffLinestring() throws ParseException {
+    Geography line = Constructors.geogFromWKT("LINESTRING (0 0, 2 0)", 4326);
+    Geography pt = Constructors.geogFromWKT("POINT (5 5)", 4326);
+    assertFalse(Functions.intersects(line, pt));
+    assertFalse(Functions.intersects(pt, line));
+  }
+
+  @Test
+  public void intersects_nullHandling() throws ParseException {
+    Geography g = Constructors.geogFromWKT("POINT (1 1)", 4326);
+    assertFalse(Functions.intersects(g, null));
+    assertFalse(Functions.intersects(null, g));
+    assertFalse(Functions.intersects(null, null));
+  }
+
   @Test
   public void contains_nullHandling() throws ParseException {
     Geography g1 = Constructors.geogFromWKT("POINT (1 1)", 4326);

docs/api/sql/geography/Geography-Functions.md

@@ -53,6 +53,7 @@ These functions operate on geography type objects.
 | [ST_Distance](Geography-Functions/ST_Distance.md) | Double | Return the minimum geodesic distance between two geographies in meters. | v1.9.0 |
 | [ST_Length](Geography-Functions/ST_Length.md) | Double | Return the spherical length of a geography in meters, summed along great-circle edges. | v1.9.1 |
 | [ST_Contains](Geography-Functions/ST_Contains.md) | Boolean | Test whether geography A fully contains geography B. | v1.9.0 |
+| [ST_Intersects](Geography-Functions/ST_Intersects.md) | Boolean | Test whether two geographies intersect. | v1.9.1 |
 | [ST_DWithin](Geography-Functions/ST_DWithin.md) | Boolean | Test whether two geographies are within a given geodesic distance (in meters) of each other. | v1.9.1 |
 | [ST_Within](Geography-Functions/ST_Within.md) | Boolean | Test whether geography A is fully within geography B. | v1.9.1 |
 | [ST_Equals](Geography-Functions/ST_Equals.md) | Boolean | Test whether two geographies are spatially equal. | v1.9.1 |

docs/api/sql/geography/Geography-Functions/ST_Intersects.md

@@ -0,0 +1,65 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements.  See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership.  The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ -->
+
+# ST_Intersects
+
+Introduction: Tests whether two geography objects intersect on the sphere using S2 spherical boolean operations. Returns `true` if `A` and `B` share any portion of space (including a single boundary point), and `false` if they are fully disjoint.
+
+Edges are interpreted as great-circle arcs, so the test is correct even when geographies cross the antimeridian or wrap around the poles — situations where a planar `ST_Intersects` would be wrong.
+
+![ST_Intersects returning true](../../../../image/ST_Intersects_geography/ST_Intersects_geography_true.svg "ST_Intersects returning true")
+![ST_Intersects returning false](../../../../image/ST_Intersects_geography/ST_Intersects_geography_false.svg "ST_Intersects returning false")
+
+Format:
+
+`ST_Intersects (A: Geography, B: Geography)`
+
+Return type: `Boolean`
+
+Since: `v1.9.1`
+
+SQL Example — overlapping polygons:
+
+```sql
+SELECT ST_Intersects(
+  ST_GeogFromWKT('POLYGON ((0 0, 2 0, 2 2, 0 2, 0 0))', 4326),
+  ST_GeogFromWKT('POLYGON ((1 1, 3 1, 3 3, 1 3, 1 1))', 4326)
+);
+```
+
+Output:
+
+```
+true
+```
+
+SQL Example — disjoint polygons:
+
+```sql
+SELECT ST_Intersects(
+  ST_GeogFromWKT('POLYGON ((0 0, 1 0, 1 1, 0 1, 0 0))', 4326),
+  ST_GeogFromWKT('POLYGON ((10 10, 11 10, 11 11, 10 11, 10 10))', 4326)
+);
+```
+
+Output:
+
+```
+false
+```

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

@@ -96,12 +96,9 @@ private[apache] case class ST_Contains(inputExpressions: Seq[Expression])
  * @param inputExpressions
  */
 private[apache] case class ST_Intersects(inputExpressions: Seq[Expression])
-    extends ST_Predicate
-    with CodegenFallback {
-
-  override def evalGeom(leftGeometry: Geometry, rightGeometry: Geometry): Boolean = {
-    Predicates.intersects(leftGeometry, rightGeometry)
-  }
+    extends InferredExpression(
+      inferrableFunction2(Predicates.intersects),
+      inferrableFunction2(org.apache.sedona.common.geography.Functions.intersects)) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)

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

@@ -61,8 +61,8 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy {
   //   * ST_Contains — broadcast joins route GeographyUDT inputs through a dedicated index/refine
   //     path (see SpatialIndexExec.geographyShape / BroadcastIndexJoinExec.geographyShape). The
   //     partition/range path still falls back to row-by-row evaluation.
-  //   * ST_Within / ST_Equals — no broadcast index path yet (the Geography refiner is
-  //     ST_Contains-specific), so we gate Geography inputs at the matcher (via
+  //   * ST_Intersects / ST_Within / ST_Equals — no broadcast index path yet (the Geography
+  //     refiner is ST_Contains-specific), so we gate Geography inputs at the matcher (via
   //     `inferredJoinDetection`) and let Spark evaluate the predicate row-by-row.
   // Other ST_Predicates reject Geography inputs at analysis time, so no guard is needed there.
   private def isGeographyInput(shape: Expression): Boolean =
@@ -98,16 +98,6 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy {
       predicate: ST_Predicate,
       extraCondition: Option[Expression] = None): Option[JoinQueryDetection] = {
     predicate match {
-      case ST_Intersects(Seq(leftShape, rightShape)) =>
-        Some(
-          JoinQueryDetection(
-            left,
-            right,
-            leftShape,
-            rightShape,
-            SpatialPredicate.INTERSECTS,
-            false,
-            extraCondition))
       case ST_Covers(Seq(leftShape, rightShape)) =>
         Some(
           JoinQueryDetection(
@@ -217,9 +207,9 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy {
       val queryDetection: Option[JoinQueryDetection] = condition.flatMap {
         case joinConditionMatcher(predicate, extraCondition) =>
           predicate match {
-            // ST_Contains / ST_Equals / ST_Within are InferredExpression (not ST_Predicate) so
-            // they can't sit inside getJoinDetection; they're also the only predicates currently
-            // accepting Geography inputs.
+            // ST_Contains / ST_Intersects / ST_Within / ST_Equals are InferredExpression (not
+            // ST_Predicate) so they can't sit inside getJoinDetection; they're also the only
+            // predicates currently accepting Geography inputs.
             //
             // ST_Contains: when either operand is GeographyUDT we still detect the join here and
             // set `geographyShape = true`; planBroadcastJoin will route the work to the
@@ -238,9 +228,17 @@ class JoinQueryDetector(sparkSession: SparkSession) extends SparkStrategy {
                   isGeography = false,
                   extraCondition,
                   geographyShape = geographyShape))
-            // ST_Within / ST_Equals on Geography have no broadcast index path yet (the Geography
-            // refiner is ST_Contains-specific), so gate Geography inputs and let them fall back
-            // to row-by-row evaluation.
+            // ST_Intersects / ST_Within / ST_Equals on Geography have no broadcast index path
+            // yet (the Geography refiner is ST_Contains-specific), so gate Geography inputs and
+            // let them fall back to row-by-row evaluation.
+            case ST_Intersects(Seq(leftShape, rightShape)) =>
+              inferredJoinDetection(
+                left,
+                right,
+                leftShape,
+                rightShape,
+                SpatialPredicate.INTERSECTS,
+                extraCondition)
             case ST_Within(Seq(leftShape, rightShape)) =>
               inferredJoinDetection(
                 left,