You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "timsaucer (via GitHub)" <gi...@apache.org> on 2025/11/24 20:43:57 UTC
[PR] Implement task context and task context provider [datafusion]
timsaucer opened a new pull request, #18918:
URL: https://github.com/apache/datafusion/pull/18918
## Which issue does this PR close?
Addresses part of https://github.com/apache/datafusion/issues/18671 but does not close it.
## Rationale for this change
We have an issue in the current FFI code in that we will need access to the `TaskContext` to do function encoding and decoding to pass logical expressions across the boundary. We use the proto crate for this operations.
We cannot encode and decode functions that have been registered since our current code creates a default session context. This causes problems in crates where we are using UDFs as inputs to either aggregations, window functions, or table filters.
With this change we keep a _weak_ reference to a new trait, a `TaskContextProvider`. By keeping a weak trait we make sure we do not create a circular dependency between function that is internally holding on to the provider, which holds the task context, which holds the function.
## What changes are included in this PR?
- Introduce the `TaskContextProvider` trait
- Implement FFI versions of `TaskContext` and `TaskContextProvider`.
This PR does _not_ use these structures in the current code. That is coming as part of a later PR in an effort to keep the size of the PRs small for effective code review.
## Are these changes tested?
- [ ] TODO add unit tests
## Are there any user-facing changes?
No.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "ntjohnson1 (via GitHub)" <gi...@apache.org>.
ntjohnson1 commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2566234506
##########
datafusion/ffi/src/session_config.rs:
##########
@@ -142,24 +136,22 @@ impl Drop for FFI_SessionConfig {
}
}
-/// A wrapper struct for accessing [`SessionConfig`] across a FFI boundary.
-/// The [`SessionConfig`] will be generated from a hash map of the config
-/// options in the provider and will be reconstructed on this side of the
-/// interface.s
-pub struct ForeignSessionConfig(pub SessionConfig);
-
-impl TryFrom<&FFI_SessionConfig> for ForeignSessionConfig {
Review Comment:
It feels like maybe a note that the from_string approach allows directly converting rather than extracting with Foreign. This breaks a little bit from the pattern but definitely seems nicer here.
##########
datafusion/ffi/src/session_config.rs:
##########
@@ -15,17 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-use abi_stable::{
- std_types::{RHashMap, RString},
- StableAbi,
-};
-use datafusion::{config::ConfigOptions, error::Result};
-use datafusion::{error::DataFusionError, prelude::SessionConfig};
-use std::sync::Arc;
-use std::{
- collections::HashMap,
- ffi::{c_char, c_void, CString},
-};
+use std::collections::HashMap;
+use std::ffi::c_void;
+
+use abi_stable::std_types::{RHashMap, RString};
+use abi_stable::StableAbi;
+use datafusion_common::error::{DataFusionError, Result};
+use datafusion_execution::config::SessionConfig;
/// A stable struct for sharing [`SessionConfig`] across FFI boundaries.
/// Instead of attempting to expose the entire SessionConfig interface, we
Review Comment:
Hmm this is no longer strong ConfigOptions as private data. So this doc string doesn't seem to be totally correct. It still uses config options to go from FFI_ to SessionConfig but doesn't convert to config options to get into FFI
##########
datafusion/ffi/src/execution/task_ctx_provider.rs:
##########
@@ -0,0 +1,213 @@
+// 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 std::ffi::c_void;
+use std::sync::{Arc, Weak};
+
+use abi_stable::std_types::{RResult, RString};
+use abi_stable::StableAbi;
+use datafusion_common::{exec_datafusion_err, DataFusionError};
+use datafusion_execution::{TaskContext, TaskContextProvider};
+
+use crate::execution::task_ctx::FFI_TaskContext;
+use crate::{df_result, rresult};
+
+/// Struct for accessing the [`TaskContext`]. This method contains a weak
+/// reference, so there are no guarantees that the [`TaskContext`] remains
+/// valid. This is used primarily for protobuf encoding and decoding of
+/// data passed across the FFI boundary. See the crate README for
+/// additional information.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContextProvider {
+ pub task_ctx: unsafe extern "C" fn(&Self) -> RResult<FFI_TaskContext, RString>,
+
+ /// Used to create a clone on the task context accessor. This should
+ /// only need to be called by the receiver of the plan.
+ pub clone: unsafe extern "C" fn(plan: &Self) -> Self,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
+
+ /// Utility to identify when FFI objects are accessed locally through
+ /// 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,
+}
+
+unsafe impl Send for FFI_TaskContextProvider {}
+unsafe impl Sync for FFI_TaskContextProvider {}
+
+struct TaskContextProviderPrivateData {
+ ctx: Weak<dyn TaskContextProvider>,
Review Comment:
Similarly ok if it makes more sense to look at in follow up PR note here. Do you have a concrete code snippet example for the circular reference this breaks? I feel like a small comment with that might make this clearer for the future.
##########
datafusion/ffi/src/execution/task_ctx_provider.rs:
##########
@@ -0,0 +1,213 @@
+// 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 std::ffi::c_void;
+use std::sync::{Arc, Weak};
+
+use abi_stable::std_types::{RResult, RString};
+use abi_stable::StableAbi;
+use datafusion_common::{exec_datafusion_err, DataFusionError};
+use datafusion_execution::{TaskContext, TaskContextProvider};
+
+use crate::execution::task_ctx::FFI_TaskContext;
+use crate::{df_result, rresult};
+
+/// Struct for accessing the [`TaskContext`]. This method contains a weak
+/// reference, so there are no guarantees that the [`TaskContext`] remains
+/// valid. This is used primarily for protobuf encoding and decoding of
+/// data passed across the FFI boundary. See the crate README for
+/// additional information.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContextProvider {
Review Comment:
You mentioned this is used primarily for data coming across protobuf so should this be set more internally? Ex. pub(crate) or something. You mentioned usage in follow up so can review this point in context there if that makes more sense.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "renato2099 (via GitHub)" <gi...@apache.org>.
renato2099 commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2586582801
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
Review Comment:
I was thinking on explicitly forbidding cloning / copying, somthing like
```
impl !Clone for FFI_TaskContext {}
impl !Copy for FFI_TaskContext {}
```
I was thinking that worst case scenario , FFI users could do a memcopy of the struct , so that it would be somehow "safer" to have something like
```
pub struct FFI_TaskContext {
// An opaque pointer to an Arc<TaskContextPrivateData>
arc_ptr: *const Arc<TaskContextPrivateData>,
}
```
and then returning a raw pointer to inner data , something along these lines
```
pub fn as_opaque(&self) -> *const c_void {
self.arc_ptr as *const c_void
}
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2585105217
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
+
+ /// Utility to identify when FFI objects are accessed locally through
+ /// 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,
+}
+
+struct TaskContextPrivateData {
+ ctx: Arc<TaskContext>,
+}
+
+impl FFI_TaskContext {
+ unsafe fn inner(&self) -> &Arc<TaskContext> {
+ let private_data = self.private_data as *const TaskContextPrivateData;
+ &(*private_data).ctx
+ }
+}
+
+unsafe extern "C" fn session_id_fn_wrapper(ctx: &FFI_TaskContext) -> RString {
+ let ctx = ctx.inner();
+ ctx.session_id().into()
+}
+
+unsafe extern "C" fn task_id_fn_wrapper(ctx: &FFI_TaskContext) -> ROption<RString> {
+ let ctx = ctx.inner();
+ ctx.task_id().map(|s| s.as_str().into()).into()
+}
+
+unsafe extern "C" fn session_config_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> FFI_SessionConfig {
+ let ctx = ctx.inner();
+ ctx.session_config().into()
+}
+
+unsafe extern "C" fn scalar_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_ScalarUDF> {
+ let ctx = ctx.inner();
+ ctx.scalar_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), Arc::clone(udf).into()))
+ .collect()
+}
+
+unsafe extern "C" fn aggregate_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_AggregateUDF> {
+ let ctx = ctx.inner();
+ ctx.aggregate_functions()
+ .iter()
+ .map(|(name, udaf)| {
+ (
+ name.to_owned().into(),
+ FFI_AggregateUDF::from(Arc::clone(udaf)),
+ )
+ })
+ .collect()
+}
+
+unsafe extern "C" fn window_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_WindowUDF> {
+ let ctx = ctx.inner();
+ ctx.window_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), FFI_WindowUDF::from(Arc::clone(udf))))
+ .collect()
+}
+
+unsafe extern "C" fn release_fn_wrapper(ctx: &mut FFI_TaskContext) {
+ let private_data = Box::from_raw(ctx.private_data as *mut TaskContextPrivateData);
+ drop(private_data);
+}
+
+impl Drop for FFI_TaskContext {
+ fn drop(&mut self) {
+ unsafe { (self.release)(self) }
+ }
+}
+
+impl From<Arc<TaskContext>> for FFI_TaskContext {
+ fn from(ctx: Arc<TaskContext>) -> Self {
+ let private_data = Box::new(TaskContextPrivateData { ctx });
+
+ FFI_TaskContext {
+ session_id: session_id_fn_wrapper,
+ task_id: task_id_fn_wrapper,
+ session_config: session_config_fn_wrapper,
+ scalar_functions: scalar_functions_fn_wrapper,
+ aggregate_functions: aggregate_functions_fn_wrapper,
+ window_functions: window_functions_fn_wrapper,
+ release: release_fn_wrapper,
+ private_data: Box::into_raw(private_data) as *mut c_void,
+ library_marker_id: crate::get_library_marker_id,
+ }
+ }
+}
+
+impl From<FFI_TaskContext> for Arc<TaskContext> {
+ fn from(ffi_ctx: FFI_TaskContext) -> Self {
+ unsafe {
+ if (ffi_ctx.library_marker_id)() == crate::get_library_marker_id() {
+ return Arc::clone(ffi_ctx.inner());
+ }
+
+ let task_id = (ffi_ctx.task_id)(&ffi_ctx).map(|s| s.to_string()).into();
+ let session_id = (ffi_ctx.session_id)(&ffi_ctx).into();
+ let session_config = (ffi_ctx.session_config)(&ffi_ctx);
+ let session_config =
+ SessionConfig::try_from(&session_config).unwrap_or_default();
+
+ let scalar_functions = (ffi_ctx.scalar_functions)(&ffi_ctx)
+ .into_iter()
+ .filter_map(|kv_pair| {
+ let udf = <Arc<dyn ScalarUDFImpl>>::try_from(&kv_pair.1);
+
+ if let Err(err) = &udf {
+ log::error!("Unable to create WindowUDF in FFI: {err}")
Review Comment:
This is an excellent question! In reviewing I discovered that I had some intermediate work where my conversion of FFI UDFs was fallible but the final versions were infallible! I updated the traits to use `From<>` instead of `TryFrom<>` so this is no longer an issue in this code snippet. Thank you for bringing it up!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2564757785
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
+
+ /// Utility to identify when FFI objects are accessed locally through
+ /// 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,
+}
+
+struct TaskContextPrivateData {
+ ctx: Arc<TaskContext>,
+}
+
+impl FFI_TaskContext {
+ unsafe fn inner(&self) -> &Arc<TaskContext> {
+ let private_data = self.private_data as *const TaskContextPrivateData;
+ &(*private_data).ctx
+ }
+}
+
+unsafe extern "C" fn session_id_fn_wrapper(ctx: &FFI_TaskContext) -> RString {
+ let ctx = ctx.inner();
+ ctx.session_id().into()
+}
+
+unsafe extern "C" fn task_id_fn_wrapper(ctx: &FFI_TaskContext) -> ROption<RString> {
+ let ctx = ctx.inner();
+ ctx.task_id().map(|s| s.as_str().into()).into()
+}
+
+unsafe extern "C" fn session_config_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> FFI_SessionConfig {
+ let ctx = ctx.inner();
+ ctx.session_config().into()
+}
+
+unsafe extern "C" fn scalar_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_ScalarUDF> {
+ let ctx = ctx.inner();
+ ctx.scalar_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), Arc::clone(udf).into()))
+ .collect()
+}
+
+unsafe extern "C" fn aggregate_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_AggregateUDF> {
+ let ctx = ctx.inner();
+ ctx.aggregate_functions()
+ .iter()
+ .map(|(name, udaf)| {
+ (
+ name.to_owned().into(),
+ FFI_AggregateUDF::from(Arc::clone(udaf)),
+ )
+ })
+ .collect()
+}
+
+unsafe extern "C" fn window_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_WindowUDF> {
+ let ctx = ctx.inner();
+ ctx.window_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), FFI_WindowUDF::from(Arc::clone(udf))))
+ .collect()
+}
+
+unsafe extern "C" fn release_fn_wrapper(ctx: &mut FFI_TaskContext) {
+ let private_data = Box::from_raw(ctx.private_data as *mut TaskContextPrivateData);
+ drop(private_data);
+}
+
+impl Drop for FFI_TaskContext {
+ fn drop(&mut self) {
+ unsafe { (self.release)(self) }
+ }
+}
+
+impl From<Arc<TaskContext>> for FFI_TaskContext {
+ fn from(ctx: Arc<TaskContext>) -> Self {
+ let private_data = Box::new(TaskContextPrivateData { ctx });
+
+ FFI_TaskContext {
+ session_id: session_id_fn_wrapper,
+ task_id: task_id_fn_wrapper,
+ session_config: session_config_fn_wrapper,
+ scalar_functions: scalar_functions_fn_wrapper,
+ aggregate_functions: aggregate_functions_fn_wrapper,
+ window_functions: window_functions_fn_wrapper,
+ release: release_fn_wrapper,
+ private_data: Box::into_raw(private_data) as *mut c_void,
+ library_marker_id: crate::get_library_marker_id,
+ }
+ }
+}
+
+impl From<FFI_TaskContext> for Arc<TaskContext> {
+ fn from(ffi_ctx: FFI_TaskContext) -> Self {
+ unsafe {
+ if (ffi_ctx.library_marker_id)() == crate::get_library_marker_id() {
+ return Arc::clone(ffi_ctx.inner());
+ }
+
+ let task_id = (ffi_ctx.task_id)(&ffi_ctx).map(|s| s.to_string()).into();
+ let session_id = (ffi_ctx.session_id)(&ffi_ctx).into();
+ let session_config = (ffi_ctx.session_config)(&ffi_ctx);
+ let session_config =
+ SessionConfig::try_from(&session_config).unwrap_or_default();
+
+ let scalar_functions = (ffi_ctx.scalar_functions)(&ffi_ctx)
+ .into_iter()
+ .filter_map(|kv_pair| {
+ let udf = <Arc<dyn ScalarUDFImpl>>::try_from(&kv_pair.1);
+
+ if let Err(err) = &udf {
+ log::error!("Unable to create WindowUDF in FFI: {err}")
Review Comment:
The error message mentions 'WindowUDF' but this is in the scalar_functions conversion code. Should be 'ScalarUDF' to accurately reflect the function type being created.
```suggestion
log::error!("Unable to create ScalarUDF in FFI: {err}")
```
##########
datafusion/ffi/src/session_config.rs:
##########
@@ -172,13 +164,15 @@ mod tests {
let session_config = SessionConfig::new();
let original_options = session_config.options().entries();
- let ffi_config: FFI_SessionConfig = (&session_config).into();
+ let mut ffi_config: FFI_SessionConfig = (&session_config).into();
+ let _ = ffi_config.clone();
+ ffi_config.library_marker_id = crate::mock_foreign_marker_id;
- let foreign_config: ForeignSessionConfig = (&ffi_config).try_into()?;
+ let foreign_config: SessionConfig = (&ffi_config).try_into()?;
- let returned_options = foreign_config.0.options().entries();
+ let returned_options = foreign_config.options().entries();
- assert!(original_options.len() == returned_options.len());
+ assert_eq!(original_options.len(), returned_options.len());
Review Comment:
The test only verifies that the number of config options matches, but doesn't verify that the actual option values round-trip correctly. Consider adding assertions to check that specific config values are preserved.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#issuecomment-3607106744
Thank you for the reviews @ntjohnson1 @renato2099 @comphead ! I've rebased now that the other PR went in and made some adjustments based on the feedback.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "ntjohnson1 (via GitHub)" <gi...@apache.org>.
ntjohnson1 commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2566215384
##########
datafusion/ffi/src/session_config.rs:
##########
@@ -15,17 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-use abi_stable::{
- std_types::{RHashMap, RString},
- StableAbi,
-};
-use datafusion::{config::ConfigOptions, error::Result};
-use datafusion::{error::DataFusionError, prelude::SessionConfig};
-use std::sync::Arc;
-use std::{
- collections::HashMap,
- ffi::{c_char, c_void, CString},
-};
+use std::collections::HashMap;
+use std::ffi::c_void;
+
+use abi_stable::std_types::{RHashMap, RString};
+use abi_stable::StableAbi;
+use datafusion_common::error::{DataFusionError, Result};
+use datafusion_execution::config::SessionConfig;
/// A stable struct for sharing [`SessionConfig`] across FFI boundaries.
/// Instead of attempting to expose the entire SessionConfig interface, we
Review Comment:
Hmm this is no longer storing ConfigOptions as private data. So this doc string doesn't seem to be totally correct. It still uses config options to go from FFI_ to SessionConfig but doesn't convert to config options to get into FFI
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "renato2099 (via GitHub)" <gi...@apache.org>.
renato2099 commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2582874985
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
+
+ /// Utility to identify when FFI objects are accessed locally through
+ /// 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,
+}
+
+struct TaskContextPrivateData {
+ ctx: Arc<TaskContext>,
+}
+
+impl FFI_TaskContext {
+ unsafe fn inner(&self) -> &Arc<TaskContext> {
+ let private_data = self.private_data as *const TaskContextPrivateData;
+ &(*private_data).ctx
+ }
+}
+
+unsafe extern "C" fn session_id_fn_wrapper(ctx: &FFI_TaskContext) -> RString {
+ let ctx = ctx.inner();
+ ctx.session_id().into()
+}
+
+unsafe extern "C" fn task_id_fn_wrapper(ctx: &FFI_TaskContext) -> ROption<RString> {
+ let ctx = ctx.inner();
+ ctx.task_id().map(|s| s.as_str().into()).into()
+}
+
+unsafe extern "C" fn session_config_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> FFI_SessionConfig {
+ let ctx = ctx.inner();
+ ctx.session_config().into()
+}
+
+unsafe extern "C" fn scalar_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_ScalarUDF> {
+ let ctx = ctx.inner();
+ ctx.scalar_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), Arc::clone(udf).into()))
+ .collect()
+}
+
+unsafe extern "C" fn aggregate_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_AggregateUDF> {
+ let ctx = ctx.inner();
+ ctx.aggregate_functions()
+ .iter()
+ .map(|(name, udaf)| {
+ (
+ name.to_owned().into(),
+ FFI_AggregateUDF::from(Arc::clone(udaf)),
+ )
+ })
+ .collect()
+}
+
+unsafe extern "C" fn window_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_WindowUDF> {
+ let ctx = ctx.inner();
+ ctx.window_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), FFI_WindowUDF::from(Arc::clone(udf))))
+ .collect()
+}
+
+unsafe extern "C" fn release_fn_wrapper(ctx: &mut FFI_TaskContext) {
+ let private_data = Box::from_raw(ctx.private_data as *mut TaskContextPrivateData);
+ drop(private_data);
+}
+
+impl Drop for FFI_TaskContext {
+ fn drop(&mut self) {
+ unsafe { (self.release)(self) }
+ }
+}
+
+impl From<Arc<TaskContext>> for FFI_TaskContext {
+ fn from(ctx: Arc<TaskContext>) -> Self {
+ let private_data = Box::new(TaskContextPrivateData { ctx });
+
+ FFI_TaskContext {
+ session_id: session_id_fn_wrapper,
+ task_id: task_id_fn_wrapper,
+ session_config: session_config_fn_wrapper,
+ scalar_functions: scalar_functions_fn_wrapper,
+ aggregate_functions: aggregate_functions_fn_wrapper,
+ window_functions: window_functions_fn_wrapper,
+ release: release_fn_wrapper,
+ private_data: Box::into_raw(private_data) as *mut c_void,
+ library_marker_id: crate::get_library_marker_id,
+ }
+ }
+}
+
+impl From<FFI_TaskContext> for Arc<TaskContext> {
+ fn from(ffi_ctx: FFI_TaskContext) -> Self {
+ unsafe {
+ if (ffi_ctx.library_marker_id)() == crate::get_library_marker_id() {
+ return Arc::clone(ffi_ctx.inner());
+ }
+
+ let task_id = (ffi_ctx.task_id)(&ffi_ctx).map(|s| s.to_string()).into();
+ let session_id = (ffi_ctx.session_id)(&ffi_ctx).into();
+ let session_config = (ffi_ctx.session_config)(&ffi_ctx);
+ let session_config =
+ SessionConfig::try_from(&session_config).unwrap_or_default();
+
+ let scalar_functions = (ffi_ctx.scalar_functions)(&ffi_ctx)
+ .into_iter()
+ .filter_map(|kv_pair| {
+ let udf = <Arc<dyn ScalarUDFImpl>>::try_from(&kv_pair.1);
+
+ if let Err(err) = &udf {
+ log::error!("Unable to create WindowUDF in FFI: {err}")
Review Comment:
I have a question on the expected behavior here, let's say that we fail while attempting `try_From` , could not failing loud cause confusion/bugs?
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
+
+ /// Utility to identify when FFI objects are accessed locally through
+ /// 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,
+}
+
+struct TaskContextPrivateData {
+ ctx: Arc<TaskContext>,
+}
+
+impl FFI_TaskContext {
+ unsafe fn inner(&self) -> &Arc<TaskContext> {
+ let private_data = self.private_data as *const TaskContextPrivateData;
+ &(*private_data).ctx
+ }
+}
+
+unsafe extern "C" fn session_id_fn_wrapper(ctx: &FFI_TaskContext) -> RString {
+ let ctx = ctx.inner();
+ ctx.session_id().into()
+}
+
+unsafe extern "C" fn task_id_fn_wrapper(ctx: &FFI_TaskContext) -> ROption<RString> {
+ let ctx = ctx.inner();
+ ctx.task_id().map(|s| s.as_str().into()).into()
+}
+
+unsafe extern "C" fn session_config_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> FFI_SessionConfig {
+ let ctx = ctx.inner();
+ ctx.session_config().into()
+}
+
+unsafe extern "C" fn scalar_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_ScalarUDF> {
+ let ctx = ctx.inner();
+ ctx.scalar_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), Arc::clone(udf).into()))
+ .collect()
+}
+
+unsafe extern "C" fn aggregate_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_AggregateUDF> {
+ let ctx = ctx.inner();
+ ctx.aggregate_functions()
+ .iter()
+ .map(|(name, udaf)| {
+ (
+ name.to_owned().into(),
+ FFI_AggregateUDF::from(Arc::clone(udaf)),
+ )
+ })
+ .collect()
+}
+
+unsafe extern "C" fn window_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_WindowUDF> {
+ let ctx = ctx.inner();
+ ctx.window_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), FFI_WindowUDF::from(Arc::clone(udf))))
+ .collect()
+}
+
+unsafe extern "C" fn release_fn_wrapper(ctx: &mut FFI_TaskContext) {
+ let private_data = Box::from_raw(ctx.private_data as *mut TaskContextPrivateData);
+ drop(private_data);
+}
+
+impl Drop for FFI_TaskContext {
+ fn drop(&mut self) {
+ unsafe { (self.release)(self) }
+ }
+}
+
+impl From<Arc<TaskContext>> for FFI_TaskContext {
+ fn from(ctx: Arc<TaskContext>) -> Self {
+ let private_data = Box::new(TaskContextPrivateData { ctx });
+
+ FFI_TaskContext {
+ session_id: session_id_fn_wrapper,
+ task_id: task_id_fn_wrapper,
+ session_config: session_config_fn_wrapper,
+ scalar_functions: scalar_functions_fn_wrapper,
+ aggregate_functions: aggregate_functions_fn_wrapper,
+ window_functions: window_functions_fn_wrapper,
+ release: release_fn_wrapper,
+ private_data: Box::into_raw(private_data) as *mut c_void,
+ library_marker_id: crate::get_library_marker_id,
+ }
+ }
+}
+
+impl From<FFI_TaskContext> for Arc<TaskContext> {
+ fn from(ffi_ctx: FFI_TaskContext) -> Self {
+ unsafe {
+ if (ffi_ctx.library_marker_id)() == crate::get_library_marker_id() {
+ return Arc::clone(ffi_ctx.inner());
+ }
+
+ let task_id = (ffi_ctx.task_id)(&ffi_ctx).map(|s| s.to_string()).into();
+ let session_id = (ffi_ctx.session_id)(&ffi_ctx).into();
+ let session_config = (ffi_ctx.session_config)(&ffi_ctx);
+ let session_config =
+ SessionConfig::try_from(&session_config).unwrap_or_default();
+
+ let scalar_functions = (ffi_ctx.scalar_functions)(&ffi_ctx)
+ .into_iter()
+ .filter_map(|kv_pair| {
+ let udf = <Arc<dyn ScalarUDFImpl>>::try_from(&kv_pair.1);
+
+ if let Err(err) = &udf {
+ log::error!("Unable to create WindowUDF in FFI: {err}")
Review Comment:
I have a question on the expected behavior here, let's say that we fail while attempting `try_From` , could not failing loud cause confusion/bugs of functions not found/missing?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer merged PR #18918:
URL: https://github.com/apache/datafusion/pull/18918
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2569920013
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
+
+ /// Utility to identify when FFI objects are accessed locally through
+ /// 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,
+}
+
+struct TaskContextPrivateData {
+ ctx: Arc<TaskContext>,
+}
+
+impl FFI_TaskContext {
+ unsafe fn inner(&self) -> &Arc<TaskContext> {
+ let private_data = self.private_data as *const TaskContextPrivateData;
+ &(*private_data).ctx
+ }
+}
+
+unsafe extern "C" fn session_id_fn_wrapper(ctx: &FFI_TaskContext) -> RString {
+ let ctx = ctx.inner();
+ ctx.session_id().into()
+}
+
+unsafe extern "C" fn task_id_fn_wrapper(ctx: &FFI_TaskContext) -> ROption<RString> {
+ let ctx = ctx.inner();
+ ctx.task_id().map(|s| s.as_str().into()).into()
+}
+
+unsafe extern "C" fn session_config_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> FFI_SessionConfig {
+ let ctx = ctx.inner();
+ ctx.session_config().into()
+}
+
+unsafe extern "C" fn scalar_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_ScalarUDF> {
+ let ctx = ctx.inner();
+ ctx.scalar_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), Arc::clone(udf).into()))
+ .collect()
+}
+
+unsafe extern "C" fn aggregate_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_AggregateUDF> {
+ let ctx = ctx.inner();
+ ctx.aggregate_functions()
+ .iter()
+ .map(|(name, udaf)| {
+ (
+ name.to_owned().into(),
+ FFI_AggregateUDF::from(Arc::clone(udaf)),
+ )
+ })
+ .collect()
+}
+
+unsafe extern "C" fn window_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_WindowUDF> {
+ let ctx = ctx.inner();
+ ctx.window_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), FFI_WindowUDF::from(Arc::clone(udf))))
+ .collect()
+}
+
+unsafe extern "C" fn release_fn_wrapper(ctx: &mut FFI_TaskContext) {
+ let private_data = Box::from_raw(ctx.private_data as *mut TaskContextPrivateData);
Review Comment:
like in other PR, any chances of double free here?
I can see
```
unsafe impl Send/Sync
```
but it is not guaranteed afaik?
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
+
+ /// Utility to identify when FFI objects are accessed locally through
+ /// 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,
+}
+
+struct TaskContextPrivateData {
+ ctx: Arc<TaskContext>,
+}
+
+impl FFI_TaskContext {
+ unsafe fn inner(&self) -> &Arc<TaskContext> {
+ let private_data = self.private_data as *const TaskContextPrivateData;
+ &(*private_data).ctx
+ }
+}
+
+unsafe extern "C" fn session_id_fn_wrapper(ctx: &FFI_TaskContext) -> RString {
+ let ctx = ctx.inner();
+ ctx.session_id().into()
+}
+
+unsafe extern "C" fn task_id_fn_wrapper(ctx: &FFI_TaskContext) -> ROption<RString> {
+ let ctx = ctx.inner();
+ ctx.task_id().map(|s| s.as_str().into()).into()
+}
+
+unsafe extern "C" fn session_config_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> FFI_SessionConfig {
+ let ctx = ctx.inner();
+ ctx.session_config().into()
+}
+
+unsafe extern "C" fn scalar_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_ScalarUDF> {
+ let ctx = ctx.inner();
+ ctx.scalar_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), Arc::clone(udf).into()))
+ .collect()
+}
+
+unsafe extern "C" fn aggregate_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_AggregateUDF> {
+ let ctx = ctx.inner();
+ ctx.aggregate_functions()
+ .iter()
+ .map(|(name, udaf)| {
+ (
+ name.to_owned().into(),
+ FFI_AggregateUDF::from(Arc::clone(udaf)),
+ )
+ })
+ .collect()
+}
+
+unsafe extern "C" fn window_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_WindowUDF> {
+ let ctx = ctx.inner();
+ ctx.window_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), FFI_WindowUDF::from(Arc::clone(udf))))
+ .collect()
+}
+
+unsafe extern "C" fn release_fn_wrapper(ctx: &mut FFI_TaskContext) {
+ let private_data = Box::from_raw(ctx.private_data as *mut TaskContextPrivateData);
Review Comment:
like in other PR, any chances of double free here for concurrent execution?
I can see
```
unsafe impl Send/Sync
```
but it is not guaranteed afaik?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "renato2099 (via GitHub)" <gi...@apache.org>.
renato2099 commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2582862875
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
Review Comment:
I am not sure, but could this lead to a double free? for example, if FFI users clone the FFI_TaskContext, we might end up holding multiple references to `private_data` .
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#issuecomment-3581033237
@ntjohnson1 a good starter FFI issue for review
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "renato2099 (via GitHub)" <gi...@apache.org>.
renato2099 commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2586582801
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
Review Comment:
Hi @timsaucer , I was thinking on explicitly forbidding cloning / copying, somthing like
```
impl !Clone for FFI_TaskContext {}
impl !Copy for FFI_TaskContext {}
```
I was thinking that worst case scenario , FFI users could do a memcopy of the struct , so that it would be somehow "safer" to have something like
```
pub struct FFI_TaskContext {
// An opaque pointer to an Arc<TaskContextPrivateData>
arc_ptr: *const Arc<TaskContextPrivateData>,
}
```
and then returning a raw pointer to inner data , something along these lines
```
pub fn as_opaque(&self) -> *const c_void {
self.arc_ptr as *const c_void
}
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2585112938
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
+
+ /// Utility to identify when FFI objects are accessed locally through
+ /// 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,
+}
+
+struct TaskContextPrivateData {
+ ctx: Arc<TaskContext>,
+}
+
+impl FFI_TaskContext {
+ unsafe fn inner(&self) -> &Arc<TaskContext> {
+ let private_data = self.private_data as *const TaskContextPrivateData;
+ &(*private_data).ctx
+ }
+}
+
+unsafe extern "C" fn session_id_fn_wrapper(ctx: &FFI_TaskContext) -> RString {
+ let ctx = ctx.inner();
+ ctx.session_id().into()
+}
+
+unsafe extern "C" fn task_id_fn_wrapper(ctx: &FFI_TaskContext) -> ROption<RString> {
+ let ctx = ctx.inner();
+ ctx.task_id().map(|s| s.as_str().into()).into()
+}
+
+unsafe extern "C" fn session_config_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> FFI_SessionConfig {
+ let ctx = ctx.inner();
+ ctx.session_config().into()
+}
+
+unsafe extern "C" fn scalar_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_ScalarUDF> {
+ let ctx = ctx.inner();
+ ctx.scalar_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), Arc::clone(udf).into()))
+ .collect()
+}
+
+unsafe extern "C" fn aggregate_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_AggregateUDF> {
+ let ctx = ctx.inner();
+ ctx.aggregate_functions()
+ .iter()
+ .map(|(name, udaf)| {
+ (
+ name.to_owned().into(),
+ FFI_AggregateUDF::from(Arc::clone(udaf)),
+ )
+ })
+ .collect()
+}
+
+unsafe extern "C" fn window_functions_fn_wrapper(
+ ctx: &FFI_TaskContext,
+) -> RHashMap<RString, FFI_WindowUDF> {
+ let ctx = ctx.inner();
+ ctx.window_functions()
+ .iter()
+ .map(|(name, udf)| (name.to_owned().into(), FFI_WindowUDF::from(Arc::clone(udf))))
+ .collect()
+}
+
+unsafe extern "C" fn release_fn_wrapper(ctx: &mut FFI_TaskContext) {
+ let private_data = Box::from_raw(ctx.private_data as *mut TaskContextPrivateData);
Review Comment:
FFI_TaskContext is not cloneable, so I don't know how that would happen. Maybe I'm missing something?
More generally, those other structs that are `Clone` I don't think could possibly double free because each one owns its own private data.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2585116291
##########
datafusion/ffi/src/execution/task_ctx_provider.rs:
##########
@@ -0,0 +1,213 @@
+// 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 std::ffi::c_void;
+use std::sync::{Arc, Weak};
+
+use abi_stable::std_types::{RResult, RString};
+use abi_stable::StableAbi;
+use datafusion_common::{exec_datafusion_err, DataFusionError};
+use datafusion_execution::{TaskContext, TaskContextProvider};
+
+use crate::execution::task_ctx::FFI_TaskContext;
+use crate::{df_result, rresult};
+
+/// Struct for accessing the [`TaskContext`]. This method contains a weak
+/// reference, so there are no guarantees that the [`TaskContext`] remains
+/// valid. This is used primarily for protobuf encoding and decoding of
+/// data passed across the FFI boundary. See the crate README for
+/// additional information.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContextProvider {
+ pub task_ctx: unsafe extern "C" fn(&Self) -> RResult<FFI_TaskContext, RString>,
+
+ /// Used to create a clone on the task context accessor. This should
+ /// only need to be called by the receiver of the plan.
+ pub clone: unsafe extern "C" fn(plan: &Self) -> Self,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
+
+ /// Utility to identify when FFI objects are accessed locally through
+ /// 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,
+}
+
+unsafe impl Send for FFI_TaskContextProvider {}
+unsafe impl Sync for FFI_TaskContextProvider {}
+
+struct TaskContextProviderPrivateData {
+ ctx: Weak<dyn TaskContextProvider>,
Review Comment:
I added more text to the README and the follow on PRs should make it more abundantly clear.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Implement FFI task context and task context provider [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #18918:
URL: https://github.com/apache/datafusion/pull/18918#discussion_r2585022245
##########
datafusion/ffi/src/execution/task_ctx.rs:
##########
@@ -0,0 +1,281 @@
+// 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 std::{ffi::c_void, sync::Arc};
+
+use abi_stable::{
+ pmr::ROption,
+ std_types::{RHashMap, RString},
+ StableAbi,
+};
+use datafusion_execution::{config::SessionConfig, runtime_env::RuntimeEnv, TaskContext};
+use datafusion_expr::{
+ AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
+};
+
+use crate::{
+ session_config::FFI_SessionConfig, udaf::FFI_AggregateUDF, udf::FFI_ScalarUDF,
+ udwf::FFI_WindowUDF,
+};
+
+/// A stable struct for sharing [`TaskContext`] across FFI boundaries.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+#[allow(non_camel_case_types)]
+pub struct FFI_TaskContext {
+ pub session_id: unsafe extern "C" fn(&Self) -> RString,
+
+ pub task_id: unsafe extern "C" fn(&Self) -> ROption<RString>,
+
+ pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
+
+ pub scalar_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_ScalarUDF>,
+
+ pub aggregate_functions:
+ unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_AggregateUDF>,
+
+ pub window_functions: unsafe extern "C" fn(&Self) -> RHashMap<RString, FFI_WindowUDF>,
+
+ /// Release the memory of the private data when it is no longer being used.
+ pub release: unsafe extern "C" fn(arg: &mut Self),
+
+ /// Internal data. This is only to be accessed by the provider of the plan.
+ /// The foreign library should never attempt to access this data.
+ pub private_data: *mut c_void,
Review Comment:
FFI_TaskContext does not implement Clone. Do you see somewhere else this could happen?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org