[SEDONA-354] Add RS_SkewX and RS_SkewY (#960)

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

@@ -201,6 +201,29 @@ public void testGridCoordXLonLat() throws FactoryException, TransformException {
         assertEquals(expectedX, actualX);
     }
 
+    @Test
+    public void testSkewX() throws FactoryException {
+        GridCoverage2D emptyRaster = RasterConstructors.makeEmptyRaster(10, 2, 4, 6, 4, 1, 1, 2, 7, 0);
+        assertEquals(2, RasterAccessors.getSkewX(emptyRaster), 0.1d);
+
+        emptyRaster = RasterConstructors.makeEmptyRaster(1, 3, 4, 100.0, 200.0,2.0, -3.0, 0.1, 0.2,0 );
+        assertEquals(0.1, RasterAccessors.getSkewX(emptyRaster), 0.01d);
+
+        emptyRaster = RasterConstructors.makeEmptyRaster(1, 5, 5, -53, 51, 5);
+        assertEquals(0, RasterAccessors.getSkewX(emptyRaster), 0.1d);
+    }
+
+    @Test
+    public void testSkewY() throws FactoryException {
+        GridCoverage2D emptyRaster = RasterConstructors.makeEmptyRaster(10, 2, 4, 6, 4, 1, 1, 2, 7, 0);
+        assertEquals(7, RasterAccessors.getSkewY(emptyRaster), 0.1d);
+
+        emptyRaster = RasterConstructors.makeEmptyRaster(1, 3, 4, 100.0, 200.0,2.0, -3.0, 0.1, 0.2,0 );
+        assertEquals(0.2, RasterAccessors.getSkewY(emptyRaster), 0.01d);
+
+        emptyRaster = RasterConstructors.makeEmptyRaster(1, 5, 5, -53, 51, 5);
+        assertEquals(0, RasterAccessors.getSkewY(emptyRaster), 0.1d);
+    }
 
     @Test
     public void testGridCoordXGeomSameSRID()  throws FactoryException, TransformException {

docs/api/sql/Raster-operators.md

@@ -172,6 +172,46 @@ Output:
 -2
 ```
 
+### RS_SkewX
+
+Introduction: Returns the X skew or rotation parameter.
+
+Format: `RS_SkewX(raster: Raster)`
+
+Since: `v1.5.0`
+
+Spark SQL Example:
+
+```sql
+SELECT RS_SkewX(RS_MakeEmptyRaster(2, 10, 10, 0.0, 0.0, 1.0, -1.0, 0.1, 0.2, 4326))
+```
+
+Output:
+
+```
+0.1
+```
+
+### RS_SkewY
+
+Introduction: Returns the Y skew or rotation parameter.
+
+Format: `RS_SkewY(raster: Raster)`
+
+Since: `v1.5.0`
+
+Spark SQL Example:
+
+```sql
+SELECT RS_SkewY(RS_MakeEmptyRaster(2, 10, 10, 0.0, 0.0, 1.0, -1.0, 0.1, 0.2, 4326))
+```
+
+Output:
+
+```
+0.2
+```
+
 ### RS_UpperLeftX
 
 Introduction: Returns the X coordinate of the upper-left corner of the raster.

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

@@ -214,6 +214,8 @@ object Catalog {
     function[RS_UpperLeftY](),
     function[RS_ScaleX](),
     function[RS_ScaleY](),
+    function[RS_SkewX](),
+    function[RS_SkewY](),
     function[RS_PixelAsPoint](),
     function[RS_ConvexHull](),
     function[RS_RasterToWorldCoordX](),

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

@@ -78,6 +78,18 @@ case class RS_ScaleY(inputExpressions: Seq[Expression]) extends InferredExpressi
   }
 }
 
+case class RS_SkewX(inputExpressions: Seq[Expression]) extends InferredExpression(RasterAccessors.getSkewX _) {
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
+case class RS_SkewY(inputExpressions: Seq[Expression]) extends InferredExpression(RasterAccessors.getSkewY _) {
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
 case class RS_RasterToWorldCoordX(inputExpressions: Seq[Expression]) extends InferredExpression(RasterAccessors.getWorldCoordX _) {
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)

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

@@ -432,6 +432,20 @@ class rasteralgebraTest extends TestBaseScala with BeforeAndAfter with GivenWhen
       assertEquals(expected, result, 1e-8)
     }
 
+    it("Passed RS_SkewX") {
+      val df = sparkSession.read.format("binaryFile").load(resourceFolder + "raster/test1.tiff")
+      val result = df.selectExpr("RS_SkewX(RS_FromGeoTiff(content))").first().getDouble(0)
+      val expected: Double = 0.0
+      assertEquals(expected, result, 0.1)
+    }
+
+    it("Passed RS_SkewY") {
+      val df = sparkSession.read.format("binaryFile").load(resourceFolder + "raster/test1.tiff")
+      val result = df.selectExpr("RS_SkewY(RS_FromGeoTiff(content))").first().getDouble(0)
+      val expected: Double = 0.0
+      assertEquals(expected, result, 0.1)
+    }
+
     it("Passed RS_Metadata") {
       val df = sparkSession.read.format("binaryFile").load(resourceFolder + "raster/test1.tiff")
       val result = df.selectExpr("RS_Metadata(RS_FromGeoTiff(content))").first().getSeq(0)