namanjain24-sudo commented on code in PR #25318:
URL: https://github.com/apache/datafusion/pull/25318#discussion_r4226833403
##########
datafusion/ffi/src/tests/mod.rs:
##########
@@ -273,6 +273,91 @@ pub extern "C" fn
datafusion_ffi_test_create_exec_with_byte_metrics() -> FFI_Exe
FFI_ExecutionPlan::new(plan, None)
}
+/// Decodes a
[`crate::proto::physical_extension_codec::fixtures::ScalarSubqueryExprExec`]
+/// from `bytes` using
+///
[`crate::proto::physical_extension_codec::fixtures::ScalarSubqueryExprExecCodec`]
+/// *inside this image*, forwarding `scalar_subquery_results` (if present)
+/// into the decode context exactly the way a real
+/// `FFI_PhysicalExtensionCodec::try_decode_with_ctx` call would, then reads
+/// back the embedded `ScalarSubqueryExpr`'s `Int64` result at `index`.
+///
+/// There is no generic `FFI_PhysicalExpr` wrapper in this crate to carry an
+/// arbitrary decoded expression back across the boundary, so the decoded
+/// `ScalarSubqueryExprExec`/`ScalarSubqueryExpr` never leave this image as
+/// Rust values - only the `i64` their results container reports does. That
+/// is still the assertion that matters: when `scalar_subquery_results` is
+/// built in a genuinely different image (see the integration test in
+/// `datafusion/ffi/tests/ffi_physical_extension_codec.rs`), reading it back
+/// here only succeeds if `get`/`set` actually round-tripped through
+/// `ForeignScalarSubqueryResultsBackend` into the caller's own results
+/// container - the cross-library coverage gap the in-process, marker-mocked
+/// unit test in `datafusion/ffi/src/proto/physical_extension_codec.rs`
+/// cannot close on its own.
+///
+/// Exported as its own top-level symbol rather than a new field on
+/// [`ForeignLibraryModule`] for the same reason as
+/// [`datafusion_ffi_test_create_exec_with_byte_metrics`]: that struct is
+/// public, `#[repr(C)]`, and has no private/gated constructor, so adding a
+/// field changes its exhaustive-construction ABI surface even for this
+/// test-only, `integration-tests`-gated struct.
+#[unsafe(no_mangle)]
+pub extern "C" fn datafusion_ffi_test_decode_scalar_subquery_expr_exec_result(
+ task_ctx_provider: crate::execution::FFI_TaskContextProvider,
+ bytes: stabby::slice::Slice<u8>,
+ scalar_subquery_results: FFI_Option<
+ crate::proto::scalar_subquery_results::FFI_ScalarSubqueryResults,
+ >,
+ index: u64,
+) -> crate::util::FFI_Result<FFI_Option<i64>> {
+ use datafusion_common::ScalarValue;
+ use datafusion_execution::TaskContext;
+ use datafusion_expr::physical_planning_context::SubqueryIndex;
+ use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr;
+ use datafusion_proto::physical_plan::{
+ DefaultPhysicalProtoConverter, PhysicalExtensionCodec,
PhysicalPlanDecodeContext,
+ };
+
+ use crate::proto::physical_extension_codec::fixtures::{
+ ScalarSubqueryExprExec, ScalarSubqueryExprExecCodec,
+ };
+ use crate::sresult_return;
+ use crate::util::FFI_Result;
+
+ let task_ctx: Arc<TaskContext> =
sresult_return!((&task_ctx_provider).try_into());
+
+ let codec = ScalarSubqueryExprExecCodec;
+ let decode_ctx = PhysicalPlanDecodeContext::new(task_ctx.as_ref(), &codec);
+ let decode_ctx = match scalar_subquery_results.into_option() {
+ Some(results) =>
decode_ctx.with_scalar_subquery_results(results.into()),
+ None => decode_ctx,
+ };
+
+ // `ScalarSubqueryExprExecCodec::try_decode_with_ctx` only uses this to
+ // read a schema; `ScalarSubqueryExpr` carries no column references.
+ let dummy_input: Arc<dyn ExecutionPlan> =
+ Arc::new(EmptyExec::new(Arc::new(Schema::empty())));
+
+ let plan = sresult_return!(codec.try_decode_with_ctx(
Review Comment:
Routed through the real callback: the codec is built inside the dlopen'd
cdylib, converted to Arc<dyn PhysicalExtensionCodec> (confirmed foreign), and
try_decode_with_ctx is called directly on that Arc with the host's active
results — the same call path
try_decode_with_ctx_fn_wrapper/ForeignPhysicalExtensionCodec dispatch through
for any foreign codec, no bespoke entry point. The decoded plan comes back
opaque (ForeignExecutionPlan), so to keep the host-value assertion I gave
ScalarSubqueryExprExec a real execute() that evaluates its expression and
returns the result as a RecordBatch, carried back via the existing
FFI_ExecutionPlan/FFI_RecordBatchStream machinery — no new wrapper needed.
Verified by reverting the production forwarding: the test now fails with the
original 'can only be deserialized as part of a surrounding ScalarSubqueryExec'
error.
--
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]
