Re: [PR] feat(c/sedona-proj): Implement item crs support for ST_Transform [sedona-db]

[via GitHub]

Copilot commented on code in PR #531:
URL: https://github.com/apache/sedona-db/pull/531#discussion_r2710369312


##########
c/sedona-proj/src/st_transform.rs:
##########
@@ -15,31 +15,353 @@
 // specific language governing permissions and limitations
 // under the License.
 use crate::transform::{ProjCrsEngine, ProjCrsEngineBuilder};
-use arrow_array::builder::BinaryBuilder;
+use arrow_array::builder::{BinaryBuilder, StringViewBuilder};
+use arrow_array::ArrayRef;
 use arrow_schema::DataType;
-use datafusion_common::{DataFusionError, Result, ScalarValue};
+use datafusion_common::cast::{as_string_view_array, as_struct_array};
+use datafusion_common::{exec_err, DataFusionError, Result, ScalarValue};
 use datafusion_expr::ColumnarValue;
-use geo_traits::to_geo::ToGeoGeometry;
+use sedona_common::sedona_internal_err;
+use sedona_expr::item_crs::make_item_crs;
 use sedona_expr::scalar_udf::{ScalarKernelRef, SedonaScalarKernel};
 use sedona_functions::executor::WkbExecutor;
 use sedona_geometry::transform::{transform, CachingCrsEngine, CrsEngine, 
CrsTransform};
 use sedona_geometry::wkb_factory::WKB_MIN_PROBABLE_BYTES;
-use sedona_schema::crs::deserialize_crs;
-use sedona_schema::datatypes::{Edges, SedonaType};
+use sedona_schema::crs::{deserialize_crs, Crs};
+use sedona_schema::datatypes::{Edges, SedonaType, WKB_GEOMETRY, 
WKB_GEOMETRY_ITEM_CRS};
 use sedona_schema::matchers::ArgMatcher;
 use std::cell::OnceCell;
-use std::rc::Rc;
+use std::io::Write;
+use std::iter::zip;
 use std::sync::{Arc, RwLock};
 use wkb::reader::Wkb;
 
-#[derive(Debug)]
-struct STTransform {}
-
 /// ST_Transform() implementation using the proj crate
 pub fn st_transform_impl() -> ScalarKernelRef {
     Arc::new(STTransform {})
 }
 
+#[derive(Debug)]
+struct STTransform {}
+
+impl SedonaScalarKernel for STTransform {
+    fn return_type_from_args_and_scalars(
+        &self,
+        arg_types: &[SedonaType],
+        scalar_args: &[Option<&ScalarValue>],
+    ) -> Result<Option<SedonaType>> {
+        let inputs = zip(arg_types, scalar_args)
+            .map(|(arg_type, arg_scalar)| 
ArgInput::from_return_type_arg(arg_type, *arg_scalar))
+            .collect::<Vec<_>>();
+
+        if inputs.len() == 2 {
+            match (inputs[0], inputs[1]) {
+                // ScalarCrs output always returns a Wkb output type with 
concrete Crs
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(scalar_value)) => {
+                    Ok(Some(output_type_from_scalar_crs_value(scalar_value)?))
+                }
+
+                // Geo or ItemCrs with ArrayCrs output always return ItemCrs 
output
+                (ArgInput::Geo(_), ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs) => {
+                    Ok(Some(WKB_GEOMETRY_ITEM_CRS.clone()))
+                }
+                _ => Ok(None),
+            }
+        } else if inputs.len() == 3 {
+            match (inputs[0], inputs[1], inputs[2]) {
+                // ScalarCrs output always returns a Wkb output type with 
concrete Crs
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(_), 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::Geo(_), ArgInput::ArrayCrs, 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(_), 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs, 
ArgInput::ScalarCrs(scalar_value)) => {
+                    Ok(Some(output_type_from_scalar_crs_value(scalar_value)?))
+                }
+
+                // Geo or ItemCrs with ArrayCrs output always return ItemCrs 
output
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(_), ArgInput::ArrayCrs)
+                | (ArgInput::Geo(_), ArgInput::ArrayCrs, ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(_), 
ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs, ArgInput::ArrayCrs) 
=> {
+                    Ok(Some(WKB_GEOMETRY_ITEM_CRS.clone()))
+                }
+                _ => Ok(None),
+            }
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn invoke_batch_from_args(
+        &self,
+        arg_types: &[SedonaType],
+        args: &[ColumnarValue],
+        _return_type: &SedonaType,
+        _num_rows: usize,
+    ) -> Result<ColumnarValue> {
+        let inputs = zip(arg_types, args)
+            .map(|(arg_type, arg)| ArgInput::from_arg(arg_type, arg))
+            .collect::<Vec<_>>();
+
+        let executor = WkbExecutor::new(arg_types, args);
+        let mut builder = BinaryBuilder::with_capacity(
+            executor.num_iterations(),
+            WKB_MIN_PROBABLE_BYTES * executor.num_iterations(),
+        );
+
+        // Optimize the easy case, where we have exactly one transformation 
and there are no
+        // null or missing CRSes to contend with.
+        let from_index = inputs.len() - 2;
+        let to_index = inputs.len() - 1;
+        let (from, to) = (inputs[from_index], inputs[to_index]);
+        if let (Some(from_constant), Some(to_constant)) = 
(from.crs_constant()?, to.crs_constant()?)
+        {
+            let maybe_from_crs = deserialize_crs(&from_constant)?;
+            let maybe_to_crs = deserialize_crs(&to_constant)?;
+            if let (Some(from_crs), Some(to_crs)) = (maybe_from_crs, 
maybe_to_crs) {
+                with_global_proj_engine(|engine| {
+                    let crs_transform = engine
+                        .get_transform_crs_to_crs(
+                            &from_crs.to_crs_string(),
+                            &to_crs.to_crs_string(),
+                            None,
+                            "",
+                        )
+                        .map_err(|e| 
DataFusionError::Execution(format!("{e}")))?;
+                    executor.execute_wkb_void(|maybe_wkb| {
+                        match maybe_wkb {
+                            Some(wkb) => {
+                                invoke_scalar(&wkb, crs_transform.as_ref(), 
&mut builder)?;
+                                builder.append_value([]);
+                            }
+                            None => builder.append_null(),
+                        }
+                        Ok(())
+                    })?;
+                    Ok(())
+                })?;
+                return executor.finish(Arc::new(builder.finish()));
+            }
+        }
+
+        // Iterate over pairs of CRS strings
+        let from_crs_array = from.crs_array(&args[from_index], 
executor.num_iterations())?;
+        let to_crs_array = to.crs_array(&args[to_index], 
executor.num_iterations())?;
+        let from_crs_string_view_array = 
as_string_view_array(&from_crs_array)?;
+        let to_crs_string_view_array = as_string_view_array(&to_crs_array)?;
+        let mut crs_to_crs_iter = zip(from_crs_string_view_array, 
to_crs_string_view_array);
+
+        // We might need to build an output array of sanitized CRS strings
+        let mut maybe_crs_ouput = if matches!(to, ArgInput::ArrayCrs) {

Review Comment:
   Corrected spelling of 'ouput' to 'output'.



##########
rust/sedona-functions/src/executor.rs:
##########
@@ -357,6 +357,15 @@ impl IterGeo for ArrayRef {
             }
             SedonaType::Wkb(_, _) => iter_wkb_binary(as_binary_array(self)?, 
func),
             SedonaType::WkbView(_, _) => 
iter_wkb_binary(as_binary_view_array(self)?, func),
+            SedonaType::Arrow(DataType::Struct(fields))
+                if fields.iter().map(|f| f.name()).collect::<Vec<_>>() == 
vec!["item", "crs"] =>

Review Comment:
   The field name validation allocates a Vec on every call. Consider using an 
iterator comparison instead: `fields.len() == 2 && fields[0].name() == \"item\" 
&& fields[1].name() == \"crs\"`
   ```suggestion
                   if fields.len() == 2
                       && fields[0].name() == "item"
                       && fields[1].name() == "crs" =>
   ```



##########
c/sedona-proj/src/st_transform.rs:
##########
@@ -15,31 +15,353 @@
 // specific language governing permissions and limitations
 // under the License.
 use crate::transform::{ProjCrsEngine, ProjCrsEngineBuilder};
-use arrow_array::builder::BinaryBuilder;
+use arrow_array::builder::{BinaryBuilder, StringViewBuilder};
+use arrow_array::ArrayRef;
 use arrow_schema::DataType;
-use datafusion_common::{DataFusionError, Result, ScalarValue};
+use datafusion_common::cast::{as_string_view_array, as_struct_array};
+use datafusion_common::{exec_err, DataFusionError, Result, ScalarValue};
 use datafusion_expr::ColumnarValue;
-use geo_traits::to_geo::ToGeoGeometry;
+use sedona_common::sedona_internal_err;
+use sedona_expr::item_crs::make_item_crs;
 use sedona_expr::scalar_udf::{ScalarKernelRef, SedonaScalarKernel};
 use sedona_functions::executor::WkbExecutor;
 use sedona_geometry::transform::{transform, CachingCrsEngine, CrsEngine, 
CrsTransform};
 use sedona_geometry::wkb_factory::WKB_MIN_PROBABLE_BYTES;
-use sedona_schema::crs::deserialize_crs;
-use sedona_schema::datatypes::{Edges, SedonaType};
+use sedona_schema::crs::{deserialize_crs, Crs};
+use sedona_schema::datatypes::{Edges, SedonaType, WKB_GEOMETRY, 
WKB_GEOMETRY_ITEM_CRS};
 use sedona_schema::matchers::ArgMatcher;
 use std::cell::OnceCell;
-use std::rc::Rc;
+use std::io::Write;
+use std::iter::zip;
 use std::sync::{Arc, RwLock};
 use wkb::reader::Wkb;
 
-#[derive(Debug)]
-struct STTransform {}
-
 /// ST_Transform() implementation using the proj crate
 pub fn st_transform_impl() -> ScalarKernelRef {
     Arc::new(STTransform {})
 }
 
+#[derive(Debug)]
+struct STTransform {}
+
+impl SedonaScalarKernel for STTransform {
+    fn return_type_from_args_and_scalars(
+        &self,
+        arg_types: &[SedonaType],
+        scalar_args: &[Option<&ScalarValue>],
+    ) -> Result<Option<SedonaType>> {
+        let inputs = zip(arg_types, scalar_args)
+            .map(|(arg_type, arg_scalar)| 
ArgInput::from_return_type_arg(arg_type, *arg_scalar))
+            .collect::<Vec<_>>();
+
+        if inputs.len() == 2 {
+            match (inputs[0], inputs[1]) {
+                // ScalarCrs output always returns a Wkb output type with 
concrete Crs
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(scalar_value)) => {
+                    Ok(Some(output_type_from_scalar_crs_value(scalar_value)?))
+                }
+
+                // Geo or ItemCrs with ArrayCrs output always return ItemCrs 
output
+                (ArgInput::Geo(_), ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs) => {
+                    Ok(Some(WKB_GEOMETRY_ITEM_CRS.clone()))
+                }
+                _ => Ok(None),
+            }
+        } else if inputs.len() == 3 {
+            match (inputs[0], inputs[1], inputs[2]) {
+                // ScalarCrs output always returns a Wkb output type with 
concrete Crs
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(_), 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::Geo(_), ArgInput::ArrayCrs, 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(_), 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs, 
ArgInput::ScalarCrs(scalar_value)) => {
+                    Ok(Some(output_type_from_scalar_crs_value(scalar_value)?))
+                }
+
+                // Geo or ItemCrs with ArrayCrs output always return ItemCrs 
output
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(_), ArgInput::ArrayCrs)
+                | (ArgInput::Geo(_), ArgInput::ArrayCrs, ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(_), 
ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs, ArgInput::ArrayCrs) 
=> {
+                    Ok(Some(WKB_GEOMETRY_ITEM_CRS.clone()))
+                }
+                _ => Ok(None),
+            }
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn invoke_batch_from_args(
+        &self,
+        arg_types: &[SedonaType],
+        args: &[ColumnarValue],
+        _return_type: &SedonaType,
+        _num_rows: usize,
+    ) -> Result<ColumnarValue> {
+        let inputs = zip(arg_types, args)
+            .map(|(arg_type, arg)| ArgInput::from_arg(arg_type, arg))
+            .collect::<Vec<_>>();
+
+        let executor = WkbExecutor::new(arg_types, args);
+        let mut builder = BinaryBuilder::with_capacity(
+            executor.num_iterations(),
+            WKB_MIN_PROBABLE_BYTES * executor.num_iterations(),
+        );
+
+        // Optimize the easy case, where we have exactly one transformation 
and there are no
+        // null or missing CRSes to contend with.
+        let from_index = inputs.len() - 2;
+        let to_index = inputs.len() - 1;
+        let (from, to) = (inputs[from_index], inputs[to_index]);
+        if let (Some(from_constant), Some(to_constant)) = 
(from.crs_constant()?, to.crs_constant()?)
+        {
+            let maybe_from_crs = deserialize_crs(&from_constant)?;
+            let maybe_to_crs = deserialize_crs(&to_constant)?;
+            if let (Some(from_crs), Some(to_crs)) = (maybe_from_crs, 
maybe_to_crs) {
+                with_global_proj_engine(|engine| {
+                    let crs_transform = engine
+                        .get_transform_crs_to_crs(
+                            &from_crs.to_crs_string(),
+                            &to_crs.to_crs_string(),
+                            None,
+                            "",
+                        )
+                        .map_err(|e| 
DataFusionError::Execution(format!("{e}")))?;
+                    executor.execute_wkb_void(|maybe_wkb| {
+                        match maybe_wkb {
+                            Some(wkb) => {
+                                invoke_scalar(&wkb, crs_transform.as_ref(), 
&mut builder)?;
+                                builder.append_value([]);
+                            }
+                            None => builder.append_null(),
+                        }
+                        Ok(())
+                    })?;
+                    Ok(())
+                })?;
+                return executor.finish(Arc::new(builder.finish()));
+            }
+        }
+
+        // Iterate over pairs of CRS strings
+        let from_crs_array = from.crs_array(&args[from_index], 
executor.num_iterations())?;
+        let to_crs_array = to.crs_array(&args[to_index], 
executor.num_iterations())?;
+        let from_crs_string_view_array = 
as_string_view_array(&from_crs_array)?;
+        let to_crs_string_view_array = as_string_view_array(&to_crs_array)?;
+        let mut crs_to_crs_iter = zip(from_crs_string_view_array, 
to_crs_string_view_array);
+
+        // We might need to build an output array of sanitized CRS strings
+        let mut maybe_crs_ouput = if matches!(to, ArgInput::ArrayCrs) {
+            Some(StringViewBuilder::with_capacity(executor.num_iterations()))
+        } else {
+            None
+        };
+
+        with_global_proj_engine(|engine| {
+            executor.execute_wkb_void(|maybe_wkb| {
+                match (maybe_wkb, crs_to_crs_iter.next().unwrap()) {
+                    (Some(wkb), (Some(from_crs_str), Some(to_crs_str))) => {
+                        let maybe_from_crs = deserialize_crs(from_crs_str)?;
+                        let maybe_to_crs = deserialize_crs(to_crs_str)?;
+
+                        if let Some(crs_output) = &mut maybe_crs_ouput {
+                            if let Some(to_crs) = &maybe_to_crs {
+                                
crs_output.append_value(to_crs.to_authority_code()?.unwrap_or_else(|| 
to_crs.to_crs_string()));
+                            } else {
+                                crs_output.append_null();
+                            }
+                        }
+
+                        if maybe_from_crs == maybe_to_crs {
+                            invoke_noop(&wkb, &mut builder)?;
+                            builder.append_value([]);
+                            return Ok(());
+                        }
+
+                        let crs_transform = match (maybe_from_crs, 
maybe_to_crs) {
+                            (Some(from_crs), Some(to_crs)) => {
+                                engine
+                                
.get_transform_crs_to_crs(&from_crs.to_crs_string(), &to_crs.to_crs_string(), 
None, "")
+                                .map_err(|e| 
DataFusionError::Execution(format!("{e}")))?
+                            },
+                            _ => return exec_err!(
+                                "Can't transform to or from an unset CRS. Do 
you need to call ST_SetSRID on the input?"
+                            )
+                        };
+
+                        invoke_scalar(&wkb, crs_transform.as_ref(), &mut 
builder)?;
+                        builder.append_value([]);
+                    }
+                    _ => {
+                        if let Some(crs_output) = &mut maybe_crs_ouput {

Review Comment:
   Corrected spelling of 'ouput' to 'output'.



##########
c/sedona-proj/src/st_transform.rs:
##########
@@ -113,503 +435,370 @@ thread_local! {
     };
 }
 
-struct TransformArgIndexes {
-    wkb: usize,
-    first_crs: usize,
-    second_crs: Option<usize>,
-    lenient: Option<usize>,
-}
-
-impl TransformArgIndexes {
-    fn new() -> Self {
-        Self {
-            wkb: 0,
-            first_crs: 1,
-            second_crs: None,
-            lenient: None,
-        }
-    }
-}
-
-fn define_arg_indexes(arg_types: &[SedonaType], indexes: &mut 
TransformArgIndexes) {
-    indexes.wkb = 0;
-    indexes.first_crs = 1;
-
-    for (i, arg_type) in arg_types.iter().enumerate().skip(2) {
-        if ArgMatcher::is_numeric().match_type(arg_type)
-            || ArgMatcher::is_string().match_type(arg_type)
-        {
-            indexes.second_crs = Some(i);
-        } else if *arg_type == SedonaType::Arrow(DataType::Boolean) {
-            indexes.lenient = Some(i);
-        }
-    }
-}
-
-impl SedonaScalarKernel for STTransform {
-    fn return_type(&self, _args: &[SedonaType]) -> Result<Option<SedonaType>, 
DataFusionError> {
-        Err(DataFusionError::Internal(
-            "Return type should only be called with args".to_string(),
-        ))
-    }
-    fn return_type_from_args_and_scalars(
-        &self,
-        arg_types: &[SedonaType],
-        scalar_args: &[Option<&ScalarValue>],
-    ) -> Result<Option<SedonaType>> {
-        let matcher = ArgMatcher::new(
-            vec![
-                ArgMatcher::is_geometry(),
-                ArgMatcher::or(vec![ArgMatcher::is_numeric(), 
ArgMatcher::is_string()]),
-                ArgMatcher::optional(ArgMatcher::or(vec![
-                    ArgMatcher::is_numeric(),
-                    ArgMatcher::is_string(),
-                ])),
-                ArgMatcher::optional(ArgMatcher::is_boolean()),
-            ],
-            SedonaType::Wkb(Edges::Planar, None),
-        );
-
-        if !matcher.matches(arg_types) {
-            return Ok(None);
-        }
-
-        let mut indexes = TransformArgIndexes::new();
-        define_arg_indexes(arg_types, &mut indexes);
-
-        let scalar_arg_opt = if let Some(second_crs_index) = 
indexes.second_crs {
-            scalar_args.get(second_crs_index).unwrap()
-        } else {
-            scalar_args.get(indexes.first_crs).unwrap()
-        };
-
-        let crs_str_opt = if let Some(scalar_crs) = scalar_arg_opt {
-            to_crs_str(scalar_crs)
-        } else {
-            None
-        };
-
-        // If there is no CRS argument, we cannot determine the return type.
-        match crs_str_opt {
-            Some(to_crs) => {
-                let crs = deserialize_crs(&to_crs)?;
-                Ok(Some(SedonaType::Wkb(Edges::Planar, crs)))
-            }
-            _ => Ok(Some(SedonaType::Wkb(Edges::Planar, None))),
-        }
-    }
-
-    fn invoke_batch(
-        &self,
-        arg_types: &[SedonaType],
-        args: &[ColumnarValue],
-    ) -> Result<ColumnarValue> {
-        let executor = WkbExecutor::new(arg_types, args);
-        let mut builder = BinaryBuilder::with_capacity(
-            executor.num_iterations(),
-            WKB_MIN_PROBABLE_BYTES * executor.num_iterations(),
-        );
-
-        let mut indexes = TransformArgIndexes::new();
-        define_arg_indexes(arg_types, &mut indexes);
-
-        let first_crs = get_crs_str(args, indexes.first_crs).ok_or_else(|| {
-            DataFusionError::Execution(
-                "First CRS argument must be a string or numeric 
scalar".to_string(),
-            )
-        })?;
-
-        let lenient = indexes
-            .lenient
-            .is_some_and(|i| get_scalar_bool(args, i).unwrap_or(false));
-
-        let second_crs = if let Some(second_crs_index) = indexes.second_crs {
-            get_crs_str(args, second_crs_index)
-        } else {
-            None
-        };
-
-        with_global_proj_engine(|engine| {
-            let crs_from_geo = parse_source_crs(&arg_types[indexes.wkb])?;
-
-            let transform = match &second_crs {
-                Some(to_crs) => get_transform_crs_to_crs(engine, &first_crs, 
to_crs)?,
-                None => get_transform_to_crs(engine, crs_from_geo, &first_crs, 
lenient)?,
-            };
-
-            executor.execute_wkb_void(|maybe_wkb| {
-                match maybe_wkb {
-                    Some(wkb) => invoke_scalar(&wkb, transform.as_ref(), &mut 
builder)?,
-                    None => builder.append_null(),
-                }
-
-                Ok(())
-            })?;
-
-            Ok(())
-        })?;
-
-        executor.finish(Arc::new(builder.finish()))
-    }
-}
-
-fn get_transform_to_crs(
-    engine: &dyn CrsEngine,
-    source_crs_opt: Option<String>,
-    to_crs: &str,
-    lenient: bool,
-) -> Result<Rc<dyn CrsTransform>, DataFusionError> {
-    let from_crs = match source_crs_opt {
-        Some(crs) => crs,
-        None if lenient => "EPSG:4326".to_string(),
-        None => {
-            return Err(DataFusionError::Execution(
-                "Source CRS is required when transforming to a 
CRS".to_string(),
-            ))
-        }
-    };
-    get_transform_crs_to_crs(engine, &from_crs, to_crs)
-}
-
-fn get_transform_crs_to_crs(
-    engine: &dyn CrsEngine,
-    from_crs: &str,
-    to_crs: &str,
-) -> Result<Rc<dyn CrsTransform>, DataFusionError> {
-    engine
-        .get_transform_crs_to_crs(from_crs, to_crs, None, "")
-        .map_err(|err| DataFusionError::Execution(format!("Transform error: 
{err}")))
-}
-
-fn invoke_scalar(wkb: &Wkb, trans: &dyn CrsTransform, builder: &mut 
BinaryBuilder) -> Result<()> {
-    let geo_geom = wkb.to_geometry();
-    transform(&geo_geom, trans, builder)
-        .map_err(|err| DataFusionError::Execution(format!("Transform error: 
{err}")))?;
-    builder.append_value([]);
-    Ok(())
-}
-
-fn parse_source_crs(source_type: &SedonaType) -> Result<Option<String>> {
-    match source_type {
-        SedonaType::Wkb(_, Some(crs)) | SedonaType::WkbView(_, Some(crs)) => {
-            Ok(Some(crs.to_crs_string()))
-        }
-        _ => Ok(None),
-    }
-}
-
-fn to_crs_str(scalar_arg: &ScalarValue) -> Option<String> {
-    if let Ok(ScalarValue::Utf8(Some(crs))) = 
scalar_arg.cast_to(&DataType::Utf8) {
-        if crs.chars().all(|c| c.is_ascii_digit()) {
-            return Some(format!("EPSG:{crs}"));
-        } else {
-            return Some(crs);
-        }
-    }
-
-    None
-}
-
-fn get_crs_str(args: &[ColumnarValue], index: usize) -> Option<String> {
-    if let ColumnarValue::Scalar(scalar_crs) = &args[index] {
-        return to_crs_str(scalar_crs);
-    }
-    None
-}
-
-fn get_scalar_bool(args: &[ColumnarValue], index: usize) -> Option<bool> {
-    if let Some(ColumnarValue::Scalar(ScalarValue::Boolean(opt_bool))) = 
args.get(index) {
-        *opt_bool
-    } else {
-        None
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
+    use arrow_array::create_array;
     use arrow_array::ArrayRef;
-    use arrow_schema::{DataType, Field};
-    use datafusion_common::config::ConfigOptions;
-    use datafusion_expr::{ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, 
ScalarUDFImpl};
+    use arrow_schema::DataType;
     use rstest::rstest;
     use sedona_expr::scalar_udf::SedonaScalarUDF;
     use sedona_schema::crs::lnglat;
     use sedona_schema::crs::Crs;
     use sedona_schema::datatypes::WKB_GEOMETRY;
-    use sedona_testing::compare::assert_value_equal;
-    use sedona_testing::{create::create_array, create::create_array_value};
+    use sedona_testing::compare::assert_array_equal;
+    use sedona_testing::create::create_array;
+    use sedona_testing::create::create_array_item_crs;
+    use sedona_testing::create::create_scalar;
+    use sedona_testing::testers::ScalarUdfTester;
 
     const NAD83ZONE6PROJ: &str = "EPSG:2230";
     const WGS84: &str = "EPSG:4326";
 
-    #[rstest]
-    fn invalid_arg_checks() {
-        let udf: SedonaScalarUDF = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
-
-        // No args
-        let result = udf.return_field_from_args(ReturnFieldArgs {
-            arg_fields: &[],
-            scalar_arguments: &[],
-        });
-        assert!(
-            result.is_err()
-                && result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("No kernel matching arguments")
+    #[test]
+    fn test_invoke_with_string() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let geometry_input = SedonaType::Wkb(Edges::Planar, lnglat());
+        let tester = ScalarUdfTester::new(
+            udf.into(),
+            vec![geometry_input.clone(), SedonaType::Arrow(DataType::Utf8)],
         );
 
-        // Too many args
-        let arg_types = [
-            WKB_GEOMETRY,
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Boolean),
-            SedonaType::Arrow(DataType::Int32),
-        ];
-        let arg_fields: Vec<Arc<Field>> = arg_types
-            .iter()
-            .map(|arg_type| Arc::new(arg_type.to_storage_field("", 
true).unwrap()))
-            .collect();
-        let result = udf.return_field_from_args(ReturnFieldArgs {
-            arg_fields: &arg_fields,
-            scalar_arguments: &[None, None, None, None, None],
-        });
-        assert!(
-            result.is_err()
-                && result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("No kernel matching arguments")
-        );
+        // Return type with scalar to argument (returns type-level CRS)
+        let expected_return_type = SedonaType::Wkb(Edges::Planar, 
get_crs(NAD83ZONE6PROJ));
+        let return_type = tester
+            .return_type_with_scalar_scalar(Option::<&str>::None, 
Some(NAD83ZONE6PROJ))
+            .unwrap();
+        assert_eq!(return_type, expected_return_type);
 
-        // First arg not geometry
-        let arg_types = [
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Utf8),
-        ];
-        let arg_fields: Vec<Arc<Field>> = arg_types
-            .iter()
-            .map(|arg_type| Arc::new(arg_type.to_storage_field("", 
true).unwrap()))
-            .collect();
-        let result = udf.return_field_from_args(ReturnFieldArgs {
-            arg_fields: &arg_fields,
-            scalar_arguments: &[None, None],
-        });
-        assert!(
-            result.is_err()
-                && result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("No kernel matching arguments")
+        // Return type with array to argument (returns item CRS)
+        let return_type = tester.return_type().unwrap();
+        assert_eq!(return_type, WKB_GEOMETRY_ITEM_CRS.clone());
+
+        // Invoke with scalar to argument (returns type-level CRS)
+        let expected_array = create_array(
+            &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
+            &expected_return_type,
         );
+        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let result = tester.invoke_array_scalar(wkb, NAD83ZONE6PROJ).unwrap();
+        assert_array_equal(&result, &expected_array);
 
-        // Second arg not string or numeric
-        let arg_types = [WKB_GEOMETRY, SedonaType::Arrow(DataType::Boolean)];
-        let arg_fields: Vec<Arc<Field>> = arg_types
-            .iter()
-            .map(|arg_type| Arc::new(arg_type.to_storage_field("", 
true).unwrap()))
-            .collect();
-        let result = udf.return_field_from_args(ReturnFieldArgs {
-            arg_fields: &arg_fields,
-            scalar_arguments: &[None, None],
-        });
-        assert!(
-            result.is_err()
-                && result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("No kernel matching arguments")
+        // Invoke with array to argument (returns item CRS)
+        let expected_array = create_array_item_crs(
+            &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
+            [None, Some(NAD83ZONE6PROJ)],
+            &WKB_GEOMETRY,
         );
+        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let crs = create_array!(Utf8, [None, Some(NAD83ZONE6PROJ)]) as 
ArrayRef;
+        let result = tester.invoke_array_array(wkb, crs).unwrap();
+        assert_array_equal(&result, &expected_array);
     }
 
-    #[rstest]
-    fn test_invoke_batch_with_geo_crs() {
-        // From-CRS pulled from sedona type
-        let arg_types = [
-            SedonaType::Wkb(Edges::Planar, lnglat()),
-            SedonaType::Arrow(DataType::Utf8),
-        ];
+    #[test]
+    fn test_invoke_with_srid() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let geometry_input = SedonaType::Wkb(Edges::Planar, lnglat());
+        let tester = ScalarUdfTester::new(
+            udf.into(),
+            vec![geometry_input.clone(), SedonaType::Arrow(DataType::UInt32)],
+        );
 
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&arg_types[0]);
+        // Return type with scalar to argument (returns type-level CRS)
+        let expected_return_type = SedonaType::Wkb(Edges::Planar, 
get_crs(NAD83ZONE6PROJ));
+        let return_type = tester
+            .return_type_with_scalar_scalar(Option::<&str>::None, Some(2230))
+            .unwrap();
+        assert_eq!(return_type, expected_return_type);
 
-        let scalar_args = 
vec![ScalarValue::Utf8(Some(NAD83ZONE6PROJ.to_string()))];
+        // Return type with array to argument (returns item CRS)
+        let return_type = tester.return_type().unwrap();
+        assert_eq!(return_type, WKB_GEOMETRY_ITEM_CRS.clone());
 
-        let expected = create_array_value(
+        // Invoke with scalar to argument (returns type-level CRS)
+        let expected_array = create_array(
             &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
-            &SedonaType::Wkb(Edges::Planar, get_crs(NAD83ZONE6PROJ)),
+            &expected_return_type,
         );
+        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let result = tester.invoke_array_scalar(wkb, 2230).unwrap();
+        assert_array_equal(&result, &expected_array);
 
-        let (result_type, result_col) =
-            invoke_udf_test(wkb, scalar_args, arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, get_crs(NAD83ZONE6PROJ))
+        // Invoke with array to argument (returns item CRS)
+        let expected_array = create_array_item_crs(
+            &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
+            [None, Some(NAD83ZONE6PROJ)],
+            &WKB_GEOMETRY,
         );
+        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let crs = create_array!(Int32, [None, Some(2230)]) as ArrayRef;
+        let result = tester.invoke_array_array(wkb, crs).unwrap();
+        assert_array_equal(&result, &expected_array);
     }
 
-    #[rstest]
-    fn test_invoke_with_srids() {
-        // Use an integer SRID for the to CRS
-        let arg_types = [
-            SedonaType::Wkb(Edges::Planar, lnglat()),
-            SedonaType::Arrow(DataType::UInt32),
-        ];
-
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&arg_types[0]);
-
-        let scalar_args = vec![ScalarValue::UInt32(Some(2230))];
+    #[test]
+    fn test_invoke_with_item_crs() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let geometry_input = WKB_GEOMETRY_ITEM_CRS.clone();
+        let tester = ScalarUdfTester::new(
+            udf.into(),
+            vec![geometry_input.clone(), SedonaType::Arrow(DataType::Utf8)],
+        );
 
-        let expected = create_array_value(
+        // Return type with scalar to argument (returns type-level CRS)
+        // This is the same as for normal input
+        let expected_return_type = SedonaType::Wkb(Edges::Planar, 
get_crs(NAD83ZONE6PROJ));
+        let return_type = tester
+            .return_type_with_scalar_scalar(Option::<&str>::None, 
Some(NAD83ZONE6PROJ))
+            .unwrap();
+        assert_eq!(return_type, expected_return_type);
+
+        // Return type with array to argument (returns item CRS)
+        // This is the same as for normal input
+        let return_type = tester.return_type().unwrap();
+        assert_eq!(return_type, WKB_GEOMETRY_ITEM_CRS.clone());
+
+        // Invoke with scalar to argument (returns type-level CRS)
+        let expected_array = create_array(
             &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
-            &SedonaType::Wkb(Edges::Planar, get_crs(NAD83ZONE6PROJ)),
+            &expected_return_type,
         );
-
-        let (result_type, result_col) =
-            invoke_udf_test(wkb, scalar_args, arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, get_crs(NAD83ZONE6PROJ))
+        let array_in = create_array_item_crs(
+            &[None, Some("POINT (79.3871 43.6426)")],
+            [None, Some("EPSG:4326")],
+            &WKB_GEOMETRY,
         );
-    }
+        let result = tester
+            .invoke_array_scalar(array_in, NAD83ZONE6PROJ)
+            .unwrap();
+        assert_array_equal(&result, &expected_array);
 
-    #[rstest]
-    fn test_invoke_batch_with_lenient() {
-        let arg_types = [
-            WKB_GEOMETRY,
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Boolean),
-        ];
-
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&WKB_GEOMETRY);
-        let scalar_args = vec![
-            ScalarValue::Utf8(Some(NAD83ZONE6PROJ.to_string())),
-            ScalarValue::Boolean(Some(true)),
-        ];
-
-        let expected = create_array_value(
+        // Invoke with array to argument (returns item CRS)
+        let expected_array = create_array_item_crs(
             &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
-            &SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap())),
+            [None, Some(NAD83ZONE6PROJ)],
+            &WKB_GEOMETRY,
         );
-
-        let (result_type, result_col) =
-            invoke_udf_test(wkb, scalar_args, arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap()))
+        let array_in = create_array_item_crs(
+            &[None, Some("POINT (79.3871 43.6426)")],
+            [None, Some("EPSG:4326")],
+            &WKB_GEOMETRY,
         );
+        let crs = create_array!(Utf8, [None, Some(NAD83ZONE6PROJ)]) as 
ArrayRef;
+        let result = tester.invoke_array_array(array_in, crs).unwrap();
+        assert_array_equal(&result, &expected_array);
     }
 
     #[rstest]
-    fn test_invoke_batch_one_crs_no_lenient() {
-        let arg_types = [WKB_GEOMETRY, SedonaType::Arrow(DataType::Utf8)];
+    fn test_invoke_source_arg() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let geometry_input = WKB_GEOMETRY;
+        let tester = ScalarUdfTester::new(
+            udf.into(),
+            vec![
+                geometry_input.clone(),
+                SedonaType::Arrow(DataType::Utf8),
+                SedonaType::Arrow(DataType::Utf8),
+            ],
+        );
 
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&WKB_GEOMETRY);
-        let scalar_args = 
vec![ScalarValue::Utf8(Some(NAD83ZONE6PROJ.to_string()))];
+        // Return type with scalar to argument (returns type-level CRS)
+        // This is the same as for normal input
+        let expected_return_type = SedonaType::Wkb(Edges::Planar, 
get_crs(NAD83ZONE6PROJ));
+        let return_type = tester
+            .return_type_with_scalar_scalar_scalar(
+                Option::<&str>::None,
+                Option::<&str>::None,
+                Some(NAD83ZONE6PROJ),
+            )
+            .unwrap();
+        assert_eq!(return_type, expected_return_type);
 
-        let err = invoke_udf_test(wkb, scalar_args, arg_types.to_vec());
-        assert!(
-            matches!(err, Err(DataFusionError::Execution(_))),
-            "Expected an Execution error"
-        );
-    }
+        // Return type with array to argument (returns item CRS)
+        // This is the same as for normal input
+        let return_type = tester.return_type().unwrap();
+        assert_eq!(return_type, WKB_GEOMETRY_ITEM_CRS.clone());
 
-    #[rstest]
-    fn test_invoke_batch_with_source_arg() {
-        let arg_types = [
-            WKB_GEOMETRY,
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Utf8),
-        ];
-
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&WKB_GEOMETRY);
-
-        let scalar_args = vec![
-            ScalarValue::Utf8(Some(WGS84.to_string())),
-            ScalarValue::Utf8(Some(NAD83ZONE6PROJ.to_string())),
-        ];
-
-        // Note: would be nice to have an epsilon of tolerance when validating
-        let expected = create_array_value(
+        // Invoke with scalar to argument (returns type-level CRS)
+        let expected_array = create_array(
             &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
-            &SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap())),
+            &expected_return_type,
         );
+        let array_in = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let crs_from = create_array!(Utf8, [None, Some(WGS84)]) as ArrayRef;
+        let result = tester
+            .invoke_array_array_scalar(array_in, crs_from, NAD83ZONE6PROJ)
+            .unwrap();
+        assert_array_equal(&result, &expected_array);
+
+        // Invoke with array to argument (returns item CRS)
+        let expected_array = create_array_item_crs(
+            &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
+            [None, Some(NAD83ZONE6PROJ)],
+            &WKB_GEOMETRY,
+        );
+        let array_in = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&WKB_GEOMETRY);
+        let crs_from = create_array!(Utf8, [None, Some(WGS84)]) as ArrayRef;
+        let crs_to = create_array!(Utf8, [None, Some(NAD83ZONE6PROJ)]) as 
ArrayRef;
+        let result = tester
+            .invoke_arrays(vec![array_in, crs_from, crs_to])
+            .unwrap();
+        assert_array_equal(&result, &expected_array);
+    }
 
-        let (result_type, result_col) =
-            invoke_udf_test(wkb.clone(), scalar_args, 
arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap()))
+    #[test]
+    fn test_invoke_null_crs_to() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let tester = ScalarUdfTester::new(
+            udf.clone().into(),
+            vec![WKB_GEOMETRY, SedonaType::Arrow(DataType::Utf8)],
         );
 
-        // Test with integer SRIDs
-        let arg_types = [
-            WKB_GEOMETRY,
-            SedonaType::Arrow(DataType::Int32),
-            SedonaType::Arrow(DataType::Int32),
-        ];
-
-        let scalar_args = vec![
-            ScalarValue::Int32(Some(4326)),
-            ScalarValue::Int32(Some(2230)),
-        ];
-
-        let (result_type, result_col) =
-            invoke_udf_test(wkb, scalar_args, arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap()))
+        // A null scalar CRS should generate WKB_GEOMETRY output with a a type

Review Comment:
   Remove duplicate 'a' in 'with a a type'.
   ```suggestion
           // A null scalar CRS should generate WKB_GEOMETRY output with a type
   ```



##########
c/sedona-proj/src/st_transform.rs:
##########
@@ -15,31 +15,353 @@
 // specific language governing permissions and limitations
 // under the License.
 use crate::transform::{ProjCrsEngine, ProjCrsEngineBuilder};
-use arrow_array::builder::BinaryBuilder;
+use arrow_array::builder::{BinaryBuilder, StringViewBuilder};
+use arrow_array::ArrayRef;
 use arrow_schema::DataType;
-use datafusion_common::{DataFusionError, Result, ScalarValue};
+use datafusion_common::cast::{as_string_view_array, as_struct_array};
+use datafusion_common::{exec_err, DataFusionError, Result, ScalarValue};
 use datafusion_expr::ColumnarValue;
-use geo_traits::to_geo::ToGeoGeometry;
+use sedona_common::sedona_internal_err;
+use sedona_expr::item_crs::make_item_crs;
 use sedona_expr::scalar_udf::{ScalarKernelRef, SedonaScalarKernel};
 use sedona_functions::executor::WkbExecutor;
 use sedona_geometry::transform::{transform, CachingCrsEngine, CrsEngine, 
CrsTransform};
 use sedona_geometry::wkb_factory::WKB_MIN_PROBABLE_BYTES;
-use sedona_schema::crs::deserialize_crs;
-use sedona_schema::datatypes::{Edges, SedonaType};
+use sedona_schema::crs::{deserialize_crs, Crs};
+use sedona_schema::datatypes::{Edges, SedonaType, WKB_GEOMETRY, 
WKB_GEOMETRY_ITEM_CRS};
 use sedona_schema::matchers::ArgMatcher;
 use std::cell::OnceCell;
-use std::rc::Rc;
+use std::io::Write;
+use std::iter::zip;
 use std::sync::{Arc, RwLock};
 use wkb::reader::Wkb;
 
-#[derive(Debug)]
-struct STTransform {}
-
 /// ST_Transform() implementation using the proj crate
 pub fn st_transform_impl() -> ScalarKernelRef {
     Arc::new(STTransform {})
 }
 
+#[derive(Debug)]
+struct STTransform {}
+
+impl SedonaScalarKernel for STTransform {
+    fn return_type_from_args_and_scalars(
+        &self,
+        arg_types: &[SedonaType],
+        scalar_args: &[Option<&ScalarValue>],
+    ) -> Result<Option<SedonaType>> {
+        let inputs = zip(arg_types, scalar_args)
+            .map(|(arg_type, arg_scalar)| 
ArgInput::from_return_type_arg(arg_type, *arg_scalar))
+            .collect::<Vec<_>>();
+
+        if inputs.len() == 2 {
+            match (inputs[0], inputs[1]) {
+                // ScalarCrs output always returns a Wkb output type with 
concrete Crs
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(scalar_value)) => {
+                    Ok(Some(output_type_from_scalar_crs_value(scalar_value)?))
+                }
+
+                // Geo or ItemCrs with ArrayCrs output always return ItemCrs 
output
+                (ArgInput::Geo(_), ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs) => {
+                    Ok(Some(WKB_GEOMETRY_ITEM_CRS.clone()))
+                }
+                _ => Ok(None),
+            }
+        } else if inputs.len() == 3 {
+            match (inputs[0], inputs[1], inputs[2]) {
+                // ScalarCrs output always returns a Wkb output type with 
concrete Crs
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(_), 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::Geo(_), ArgInput::ArrayCrs, 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(_), 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs, 
ArgInput::ScalarCrs(scalar_value)) => {
+                    Ok(Some(output_type_from_scalar_crs_value(scalar_value)?))
+                }
+
+                // Geo or ItemCrs with ArrayCrs output always return ItemCrs 
output
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(_), ArgInput::ArrayCrs)
+                | (ArgInput::Geo(_), ArgInput::ArrayCrs, ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(_), 
ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs, ArgInput::ArrayCrs) 
=> {
+                    Ok(Some(WKB_GEOMETRY_ITEM_CRS.clone()))
+                }
+                _ => Ok(None),
+            }
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn invoke_batch_from_args(
+        &self,
+        arg_types: &[SedonaType],
+        args: &[ColumnarValue],
+        _return_type: &SedonaType,
+        _num_rows: usize,
+    ) -> Result<ColumnarValue> {
+        let inputs = zip(arg_types, args)
+            .map(|(arg_type, arg)| ArgInput::from_arg(arg_type, arg))
+            .collect::<Vec<_>>();
+
+        let executor = WkbExecutor::new(arg_types, args);
+        let mut builder = BinaryBuilder::with_capacity(
+            executor.num_iterations(),
+            WKB_MIN_PROBABLE_BYTES * executor.num_iterations(),
+        );
+
+        // Optimize the easy case, where we have exactly one transformation 
and there are no
+        // null or missing CRSes to contend with.
+        let from_index = inputs.len() - 2;
+        let to_index = inputs.len() - 1;
+        let (from, to) = (inputs[from_index], inputs[to_index]);
+        if let (Some(from_constant), Some(to_constant)) = 
(from.crs_constant()?, to.crs_constant()?)
+        {
+            let maybe_from_crs = deserialize_crs(&from_constant)?;
+            let maybe_to_crs = deserialize_crs(&to_constant)?;
+            if let (Some(from_crs), Some(to_crs)) = (maybe_from_crs, 
maybe_to_crs) {
+                with_global_proj_engine(|engine| {
+                    let crs_transform = engine
+                        .get_transform_crs_to_crs(
+                            &from_crs.to_crs_string(),
+                            &to_crs.to_crs_string(),
+                            None,
+                            "",
+                        )
+                        .map_err(|e| 
DataFusionError::Execution(format!("{e}")))?;
+                    executor.execute_wkb_void(|maybe_wkb| {
+                        match maybe_wkb {
+                            Some(wkb) => {
+                                invoke_scalar(&wkb, crs_transform.as_ref(), 
&mut builder)?;
+                                builder.append_value([]);
+                            }
+                            None => builder.append_null(),
+                        }
+                        Ok(())
+                    })?;
+                    Ok(())
+                })?;
+                return executor.finish(Arc::new(builder.finish()));
+            }
+        }
+
+        // Iterate over pairs of CRS strings
+        let from_crs_array = from.crs_array(&args[from_index], 
executor.num_iterations())?;
+        let to_crs_array = to.crs_array(&args[to_index], 
executor.num_iterations())?;
+        let from_crs_string_view_array = 
as_string_view_array(&from_crs_array)?;
+        let to_crs_string_view_array = as_string_view_array(&to_crs_array)?;
+        let mut crs_to_crs_iter = zip(from_crs_string_view_array, 
to_crs_string_view_array);
+
+        // We might need to build an output array of sanitized CRS strings
+        let mut maybe_crs_ouput = if matches!(to, ArgInput::ArrayCrs) {
+            Some(StringViewBuilder::with_capacity(executor.num_iterations()))
+        } else {
+            None
+        };
+
+        with_global_proj_engine(|engine| {
+            executor.execute_wkb_void(|maybe_wkb| {
+                match (maybe_wkb, crs_to_crs_iter.next().unwrap()) {
+                    (Some(wkb), (Some(from_crs_str), Some(to_crs_str))) => {
+                        let maybe_from_crs = deserialize_crs(from_crs_str)?;
+                        let maybe_to_crs = deserialize_crs(to_crs_str)?;
+
+                        if let Some(crs_output) = &mut maybe_crs_ouput {
+                            if let Some(to_crs) = &maybe_to_crs {
+                                
crs_output.append_value(to_crs.to_authority_code()?.unwrap_or_else(|| 
to_crs.to_crs_string()));
+                            } else {
+                                crs_output.append_null();
+                            }
+                        }
+
+                        if maybe_from_crs == maybe_to_crs {
+                            invoke_noop(&wkb, &mut builder)?;
+                            builder.append_value([]);
+                            return Ok(());
+                        }
+
+                        let crs_transform = match (maybe_from_crs, 
maybe_to_crs) {
+                            (Some(from_crs), Some(to_crs)) => {
+                                engine
+                                
.get_transform_crs_to_crs(&from_crs.to_crs_string(), &to_crs.to_crs_string(), 
None, "")
+                                .map_err(|e| 
DataFusionError::Execution(format!("{e}")))?
+                            },
+                            _ => return exec_err!(
+                                "Can't transform to or from an unset CRS. Do 
you need to call ST_SetSRID on the input?"
+                            )
+                        };
+
+                        invoke_scalar(&wkb, crs_transform.as_ref(), &mut 
builder)?;
+                        builder.append_value([]);
+                    }
+                    _ => {
+                        if let Some(crs_output) = &mut maybe_crs_ouput {
+                            crs_output.append_null();
+                        }
+
+                        builder.append_null()
+                    },
+                }
+                Ok(())
+            })?;
+            Ok(())
+        })?;
+
+        let output_geometry = executor.finish(Arc::new(builder.finish()))?;
+        if let Some(mut crs_output) = maybe_crs_ouput {

Review Comment:
   Corrected spelling of 'ouput' to 'output'.



##########
c/sedona-proj/src/st_transform.rs:
##########
@@ -15,31 +15,353 @@
 // specific language governing permissions and limitations
 // under the License.
 use crate::transform::{ProjCrsEngine, ProjCrsEngineBuilder};
-use arrow_array::builder::BinaryBuilder;
+use arrow_array::builder::{BinaryBuilder, StringViewBuilder};
+use arrow_array::ArrayRef;
 use arrow_schema::DataType;
-use datafusion_common::{DataFusionError, Result, ScalarValue};
+use datafusion_common::cast::{as_string_view_array, as_struct_array};
+use datafusion_common::{exec_err, DataFusionError, Result, ScalarValue};
 use datafusion_expr::ColumnarValue;
-use geo_traits::to_geo::ToGeoGeometry;
+use sedona_common::sedona_internal_err;
+use sedona_expr::item_crs::make_item_crs;
 use sedona_expr::scalar_udf::{ScalarKernelRef, SedonaScalarKernel};
 use sedona_functions::executor::WkbExecutor;
 use sedona_geometry::transform::{transform, CachingCrsEngine, CrsEngine, 
CrsTransform};
 use sedona_geometry::wkb_factory::WKB_MIN_PROBABLE_BYTES;
-use sedona_schema::crs::deserialize_crs;
-use sedona_schema::datatypes::{Edges, SedonaType};
+use sedona_schema::crs::{deserialize_crs, Crs};
+use sedona_schema::datatypes::{Edges, SedonaType, WKB_GEOMETRY, 
WKB_GEOMETRY_ITEM_CRS};
 use sedona_schema::matchers::ArgMatcher;
 use std::cell::OnceCell;
-use std::rc::Rc;
+use std::io::Write;
+use std::iter::zip;
 use std::sync::{Arc, RwLock};
 use wkb::reader::Wkb;
 
-#[derive(Debug)]
-struct STTransform {}
-
 /// ST_Transform() implementation using the proj crate
 pub fn st_transform_impl() -> ScalarKernelRef {
     Arc::new(STTransform {})
 }
 
+#[derive(Debug)]
+struct STTransform {}
+
+impl SedonaScalarKernel for STTransform {
+    fn return_type_from_args_and_scalars(
+        &self,
+        arg_types: &[SedonaType],
+        scalar_args: &[Option<&ScalarValue>],
+    ) -> Result<Option<SedonaType>> {
+        let inputs = zip(arg_types, scalar_args)
+            .map(|(arg_type, arg_scalar)| 
ArgInput::from_return_type_arg(arg_type, *arg_scalar))
+            .collect::<Vec<_>>();
+
+        if inputs.len() == 2 {
+            match (inputs[0], inputs[1]) {
+                // ScalarCrs output always returns a Wkb output type with 
concrete Crs
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(scalar_value)) => {
+                    Ok(Some(output_type_from_scalar_crs_value(scalar_value)?))
+                }
+
+                // Geo or ItemCrs with ArrayCrs output always return ItemCrs 
output
+                (ArgInput::Geo(_), ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs) => {
+                    Ok(Some(WKB_GEOMETRY_ITEM_CRS.clone()))
+                }
+                _ => Ok(None),
+            }
+        } else if inputs.len() == 3 {
+            match (inputs[0], inputs[1], inputs[2]) {
+                // ScalarCrs output always returns a Wkb output type with 
concrete Crs
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(_), 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::Geo(_), ArgInput::ArrayCrs, 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(_), 
ArgInput::ScalarCrs(scalar_value))
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs, 
ArgInput::ScalarCrs(scalar_value)) => {
+                    Ok(Some(output_type_from_scalar_crs_value(scalar_value)?))
+                }
+
+                // Geo or ItemCrs with ArrayCrs output always return ItemCrs 
output
+                (ArgInput::Geo(_), ArgInput::ScalarCrs(_), ArgInput::ArrayCrs)
+                | (ArgInput::Geo(_), ArgInput::ArrayCrs, ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ScalarCrs(_), 
ArgInput::ArrayCrs)
+                | (ArgInput::ItemCrs, ArgInput::ArrayCrs, ArgInput::ArrayCrs) 
=> {
+                    Ok(Some(WKB_GEOMETRY_ITEM_CRS.clone()))
+                }
+                _ => Ok(None),
+            }
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn invoke_batch_from_args(
+        &self,
+        arg_types: &[SedonaType],
+        args: &[ColumnarValue],
+        _return_type: &SedonaType,
+        _num_rows: usize,
+    ) -> Result<ColumnarValue> {
+        let inputs = zip(arg_types, args)
+            .map(|(arg_type, arg)| ArgInput::from_arg(arg_type, arg))
+            .collect::<Vec<_>>();
+
+        let executor = WkbExecutor::new(arg_types, args);
+        let mut builder = BinaryBuilder::with_capacity(
+            executor.num_iterations(),
+            WKB_MIN_PROBABLE_BYTES * executor.num_iterations(),
+        );
+
+        // Optimize the easy case, where we have exactly one transformation 
and there are no
+        // null or missing CRSes to contend with.
+        let from_index = inputs.len() - 2;
+        let to_index = inputs.len() - 1;
+        let (from, to) = (inputs[from_index], inputs[to_index]);
+        if let (Some(from_constant), Some(to_constant)) = 
(from.crs_constant()?, to.crs_constant()?)
+        {
+            let maybe_from_crs = deserialize_crs(&from_constant)?;
+            let maybe_to_crs = deserialize_crs(&to_constant)?;
+            if let (Some(from_crs), Some(to_crs)) = (maybe_from_crs, 
maybe_to_crs) {
+                with_global_proj_engine(|engine| {
+                    let crs_transform = engine
+                        .get_transform_crs_to_crs(
+                            &from_crs.to_crs_string(),
+                            &to_crs.to_crs_string(),
+                            None,
+                            "",
+                        )
+                        .map_err(|e| 
DataFusionError::Execution(format!("{e}")))?;
+                    executor.execute_wkb_void(|maybe_wkb| {
+                        match maybe_wkb {
+                            Some(wkb) => {
+                                invoke_scalar(&wkb, crs_transform.as_ref(), 
&mut builder)?;
+                                builder.append_value([]);
+                            }
+                            None => builder.append_null(),
+                        }
+                        Ok(())
+                    })?;
+                    Ok(())
+                })?;
+                return executor.finish(Arc::new(builder.finish()));
+            }
+        }
+
+        // Iterate over pairs of CRS strings
+        let from_crs_array = from.crs_array(&args[from_index], 
executor.num_iterations())?;
+        let to_crs_array = to.crs_array(&args[to_index], 
executor.num_iterations())?;
+        let from_crs_string_view_array = 
as_string_view_array(&from_crs_array)?;
+        let to_crs_string_view_array = as_string_view_array(&to_crs_array)?;
+        let mut crs_to_crs_iter = zip(from_crs_string_view_array, 
to_crs_string_view_array);
+
+        // We might need to build an output array of sanitized CRS strings
+        let mut maybe_crs_ouput = if matches!(to, ArgInput::ArrayCrs) {
+            Some(StringViewBuilder::with_capacity(executor.num_iterations()))
+        } else {
+            None
+        };
+
+        with_global_proj_engine(|engine| {
+            executor.execute_wkb_void(|maybe_wkb| {
+                match (maybe_wkb, crs_to_crs_iter.next().unwrap()) {
+                    (Some(wkb), (Some(from_crs_str), Some(to_crs_str))) => {
+                        let maybe_from_crs = deserialize_crs(from_crs_str)?;
+                        let maybe_to_crs = deserialize_crs(to_crs_str)?;
+
+                        if let Some(crs_output) = &mut maybe_crs_ouput {

Review Comment:
   Corrected spelling of 'ouput' to 'output'.



##########
c/sedona-proj/src/st_transform.rs:
##########
@@ -113,503 +435,370 @@ thread_local! {
     };
 }
 
-struct TransformArgIndexes {
-    wkb: usize,
-    first_crs: usize,
-    second_crs: Option<usize>,
-    lenient: Option<usize>,
-}
-
-impl TransformArgIndexes {
-    fn new() -> Self {
-        Self {
-            wkb: 0,
-            first_crs: 1,
-            second_crs: None,
-            lenient: None,
-        }
-    }
-}
-
-fn define_arg_indexes(arg_types: &[SedonaType], indexes: &mut 
TransformArgIndexes) {
-    indexes.wkb = 0;
-    indexes.first_crs = 1;
-
-    for (i, arg_type) in arg_types.iter().enumerate().skip(2) {
-        if ArgMatcher::is_numeric().match_type(arg_type)
-            || ArgMatcher::is_string().match_type(arg_type)
-        {
-            indexes.second_crs = Some(i);
-        } else if *arg_type == SedonaType::Arrow(DataType::Boolean) {
-            indexes.lenient = Some(i);
-        }
-    }
-}
-
-impl SedonaScalarKernel for STTransform {
-    fn return_type(&self, _args: &[SedonaType]) -> Result<Option<SedonaType>, 
DataFusionError> {
-        Err(DataFusionError::Internal(
-            "Return type should only be called with args".to_string(),
-        ))
-    }
-    fn return_type_from_args_and_scalars(
-        &self,
-        arg_types: &[SedonaType],
-        scalar_args: &[Option<&ScalarValue>],
-    ) -> Result<Option<SedonaType>> {
-        let matcher = ArgMatcher::new(
-            vec![
-                ArgMatcher::is_geometry(),
-                ArgMatcher::or(vec![ArgMatcher::is_numeric(), 
ArgMatcher::is_string()]),
-                ArgMatcher::optional(ArgMatcher::or(vec![
-                    ArgMatcher::is_numeric(),
-                    ArgMatcher::is_string(),
-                ])),
-                ArgMatcher::optional(ArgMatcher::is_boolean()),
-            ],
-            SedonaType::Wkb(Edges::Planar, None),
-        );
-
-        if !matcher.matches(arg_types) {
-            return Ok(None);
-        }
-
-        let mut indexes = TransformArgIndexes::new();
-        define_arg_indexes(arg_types, &mut indexes);
-
-        let scalar_arg_opt = if let Some(second_crs_index) = 
indexes.second_crs {
-            scalar_args.get(second_crs_index).unwrap()
-        } else {
-            scalar_args.get(indexes.first_crs).unwrap()
-        };
-
-        let crs_str_opt = if let Some(scalar_crs) = scalar_arg_opt {
-            to_crs_str(scalar_crs)
-        } else {
-            None
-        };
-
-        // If there is no CRS argument, we cannot determine the return type.
-        match crs_str_opt {
-            Some(to_crs) => {
-                let crs = deserialize_crs(&to_crs)?;
-                Ok(Some(SedonaType::Wkb(Edges::Planar, crs)))
-            }
-            _ => Ok(Some(SedonaType::Wkb(Edges::Planar, None))),
-        }
-    }
-
-    fn invoke_batch(
-        &self,
-        arg_types: &[SedonaType],
-        args: &[ColumnarValue],
-    ) -> Result<ColumnarValue> {
-        let executor = WkbExecutor::new(arg_types, args);
-        let mut builder = BinaryBuilder::with_capacity(
-            executor.num_iterations(),
-            WKB_MIN_PROBABLE_BYTES * executor.num_iterations(),
-        );
-
-        let mut indexes = TransformArgIndexes::new();
-        define_arg_indexes(arg_types, &mut indexes);
-
-        let first_crs = get_crs_str(args, indexes.first_crs).ok_or_else(|| {
-            DataFusionError::Execution(
-                "First CRS argument must be a string or numeric 
scalar".to_string(),
-            )
-        })?;
-
-        let lenient = indexes
-            .lenient
-            .is_some_and(|i| get_scalar_bool(args, i).unwrap_or(false));
-
-        let second_crs = if let Some(second_crs_index) = indexes.second_crs {
-            get_crs_str(args, second_crs_index)
-        } else {
-            None
-        };
-
-        with_global_proj_engine(|engine| {
-            let crs_from_geo = parse_source_crs(&arg_types[indexes.wkb])?;
-
-            let transform = match &second_crs {
-                Some(to_crs) => get_transform_crs_to_crs(engine, &first_crs, 
to_crs)?,
-                None => get_transform_to_crs(engine, crs_from_geo, &first_crs, 
lenient)?,
-            };
-
-            executor.execute_wkb_void(|maybe_wkb| {
-                match maybe_wkb {
-                    Some(wkb) => invoke_scalar(&wkb, transform.as_ref(), &mut 
builder)?,
-                    None => builder.append_null(),
-                }
-
-                Ok(())
-            })?;
-
-            Ok(())
-        })?;
-
-        executor.finish(Arc::new(builder.finish()))
-    }
-}
-
-fn get_transform_to_crs(
-    engine: &dyn CrsEngine,
-    source_crs_opt: Option<String>,
-    to_crs: &str,
-    lenient: bool,
-) -> Result<Rc<dyn CrsTransform>, DataFusionError> {
-    let from_crs = match source_crs_opt {
-        Some(crs) => crs,
-        None if lenient => "EPSG:4326".to_string(),
-        None => {
-            return Err(DataFusionError::Execution(
-                "Source CRS is required when transforming to a 
CRS".to_string(),
-            ))
-        }
-    };
-    get_transform_crs_to_crs(engine, &from_crs, to_crs)
-}
-
-fn get_transform_crs_to_crs(
-    engine: &dyn CrsEngine,
-    from_crs: &str,
-    to_crs: &str,
-) -> Result<Rc<dyn CrsTransform>, DataFusionError> {
-    engine
-        .get_transform_crs_to_crs(from_crs, to_crs, None, "")
-        .map_err(|err| DataFusionError::Execution(format!("Transform error: 
{err}")))
-}
-
-fn invoke_scalar(wkb: &Wkb, trans: &dyn CrsTransform, builder: &mut 
BinaryBuilder) -> Result<()> {
-    let geo_geom = wkb.to_geometry();
-    transform(&geo_geom, trans, builder)
-        .map_err(|err| DataFusionError::Execution(format!("Transform error: 
{err}")))?;
-    builder.append_value([]);
-    Ok(())
-}
-
-fn parse_source_crs(source_type: &SedonaType) -> Result<Option<String>> {
-    match source_type {
-        SedonaType::Wkb(_, Some(crs)) | SedonaType::WkbView(_, Some(crs)) => {
-            Ok(Some(crs.to_crs_string()))
-        }
-        _ => Ok(None),
-    }
-}
-
-fn to_crs_str(scalar_arg: &ScalarValue) -> Option<String> {
-    if let Ok(ScalarValue::Utf8(Some(crs))) = 
scalar_arg.cast_to(&DataType::Utf8) {
-        if crs.chars().all(|c| c.is_ascii_digit()) {
-            return Some(format!("EPSG:{crs}"));
-        } else {
-            return Some(crs);
-        }
-    }
-
-    None
-}
-
-fn get_crs_str(args: &[ColumnarValue], index: usize) -> Option<String> {
-    if let ColumnarValue::Scalar(scalar_crs) = &args[index] {
-        return to_crs_str(scalar_crs);
-    }
-    None
-}
-
-fn get_scalar_bool(args: &[ColumnarValue], index: usize) -> Option<bool> {
-    if let Some(ColumnarValue::Scalar(ScalarValue::Boolean(opt_bool))) = 
args.get(index) {
-        *opt_bool
-    } else {
-        None
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
+    use arrow_array::create_array;
     use arrow_array::ArrayRef;
-    use arrow_schema::{DataType, Field};
-    use datafusion_common::config::ConfigOptions;
-    use datafusion_expr::{ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, 
ScalarUDFImpl};
+    use arrow_schema::DataType;
     use rstest::rstest;
     use sedona_expr::scalar_udf::SedonaScalarUDF;
     use sedona_schema::crs::lnglat;
     use sedona_schema::crs::Crs;
     use sedona_schema::datatypes::WKB_GEOMETRY;
-    use sedona_testing::compare::assert_value_equal;
-    use sedona_testing::{create::create_array, create::create_array_value};
+    use sedona_testing::compare::assert_array_equal;
+    use sedona_testing::create::create_array;
+    use sedona_testing::create::create_array_item_crs;
+    use sedona_testing::create::create_scalar;
+    use sedona_testing::testers::ScalarUdfTester;
 
     const NAD83ZONE6PROJ: &str = "EPSG:2230";
     const WGS84: &str = "EPSG:4326";
 
-    #[rstest]
-    fn invalid_arg_checks() {
-        let udf: SedonaScalarUDF = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
-
-        // No args
-        let result = udf.return_field_from_args(ReturnFieldArgs {
-            arg_fields: &[],
-            scalar_arguments: &[],
-        });
-        assert!(
-            result.is_err()
-                && result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("No kernel matching arguments")
+    #[test]
+    fn test_invoke_with_string() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let geometry_input = SedonaType::Wkb(Edges::Planar, lnglat());
+        let tester = ScalarUdfTester::new(
+            udf.into(),
+            vec![geometry_input.clone(), SedonaType::Arrow(DataType::Utf8)],
         );
 
-        // Too many args
-        let arg_types = [
-            WKB_GEOMETRY,
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Boolean),
-            SedonaType::Arrow(DataType::Int32),
-        ];
-        let arg_fields: Vec<Arc<Field>> = arg_types
-            .iter()
-            .map(|arg_type| Arc::new(arg_type.to_storage_field("", 
true).unwrap()))
-            .collect();
-        let result = udf.return_field_from_args(ReturnFieldArgs {
-            arg_fields: &arg_fields,
-            scalar_arguments: &[None, None, None, None, None],
-        });
-        assert!(
-            result.is_err()
-                && result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("No kernel matching arguments")
-        );
+        // Return type with scalar to argument (returns type-level CRS)
+        let expected_return_type = SedonaType::Wkb(Edges::Planar, 
get_crs(NAD83ZONE6PROJ));
+        let return_type = tester
+            .return_type_with_scalar_scalar(Option::<&str>::None, 
Some(NAD83ZONE6PROJ))
+            .unwrap();
+        assert_eq!(return_type, expected_return_type);
 
-        // First arg not geometry
-        let arg_types = [
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Utf8),
-        ];
-        let arg_fields: Vec<Arc<Field>> = arg_types
-            .iter()
-            .map(|arg_type| Arc::new(arg_type.to_storage_field("", 
true).unwrap()))
-            .collect();
-        let result = udf.return_field_from_args(ReturnFieldArgs {
-            arg_fields: &arg_fields,
-            scalar_arguments: &[None, None],
-        });
-        assert!(
-            result.is_err()
-                && result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("No kernel matching arguments")
+        // Return type with array to argument (returns item CRS)
+        let return_type = tester.return_type().unwrap();
+        assert_eq!(return_type, WKB_GEOMETRY_ITEM_CRS.clone());
+
+        // Invoke with scalar to argument (returns type-level CRS)
+        let expected_array = create_array(
+            &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
+            &expected_return_type,
         );
+        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let result = tester.invoke_array_scalar(wkb, NAD83ZONE6PROJ).unwrap();
+        assert_array_equal(&result, &expected_array);
 
-        // Second arg not string or numeric
-        let arg_types = [WKB_GEOMETRY, SedonaType::Arrow(DataType::Boolean)];
-        let arg_fields: Vec<Arc<Field>> = arg_types
-            .iter()
-            .map(|arg_type| Arc::new(arg_type.to_storage_field("", 
true).unwrap()))
-            .collect();
-        let result = udf.return_field_from_args(ReturnFieldArgs {
-            arg_fields: &arg_fields,
-            scalar_arguments: &[None, None],
-        });
-        assert!(
-            result.is_err()
-                && result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("No kernel matching arguments")
+        // Invoke with array to argument (returns item CRS)
+        let expected_array = create_array_item_crs(
+            &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
+            [None, Some(NAD83ZONE6PROJ)],
+            &WKB_GEOMETRY,
         );
+        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let crs = create_array!(Utf8, [None, Some(NAD83ZONE6PROJ)]) as 
ArrayRef;
+        let result = tester.invoke_array_array(wkb, crs).unwrap();
+        assert_array_equal(&result, &expected_array);
     }
 
-    #[rstest]
-    fn test_invoke_batch_with_geo_crs() {
-        // From-CRS pulled from sedona type
-        let arg_types = [
-            SedonaType::Wkb(Edges::Planar, lnglat()),
-            SedonaType::Arrow(DataType::Utf8),
-        ];
+    #[test]
+    fn test_invoke_with_srid() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let geometry_input = SedonaType::Wkb(Edges::Planar, lnglat());
+        let tester = ScalarUdfTester::new(
+            udf.into(),
+            vec![geometry_input.clone(), SedonaType::Arrow(DataType::UInt32)],
+        );
 
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&arg_types[0]);
+        // Return type with scalar to argument (returns type-level CRS)
+        let expected_return_type = SedonaType::Wkb(Edges::Planar, 
get_crs(NAD83ZONE6PROJ));
+        let return_type = tester
+            .return_type_with_scalar_scalar(Option::<&str>::None, Some(2230))
+            .unwrap();
+        assert_eq!(return_type, expected_return_type);
 
-        let scalar_args = 
vec![ScalarValue::Utf8(Some(NAD83ZONE6PROJ.to_string()))];
+        // Return type with array to argument (returns item CRS)
+        let return_type = tester.return_type().unwrap();
+        assert_eq!(return_type, WKB_GEOMETRY_ITEM_CRS.clone());
 
-        let expected = create_array_value(
+        // Invoke with scalar to argument (returns type-level CRS)
+        let expected_array = create_array(
             &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
-            &SedonaType::Wkb(Edges::Planar, get_crs(NAD83ZONE6PROJ)),
+            &expected_return_type,
         );
+        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let result = tester.invoke_array_scalar(wkb, 2230).unwrap();
+        assert_array_equal(&result, &expected_array);
 
-        let (result_type, result_col) =
-            invoke_udf_test(wkb, scalar_args, arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, get_crs(NAD83ZONE6PROJ))
+        // Invoke with array to argument (returns item CRS)
+        let expected_array = create_array_item_crs(
+            &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
+            [None, Some(NAD83ZONE6PROJ)],
+            &WKB_GEOMETRY,
         );
+        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let crs = create_array!(Int32, [None, Some(2230)]) as ArrayRef;
+        let result = tester.invoke_array_array(wkb, crs).unwrap();
+        assert_array_equal(&result, &expected_array);
     }
 
-    #[rstest]
-    fn test_invoke_with_srids() {
-        // Use an integer SRID for the to CRS
-        let arg_types = [
-            SedonaType::Wkb(Edges::Planar, lnglat()),
-            SedonaType::Arrow(DataType::UInt32),
-        ];
-
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&arg_types[0]);
-
-        let scalar_args = vec![ScalarValue::UInt32(Some(2230))];
+    #[test]
+    fn test_invoke_with_item_crs() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let geometry_input = WKB_GEOMETRY_ITEM_CRS.clone();
+        let tester = ScalarUdfTester::new(
+            udf.into(),
+            vec![geometry_input.clone(), SedonaType::Arrow(DataType::Utf8)],
+        );
 
-        let expected = create_array_value(
+        // Return type with scalar to argument (returns type-level CRS)
+        // This is the same as for normal input
+        let expected_return_type = SedonaType::Wkb(Edges::Planar, 
get_crs(NAD83ZONE6PROJ));
+        let return_type = tester
+            .return_type_with_scalar_scalar(Option::<&str>::None, 
Some(NAD83ZONE6PROJ))
+            .unwrap();
+        assert_eq!(return_type, expected_return_type);
+
+        // Return type with array to argument (returns item CRS)
+        // This is the same as for normal input
+        let return_type = tester.return_type().unwrap();
+        assert_eq!(return_type, WKB_GEOMETRY_ITEM_CRS.clone());
+
+        // Invoke with scalar to argument (returns type-level CRS)
+        let expected_array = create_array(
             &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
-            &SedonaType::Wkb(Edges::Planar, get_crs(NAD83ZONE6PROJ)),
+            &expected_return_type,
         );
-
-        let (result_type, result_col) =
-            invoke_udf_test(wkb, scalar_args, arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, get_crs(NAD83ZONE6PROJ))
+        let array_in = create_array_item_crs(
+            &[None, Some("POINT (79.3871 43.6426)")],
+            [None, Some("EPSG:4326")],
+            &WKB_GEOMETRY,
         );
-    }
+        let result = tester
+            .invoke_array_scalar(array_in, NAD83ZONE6PROJ)
+            .unwrap();
+        assert_array_equal(&result, &expected_array);
 
-    #[rstest]
-    fn test_invoke_batch_with_lenient() {
-        let arg_types = [
-            WKB_GEOMETRY,
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Boolean),
-        ];
-
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&WKB_GEOMETRY);
-        let scalar_args = vec![
-            ScalarValue::Utf8(Some(NAD83ZONE6PROJ.to_string())),
-            ScalarValue::Boolean(Some(true)),
-        ];
-
-        let expected = create_array_value(
+        // Invoke with array to argument (returns item CRS)
+        let expected_array = create_array_item_crs(
             &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
-            &SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap())),
+            [None, Some(NAD83ZONE6PROJ)],
+            &WKB_GEOMETRY,
         );
-
-        let (result_type, result_col) =
-            invoke_udf_test(wkb, scalar_args, arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap()))
+        let array_in = create_array_item_crs(
+            &[None, Some("POINT (79.3871 43.6426)")],
+            [None, Some("EPSG:4326")],
+            &WKB_GEOMETRY,
         );
+        let crs = create_array!(Utf8, [None, Some(NAD83ZONE6PROJ)]) as 
ArrayRef;
+        let result = tester.invoke_array_array(array_in, crs).unwrap();
+        assert_array_equal(&result, &expected_array);
     }
 
     #[rstest]
-    fn test_invoke_batch_one_crs_no_lenient() {
-        let arg_types = [WKB_GEOMETRY, SedonaType::Arrow(DataType::Utf8)];
+    fn test_invoke_source_arg() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let geometry_input = WKB_GEOMETRY;
+        let tester = ScalarUdfTester::new(
+            udf.into(),
+            vec![
+                geometry_input.clone(),
+                SedonaType::Arrow(DataType::Utf8),
+                SedonaType::Arrow(DataType::Utf8),
+            ],
+        );
 
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&WKB_GEOMETRY);
-        let scalar_args = 
vec![ScalarValue::Utf8(Some(NAD83ZONE6PROJ.to_string()))];
+        // Return type with scalar to argument (returns type-level CRS)
+        // This is the same as for normal input
+        let expected_return_type = SedonaType::Wkb(Edges::Planar, 
get_crs(NAD83ZONE6PROJ));
+        let return_type = tester
+            .return_type_with_scalar_scalar_scalar(
+                Option::<&str>::None,
+                Option::<&str>::None,
+                Some(NAD83ZONE6PROJ),
+            )
+            .unwrap();
+        assert_eq!(return_type, expected_return_type);
 
-        let err = invoke_udf_test(wkb, scalar_args, arg_types.to_vec());
-        assert!(
-            matches!(err, Err(DataFusionError::Execution(_))),
-            "Expected an Execution error"
-        );
-    }
+        // Return type with array to argument (returns item CRS)
+        // This is the same as for normal input
+        let return_type = tester.return_type().unwrap();
+        assert_eq!(return_type, WKB_GEOMETRY_ITEM_CRS.clone());
 
-    #[rstest]
-    fn test_invoke_batch_with_source_arg() {
-        let arg_types = [
-            WKB_GEOMETRY,
-            SedonaType::Arrow(DataType::Utf8),
-            SedonaType::Arrow(DataType::Utf8),
-        ];
-
-        let wkb = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&WKB_GEOMETRY);
-
-        let scalar_args = vec![
-            ScalarValue::Utf8(Some(WGS84.to_string())),
-            ScalarValue::Utf8(Some(NAD83ZONE6PROJ.to_string())),
-        ];
-
-        // Note: would be nice to have an epsilon of tolerance when validating
-        let expected = create_array_value(
+        // Invoke with scalar to argument (returns type-level CRS)
+        let expected_array = create_array(
             &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
-            &SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap())),
+            &expected_return_type,
         );
+        let array_in = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&geometry_input);
+        let crs_from = create_array!(Utf8, [None, Some(WGS84)]) as ArrayRef;
+        let result = tester
+            .invoke_array_array_scalar(array_in, crs_from, NAD83ZONE6PROJ)
+            .unwrap();
+        assert_array_equal(&result, &expected_array);
+
+        // Invoke with array to argument (returns item CRS)
+        let expected_array = create_array_item_crs(
+            &[None, Some("POINT (-21508577.363421552 34067918.06097863)")],
+            [None, Some(NAD83ZONE6PROJ)],
+            &WKB_GEOMETRY,
+        );
+        let array_in = create_array(&[None, Some("POINT (79.3871 43.6426)")], 
&WKB_GEOMETRY);
+        let crs_from = create_array!(Utf8, [None, Some(WGS84)]) as ArrayRef;
+        let crs_to = create_array!(Utf8, [None, Some(NAD83ZONE6PROJ)]) as 
ArrayRef;
+        let result = tester
+            .invoke_arrays(vec![array_in, crs_from, crs_to])
+            .unwrap();
+        assert_array_equal(&result, &expected_array);
+    }
 
-        let (result_type, result_col) =
-            invoke_udf_test(wkb.clone(), scalar_args, 
arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap()))
+    #[test]
+    fn test_invoke_null_crs_to() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let tester = ScalarUdfTester::new(
+            udf.clone().into(),
+            vec![WKB_GEOMETRY, SedonaType::Arrow(DataType::Utf8)],
         );
 
-        // Test with integer SRIDs
-        let arg_types = [
-            WKB_GEOMETRY,
-            SedonaType::Arrow(DataType::Int32),
-            SedonaType::Arrow(DataType::Int32),
-        ];
-
-        let scalar_args = vec![
-            ScalarValue::Int32(Some(4326)),
-            ScalarValue::Int32(Some(2230)),
-        ];
-
-        let (result_type, result_col) =
-            invoke_udf_test(wkb, scalar_args, arg_types.to_vec()).unwrap();
-        assert_value_equal(&result_col, &expected);
-        assert_eq!(
-            result_type,
-            SedonaType::Wkb(Edges::Planar, 
Some(get_crs(NAD83ZONE6PROJ).unwrap()))
+        // A null scalar CRS should generate WKB_GEOMETRY output with a a type
+        // level CRS that is unset; however, all the output will be null.
+        let result = tester
+            .invoke_scalar_scalar("POINT (0 1)", ScalarValue::Null)
+            .unwrap();
+        assert_eq!(result, create_scalar(None, &WKB_GEOMETRY));
+
+        let expected_array = create_array(&[None, None, None], &WKB_GEOMETRY);
+        let array_in = create_array(
+            &[
+                Some("POINT (0 1)"),
+                Some("POINT (1 2)"),
+                Some("POINT (2 3)"),
+            ],
+            &WKB_GEOMETRY,
+        );
+        let result = tester
+            .invoke_array_scalar(array_in, ScalarValue::Null)
+            .unwrap();
+        assert_array_equal(&result, &expected_array);
+
+        // This currently has a side effect of working even though there is not
+        // valid transform from lnglat() to an unset CRS (because no 
transformations
+        // will ever take place).
+        let geometry_input = SedonaType::Wkb(Edges::Planar, lnglat());
+        let tester = ScalarUdfTester::new(
+            udf.clone().into(),
+            vec![geometry_input, SedonaType::Arrow(DataType::Utf8)],
         );
+        let result = tester
+            .invoke_scalar_scalar("POINT (0 1)", ScalarValue::Null)
+            .unwrap();
+        assert_eq!(result, create_scalar(None, &WKB_GEOMETRY));
     }
 
-    fn get_crs(auth_code: &str) -> Crs {
-        deserialize_crs(auth_code).unwrap()
+    #[test]
+    fn test_invoke_unset_crs_to() {
+        let udf = SedonaScalarUDF::from_impl("st_transform", 
st_transform_impl());
+        let tester = ScalarUdfTester::new(
+            udf.clone().into(),
+            vec![WKB_GEOMETRY, SedonaType::Arrow(DataType::Int32)],
+        );
+
+        // A unset scalar CRS should generate WKB_GEOMETRY output with a a type

Review Comment:
   Remove duplicate 'a' in 'with a a type', and change 'A unset' to 'An unset'.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscr...@sedona.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org