[GH-2971] Add ST_DWithin(Box2D, Box2D, distance) overload

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

@@ -134,6 +134,25 @@ public static boolean dWithin(
     }
   }
 
+  /**
+   * Closed-interval planar distance test between two Box2D rectangles. Returns true if the minimum
+   * Euclidean distance between the rectangles is less than or equal to {@code distance}.
+   *
+   * <p>Overlapping or edge/corner-touching boxes have distance 0 and therefore match for any {@code
+   * distance >= 0}. Inverted bounds throw for the same reason {@link #boxIntersects(Box2D, Box2D)}
+   * does — planar predicates have no defined meaning on inverted intervals.
+   */
+  public static boolean dWithin(Box2D a, Box2D b, double distance) {
+    requireOrderedPlanarBox(a, "a");
+    requireOrderedPlanarBox(b, "b");
+    double dx = Math.max(0.0, Math.max(a.getXMin() - b.getXMax(), b.getXMin() - a.getXMax()));
+    double dy = Math.max(0.0, Math.max(a.getYMin() - b.getYMax(), b.getYMin() - a.getYMax()));
+    // Compare squared distance to avoid a sqrt; bail out fast if either delta already exceeds
+    // the supplied radius.
+    if (dx > distance || dy > distance) return false;
+    return dx * dx + dy * dy <= distance * distance;
+  }
+
   public static String relate(Geometry leftGeometry, Geometry rightGeometry) {
     return RelateOp.relate(leftGeometry, rightGeometry).toString();
   }

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

@@ -65,6 +65,37 @@ public void testBoxContains() {
     assertFalse(Predicates.boxContains(outer, new Box2D(5.0, 0.0, 11.0, 5.0)));
   }
 
+  @Test
+  public void testDWithinBox2D() {
+    Box2D a = new Box2D(0.0, 0.0, 10.0, 10.0);
+
+    // Overlapping → distance 0, matches any non-negative radius.
+    assertTrue(Predicates.dWithin(a, new Box2D(5.0, 5.0, 15.0, 15.0), 0.0));
+    // Edge-touching → distance 0 (closed-interval).
+    assertTrue(Predicates.dWithin(a, new Box2D(10.0, 0.0, 20.0, 10.0), 0.0));
+    // Corner-touching diagonally → distance 0.
+    assertTrue(Predicates.dWithin(a, new Box2D(10.0, 10.0, 20.0, 20.0), 0.0));
+    // Separated by 1 on X only → distance 1.0.
+    Box2D rightOf = new Box2D(11.0, 0.0, 20.0, 10.0);
+    assertTrue(Predicates.dWithin(a, rightOf, 1.0));
+    assertFalse(Predicates.dWithin(a, rightOf, 0.999));
+    // Separated by (3, 4) → distance 5 (Pythagorean).
+    Box2D diagonal = new Box2D(13.0, 14.0, 20.0, 20.0);
+    assertTrue(Predicates.dWithin(a, diagonal, 5.0));
+    assertFalse(Predicates.dWithin(a, diagonal, 4.999));
+    // Negative radius never matches, even for overlapping boxes.
+    assertFalse(Predicates.dWithin(a, a, -1.0));
+  }
+
+  @Test
+  public void testDWithinBox2DRejectInvertedBounds() {
+    Box2D normal = new Box2D(0.0, 0.0, 5.0, 5.0);
+    Box2D wrapX = new Box2D(170.0, 10.0, -170.0, 20.0);
+    IllegalArgumentException ex =
+        assertThrows(IllegalArgumentException.class, () -> Predicates.dWithin(wrapX, normal, 1.0));
+    assertTrue(ex.getMessage().contains("inverted bounds"));
+  }
+
   @Test
   public void testBoxPredicatesRejectInvertedBounds() {
     // Box2D allows xmin > xmax (reserved for future antimeridian wraparound); planar predicates

docs/api/sql/box2d/Box2D-Functions.md

@@ -55,6 +55,7 @@ The same `ST_XMin` / `ST_YMin` / `ST_XMax` / `ST_YMax` functions also accept `Ge
 | :--- | :--- | :--- | :--- |
 | [ST_BoxIntersects](Box2D-Predicates/ST_BoxIntersects.md) | Boolean | Closed-interval bbox intersection over two Box2D arguments. Matches PostGIS `&&` on `box2d`. | v1.9.1 |
 | [ST_BoxContains](Box2D-Predicates/ST_BoxContains.md) | Boolean | Closed-interval bbox containment over two Box2D arguments. Matches PostGIS `~` on `box2d`. | v1.9.1 |
+| [ST_DWithin](Box2D-Predicates/ST_DWithin.md) | Boolean | Closed-interval planar distance test between two Box2D rectangles. | v1.9.1 |
 
 ## Box2D Functions
 

docs/api/sql/box2d/Box2D-Predicates/ST_DWithin.md

@@ -0,0 +1,55 @@
+<!--
+ 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_DWithin
+
+Introduction: Closed-interval planar distance test between two `Box2D` rectangles. Returns `true` if the minimum Euclidean distance between `a` and `b` is less than or equal to `distance`.
+
+Overlapping or edge/corner-touching boxes have distance `0` and therefore match for any non-negative radius. This is the `Box2D` overload of `ST_DWithin`; the [Geometry input form](../../Predicates/ST_DWithin.md) and the [Geography input form](../../geography/Geography-Functions/ST_DWithin.md) handle the other types.
+
+Format: `ST_DWithin(a: Box2D, b: Box2D, distance: Double)`
+
+Return type: `Boolean`
+
+Since: `v1.9.1`
+
+SQL Example
+
+```sql
+SELECT ST_DWithin(
+    ST_MakeBox2D(ST_Point(0.0, 0.0), ST_Point(10.0, 10.0)),
+    ST_MakeBox2D(ST_Point(11.0, 0.0), ST_Point(12.0, 1.0)),
+    1.0)
+```
+
+Output:
+
+```
+true
+```
+
+## Optimization
+
+`ST_DWithin(box_a, box_b, distance)` between two `Box2D` columns routes through Sedona's distance-join planner: the broadcast-index path (with a hinted side) or the partition-based `DistanceJoinExec`. The Box2D values materialise as rectangular polygons at the join boundary, after which the existing distance-expansion + index-probe + refine pipeline runs unchanged. See [Box2D spatial join](../../Optimizer.md#box2d-spatial-join).
+
+## Errors
+
+Throws `IllegalArgumentException` if either box has inverted bounds (`xmin > xmax` or `ymin > ymax`). Inverted-bound values are reserved for a future antimeridian-wraparound semantics.
+
+Returns `NULL` if any argument is `NULL`.

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

@@ -315,9 +315,14 @@ private[apache] case class ST_OrderingEquals(inputExpressions: Seq[Expression])
 
 private[apache] case class ST_DWithin(inputExpressions: Seq[Expression])
     extends InferredExpression(
-      inferrableFunction3(Predicates.dWithin),
+      inferrableFunction3((l: Geometry, r: Geometry, d: Double) => Predicates.dWithin(l, r, d)),
       inferrableFunction4(Predicates.dWithin),
-      inferrableFunction3(org.apache.sedona.common.geography.Functions.dWithin)) {
+      inferrableFunction3(org.apache.sedona.common.geography.Functions.dWithin),
+      inferrableFunction3(
+        (
+            a: org.apache.sedona.common.geometryObjects.Box2D,
+            b: org.apache.sedona.common.geometryObjects.Box2D,
+            distance: Double) => Predicates.dWithin(a, b, distance))) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)

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

@@ -20,7 +20,7 @@ package org.apache.sedona.sql
 
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions.{broadcast, expr}
-import org.apache.spark.sql.sedona_sql.strategy.join.{BroadcastIndexJoinExec, RangeJoinExec}
+import org.apache.spark.sql.sedona_sql.strategy.join.{BroadcastIndexJoinExec, DistanceJoinExec, RangeJoinExec}
 
 class Box2DJoinSuite extends TestBaseScala {
 
@@ -184,6 +184,73 @@ class Box2DJoinSuite extends TestBaseScala {
     }
   }
 
+  describe("Box2D distance join") {
+
+    /**
+     * Boxes wired so the L↔R distances are exactly predictable:
+     *   - L1=(0,0,10,10), R1=(11,0,12,1) → distance 1.0 (separated by 1 on X)
+     *   - L1=(0,0,10,10), R2=(5,15,8,18) → distance 5.0 (separated by 5 on Y)
+     *   - L2=(0,0,1,1), R1=(11,0,12,1) → distance 10.0
+     *   - L2=(0,0,1,1), R2=(5,15,8,18) → sqrt(16 + 196) ≈ 14.56
+     */
+    def distLeft: DataFrame = {
+      import sparkSession.implicits._
+      Seq(TestBox(1, 0.0, 0.0, 10.0, 10.0), TestBox(2, 0.0, 0.0, 1.0, 1.0))
+        .toDF("id", "xmin", "ymin", "xmax", "ymax")
+        .selectExpr("id", "ST_MakeBox2D(ST_Point(xmin, ymin), ST_Point(xmax, ymax)) AS box")
+    }
+
+    def distRight: DataFrame = {
+      import sparkSession.implicits._
+      Seq(TestBox(11, 11.0, 0.0, 12.0, 1.0), TestBox(12, 5.0, 15.0, 8.0, 18.0))
+        .toDF("id", "xmin", "ymin", "xmax", "ymax")
+        .selectExpr("id", "ST_MakeBox2D(ST_Point(xmin, ymin), ST_Point(xmax, ymax)) AS box")
+    }
+
+    it("ST_DWithin on Box2D inputs: broadcast index join, radius 1.0") {
+      val df = distLeft
+        .alias("L")
+        .join(broadcast(distRight.alias("R")), expr("ST_DWithin(L.box, R.box, 1.0)"))
+      assert(
+        df.queryExecution.sparkPlan.collect { case b: BroadcastIndexJoinExec => b }.size == 1,
+        "Expected BroadcastIndexJoinExec in the plan")
+      // Only (L1, R1) is within 1.0.
+      assert(df.count() == 1)
+    }
+
+    it("ST_DWithin on Box2D inputs: broadcast index join, radius 6.0") {
+      val df = distLeft
+        .alias("L")
+        .join(broadcast(distRight.alias("R")), expr("ST_DWithin(L.box, R.box, 6.0)"))
+      // (L1, R1) distance=1 ✓, (L1, R2) distance=5 ✓; (L2, R1) distance=10 ✗, (L2, R2) ≈14.56 ✗.
+      assert(df.count() == 2)
+    }
+
+    it("ST_DWithin on Box2D inputs: distance join (non-broadcast) produces the same count") {
+      val df = distLeft
+        .alias("L")
+        .join(distRight.alias("R"), expr("ST_DWithin(L.box, R.box, 6.0)"))
+      assert(
+        df.queryExecution.sparkPlan.collect { case d: DistanceJoinExec => d }.size == 1,
+        "Expected DistanceJoinExec in the plan")
+      assert(df.count() == 2)
+    }
+
+    it("ST_DWithin on Box2D inputs: zero radius matches only edge/corner-touching pairs") {
+      import sparkSession.implicits._
+      val touching = Seq(TestBox(1, 0.0, 0.0, 10.0, 10.0))
+        .toDF("id", "xmin", "ymin", "xmax", "ymax")
+        .selectExpr("id", "ST_MakeBox2D(ST_Point(xmin, ymin), ST_Point(xmax, ymax)) AS box")
+      val adjacent = Seq(TestBox(11, 10.0, 0.0, 20.0, 10.0)) // shares edge x=10
+        .toDF("id", "xmin", "ymin", "xmax", "ymax")
+        .selectExpr("id", "ST_MakeBox2D(ST_Point(xmin, ymin), ST_Point(xmax, ymax)) AS box")
+      val df = touching
+        .alias("L")
+        .join(broadcast(adjacent.alias("R")), expr("ST_DWithin(L.box, R.box, 0.0)"))
+      assert(df.count() == 1)
+    }
+  }
+
 }
 
 object Box2DJoinSuite {