Remove core dependency from ffi (#19422)

## Which issue does this PR close?

Closes #18671 

## Rationale for this change

With the latest changes for #18671 we no longer require `datafusion`
crate as a dependency. This will reduce build times for users. Also it
guarantees we do not accidentally introduce code that will create a
`SessionContext` or any other large binary inside our FFI
implementations.

## What changes are included in this PR?

- Remove `datafusion` crate from Cargo.toml
- Update paths
- Apply consistent formatting

## Are these changes tested?

Existing unit tests.

## Are there any user-facing changes?

No. This only updates paths.

Author: timsaucer

Cargo.lock

@@ -40,19 +40,25 @@ workspace = true
 name = "datafusion_ffi"
 crate-type = ["cdylib", "rlib"]
 
+# Note to developers: do *not* add `datafusion` as a dependency in this crate.
+# It increases build times and library binary size for users.
+
 [dependencies]
 abi_stable = "0.11.3"
 arrow = { workspace = true, features = ["ffi"] }
 arrow-schema = { workspace = true }
 async-ffi = { version = "0.5.0", features = ["abi_stable"] }
 async-trait = { workspace = true }
-datafusion = { workspace = true, default-features = false }
 datafusion-catalog = { workspace = true }
 datafusion-common = { workspace = true }
 datafusion-datasource = { workspace = true }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
+datafusion-functions = { workspace = true, optional = true }
+datafusion-functions-aggregate = { workspace = true, optional = true }
 datafusion-functions-aggregate-common = { workspace = true }
+datafusion-functions-table = { workspace = true, optional = true }
+datafusion-functions-window = { workspace = true, optional = true }
 datafusion-physical-expr = { workspace = true }
 datafusion-physical-expr-common = { workspace = true }
 datafusion-physical-plan = { workspace = true }
@@ -74,5 +80,10 @@ datafusion-functions-window = { workspace = true }
 doc-comment = { workspace = true }
 
 [features]
-integration-tests = []
+integration-tests = [
+    "datafusion-functions",
+    "datafusion-functions-aggregate",
+    "datafusion-functions-table",
+    "datafusion-functions-window",
+]
 tarpaulin_include = [] # Exists only to prevent warnings on stable and still have accurate coverage

datafusion/ffi/Cargo.toml

@@ -18,12 +18,10 @@
 use std::sync::Arc;
 
 use abi_stable::StableAbi;
-use arrow::{
-    array::{ArrayRef, make_array},
-    datatypes::{Schema, SchemaRef},
-    error::ArrowError,
-    ffi::{FFI_ArrowArray, FFI_ArrowSchema, from_ffi, to_ffi},
-};
+use arrow::array::{ArrayRef, make_array};
+use arrow::datatypes::{Schema, SchemaRef};
+use arrow::error::ArrowError;
+use arrow::ffi::{FFI_ArrowArray, FFI_ArrowSchema, from_ffi, to_ffi};
 use datafusion_common::{DataFusionError, ScalarValue};
 use log::error;
 

datafusion/ffi/src/arrow_wrappers.rs

@@ -15,26 +15,24 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::{ffi::c_void, pin::Pin, sync::Arc};
-
-use abi_stable::{
-    StableAbi,
-    std_types::{RString, RVec},
-};
-use datafusion::{
-    error::DataFusionError,
-    execution::{SendableRecordBatchStream, TaskContext},
-    physical_plan::{DisplayAs, ExecutionPlan, PlanProperties},
+use std::ffi::c_void;
+use std::pin::Pin;
+use std::sync::Arc;
+
+use abi_stable::StableAbi;
+use abi_stable::std_types::{RString, RVec};
+use datafusion_common::{DataFusionError, Result};
+use datafusion_execution::{SendableRecordBatchStream, TaskContext};
+use datafusion_physical_plan::{
+    DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
 };
-use datafusion::{error::Result, physical_plan::DisplayFormatType};
 use tokio::runtime::Handle;
 
 use crate::execution::FFI_TaskContext;
+use crate::plan_properties::FFI_PlanProperties;
+use crate::record_batch_stream::FFI_RecordBatchStream;
 use crate::util::FFIResult;
-use crate::{
-    df_result, plan_properties::FFI_PlanProperties,
-    record_batch_stream::FFI_RecordBatchStream, rresult,
-};
+use crate::{df_result, rresult};
 
 /// A stable struct for sharing a [`ExecutionPlan`] across FFI boundaries.
 #[repr(C)]
@@ -300,12 +298,11 @@ impl ExecutionPlan for ForeignExecutionPlan {
 
 #[cfg(test)]
 pub(crate) mod tests {
-    use super::*;
     use arrow::datatypes::{DataType, Field, Schema};
-    use datafusion::physical_plan::{
-        Partitioning,
-        execution_plan::{Boundedness, EmissionType},
-    };
+    use datafusion::physical_plan::Partitioning;
+    use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
+
+    use super::*;
 
     #[derive(Debug)]
     pub struct EmptyExec {

datafusion/ffi/src/execution_plan.rs

@@ -15,15 +15,16 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::arrow_wrappers::WrappedArray;
-use crate::expr::interval::FFI_Interval;
 use abi_stable::StableAbi;
 use datafusion_common::DataFusionError;
 use datafusion_expr::statistics::{
     BernoulliDistribution, Distribution, ExponentialDistribution, GaussianDistribution,
     GenericDistribution, UniformDistribution,
 };
 
+use crate::arrow_wrappers::WrappedArray;
+use crate::expr::interval::FFI_Interval;
+
 /// A stable struct for sharing [`Distribution`] across FFI boundaries.
 /// See ['Distribution'] for the meaning of each variant.
 #[repr(C)]

datafusion/ffi/src/expr/distribution.rs

@@ -15,11 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::arrow_wrappers::WrappedArray;
 use abi_stable::StableAbi;
 use datafusion_common::DataFusionError;
 use datafusion_expr::interval_arithmetic::Interval;
 
+use crate::arrow_wrappers::WrappedArray;
+
 /// A stable struct for sharing [`Interval`] across FFI boundaries.
 /// See [`Interval`] for the meaning of each field. Scalar values
 /// are passed as Arrow arrays of length 1.

datafusion/ffi/src/expr/interval.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use abi_stable::StableAbi;
-use datafusion::logical_expr::logical_plan::dml::InsertOp;
+use datafusion_expr::logical_plan::dml::InsertOp;
 
 /// FFI safe version of [`InsertOp`].
 #[repr(C)]

datafusion/ffi/src/insert_op.rs

@@ -18,42 +18,36 @@
 pub(crate) mod partitioning;
 pub(crate) mod sort;
 
-use std::{
-    any::Any,
-    ffi::c_void,
-    fmt::{Display, Formatter},
-    hash::{DefaultHasher, Hash, Hasher},
-    sync::Arc,
-};
-
-use abi_stable::{
-    StableAbi,
-    std_types::{ROption, RResult, RString, RVec},
-};
-use arrow::{
-    array::{ArrayRef, BooleanArray, RecordBatch},
-    datatypes::SchemaRef,
-};
-use arrow_schema::{DataType, Field, FieldRef, Schema, ffi::FFI_ArrowSchema};
+use std::any::Any;
+use std::ffi::c_void;
+use std::fmt::{Display, Formatter};
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use abi_stable::StableAbi;
+use abi_stable::std_types::{ROption, RResult, RString, RVec};
+use arrow::array::{ArrayRef, BooleanArray, RecordBatch};
+use arrow::datatypes::SchemaRef;
+use arrow_schema::ffi::FFI_ArrowSchema;
+use arrow_schema::{DataType, Field, FieldRef, Schema};
 use datafusion_common::{Result, ffi_datafusion_err};
-use datafusion_expr::{
-    ColumnarValue, interval_arithmetic::Interval, sort_properties::ExprProperties,
-    statistics::Distribution,
-};
+use datafusion_expr::ColumnarValue;
+use datafusion_expr::interval_arithmetic::Interval;
+use datafusion_expr::sort_properties::ExprProperties;
+use datafusion_expr::statistics::Distribution;
 use datafusion_physical_expr::PhysicalExpr;
 use datafusion_physical_expr_common::physical_expr::fmt_sql;
 
-use crate::{
-    arrow_wrappers::{WrappedArray, WrappedSchema},
-    df_result,
-    expr::{
-        columnar_value::FFI_ColumnarValue, distribution::FFI_Distribution,
-        expr_properties::FFI_ExprProperties, interval::FFI_Interval,
-    },
-    record_batch_stream::{record_batch_to_wrapped_array, wrapped_array_to_record_batch},
-    rresult, rresult_return,
-    util::FFIResult,
+use crate::arrow_wrappers::{WrappedArray, WrappedSchema};
+use crate::expr::columnar_value::FFI_ColumnarValue;
+use crate::expr::distribution::FFI_Distribution;
+use crate::expr::expr_properties::FFI_ExprProperties;
+use crate::expr::interval::FFI_Interval;
+use crate::record_batch_stream::{
+    record_batch_to_wrapped_array, wrapped_array_to_record_batch,
 };
+use crate::util::FFIResult;
+use crate::{df_result, rresult, rresult_return};
 
 #[repr(C)]
 #[derive(Debug, StableAbi)]
@@ -737,14 +731,14 @@ impl Display for ForeignPhysicalExpr {
 
 #[cfg(test)]
 mod tests {
-    use std::{
-        hash::{DefaultHasher, Hash, Hasher},
-        sync::Arc,
-    };
+    use std::hash::{DefaultHasher, Hash, Hasher};
+    use std::sync::Arc;
 
     use arrow::array::{BooleanArray, RecordBatch, record_batch};
-    use datafusion_common::{DataFusionError, ScalarValue, tree_node::DynTreeNode};
-    use datafusion_expr::{interval_arithmetic::Interval, statistics::Distribution};
+    use datafusion_common::tree_node::DynTreeNode;
+    use datafusion_common::{DataFusionError, ScalarValue};
+    use datafusion_expr::interval_arithmetic::Interval;
+    use datafusion_expr::statistics::Distribution;
     use datafusion_physical_expr::expressions::{Column, NegativeExpr, NotExpr};
     use datafusion_physical_expr_common::physical_expr::{PhysicalExpr, fmt_sql};
 

datafusion/ffi/src/physical_expr/mod.rs

@@ -17,7 +17,8 @@
 
 use std::sync::Arc;
 
-use abi_stable::{StableAbi, std_types::RVec};
+use abi_stable::StableAbi;
+use abi_stable::std_types::RVec;
 use datafusion_physical_expr::Partitioning;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 
@@ -70,7 +71,8 @@ impl From<&FFI_Partitioning> for Partitioning {
 
 #[cfg(test)]
 mod tests {
-    use datafusion_physical_expr::{Partitioning, expressions::lit};
+    use datafusion_physical_expr::Partitioning;
+    use datafusion_physical_expr::expressions::lit;
 
     use crate::physical_expr::partitioning::FFI_Partitioning;
 

datafusion/ffi/src/physical_expr/partitioning.rs

@@ -55,12 +55,14 @@ impl From<&FFI_PhysicalSortExpr> for PhysicalSortExpr {
 
 #[cfg(test)]
 mod tests {
-    use crate::physical_expr::sort::FFI_PhysicalSortExpr;
+    use std::sync::Arc;
+
     use arrow_schema::SortOptions;
     use datafusion_physical_expr::expressions::Column;
     use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
     use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
-    use std::sync::Arc;
+
+    use crate::physical_expr::sort::FFI_PhysicalSortExpr;
 
     #[test]
     fn ffi_sort_expr_round_trip() {