(datafusion) branch main updated: FFI: forward ScalarUDF preserves_lex_ordering (#23069)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git


The following commit(s) were added to refs/heads/main by this push:
     new 9facaaabd0 FFI: forward ScalarUDF preserves_lex_ordering (#23069)
9facaaabd0 is described below

commit 9facaaabd001d0f82ecc6bcd4d09adcc6f731109
Author: Amogh Ramesh <ra...@gmail.com>
AuthorDate: Mon Jul 27 19:58:38 2026 +0530

    FFI: forward ScalarUDF preserves_lex_ordering (#23069)
    
    ## Which issue does this PR close?
    
    Part of #22330.
    
    ## Rationale for this change
    
    `ForeignScalarUDF` inherits the default `preserves_lex_ordering`, so
    producer overrides are lost across the FFI boundary.
    
    ## What changes are included in this PR?
    
    - Forward `preserves_lex_ordering` through `FFI_ScalarUDF`.
    - Reuse the existing placement UDF for unit and dynamic-library
    coverage.
    
    ## Are these changes tested?
    
    - `cargo test -p datafusion-ffi --features integration-tests`
    - `cargo clippy --all-targets --all-features -- -D warnings`
    
    ## Are there any user-facing changes?
    
    The FFI ABI changes. Foreign libraries must rebuild against the new
    DataFusion version.
    
    ---------
    
    Signed-off-by: Amogh Ramesh <ra...@gmail.com>
---
 datafusion/ffi/src/tests/udf_udaf_udwf.rs |  8 +++++
 datafusion/ffi/src/udf/mod.rs             | 57 +++++++++++++++++++++++++++++++
 datafusion/ffi/tests/ffi_udf.rs           | 10 ++++--
 3 files changed, 73 insertions(+), 2 deletions(-)

diff --git a/datafusion/ffi/src/tests/udf_udaf_udwf.rs b/datafusion/ffi/src/tests/udf_udaf_udwf.rs
index b393f5db3a..a84df52b8d 100644
--- a/datafusion/ffi/src/tests/udf_udaf_udwf.rs
+++ b/datafusion/ffi/src/tests/udf_udaf_udwf.rs
@@ -20,6 +20,7 @@ use std::sync::Arc;
 use arrow_schema::DataType;
 use datafusion_catalog::TableFunctionImpl;
 use datafusion_common::ScalarValue;
+use datafusion_expr::sort_properties::ExprProperties;
 use datafusion_expr::{
     AggregateUDF, ColumnarValue, ExpressionPlacement, ScalarFunctionArgs, ScalarUDF,
     ScalarUDFImpl, Signature, Volatility, WindowUDF,
@@ -152,6 +153,13 @@ impl ScalarUDFImpl for PlacementUDF {
             ExpressionPlacement::KeepInPlace
         }
     }
+
+    fn preserves_lex_ordering(
+        &self,
+        inputs: &[ExprProperties],
+    ) -> datafusion_common::Result<bool> {
+        Ok(inputs.iter().all(|input| input.preserves_lex_ordering))
+    }
 }
 
 pub(crate) extern "C" fn create_placement_func() -> FFI_ScalarUDF {
diff --git a/datafusion/ffi/src/udf/mod.rs b/datafusion/ffi/src/udf/mod.rs
index 4fc22e859f..8e96dd9013 100644
--- a/datafusion/ffi/src/udf/mod.rs
+++ b/datafusion/ffi/src/udf/mod.rs
@@ -26,6 +26,7 @@ use arrow::ffi::{FFI_ArrowSchema, from_ffi, to_ffi};
 use arrow_schema::FieldRef;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::{DataFusionError, Result, internal_err};
+use datafusion_expr::sort_properties::ExprProperties;
 use datafusion_expr::type_coercion::functions::fields_with_udf;
 use datafusion_expr::{
     ColumnarValue, ExpressionPlacement, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF,
@@ -41,6 +42,7 @@ use stabby::vec::Vec as SVec;
 use crate::arrow_wrappers::{WrappedArray, WrappedSchema};
 use crate::config::FFI_ConfigOptions;
 use crate::expr::columnar_value::FFI_ColumnarValue;
+use crate::expr::expr_properties::FFI_ExprProperties;
 use crate::placement::FFI_ExpressionPlacement;
 use crate::util::{
     FFI_Result, rvec_wrapped_to_vec_datatype, vec_datatype_to_rvec_wrapped,
@@ -115,6 +117,12 @@ pub struct FFI_ScalarUDF {
     /// the foreign interface. See [`crate::get_library_marker_id`] and
     /// the crate's `README.md` for more information.
     pub library_marker_id: extern "C" fn() -> usize,
+
+    /// FFI equivalent to [`ScalarUDFImpl::preserves_lex_ordering`].
+    pub preserves_lex_ordering: unsafe extern "C" fn(
+        udf: &Self,
+        inputs: SVec<FFI_ExprProperties>,
+    ) -> FFI_Result<bool>,
 }
 
 unsafe impl Send for FFI_ScalarUDF {}
@@ -178,6 +186,19 @@ unsafe extern "C" fn placement_fn_wrapper(
     udf.inner().placement(&args).into()
 }
 
+unsafe extern "C" fn preserves_lex_ordering_fn_wrapper(
+    udf: &FFI_ScalarUDF,
+    inputs: SVec<FFI_ExprProperties>,
+) -> FFI_Result<bool> {
+    let result = inputs
+        .into_iter()
+        .map(ExprProperties::try_from)
+        .collect::<Result<Vec<_>>>()
+        .and_then(|inputs| udf.inner().preserves_lex_ordering(&inputs));
+
+    sresult!(result)
+}
+
 unsafe extern "C" fn invoke_with_args_fn_wrapper(
     udf: &FFI_ScalarUDF,
     args: SVec<WrappedArray>,
@@ -276,6 +297,7 @@ impl From<Arc<ScalarUDF>> for FFI_ScalarUDF {
             release: release_fn_wrapper,
             private_data: Box::into_raw(private_data) as *mut c_void,
             library_marker_id: crate::get_library_marker_id,
+            preserves_lex_ordering: preserves_lex_ordering_fn_wrapper,
         }
     }
 }
@@ -460,6 +482,18 @@ impl ScalarUDFImpl for ForeignScalarUDF {
 
         result.into()
     }
+
+    fn preserves_lex_ordering(&self, inputs: &[ExprProperties]) -> Result<bool> {
+        inputs
+            .iter()
+            .map(FFI_ExprProperties::try_from)
+            .collect::<Result<SVec<_>>>()
+            .and_then(|inputs| {
+                let result =
+                    unsafe { (self.udf.preserves_lex_ordering)(&self.udf, inputs) };
+                df_result!(result)
+            })
+    }
 }
 
 #[cfg(test)]
@@ -500,6 +534,14 @@ mod tests {
                 ExpressionPlacement::KeepInPlace
             }
         }
+
+        fn preserves_lex_ordering(&self, inputs: &[ExprProperties]) -> Result<bool> {
+            if inputs.is_empty() {
+                return internal_err!("preserves_lex_ordering requires an input");
+            }
+
+            Ok(inputs.iter().all(|input| input.preserves_lex_ordering))
+        }
     }
 
     #[test]
@@ -572,6 +614,21 @@ mod tests {
         );
         assert_eq!(foreign_udf.placement(&[]), ExpressionPlacement::KeepInPlace);
 
+        let preserves = ExprProperties::new_unknown().with_preserves_lex_ordering(true);
+        let does_not_preserve = ExprProperties::new_unknown();
+
+        assert!(
+            foreign_udf
+                .preserves_lex_ordering(std::slice::from_ref(&preserves))
+                .unwrap()
+        );
+        assert!(
+            !foreign_udf
+                .preserves_lex_ordering(&[preserves, does_not_preserve])
+                .unwrap()
+        );
+        assert!(foreign_udf.preserves_lex_ordering(&[]).is_err());
+
         Ok(())
     }
 }
diff --git a/datafusion/ffi/tests/ffi_udf.rs b/datafusion/ffi/tests/ffi_udf.rs
index 617cbc196b..d9e7263ccd 100644
--- a/datafusion/ffi/tests/ffi_udf.rs
+++ b/datafusion/ffi/tests/ffi_udf.rs
@@ -26,6 +26,7 @@ mod tests {
     use datafusion::prelude::{SessionContext, col};
     use datafusion_execution::config::SessionConfig;
     use datafusion_expr::lit;
+    use datafusion_expr::sort_properties::ExprProperties;
     use datafusion_ffi::tests::create_record_batch;
     use datafusion_ffi::tests::utils::get_module;
     use std::sync::Arc;
@@ -90,8 +91,7 @@ mod tests {
         Ok(())
     }
 
-    /// This test validates that a producer's `placement` override survives the
-    /// FFI boundary instead of collapsing to the default `KeepInPlace`.
+    /// Checks planning-property overrides across the FFI boundary.
     #[tokio::test]
     async fn test_scalar_udf_placement() -> Result<()> {
         let module = get_module()?;
@@ -112,6 +112,12 @@ mod tests {
             ExpressionPlacement::KeepInPlace
         );
 
+        let preserves = ExprProperties::new_unknown().with_preserves_lex_ordering(true);
+        let does_not_preserve = ExprProperties::new_unknown();
+
+        assert!(foreign_func.preserves_lex_ordering(std::slice::from_ref(&preserves))?);
+        assert!(!foreign_func.preserves_lex_ordering(&[preserves, does_not_preserve])?);
+
         Ok(())
     }
 


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org