[GH-2652] Add RS_AsCOG SQL function for Cloud Optimized GeoTiff output (#2669)

Author: jiayuasu

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

@@ -108,6 +108,62 @@ public static byte[] asCloudOptimizedGeoTiff(GridCoverage2D raster, CogOptions o
     }
   }
 
+  // ---------- asCOG overloads (used by RS_AsCOG via InferredExpression) ----------
+
+  public static byte[] asCOG(GridCoverage2D raster) {
+    return asCloudOptimizedGeoTiff(raster, CogOptions.defaults());
+  }
+
+  public static byte[] asCOG(GridCoverage2D raster, String compression) {
+    return asCloudOptimizedGeoTiff(raster, CogOptions.builder().compression(compression).build());
+  }
+
+  public static byte[] asCOG(GridCoverage2D raster, String compression, int tileSize) {
+    return asCloudOptimizedGeoTiff(
+        raster, CogOptions.builder().compression(compression).tileSize(tileSize).build());
+  }
+
+  public static byte[] asCOG(
+      GridCoverage2D raster, String compression, int tileSize, double quality) {
+    return asCloudOptimizedGeoTiff(
+        raster,
+        CogOptions.builder()
+            .compression(compression)
+            .tileSize(tileSize)
+            .compressionQuality(quality)
+            .build());
+  }
+
+  public static byte[] asCOG(
+      GridCoverage2D raster, String compression, int tileSize, double quality, String resampling) {
+    return asCloudOptimizedGeoTiff(
+        raster,
+        CogOptions.builder()
+            .compression(compression)
+            .tileSize(tileSize)
+            .compressionQuality(quality)
+            .resampling(resampling)
+            .build());
+  }
+
+  public static byte[] asCOG(
+      GridCoverage2D raster,
+      String compression,
+      int tileSize,
+      double quality,
+      String resampling,
+      int overviewCount) {
+    return asCloudOptimizedGeoTiff(
+        raster,
+        CogOptions.builder()
+            .compression(compression)
+            .tileSize(tileSize)
+            .compressionQuality(quality)
+            .resampling(resampling)
+            .overviewCount(overviewCount)
+            .build());
+  }
+
   /**
    * Creates a GeoTiff file with the provided raster. Primarily used for testing.
    *

common/src/main/java/org/apache/sedona/common/raster/cog/CogOptions.java

@@ -20,7 +20,6 @@
 
 import java.util.Arrays;
 import java.util.List;
-import java.util.Locale;
 
 /**
  * Options for Cloud Optimized GeoTIFF (COG) generation.
@@ -206,10 +205,19 @@ public CogOptions build() {
       if (compression == null || compression.isEmpty()) {
         throw new IllegalArgumentException("compression must not be null or empty");
       }
-      if (!VALID_COMPRESSION.contains(compression)) {
+      // Preserve the original input for error reporting
+      String originalCompression = compression;
+      // Case-insensitive matching: find the canonical value from the valid list
+      String normalizedCompression = matchIgnoreCase(VALID_COMPRESSION, originalCompression);
+      if (normalizedCompression == null) {
         throw new IllegalArgumentException(
-            "compression must be one of " + VALID_COMPRESSION + ", got: '" + compression + "'");
+            "compression must be one of "
+                + VALID_COMPRESSION
+                + ", got: '"
+                + originalCompression
+                + "'");
       }
+      this.compression = normalizedCompression;
       if (compressionQuality < 0 || compressionQuality > 1.0) {
         throw new IllegalArgumentException(
             "compressionQuality must be between 0.0 and 1.0, got: " + compressionQuality);
@@ -225,27 +233,40 @@ public CogOptions build() {
             "overviewCount must be -1 (auto), 0 (none), or positive, got: " + overviewCount);
       }
 
-      // Normalize resampling to title-case for matching
-      String normalized = normalizeResampling(resampling);
-      if (!VALID_RESAMPLING.contains(normalized)) {
-        throw new IllegalArgumentException(
-            "resampling must be one of " + VALID_RESAMPLING + ", got: '" + resampling + "'");
+      // Case-insensitive matching for resampling; treat null/blank as default (Nearest)
+      if (resampling == null || resampling.isEmpty()) {
+        this.resampling = "Nearest";
+      } else {
+        String originalResampling = resampling;
+        String normalizedResampling = matchIgnoreCase(VALID_RESAMPLING, originalResampling);
+        if (normalizedResampling == null) {
+          throw new IllegalArgumentException(
+              "resampling must be one of "
+                  + VALID_RESAMPLING
+                  + ", got: '"
+                  + originalResampling
+                  + "'");
+        }
+        this.resampling = normalizedResampling;
       }
-      this.resampling = normalized;
 
       return new CogOptions(this);
     }
 
     /**
-     * Normalize the resampling string to title-case (first letter uppercase, rest lowercase) so
-     * callers can pass "nearest", "BILINEAR", etc.
+     * Find the canonical value from a list that matches the input case-insensitively. Returns null
+     * if no match found. This allows callers to pass "lzw", "PACKBITS", "bilinear", etc.
      */
-    private static String normalizeResampling(String value) {
-      if (value == null || value.isEmpty()) {
-        return "Nearest";
+    private static String matchIgnoreCase(List<String> validValues, String input) {
+      if (input == null) {
+        return null;
+      }
+      for (String valid : validValues) {
+        if (valid.equalsIgnoreCase(input)) {
+          return valid;
+        }
       }
-      String lower = value.toLowerCase(Locale.ROOT);
-      return Character.toUpperCase(lower.charAt(0)) + lower.substring(1);
+      return null;
     }
   }
 }

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

@@ -254,4 +254,103 @@ public void testAsImageVariousBandDataType() throws IOException, FactoryExceptio
       }
     }
   }
+
+  // ---- RS_AsCOG / asCOG tests ----
+
+  @Test
+  public void testAsCOGDefaults() throws IOException {
+    GridCoverage2D raster = rasterFromGeoTiff(resourceFolder + "raster/test1.tiff");
+    byte[] cogBytes = RasterOutputs.asCOG(raster);
+    assertNotNull(cogBytes);
+    assertTrue(cogBytes.length >= 2);
+    // Verify it is a valid TIFF (starts with II or MM)
+    assertTrue(
+        (cogBytes[0] == 'I' && cogBytes[1] == 'I') || (cogBytes[0] == 'M' && cogBytes[1] == 'M'));
+  }
+
+  @Test
+  public void testAsCOGRoundTrip() throws IOException {
+    GridCoverage2D original = rasterFromGeoTiff(resourceFolder + "raster/test1.tiff");
+    byte[] cogBytes = RasterOutputs.asCOG(original, "LZW", 256);
+    // Read COG bytes back as a raster via GeoTiff reader
+    GridCoverage2D roundTripped = RasterConstructors.fromGeoTiff(cogBytes);
+    assertNotNull(roundTripped);
+    // Verify envelope is preserved
+    assertEquals(original.getEnvelope2D().toString(), roundTripped.getEnvelope2D().toString());
+    // Verify dimensions are preserved
+    assertEquals(
+        original.getRenderedImage().getWidth(), roundTripped.getRenderedImage().getWidth());
+    assertEquals(
+        original.getRenderedImage().getHeight(), roundTripped.getRenderedImage().getHeight());
+    // Verify number of bands is preserved
+    assertEquals(original.getNumSampleDimensions(), roundTripped.getNumSampleDimensions());
+  }
+
+  @Test
+  public void testAsCOGWithCompression() throws IOException {
+    GridCoverage2D raster = rasterFromGeoTiff(resourceFolder + "raster/test1.tiff");
+    byte[] cogLZW = RasterOutputs.asCOG(raster, "LZW");
+    byte[] cogDeflate = RasterOutputs.asCOG(raster, "Deflate");
+    assertNotNull(cogLZW);
+    assertNotNull(cogDeflate);
+    assertTrue(cogLZW.length > 0);
+    assertTrue(cogDeflate.length > 0);
+    // Different compressions should produce different sizes
+    assertNotEquals(cogLZW.length, cogDeflate.length);
+  }
+
+  @Test
+  public void testAsCOGWithCompressionAndTileSize() throws IOException {
+    GridCoverage2D raster = rasterFromGeoTiff(resourceFolder + "raster/test1.tiff");
+    byte[] cog256 = RasterOutputs.asCOG(raster, "Deflate", 256);
+    byte[] cog512 = RasterOutputs.asCOG(raster, "Deflate", 512);
+    assertNotNull(cog256);
+    assertNotNull(cog512);
+    assertTrue(cog256.length > 0);
+    assertTrue(cog512.length > 0);
+  }
+
+  @Test
+  public void testAsCOGWithCompressionTileSizeAndQuality() throws IOException {
+    GridCoverage2D raster = rasterFromGeoTiff(resourceFolder + "raster/test1.tiff");
+    byte[] cogHighQ = RasterOutputs.asCOG(raster, "Deflate", 256, 1.0);
+    byte[] cogLowQ = RasterOutputs.asCOG(raster, "Deflate", 256, 0.1);
+    assertNotNull(cogHighQ);
+    assertNotNull(cogLowQ);
+    assertTrue(cogHighQ.length > 0);
+    assertTrue(cogLowQ.length > 0);
+  }
+
+  @Test
+  public void testAsCOGWithResampling() throws IOException {
+    GridCoverage2D raster = rasterFromGeoTiff(resourceFolder + "raster/test1.tiff");
+    byte[] cog = RasterOutputs.asCOG(raster, "Deflate", 256, 0.2, "Bilinear");
+    assertNotNull(cog);
+    assertTrue(cog.length > 0);
+  }
+
+  @Test
+  public void testAsCOGAllArgs() throws IOException {
+    GridCoverage2D raster = rasterFromGeoTiff(resourceFolder + "raster/test1.tiff");
+    byte[] cog = RasterOutputs.asCOG(raster, "LZW", 256, 0.5, "Nearest", 2);
+    assertNotNull(cog);
+    assertTrue(cog.length > 0);
+  }
+
+  @Test
+  public void testAsCOGCaseInsensitive() throws IOException {
+    GridCoverage2D raster = rasterFromGeoTiff(resourceFolder + "raster/test1.tiff");
+    // compression and resampling should be case-insensitive
+    byte[] cog = RasterOutputs.asCOG(raster, "lzw", 256, 0.5, "bilinear", 2);
+    assertNotNull(cog);
+    assertTrue(cog.length > 0);
+    // uppercase
+    byte[] cog2 = RasterOutputs.asCOG(raster, "DEFLATE", 256, 0.5, "NEAREST", 2);
+    assertNotNull(cog2);
+    assertTrue(cog2.length > 0);
+    // mixed case: packbits
+    byte[] cog3 = RasterOutputs.asCOG(raster, "packbits");
+    assertNotNull(cog3);
+    assertTrue(cog3.length > 0);
+  }
 }

docs/api/sql/Raster-writer.md

@@ -116,6 +116,71 @@ root
  |-- geotiff: binary (nullable = true)
 ```
 
+#### RS_AsCOG
+
+Introduction: Returns a binary DataFrame from a Raster DataFrame. Each raster object in the resulting DataFrame is a [Cloud Optimized GeoTIFF](https://www.cogeo.org/) (COG) image in binary format. COG is a GeoTIFF that is internally organized to enable efficient range-read access over HTTP, making it ideal for cloud-hosted raster data.
+
+Possible values for `compression`: `Deflate` (default), `LZW`, `JPEG`, `PackBits`. Case-insensitive.
+
+`tileSize` must be a power of 2 (e.g., 128, 256, 512). Default value: `256`
+
+Possible values for `quality`: any decimal number between 0 and 1. 0 means maximum compression and 1 means minimum compression. Default value: `0.2`
+
+Possible values for `resampling`: `Nearest` (default), `Bilinear`, `Bicubic`. Case-insensitive. This controls the resampling algorithm used to build overview levels.
+
+`overviewCount` controls the number of overview levels. Use `-1` for automatic (default), `0` for no overviews, or any positive integer for a specific count.
+
+Format:
+
+`RS_AsCOG(raster: Raster)`
+
+`RS_AsCOG(raster: Raster, compression: String)`
+
+`RS_AsCOG(raster: Raster, compression: String, tileSize: Integer)`
+
+`RS_AsCOG(raster: Raster, compression: String, tileSize: Integer, quality: Double)`
+
+`RS_AsCOG(raster: Raster, compression: String, tileSize: Integer, quality: Double, resampling: String)`
+
+`RS_AsCOG(raster: Raster, compression: String, tileSize: Integer, quality: Double, resampling: String, overviewCount: Integer)`
+
+Since: `v1.9.0`
+
+SQL Example
+
+```sql
+SELECT RS_AsCOG(raster) FROM my_raster_table
+```
+
+SQL Example
+
+```sql
+SELECT RS_AsCOG(raster, 'LZW') FROM my_raster_table
+```
+
+SQL Example
+
+```sql
+SELECT RS_AsCOG(raster, 'LZW', 512, 0.75, 'Bilinear', 3) FROM my_raster_table
+```
+
+Output:
+
+```html
++--------------------+
+|                 cog|
++--------------------+
+|[4D 4D 00 2A 00 0...|
++--------------------+
+```
+
+Output schema:
+
+```sql
+root
+ |-- cog: binary (nullable = true)
+```
+
 #### RS_AsPNG
 
 Introduction: Returns a PNG byte array, that can be written to raster files as PNGs using the [sedona function](#write-a-binary-dataframe-to-raster-files). This function can only accept pixel data type of unsigned integer. PNG can accept 1 or 3 bands of data from the raster, refer to [RS_Band](Raster-operators.md#rs_band) for more details.

docs/tutorial/raster.md

@@ -527,6 +527,14 @@ Use [RS_AsGeoTiff](../api/sql/Raster-writer.md#rs_asgeotiff) to get the binary D
 SELECT RS_AsGeoTiff(raster)
 ```
 
+### As Cloud Optimized GeoTiff
+
+Use [RS_AsCOG](../api/sql/Raster-writer.md#rs_ascog) to get the binary Dataframe of the raster in [Cloud Optimized GeoTiff](https://www.cogeo.org/) (COG) format. COG is ideal for cloud-hosted raster data because it supports efficient range-read access over HTTP.
+
+```sql
+SELECT RS_AsCOG(raster)
+```
+
 ### As PNG
 
 Use [RS_AsPNG](../api/sql/Raster-writer.md#rs_aspng) to get the binary Dataframe of the raster in PNG format.

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

@@ -314,6 +314,7 @@ object Catalog extends AbstractCatalog with Logging {
     function[RS_Intersects](),
     function[RS_Interpolate](),
     function[RS_AsGeoTiff](),
+    function[RS_AsCOG](),
     function[RS_AsRaster](),
     function[RS_AsArcGrid](),
     function[RS_AsBase64](),

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

@@ -78,3 +78,16 @@ private[apache] case class RS_AsImage(inputExpressions: Seq[Expression])
     copy(inputExpressions = newChildren)
   }
 }
+
+private[apache] case class RS_AsCOG(inputExpressions: Seq[Expression])
+    extends InferredExpression(
+      inferrableFunction6(RasterOutputs.asCOG),
+      inferrableFunction5(RasterOutputs.asCOG),
+      inferrableFunction4(RasterOutputs.asCOG),
+      inferrableFunction3(RasterOutputs.asCOG),
+      inferrableFunction2(RasterOutputs.asCOG),
+      inferrableFunction1(RasterOutputs.asCOG)) {
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}