[SEDONA-334] Add RS_ScaleX and RS_ScaleY (#932)

common/src/main/java/org/apache/sedona/common/raster/RasterAccessors.java

@@ -19,6 +19,8 @@
 package org.apache.sedona.common.raster;
 
 import org.geotools.coverage.grid.GridCoverage2D;
+import org.geotools.coverage.grid.GridGeometry2D;
+import org.geotools.coverage.processing.operation.Affine;
 import org.geotools.geometry.Envelope2D;
 import org.geotools.referencing.CRS;
 import org.geotools.referencing.crs.DefaultEngineeringCRS;
@@ -62,6 +64,23 @@ public static int getHeight(GridCoverage2D raster) {
     }
 
 
+    public static double getScaleX(GridCoverage2D raster) {
+        return getAffineTransform(raster).getScaleX();
+    }
+
+    public static double getScaleY(GridCoverage2D raster) {
+        return getAffineTransform(raster).getScaleY();
+    }
+
+    private static AffineTransform2D getAffineTransform(GridCoverage2D raster) throws UnsupportedOperationException {
+        GridGeometry2D gridGeometry2D = raster.getGridGeometry();
+        MathTransform crsTransform = gridGeometry2D.getGridToCRS2D();
+        if (!(crsTransform instanceof AffineTransform2D)) {
+            throw new UnsupportedOperationException("Only AffineTransform2D is supported");
+        }
+        return (AffineTransform2D) crsTransform;
+    }
+
 
     public static Geometry envelope(GridCoverage2D raster) throws FactoryException {
         Envelope2D envelope2D = raster.getEnvelope2D();

common/src/test/java/org/apache/sedona/common/raster/RasterAccessorsTest.java

@@ -56,6 +56,18 @@ public void testSrid() throws FactoryException {
         assertEquals(4326, RasterAccessors.srid(multiBandRaster));
     }
 
+    @Test
+    public void testScaleX() throws UnsupportedOperationException, FactoryException {
+        GridCoverage2D emptyRaster = RasterConstructors.makeEmptyRaster(2, 10, 15, 0, 0, 1, 2, 0, 0, 0);
+        assertEquals(1, RasterAccessors.getScaleX(emptyRaster), 1e-9);
+    }
+
+    @Test
+    public void testScaleY() throws UnsupportedOperationException, FactoryException {
+        GridCoverage2D emptyRaster = RasterConstructors.makeEmptyRaster(2, 10, 15, 0, 0, 1, 2, 0, 0, 0);
+        assertEquals(-2, RasterAccessors.getScaleY(emptyRaster), 1e-9);
+    }
+
     @Test
     public void testMetaData()
             throws FactoryException

docs/api/sql/Raster-operators.md

@@ -18,6 +18,52 @@ Output:
 512
 ```
 
+### RS_ScaleX
+
+Introduction: Returns the pixel width of the raster in CRS units.
+!!!Note
+    RS_ScaleX attempts to get an Affine transform on the grid in order to return scaleX (See [World File](https://en.wikipedia.org/wiki/World_file) for more details). If the transform on the geometry is not an Affine transform, RS_ScaleX will throw an UnsupportedException:
+    ```
+    UnsupportedOperationException("Only AffineTransform2D is supported")
+    ```
+
+Format: `RS_ScaleX(raster: Raster)`
+
+Since: `1.5.0`
+
+Spark SQL example:
+```sql
+SELECT RS_ScaleX(raster) FROM rasters
+```
+
+Output:
+```
+1
+```
+
+### RS_ScaleY
+
+Introduction: Returns the pixel height of the raster in CRS units.
+!!!Note
+    RS_ScaleY attempts to get an Affine transform on the grid in order to return scaleX (See [World File](https://en.wikipedia.org/wiki/World_file) for more details). If the transform on the geometry is not an Affine transform, RS_ScaleY will throw an UnsupportedException:
+    ```
+    UnsupportedOperationException("Only AffineTransform2D is supported")
+    ```
+
+Format: `RS_ScaleY(raster: Raster)`
+
+Since: `1.5.0`
+
+Spark SQL example:
+```sql
+SELECT RS_ScaleY(raster) FROM rasters
+```
+
+Output:
+```
+-2
+```
+
 ### RS_Width
 
 Introduction: Returns the width of the raster.

sql/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala

@@ -205,7 +205,9 @@ object Catalog {
     function[RS_AsGeoTiff](),
     function[RS_AsArcGrid](),
     function[RS_Width](),
-    function[RS_Height]()
+    function[RS_Height](),
+    function[RS_ScaleX](),
+    function[RS_ScaleY]()
   )
 
   val aggregateExpressions: Seq[Aggregator[Geometry, Geometry, Geometry]] = Seq(

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

@@ -57,4 +57,18 @@ case class RS_Height(inputExpressions: Seq[Expression]) extends InferredExpressi
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)
   }
-}
+}
+
+
+case class RS_ScaleX(inputExpressions: Seq[Expression]) extends InferredExpression(RasterAccessors.getScaleX _) {
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
+case class RS_ScaleY(inputExpressions: Seq[Expression]) extends InferredExpression(RasterAccessors.getScaleY _) {
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}
+

sql/common/src/test/scala/org/apache/sedona/sql/rasteralgebraTest.scala

@@ -488,5 +488,19 @@ class rasteralgebraTest extends TestBaseScala with BeforeAndAfter with GivenWhen
       raster = df.collect().head.getAs[GridCoverage2D](0)
       assert(raster.getNumSampleDimensions == 2)
     }
+
+    it("Passed RS_ScaleX with raster") {
+      val df = sparkSession.read.format("binaryFile").load(resourceFolder + "raster/test1.tiff")
+      val result = df.selectExpr("RS_ScaleX(RS_FromGeoTiff(content))").first().getDouble(0)
+      val expected: Double = 72.32861272132695
+      assertEquals(expected, result, 1e-9)
+    }
+
+    it("Passed RS_ScaleY with raster") {
+      val df = sparkSession.read.format("binaryFile").load(resourceFolder + "raster/test1.tiff")
+      val result = df.selectExpr("RS_ScaleY(RS_FromGeoTiff(content))").first().getDouble(0)
+      val expected: Double = -72.32861272132695
+      assertEquals(expected, result, 1e-9)
+    }
   }
 }