refactor(raster): return StringViewArray from CRS conversion helpers

srid_to_crs_columnar and normalize_crs_columnar now return StringViewArray
directly instead of ColumnarValue, removing redundant Scalar/Array
re-wrapping and the cast_to(Utf8View) in replace_raster_crs. Broadcasting
for scalar CRS applied to array raster is handled explicitly.

Author: Kontinuation

rust/sedona-raster-functions/src/rs_setsrid.rs

@@ -190,7 +190,7 @@ impl SedonaScalarKernel for RsSetCrs {
 /// null will have the entire raster nulled out (not just the CRS column).
 fn replace_raster_crs(
     raster_arg: &ColumnarValue,
-    crs_value: &ColumnarValue,
+    crs_array: &StringViewArray,
     input_nulls: Option<&NullBuffer>,
 ) -> Result<ColumnarValue> {
     match raster_arg {
@@ -205,10 +205,8 @@ fn replace_raster_crs(
                 })?;
 
             let num_rows = raster_struct.len();
-            let new_crs_array = crs_value
-                .cast_to(&DataType::Utf8View, None)?
-                .to_array(num_rows)?;
-            let new_struct = swap_crs_column(raster_struct, new_crs_array)?;
+            let new_crs: ArrayRef = broadcast_string_view(crs_array, num_rows);
+            let new_struct = swap_crs_column(raster_struct, new_crs)?;
 
             // Merge input nulls: rows where the SRID/CRS input was null become null rasters
             let merged_nulls = NullBuffer::union(new_struct.nulls(), input_nulls);
@@ -221,11 +219,8 @@ fn replace_raster_crs(
             Ok(ColumnarValue::Array(Arc::new(new_struct)))
         }
         ColumnarValue::Scalar(ScalarValue::Struct(arc_struct)) => {
-            let num_rows = arc_struct.len();
-            let new_crs_array = crs_value
-                .cast_to(&DataType::Utf8View, None)?
-                .to_array(num_rows)?;
-            let new_struct = swap_crs_column(arc_struct.as_ref(), new_crs_array)?;
+            let new_crs: ArrayRef = Arc::new(crs_array.clone());
+            let new_struct = swap_crs_column(arc_struct.as_ref(), new_crs)?;
 
             // Merge input nulls: null SRID/CRS input produces a null raster
             let merged_nulls = NullBuffer::union(new_struct.nulls(), input_nulls);
@@ -244,6 +239,23 @@ fn replace_raster_crs(
     }
 }
 
+/// Broadcast a `StringViewArray` to a target length.
+///
+/// If the array already has the target length, it is returned as-is (clone of Arc).
+/// Otherwise the array must have length 1, and its single value is repeated.
+fn broadcast_string_view(array: &StringViewArray, len: usize) -> ArrayRef {
+    if array.len() == len {
+        return Arc::new(array.clone());
+    }
+    debug_assert_eq!(array.len(), 1);
+    if array.is_null(0) {
+        Arc::new(StringViewArray::new_null(len))
+    } else {
+        let value = array.value(0);
+        Arc::new(std::iter::repeat_n(Some(value), len).collect::<StringViewArray>())
+    }
+}
+
 /// Swap only the CRS column of a raster StructArray, keeping all other columns intact.
 fn swap_crs_column(raster_struct: &StructArray, new_crs_array: ArrayRef) -> Result<StructArray> {
     let mut columns: Vec<ArrayRef> = raster_struct.columns().to_vec();
@@ -291,7 +303,7 @@ fn extract_input_nulls(input: &ColumnarValue) -> Option<NullBuffer> {
 fn srid_to_crs_columnar(
     srid_arg: &ColumnarValue,
     maybe_engine: Option<&Arc<dyn CrsEngine + Send + Sync>>,
-) -> Result<ColumnarValue> {
+) -> Result<StringViewArray> {
     let mut srid_to_crs = CachedSRIDToCrs::new();
 
     // Cast to Int64 for uniform handling
@@ -314,17 +326,7 @@ fn srid_to_crs_columnar(
         })
         .collect::<Result<_>>()?;
 
-    // Return as Scalar if the original was scalar, Array otherwise
-    if matches!(srid_arg, ColumnarValue::Scalar(_)) {
-        let scalar = if crs_array.is_null(0) {
-            ScalarValue::Utf8View(None)
-        } else {
-            ScalarValue::Utf8View(Some(crs_array.value(0).to_string()))
-        };
-        Ok(ColumnarValue::Scalar(scalar))
-    } else {
-        Ok(ColumnarValue::Array(Arc::new(crs_array)))
-    }
+    Ok(crs_array)
 }
 
 // ---------------------------------------------------------------------------
@@ -339,7 +341,7 @@ fn srid_to_crs_columnar(
 fn normalize_crs_columnar(
     crs_arg: &ColumnarValue,
     _maybe_engine: Option<&Arc<dyn CrsEngine + Send + Sync>>,
-) -> Result<ColumnarValue> {
+) -> Result<StringViewArray> {
     let mut crs_norm = CachedCrsNormalization::new();
 
     let string_value = crs_arg.cast_to(&DataType::Utf8View, None)?;
@@ -358,17 +360,7 @@ fn normalize_crs_columnar(
         })
         .collect::<Result<_>>()?;
 
-    // Return as Scalar if the original was scalar, Array otherwise
-    if matches!(crs_arg, ColumnarValue::Scalar(_)) {
-        let scalar = if crs_array.is_null(0) {
-            ScalarValue::Utf8View(None)
-        } else {
-            ScalarValue::Utf8View(Some(crs_array.value(0).to_string()))
-        };
-        Ok(ColumnarValue::Scalar(scalar))
-    } else {
-        Ok(ColumnarValue::Array(Arc::new(crs_array)))
-    }
+    Ok(crs_array)
 }
 
 /// Validate a CRS string