[GH-2882] Overload ST_XMin/XMax/YMin/YMax for Box2D

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

@@ -699,6 +699,10 @@ public static Double xMin(Geometry geometry) {
     return min == Double.MAX_VALUE ? null : min;
   }
 
+  public static Double xMin(Box2D box) {
+    return box == null ? null : box.getXMin();
+  }
+
   public static Double xMax(Geometry geometry) {
     Coordinate[] points = geometry.getCoordinates();
     double max = -Double.MAX_VALUE;
@@ -708,6 +712,10 @@ public static Double xMax(Geometry geometry) {
     return max == -Double.MAX_VALUE ? null : max;
   }
 
+  public static Double xMax(Box2D box) {
+    return box == null ? null : box.getXMax();
+  }
+
   public static Double yMin(Geometry geometry) {
     Coordinate[] points = geometry.getCoordinates();
     double min = Double.MAX_VALUE;
@@ -717,6 +725,10 @@ public static Double yMin(Geometry geometry) {
     return min == Double.MAX_VALUE ? null : min;
   }
 
+  public static Double yMin(Box2D box) {
+    return box == null ? null : box.getYMin();
+  }
+
   public static Double yMax(Geometry geometry) {
     Coordinate[] points = geometry.getCoordinates();
     double max = -Double.MAX_VALUE;
@@ -726,6 +738,10 @@ public static Double yMax(Geometry geometry) {
     return max == -Double.MAX_VALUE ? null : max;
   }
 
+  public static Double yMax(Box2D box) {
+    return box == null ? null : box.getYMax();
+  }
+
   public static Double zMax(Geometry geometry) {
     Coordinate[] points = geometry.getCoordinates();
     double max = -Double.MAX_VALUE;

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

@@ -19,6 +19,7 @@
 package org.apache.spark.sql.sedona_sql.expressions
 
 import org.apache.sedona.common.{Functions, FunctionsGeoTools, FunctionsProj4}
+import org.apache.sedona.common.geometryObjects.Box2D
 import org.apache.sedona.common.sphere.{Haversine, Spheroid}
 import org.apache.sedona.common.utils.{InscribedCircle, ValidDetail}
 import org.apache.sedona.core.utils.SedonaConf
@@ -69,15 +70,19 @@ private[apache] case class ST_Distance(inputExpressions: Seq[Expression])
 }
 
 private[apache] case class ST_YMax(inputExpressions: Seq[Expression])
-    extends InferredExpression(Functions.yMax _) {
+    extends InferredExpression(
+      inferrableFunction1((g: Geometry) => Functions.yMax(g)),
+      inferrableFunction1((b: Box2D) => Functions.yMax(b))) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)
   }
 }
 
 private[apache] case class ST_YMin(inputExpressions: Seq[Expression])
-    extends InferredExpression(Functions.yMin _) {
+    extends InferredExpression(
+      inferrableFunction1((g: Geometry) => Functions.yMin(g)),
+      inferrableFunction1((b: Box2D) => Functions.yMin(b))) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)
@@ -1489,7 +1494,9 @@ private[apache] case class ST_IsEmpty(inputExpressions: Seq[Expression])
  * @param inputExpressions
  */
 private[apache] case class ST_XMax(inputExpressions: Seq[Expression])
-    extends InferredExpression(Functions.xMax _) {
+    extends InferredExpression(
+      inferrableFunction1((g: Geometry) => Functions.xMax(g)),
+      inferrableFunction1((b: Box2D) => Functions.xMax(b))) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)
@@ -1502,7 +1509,9 @@ private[apache] case class ST_XMax(inputExpressions: Seq[Expression])
  * @param inputExpressions
  */
 private[apache] case class ST_XMin(inputExpressions: Seq[Expression])
-    extends InferredExpression(Functions.xMin _) {
+    extends InferredExpression(
+      inferrableFunction1((g: Geometry) => Functions.xMin(g)),
+      inferrableFunction1((b: Box2D) => Functions.xMin(b))) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)

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

@@ -217,6 +217,8 @@ object InferredTypes {
       expr.toGeography(input)
     } else if (t =:= typeOf[Array[Geography]]) { expr => input =>
       expr.toGeographyArray(input)
+    } else if (t =:= typeOf[Box2D]) { expr => input =>
+      expr.toBox2D(input)
     } else if (InferredRasterExpression.isRasterType(t)) {
       InferredRasterExpression.rasterExtractor
     } else if (t =:= typeOf[Array[Double]]) { expr => input =>

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

@@ -19,6 +19,7 @@
 package org.apache.spark.sql.sedona_sql.expressions
 
 import org.apache.sedona.common.S2Geography.{Geography, GeographyWKBSerializer}
+import org.apache.sedona.common.geometryObjects.Box2D
 import org.apache.sedona.sql.utils.GeometrySerializer
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.Expression
@@ -73,6 +74,19 @@ object implicits {
       }
     }
 
+    def toBox2D(input: InternalRow): Box2D = {
+      inputExpression match {
+        case serdeAware: SerdeAware =>
+          serdeAware.evalWithoutSerialization(input).asInstanceOf[Box2D]
+        case _ =>
+          inputExpression.eval(input) match {
+            case row: InternalRow =>
+              new Box2D(row.getDouble(0), row.getDouble(1), row.getDouble(2), row.getDouble(3))
+            case _ => null
+          }
+      }
+    }
+
     def toGeographyArray(input: InternalRow): Array[Geography] = {
       inputExpression match {
         case aware: SerdeAware =>

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

@@ -274,6 +274,28 @@ class functionTestScala
       assert(test.take(1)(0).get(0).asInstanceOf[Double] == -3.0)
     }
 
+    it("Passed ST_XMin / XMax / YMin / YMax for Box2D") {
+      val df = sparkSession.sql("""
+        WITH t AS (
+          SELECT ST_Box2D(ST_GeomFromText('POLYGON((1 2, 1 5, 4 5, 4 2, 1 2))')) AS bbox,
+                 ST_Box2D(ST_GeomFromText(NULL))                                  AS bbox_null
+        )
+        SELECT
+          ST_XMin(bbox),  ST_YMin(bbox),  ST_XMax(bbox),  ST_YMax(bbox),
+          ST_XMin(bbox_null), ST_YMin(bbox_null), ST_XMax(bbox_null), ST_YMax(bbox_null)
+        FROM t
+      """)
+      val row = df.collect()(0)
+      assert(row.getDouble(0) == 1.0)
+      assert(row.getDouble(1) == 2.0)
+      assert(row.getDouble(2) == 4.0)
+      assert(row.getDouble(3) == 5.0)
+      assert(row.isNullAt(4))
+      assert(row.isNullAt(5))
+      assert(row.isNullAt(6))
+      assert(row.isNullAt(7))
+    }
+
     it("Passed ST_ZMax") {
       val test = sparkSession.sql(
         "SELECT ST_ZMax(ST_GeomFromWKT('POLYGON((0 0 0,0 5 0,5 0 0,0 0 5),(1 1 0,3 1 0,1 3 0,1 1 0))'))")