FFI: plumb placement for FFI_ScalarUDF (#22608)

## Which issue does this PR close?

- Part of #22330.

This is the first of the per-method PRs that issue describes. It plumbs
`placement` only; the remaining defaulted methods follow separately, so
the umbrella issue stays open.

## Rationale for this change

`FFI_ScalarUDF` (`datafusion/ffi/src/udf/mod.rs`) carried no function
pointer for `placement`, and `ForeignScalarUDF` did not override it, so
a producer's override of `ScalarUDFImpl::placement` (default body at
`datafusion/expr/src/udf.rs:1028`) was dropped on the consumer side and
every foreign UDF fell back to `KeepInPlace`. A UDF loaded over FFI
never delivered its leaf-pushdown hint to the optimizer.

## What changes are included in this PR?

- New `FFI_ExpressionPlacement` enum bridge in
`datafusion/ffi/src/placement.rs`, in the shape of `FFI_Volatility`:
`#[repr(u8)]` with `From` impls both ways and a round-trip test over
every variant.
- A `placement` function pointer on `FFI_ScalarUDF`, populated in the
`From<Arc<ScalarUDF>>` constructor, with `placement_fn_wrapper` on the
producer side and a forwarding `ForeignScalarUDF::placement` on the
consumer side. `placement` is infallible, so the pointer returns the
enum directly rather than `FFI_Result`.

Adding a field to the `#[repr(C)]` struct changes its layout, so this is
an API change and should carry the `api change` label (I can't add it
myself). It targets `main` and should not be back-ported to a release
branch.

`display_name` is also on the issue's list, but it has been deprecated
since 50.0.0, so it should be dropped from the gap list rather than
plumbed. I have left it and the remaining methods to follow-up PRs.

## Are these changes tested?

Yes.

- Unit: a round-trip test over all four `ExpressionPlacement` variants,
plus a forced-foreign test (`mock_foreign_marker_id`) using a UDF whose
`placement` override depends on its arguments. The assertions cover
ordered, reordered, and empty argument slices, so argument marshalling
is checked, not just the return value.
- Integration: `tests/ffi_udf.rs` loads the UDF from the real cdylib and
asserts the override survives the boundary, which is the surface a
layout change needs.

Run with `cargo test -p datafusion-ffi` and `cargo test -p
datafusion-ffi --features integration-tests`.

## Are there any user-facing changes?

A `placement` override on a `ScalarUDFImpl` is now preserved across the
FFI boundary instead of being silently replaced by the default. This is
an ABI change to `FFI_ScalarUDF`; consumers must be recompiled against
the new layout.

---------

Signed-off-by: Amogh Ramesh <ramogh2404@gmail.com>

Author: Amogh-2404

datafusion/ffi/src/lib.rs

@@ -36,6 +36,7 @@ pub mod ffi_option;
 pub mod insert_op;
 pub mod physical_expr;
 pub mod physical_optimizer;
+pub mod placement;
 pub mod plan_properties;
 pub mod proto;
 pub mod record_batch_stream;

datafusion/ffi/src/placement.rs

@@ -0,0 +1,72 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion_expr::ExpressionPlacement;
+
+#[expect(non_camel_case_types)]
+#[repr(u8)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FFI_ExpressionPlacement {
+    Literal,
+    Column,
+    MoveTowardsLeafNodes,
+    KeepInPlace,
+}
+
+impl From<ExpressionPlacement> for FFI_ExpressionPlacement {
+    fn from(value: ExpressionPlacement) -> Self {
+        match value {
+            ExpressionPlacement::Literal => Self::Literal,
+            ExpressionPlacement::Column => Self::Column,
+            ExpressionPlacement::MoveTowardsLeafNodes => Self::MoveTowardsLeafNodes,
+            ExpressionPlacement::KeepInPlace => Self::KeepInPlace,
+        }
+    }
+}
+
+impl From<FFI_ExpressionPlacement> for ExpressionPlacement {
+    fn from(value: FFI_ExpressionPlacement) -> Self {
+        match value {
+            FFI_ExpressionPlacement::Literal => Self::Literal,
+            FFI_ExpressionPlacement::Column => Self::Column,
+            FFI_ExpressionPlacement::MoveTowardsLeafNodes => Self::MoveTowardsLeafNodes,
+            FFI_ExpressionPlacement::KeepInPlace => Self::KeepInPlace,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use datafusion::logical_expr::ExpressionPlacement;
+
+    use super::FFI_ExpressionPlacement;
+
+    fn test_round_trip_placement(placement: ExpressionPlacement) {
+        let ffi_placement: FFI_ExpressionPlacement = placement.into();
+        let round_trip: ExpressionPlacement = ffi_placement.into();
+
+        assert_eq!(placement, round_trip);
+    }
+
+    #[test]
+    fn test_all_round_trip_placement() {
+        test_round_trip_placement(ExpressionPlacement::Literal);
+        test_round_trip_placement(ExpressionPlacement::Column);
+        test_round_trip_placement(ExpressionPlacement::MoveTowardsLeafNodes);
+        test_round_trip_placement(ExpressionPlacement::KeepInPlace);
+    }
+}

datafusion/ffi/src/tests/mod.rs

@@ -90,6 +90,8 @@ pub struct ForeignLibraryModule {
 
     pub create_timezone_udf: extern "C" fn() -> FFI_ScalarUDF,
 
+    pub create_placement_udf: extern "C" fn() -> FFI_ScalarUDF,
+
     pub create_table_function:
         extern "C" fn(FFI_LogicalExtensionCodec) -> FFI_TableFunction,
 
@@ -251,6 +253,7 @@ pub extern "C" fn datafusion_ffi_get_module() -> ForeignLibraryModule {
         create_scalar_udf: create_ffi_abs_func,
         create_nullary_udf: create_ffi_random_func,
         create_timezone_udf: udf_udaf_udwf::create_timezone_func,
+        create_placement_udf: udf_udaf_udwf::create_placement_func,
         create_table_function: create_ffi_table_func,
         create_sum_udaf: create_ffi_sum_func,
         create_stddev_udaf: create_ffi_stddev_func,

datafusion/ffi/src/tests/udf_udaf_udwf.rs

@@ -21,8 +21,8 @@ use arrow_schema::DataType;
 use datafusion_catalog::TableFunctionImpl;
 use datafusion_common::ScalarValue;
 use datafusion_expr::{
-    AggregateUDF, ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
-    Volatility, WindowUDF,
+    AggregateUDF, ColumnarValue, ExpressionPlacement, ScalarFunctionArgs, ScalarUDF,
+    ScalarUDFImpl, Signature, Volatility, WindowUDF,
 };
 use datafusion_functions::math::abs::AbsFunc;
 use datafusion_functions::math::random::RandomFunc;
@@ -112,6 +112,56 @@ pub(crate) extern "C" fn create_timezone_func() -> FFI_ScalarUDF {
     udf.into()
 }
 
+#[derive(Debug, PartialEq, Eq, Hash)]
+struct PlacementUDF {
+    signature: Signature,
+}
+
+impl ScalarUDFImpl for PlacementUDF {
+    fn name(&self) -> &str {
+        "placement_udf"
+    }
+
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn return_type(
+        &self,
+        _arg_types: &[DataType],
+    ) -> datafusion_common::Result<DataType> {
+        Ok(DataType::Int64)
+    }
+
+    fn invoke_with_args(
+        &self,
+        _args: ScalarFunctionArgs,
+    ) -> datafusion_common::Result<ColumnarValue> {
+        datafusion_common::internal_err!("placement_udf is not meant to be invoked")
+    }
+
+    fn placement(&self, args: &[ExpressionPlacement]) -> ExpressionPlacement {
+        // Push to the leaves only for a (Column, Literal) pairing, so the
+        // test catches dropped, reordered, or truncated arguments.
+        if matches!(
+            args,
+            [ExpressionPlacement::Column, ExpressionPlacement::Literal]
+        ) {
+            ExpressionPlacement::MoveTowardsLeafNodes
+        } else {
+            ExpressionPlacement::KeepInPlace
+        }
+    }
+}
+
+pub(crate) extern "C" fn create_placement_func() -> FFI_ScalarUDF {
+    let udf: Arc<ScalarUDF> = Arc::new(ScalarUDF::from(PlacementUDF {
+        signature: Signature::uniform(1, vec![DataType::Int64], Volatility::Immutable),
+    }));
+
+    udf.into()
+}
+
 pub(crate) extern "C" fn create_ffi_table_func(
     codec: FFI_LogicalExtensionCodec,
 ) -> FFI_TableFunction {

datafusion/ffi/src/udf/mod.rs

@@ -28,8 +28,8 @@ use datafusion_common::config::ConfigOptions;
 use datafusion_common::{DataFusionError, Result, internal_err};
 use datafusion_expr::type_coercion::functions::fields_with_udf;
 use datafusion_expr::{
-    ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl,
-    Signature,
+    ColumnarValue, ExpressionPlacement, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF,
+    ScalarUDFImpl, Signature,
 };
 use return_type_args::{
     FFI_ReturnFieldArgs, ForeignReturnFieldArgs, ForeignReturnFieldArgsOwned,
@@ -41,6 +41,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::placement::FFI_ExpressionPlacement;
 use crate::util::{
     FFI_Result, rvec_wrapped_to_vec_datatype, vec_datatype_to_rvec_wrapped,
 };
@@ -91,6 +92,14 @@ pub struct FFI_ScalarUDF {
         arg_types: SVec<WrappedSchema>,
     ) -> FFI_Result<SVec<WrappedSchema>>,
 
+    /// FFI equivalent to the `placement` of a [`ScalarUDFImpl`]. Returns the
+    /// placement hint for the underlying [`ScalarUDF`] given each argument's
+    /// placement. Infallible, so it returns the value directly, not an `FFI_Result`.
+    pub placement: unsafe extern "C" fn(
+        udf: &Self,
+        args: SVec<FFI_ExpressionPlacement>,
+    ) -> FFI_ExpressionPlacement,
+
     /// Used to create a clone on the provider of the udf. This should
     /// only need to be called by the receiver of the udf.
     pub clone: unsafe extern "C" fn(udf: &Self) -> Self,
@@ -157,6 +166,18 @@ unsafe extern "C" fn coerce_types_fn_wrapper(
     sresult!(vec_datatype_to_rvec_wrapped(&return_types))
 }
 
+unsafe extern "C" fn placement_fn_wrapper(
+    udf: &FFI_ScalarUDF,
+    args: SVec<FFI_ExpressionPlacement>,
+) -> FFI_ExpressionPlacement {
+    let args = args
+        .into_iter()
+        .map(ExpressionPlacement::from)
+        .collect::<Vec<_>>();
+
+    udf.inner().placement(&args).into()
+}
+
 unsafe extern "C" fn invoke_with_args_fn_wrapper(
     udf: &FFI_ScalarUDF,
     args: SVec<WrappedArray>,
@@ -250,6 +271,7 @@ impl From<Arc<ScalarUDF>> for FFI_ScalarUDF {
             invoke_with_args: invoke_with_args_fn_wrapper,
             return_field_from_args: return_field_from_args_fn_wrapper,
             coerce_types: coerce_types_fn_wrapper,
+            placement: placement_fn_wrapper,
             clone: clone_fn_wrapper,
             release: release_fn_wrapper,
             private_data: Box::into_raw(private_data) as *mut c_void,
@@ -427,12 +449,59 @@ impl ScalarUDFImpl for ForeignScalarUDF {
             Ok(rvec_wrapped_to_vec_datatype(&result_types)?)
         }
     }
+
+    fn placement(&self, args: &[ExpressionPlacement]) -> ExpressionPlacement {
+        let args = args
+            .iter()
+            .map(|p| FFI_ExpressionPlacement::from(*p))
+            .collect::<SVec<_>>();
+
+        let result = unsafe { (self.udf.placement)(&self.udf, args) };
+
+        result.into()
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[derive(Debug, PartialEq, Eq, Hash)]
+    struct PlacementUDF {
+        signature: Signature,
+    }
+
+    impl ScalarUDFImpl for PlacementUDF {
+        fn name(&self) -> &str {
+            "placement_udf"
+        }
+
+        fn signature(&self) -> &Signature {
+            &self.signature
+        }
+
+        fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+            Ok(DataType::Int64)
+        }
+
+        fn invoke_with_args(&self, _args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+            internal_err!("placement_udf is not meant to be invoked")
+        }
+
+        fn placement(&self, args: &[ExpressionPlacement]) -> ExpressionPlacement {
+            // Push to the leaves only for a (Column, Literal) pairing, so the
+            // test catches dropped, reordered, or truncated arguments.
+            if matches!(
+                args,
+                [ExpressionPlacement::Column, ExpressionPlacement::Literal]
+            ) {
+                ExpressionPlacement::MoveTowardsLeafNodes
+            } else {
+                ExpressionPlacement::KeepInPlace
+            }
+        }
+    }
+
     #[test]
     fn test_round_trip_scalar_udf() -> Result<()> {
         let original_udf = datafusion::functions::math::abs::AbsFunc::new();
@@ -467,4 +536,42 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_ffi_udf_placement_round_trip() -> Result<()> {
+        use datafusion_expr::Volatility;
+
+        let original_udf = Arc::new(ScalarUDF::from(PlacementUDF {
+            signature: Signature::uniform(
+                1,
+                vec![DataType::Int64],
+                Volatility::Immutable,
+            ),
+        }));
+
+        let mut ffi_udf = FFI_ScalarUDF::from(original_udf);
+
+        // Force the foreign path so the call travels through the FFI vtable
+        // rather than downcasting back to the original local type.
+        ffi_udf.library_marker_id = crate::mock_foreign_marker_id;
+        let foreign_udf: Arc<dyn ScalarUDFImpl> = (&ffi_udf).into();
+        assert!(foreign_udf.is::<ForeignScalarUDF>());
+
+        // Without the plumbing the override is dropped and every call is
+        // KeepInPlace. The three cases also check the arguments survive the
+        // round trip in order.
+        assert_eq!(
+            foreign_udf
+                .placement(&[ExpressionPlacement::Column, ExpressionPlacement::Literal]),
+            ExpressionPlacement::MoveTowardsLeafNodes
+        );
+        assert_eq!(
+            foreign_udf
+                .placement(&[ExpressionPlacement::Literal, ExpressionPlacement::Column]),
+            ExpressionPlacement::KeepInPlace
+        );
+        assert_eq!(foreign_udf.placement(&[]), ExpressionPlacement::KeepInPlace);
+
+        Ok(())
+    }
 }

datafusion/ffi/tests/ffi_udf.rs

@@ -23,7 +23,7 @@ mod tests {
     use arrow::datatypes::DataType;
     use datafusion::common::record_batch;
     use datafusion::error::Result;
-    use datafusion::logical_expr::{ScalarUDF, ScalarUDFImpl};
+    use datafusion::logical_expr::{ExpressionPlacement, ScalarUDF, ScalarUDFImpl};
     use datafusion::prelude::{SessionContext, col};
     use datafusion_execution::config::SessionConfig;
     use datafusion_expr::lit;
@@ -91,6 +91,31 @@ mod tests {
         Ok(())
     }
 
+    /// This test validates that a producer's `placement` override survives the
+    /// FFI boundary instead of collapsing to the default `KeepInPlace`.
+    #[tokio::test]
+    async fn test_scalar_udf_placement() -> Result<()> {
+        let module = get_module()?;
+
+        let ffi_placement_func = (module.create_placement_udf)();
+        let foreign_func: Arc<dyn ScalarUDFImpl> = (&ffi_placement_func).into();
+
+        // The override pushes to the leaves only for (Column, Literal), so these
+        // also check the arguments cross the boundary in order.
+        assert_eq!(
+            foreign_func
+                .placement(&[ExpressionPlacement::Column, ExpressionPlacement::Literal]),
+            ExpressionPlacement::MoveTowardsLeafNodes
+        );
+        assert_eq!(
+            foreign_func
+                .placement(&[ExpressionPlacement::Literal, ExpressionPlacement::Column]),
+            ExpressionPlacement::KeepInPlace
+        );
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_config_on_scalar_udf() -> Result<()> {
         let module = get_module()?;