alamb commented on code in PR #25318: URL: https://github.com/apache/datafusion/pull/25318#discussion_r4501927448
########## docs/source/library-user-guide/upgrading/56.0.0.md: ########## @@ -1486,6 +1486,62 @@ If a registered decoder fails, the decode fails. It does not fall back to the co See [issue #24626](https://github.com/apache/datafusion/issues/24626) for details. +### `PhysicalExtensionCodec` can decode with the full decode context Review Comment: actually since we cut the brach, this should now go into the 57 upgrade guide ########## docs/source/library-user-guide/upgrading/56.0.0.md: ########## @@ -1486,6 +1486,62 @@ If a registered decoder fails, the decode fails. It does not fall back to the co See [issue #24626](https://github.com/apache/datafusion/issues/24626) for details. +### `PhysicalExtensionCodec` can decode with the full decode context + +`PhysicalExtensionCodec::try_decode` received a `TaskContext`, which carries no Review Comment: I din't think we have to reiterate the problem here -- we need to just say it didn't have enough context and then refer to the issue for more details ########## docs/source/library-user-guide/upgrading/56.0.0.md: ########## @@ -1486,6 +1486,62 @@ If a registered decoder fails, the decode fails. It does not fall back to the co See [issue #24626](https://github.com/apache/datafusion/issues/24626) for details. +### `PhysicalExtensionCodec` can decode with the full decode context + +`PhysicalExtensionCodec::try_decode` received a `TaskContext`, which carries no +state that exists only while the surrounding plan is being decoded. A codec that +decodes nested expressions itself — through +`PhysicalProtoConverterExtension::proto_to_physical_expr`, say — had to build a +fresh context, so a `ScalarSubqueryExpr` lost the results of the +`ScalarSubqueryExec` it is bound to. + +The trait gains `try_decode_with_ctx`, which takes the `PhysicalPlanDecodeContext` +the surrounding decode is running under: + +```rust,ignore +fn try_decode_with_ctx( + &self, + buf: &[u8], + inputs: &[Arc<dyn ExecutionPlan>], + ctx: &PhysicalPlanDecodeContext<'_>, + proto_converter: &dyn PhysicalProtoConverterExtension, +) -> Result<Arc<dyn ExecutionPlan>> { + let proto = MyExecProto::decode(buf)?; + let input_schema = inputs[0].schema(); + // Pass `ctx` on to the nested decode rather than building a new context. + let expr = + proto_converter.proto_to_physical_expr(&proto.expr, &input_schema, ctx)?; + Ok(Arc::new(MyExec::new(expr, Arc::clone(&inputs[0])))) +} +``` + +**Who is affected:** + +- Nobody is required to change anything. `try_decode_with_ctx` has a default + implementation that calls `try_decode` with `ctx.task_ctx()`, so a codec that + only implements `try_decode` keeps compiling and keeps its current behavior. +- A codec that decodes nested expressions or plans itself should implement + `try_decode_with_ctx` and forward `ctx`. Keeping only `try_decode` means a + nested `ScalarSubqueryExpr` is decoded against an empty scope and cannot see + the results of an enclosing `ScalarSubqueryExec`. +- `ComposedPhysicalExtensionCodec` forwards `try_decode_with_ctx` to the codec + that wrote the payload, so composing codecs preserves the context. + +**FFI users must rebuild both sides.** `FFI_PhysicalExtensionCodec` gained a +`try_decode_with_ctx` entry point, added at the end of the `repr(C)` struct so +every existing field keeps its offset. The struct still grows, so a host built +against this version would read that field from a struct an older provider never +wrote. The change lands in a major release, so the number +`datafusion_ffi::version()` reports changes with it: a provider and host that +compare it, which is what that function is for, see the mismatch instead. + +`ScalarSubqueryResults` is no longer backed only by in-process state. It can be Review Comment: is this important to state? This just reiterates how things are implemented. I think we could remove most/all of this last paragraph ########## datafusion/ffi/src/proto/physical_extension_codec.rs: ########## @@ -431,25 +514,248 @@ impl PhysicalExtensionCodec for ForeignPhysicalExtensionCodec { } } +/// Fixtures shared by the in-process unit tests in this module and the +/// cross-library integration test in +/// `datafusion/ffi/tests/ffi_physical_extension_codec.rs`. +/// +/// Gated on `feature = "integration-tests"` (in addition to `test`) so the +/// external integration-test binary, which links this crate as an ordinary +/// dependency rather than compiling it with `cfg(test)`, can still reach +/// these types through `datafusion_ffi::proto::physical_extension_codec::fixtures`. +#[cfg(any(test, feature = "integration-tests"))] Review Comment: Why is this in the main ffi crate (and not in the test binary?)? This doesn't make sense to me ########## datafusion/ffi/integration-tests/tests/cases/physical_extension_codec.rs: ########## @@ -0,0 +1,140 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +mod tests { + use std::any::Any; + use std::sync::Arc; + + use arrow::array::{AsArray, RecordBatch}; + use arrow::datatypes::{DataType, Int64Type, Schema}; + use datafusion::prelude::SessionContext; + use datafusion_common::{DataFusionError, ScalarValue}; + use datafusion_execution::TaskContextProvider; + use datafusion_expr::physical_planning_context::{ + ScalarSubqueryResults, SubqueryIndex, + }; + use datafusion_ffi::execution::FFI_TaskContextProvider; + use datafusion_ffi::execution_plan::ForeignExecutionPlan; + use datafusion_ffi::proto::physical_extension_codec::ForeignPhysicalExtensionCodec; + use datafusion_ffi::proto::physical_extension_codec::fixtures::{ + ScalarSubqueryExprExec, ScalarSubqueryExprExecCodec, + }; + use datafusion_ffi::tests::utils::get_scalar_subquery_expr_exec_codec; + use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr; + use datafusion_physical_plan::empty::EmptyExec; + use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr}; + use datafusion_proto::physical_plan::{ + DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext, + }; + + /// The decode context's active scalar-subquery results scope must reach Review Comment: Please cut this comment down to explain the purpose of the test, rahter than restating in english how it is implemented ########## datafusion/ffi/src/proto/scalar_subquery_results.rs: ########## @@ -0,0 +1,305 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +use std::ffi::c_void; +use std::sync::Arc; + +use datafusion_common::error::Result; +use datafusion_common::{DataFusionError, ScalarValue, internal_datafusion_err}; +use datafusion_expr::physical_planning_context::{ + ScalarSubqueryResults, ScalarSubqueryResultsBackend, SubqueryIndex, +}; +use datafusion_proto::protobuf::ArrowType; +use prost::Message; + +use stabby::vec::Vec as SVec; + +use crate::util::{FFI_Option, FFI_Result}; +use crate::{df_result, sresult, sresult_return}; + +/// A stable struct for sharing the shared results container of a +/// `ScalarSubqueryExec` across FFI boundaries. +/// +/// Unlike most FFI wrappers in this crate, which stand in for a whole Review Comment: And remove details about its implementation -- instead focus on its relevance ########## datafusion/ffi/src/proto/scalar_subquery_results.rs: ########## @@ -0,0 +1,305 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +use std::ffi::c_void; +use std::sync::Arc; + +use datafusion_common::error::Result; +use datafusion_common::{DataFusionError, ScalarValue, internal_datafusion_err}; +use datafusion_expr::physical_planning_context::{ + ScalarSubqueryResults, ScalarSubqueryResultsBackend, SubqueryIndex, +}; +use datafusion_proto::protobuf::ArrowType; +use prost::Message; + +use stabby::vec::Vec as SVec; + +use crate::util::{FFI_Option, FFI_Result}; +use crate::{df_result, sresult, sresult_return}; + +/// A stable struct for sharing the shared results container of a +/// `ScalarSubqueryExec` across FFI boundaries. +/// +/// Unlike most FFI wrappers in this crate, which stand in for a whole +/// `dyn Trait` object, this one stands in for a single +/// [`ScalarSubqueryResults`] value: the active scope that a +/// `PhysicalExtensionCodec` on the near side of the boundary passes to +/// `try_decode_with_ctx` so a `ScalarSubqueryExpr` decoded on the far side +/// reads the same populated values as the `ScalarSubqueryExec` that owns +/// them. +/// +/// `ScalarValue`s cross the boundary as prost-encoded bytes, the same wire +/// format `PhysicalExtensionCodec::try_decode`/`try_encode` already use, so +/// this struct never assumes a stable memory layout for `ScalarValue` itself. +#[repr(C)] +#[derive(Debug)] +pub struct FFI_ScalarSubqueryResults { + get: unsafe extern "C" fn(&Self, index: u64) -> FFI_Result<FFI_Option<SVec<u8>>>, + + set: unsafe extern "C" fn(&Self, index: u64, value: SVec<u8>) -> FFI_Result<()>, + + clear: unsafe extern "C" fn(&Self), + + /// Used to create a clone on the provider of the results container. This + /// should only need to be called by the receiver of the handle. + clone: unsafe extern "C" fn(&Self) -> Self, + + /// Release the memory of the private data when it is no longer being used. + release: unsafe extern "C" fn(&mut Self), + + private_data: *mut c_void, + + /// Utility to identify when FFI objects are accessed locally through + /// the foreign interface. + library_marker_id: extern "C" fn() -> usize, +} + +unsafe impl Send for FFI_ScalarSubqueryResults {} +unsafe impl Sync for FFI_ScalarSubqueryResults {} + +impl FFI_ScalarSubqueryResults { + fn inner(&self) -> &ScalarSubqueryResults { + let private_data = self.private_data as *const ScalarSubqueryResults; + unsafe { &*private_data } + } + + /// Creates a new [`FFI_ScalarSubqueryResults`] sharing `results`. + pub fn new(results: ScalarSubqueryResults) -> Self { + let private_data = Box::new(results); + + Self { + get: get_fn_wrapper, + set: set_fn_wrapper, + clear: clear_fn_wrapper, + clone: clone_fn_wrapper, + release: release_fn_wrapper, + private_data: Box::into_raw(private_data).cast::<c_void>(), + library_marker_id: crate::get_library_marker_id, + } + } +} + +/// The wire form a [`ScalarValue`] crosses the FFI boundary in. +/// +/// `datafusion_proto::protobuf::ScalarValue` alone is not a type-preserving +/// transport: it has no dedicated variant for every `DataType` (`Float16`, Review Comment: THis seems like a rationale for expanding the coverage of `datafusion_proto::protobuf::ScalarValue`, ot adding a whole new wrapper ########## docs/source/library-user-guide/upgrading/56.0.0.md: ########## @@ -1486,6 +1486,62 @@ If a registered decoder fails, the decode fails. It does not fall back to the co See [issue #24626](https://github.com/apache/datafusion/issues/24626) for details. +### `PhysicalExtensionCodec` can decode with the full decode context + +`PhysicalExtensionCodec::try_decode` received a `TaskContext`, which carries no +state that exists only while the surrounding plan is being decoded. A codec that +decodes nested expressions itself — through +`PhysicalProtoConverterExtension::proto_to_physical_expr`, say — had to build a +fresh context, so a `ScalarSubqueryExpr` lost the results of the +`ScalarSubqueryExec` it is bound to. + +The trait gains `try_decode_with_ctx`, which takes the `PhysicalPlanDecodeContext` +the surrounding decode is running under: + +```rust,ignore +fn try_decode_with_ctx( + &self, + buf: &[u8], + inputs: &[Arc<dyn ExecutionPlan>], + ctx: &PhysicalPlanDecodeContext<'_>, + proto_converter: &dyn PhysicalProtoConverterExtension, +) -> Result<Arc<dyn ExecutionPlan>> { + let proto = MyExecProto::decode(buf)?; + let input_schema = inputs[0].schema(); + // Pass `ctx` on to the nested decode rather than building a new context. + let expr = + proto_converter.proto_to_physical_expr(&proto.expr, &input_schema, ctx)?; + Ok(Arc::new(MyExec::new(expr, Arc::clone(&inputs[0])))) +} +``` + +**Who is affected:** Review Comment: Can you please trim this down from AI speak For example this first sentence doesn't need to restate how the code works, it can just say "no changes are required" The second can say "if you need the additional context, use the new function" The detail about `ComposedPhysical...` is an implementation detail that isn't relevant for implmenters, is it? ########## datafusion/ffi/src/proto/physical_extension_codec.rs: ########## @@ -98,6 +99,21 @@ pub struct FFI_PhysicalExtensionCodec { /// Utility to identify when FFI objects are accessed locally through /// the foreign interface. pub library_marker_id: extern "C" fn() -> usize, + + /// Decode bytes into an execution plan, forwarding the active scalar + /// subquery results scope (if any) from the caller's decode context so a + /// `ScalarSubqueryExpr` decoded on this side of the boundary shares the + /// same populated results as the `ScalarSubqueryExec` that owns them. + /// + /// Added after the original fields, at the end of the struct, so that Review Comment: this note about original fields will not be relevant after this PR -- please trim the comments ########## docs/source/library-user-guide/upgrading/56.0.0.md: ########## @@ -1486,6 +1486,62 @@ If a registered decoder fails, the decode fails. It does not fall back to the co See [issue #24626](https://github.com/apache/datafusion/issues/24626) for details. +### `PhysicalExtensionCodec` can decode with the full decode context + +`PhysicalExtensionCodec::try_decode` received a `TaskContext`, which carries no +state that exists only while the surrounding plan is being decoded. A codec that +decodes nested expressions itself — through +`PhysicalProtoConverterExtension::proto_to_physical_expr`, say — had to build a +fresh context, so a `ScalarSubqueryExpr` lost the results of the +`ScalarSubqueryExec` it is bound to. + +The trait gains `try_decode_with_ctx`, which takes the `PhysicalPlanDecodeContext` +the surrounding decode is running under: + +```rust,ignore +fn try_decode_with_ctx( + &self, + buf: &[u8], + inputs: &[Arc<dyn ExecutionPlan>], + ctx: &PhysicalPlanDecodeContext<'_>, + proto_converter: &dyn PhysicalProtoConverterExtension, +) -> Result<Arc<dyn ExecutionPlan>> { + let proto = MyExecProto::decode(buf)?; + let input_schema = inputs[0].schema(); + // Pass `ctx` on to the nested decode rather than building a new context. + let expr = + proto_converter.proto_to_physical_expr(&proto.expr, &input_schema, ctx)?; + Ok(Arc::new(MyExec::new(expr, Arc::clone(&inputs[0])))) +} +``` + +**Who is affected:** + +- Nobody is required to change anything. `try_decode_with_ctx` has a default + implementation that calls `try_decode` with `ctx.task_ctx()`, so a codec that + only implements `try_decode` keeps compiling and keeps its current behavior. +- A codec that decodes nested expressions or plans itself should implement + `try_decode_with_ctx` and forward `ctx`. Keeping only `try_decode` means a + nested `ScalarSubqueryExpr` is decoded against an empty scope and cannot see + the results of an enclosing `ScalarSubqueryExec`. +- `ComposedPhysicalExtensionCodec` forwards `try_decode_with_ctx` to the codec + that wrote the payload, so composing codecs preserves the context. + +**FFI users must rebuild both sides.** `FFI_PhysicalExtensionCodec` gained a Review Comment: is this specific to this change? It seem generic to all FFI ########## datafusion/expr/src/physical_planning_context.rs: ########## @@ -110,70 +110,150 @@ impl SubqueryIndex { } } +/// An alternate backing store for a [`ScalarSubqueryResults`] container. +/// +/// A results container normally owns its slots directly (see +/// [`ScalarSubqueryResults::new`]). This trait lets a container instead +/// forward `get`/`set`/`clear` to some other shared container, which is how +/// `datafusion-ffi` bridges the results of a `ScalarSubqueryExec` across a +/// non-Rust-ABI boundary: the far side of the boundary gets a +/// [`ScalarSubqueryResults`] backed by a proxy that calls back into the +/// near side's real container for every operation, so both sides observe the +/// same populated values. +pub trait ScalarSubqueryResultsBackend: + fmt::Debug + Send + Sync + std::panic::RefUnwindSafe +{ + /// Returns the scalar value stored at `index`, if it has been populated. + fn get(&self, index: usize) -> Option<ScalarValue>; + + /// Stores `value` in the slot at `index`. + fn set(&self, index: usize, value: ScalarValue) -> Result<()>; + + /// Clears all populated results so the container can be reused. + fn clear(&self); +} + +#[derive(Clone)] +enum ScalarSubqueryResultsRepr { + Local(Arc<Vec<Mutex<Option<ScalarValue>>>>), + Remote(Arc<dyn ScalarSubqueryResultsBackend>), Review Comment: Instaed of "remote" erhaps can call this "callback" or function or something like that. What is "remote" about it (other than it is across the FFI boundary)? Also, if we already have a trait, why are we also wrapping it in an Enum? We could change `ScalarSubqueryResultsRepr` to `Arc<dyn ScalarSubqueryResultsBackend>` and then implement `ScalarSubqueryResultsBackend` for `Vec<Mutex<Option<ScalarValue>>>` (or some wrapper type), right? ########## datafusion/ffi/src/proto/scalar_subquery_results.rs: ########## @@ -0,0 +1,305 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +use std::ffi::c_void; +use std::sync::Arc; + +use datafusion_common::error::Result; +use datafusion_common::{DataFusionError, ScalarValue, internal_datafusion_err}; +use datafusion_expr::physical_planning_context::{ + ScalarSubqueryResults, ScalarSubqueryResultsBackend, SubqueryIndex, +}; +use datafusion_proto::protobuf::ArrowType; +use prost::Message; + +use stabby::vec::Vec as SVec; + +use crate::util::{FFI_Option, FFI_Result}; +use crate::{df_result, sresult, sresult_return}; + +/// A stable struct for sharing the shared results container of a +/// `ScalarSubqueryExec` across FFI boundaries. +/// +/// Unlike most FFI wrappers in this crate, which stand in for a whole Review Comment: Why is it important that this is different? Please clarify this comment ########## datafusion/ffi/src/proto/physical_extension_codec.rs: ########## @@ -743,4 +1113,99 @@ pub(crate) mod tests { .expect("rebound codec resolves"); assert_eq!(task_ctx.session_id(), ctx_b.task_ctx().session_id()); } + + /// The decode context's active scalar subquery results scope must reach a + /// `ScalarSubqueryExpr` decoded by a codec that is forced foreign through + /// the FFI boundary, not just a local one. Without the fix, the far side + /// always decodes with a root context (no scope), so the embedded + /// `ScalarSubqueryExpr` fails to deserialize. + /// + /// This only mocks the foreign path in-process (see Review Comment: similar, please trim this comment to describe what this is testing, not restate how it is implemented Also, why is this test needed if we already have a integration test? -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
