paleolimbot commented on code in PR #19281:
URL: https://github.com/apache/datafusion/pull/19281#discussion_r2614659665
##########
datafusion/ffi/src/table_provider.rs:
##########
@@ -267,53 +266,40 @@ unsafe extern "C" fn scan_fn_wrapper(
let plan = rresult_return!(
internal_provider
- .scan(&ctx.state(), Some(&projections), &filters, limit.into())
+ .scan(session, Some(&projections), &filters, limit.into())
.await
);
- RResult::ROk(FFI_ExecutionPlan::new(
- plan,
- ctx.task_ctx(),
- runtime.clone(),
- ))
+ RResult::ROk(FFI_ExecutionPlan::new(plan, runtime.clone()))
}
.into_ffi()
}
unsafe extern "C" fn insert_into_fn_wrapper(
provider: &FFI_TableProvider,
- session_config: &FFI_SessionConfig,
+ session: FFI_SessionRef,
input: &FFI_ExecutionPlan,
insert_op: FFI_InsertOp,
) -> FfiFuture<FFIResult<FFI_ExecutionPlan>> {
let runtime = provider.runtime().clone();
let internal_provider = Arc::clone(provider.inner());
- let session_config = session_config.clone();
let input = input.clone();
async move {
- let config = rresult_return!(SessionConfig::try_from(&session_config));
- let session = SessionStateBuilder::new()
- .with_default_features()
- .with_config(config)
- .build();
- let ctx = SessionContext::new_with_state(session);
+ let foreign_session =
rresult_return!(ForeignSession::try_from(&session));
Review Comment:
You could possibly push this type of optimization into the `ForeignSession`
by only doing expensive initialization if one of the methods is actually called
using a OnceLock or something. If I'm remembering that PR correctly, the main
issue there is that some of the session trait items don't return a Result so
there's no opportunity for failure.
##########
datafusion/ffi/src/plan_properties.rs:
##########
@@ -92,15 +78,8 @@ impl FFI_PlanProperties {
unsafe extern "C" fn output_partitioning_fn_wrapper(
properties: &FFI_PlanProperties,
-) -> FFIResult<RVec<u8>> {
- let codec = DefaultPhysicalExtensionCodec {};
- let partitioning_data = rresult_return!(serialize_partitioning(
- properties.inner().output_partitioning(),
- &codec
- ));
- let output_partitioning = partitioning_data.encode_to_vec();
-
- ROk(output_partitioning.into())
Review Comment:
This is cool, although my reading of this means that any ExecutionPlan
passed via FFI will not benefit from any physical optimization in the same way
it would have if serialized/unserialized via protobuf. It seems like these are
mostly for TableProviders where the plan being passed around would normally not
participate in that anyway but I thought I would point that out.
##########
datafusion/ffi/src/table_provider.rs:
##########
@@ -221,44 +215,49 @@ fn supports_filters_pushdown_internal(
unsafe extern "C" fn supports_filters_pushdown_fn_wrapper(
provider: &FFI_TableProvider,
filters_serialized: RVec<u8>,
-) -> FFIResult<RVec<FFI_TableProviderFilterPushDown>> {
- supports_filters_pushdown_internal(provider.inner(), &filters_serialized)
- .map_err(|e| e.to_string().into())
- .into()
+) -> RResult<RVec<FFI_TableProviderFilterPushDown>, RString> {
Review Comment:
Does this need updating?
--
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]
