Copilot commented on code in PR #23906:
URL: https://github.com/apache/datafusion/pull/23906#discussion_r3703965757
##########
datafusion/ffi/src/table_provider.rs:
##########
@@ -671,11 +906,161 @@ mod tests {
Ok(())
}
+ #[derive(Debug, Default)]
+ struct DmlCalls {
+ delete_filters: Option<Vec<Expr>>,
+ update_assignments: Option<Vec<(String, Expr)>>,
+ update_filters: Option<Vec<Expr>>,
+ truncated: bool,
+ }
+
+ #[derive(Debug)]
+ struct DmlTableProvider {
+ calls: Arc<Mutex<DmlCalls>>,
+ schema: SchemaRef,
+ }
Review Comment:
`std::sync::Mutex` is used from within async `TableProvider` methods
(`delete_from`/`update`/`truncate`). Locking a std mutex inside async code can
block a Tokio worker thread. Recommendation (mandatory): switch to
`tokio::sync::Mutex` (or another async-aware lock) for `calls`, or collect call
state via channels/atomics appropriate for async contexts.
##########
datafusion/ffi/src/table_provider.rs:
##########
@@ -671,11 +906,161 @@ mod tests {
Ok(())
}
+ #[derive(Debug, Default)]
+ struct DmlCalls {
+ delete_filters: Option<Vec<Expr>>,
+ update_assignments: Option<Vec<(String, Expr)>>,
+ update_filters: Option<Vec<Expr>>,
+ truncated: bool,
+ }
+
+ #[derive(Debug)]
+ struct DmlTableProvider {
+ calls: Arc<Mutex<DmlCalls>>,
+ schema: SchemaRef,
+ }
+
+ fn dml_count_plan() -> Arc<dyn ExecutionPlan> {
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "count",
+ DataType::UInt64,
+ false,
+ )]));
+ Arc::new(crate::execution_plan::tests::EmptyExec::new(schema))
+ }
+
+ #[async_trait]
+ impl TableProvider for DmlTableProvider {
+ fn schema(&self) -> SchemaRef {
+ Arc::clone(&self.schema)
+ }
+
+ fn table_type(&self) -> TableType {
+ TableType::Base
+ }
+
+ async fn scan(
+ &self,
+ _session: &dyn Session,
+ _projection: Option<&Vec<usize>>,
+ _filters: &[Expr],
+ _limit: Option<usize>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ Err(DataFusionError::Internal(
+ "DmlTableProvider scan should not be called".to_string(),
+ ))
+ }
+
+ async fn delete_from(
+ &self,
+ _state: &dyn Session,
+ filters: Vec<Expr>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ self.calls.lock().unwrap().delete_filters = Some(filters);
+ Ok(dml_count_plan())
+ }
Review Comment:
`std::sync::Mutex` is used from within async `TableProvider` methods
(`delete_from`/`update`/`truncate`). Locking a std mutex inside async code can
block a Tokio worker thread. Recommendation (mandatory): switch to
`tokio::sync::Mutex` (or another async-aware lock) for `calls`, or collect call
state via channels/atomics appropriate for async contexts.
##########
datafusion/ffi/src/table_provider.rs:
##########
@@ -336,6 +378,133 @@ unsafe extern "C" fn insert_into_fn_wrapper(
.into_ffi()
}
+unsafe extern "C" fn delete_from_fn_wrapper(
+ provider: &FFI_TableProvider,
+ session: FFI_SessionRef,
+ filters_serialized: SVec<u8>,
+) -> FfiFuture<FFI_Result<FFI_ExecutionPlan>> {
+ let task_ctx: Result<Arc<TaskContext>, DataFusionError> =
+ (&provider.logical_codec.task_ctx_provider).try_into();
+ let runtime = provider.runtime().clone();
+ let logical_codec: Arc<dyn LogicalExtensionCodec> =
(&provider.logical_codec).into();
+ let internal_provider = Arc::clone(provider.inner());
+
+ async move {
+ let mut foreign_session = None;
+ let session = sresult_return!(
+ session
+ .as_local()
+ .map(Ok::<&(dyn Session + Send + Sync), DataFusionError>)
+ .unwrap_or_else(|| {
+ foreign_session =
Some(ForeignSession::try_from(&session)?);
+ Ok(foreign_session.as_ref().unwrap())
+ })
+ );
+
+ let task_ctx = sresult_return!(task_ctx);
+ let filters = sresult_return!(parse_serialized_exprs(
+ &filters_serialized,
+ &task_ctx,
+ logical_codec.as_ref(),
+ ));
+
+ let plan = sresult_return!(internal_provider.delete_from(session,
filters).await);
+
+ FFI_Result::Ok(FFI_ExecutionPlan::new(plan, runtime))
+ }
+ .into_ffi()
+}
+
+unsafe extern "C" fn update_fn_wrapper(
+ provider: &FFI_TableProvider,
+ session: FFI_SessionRef,
+ assignments: SVec<FFI_TableProviderUpdateAssignment>,
+ filters_serialized: SVec<u8>,
+) -> FfiFuture<FFI_Result<FFI_ExecutionPlan>> {
+ let task_ctx: Result<Arc<TaskContext>, DataFusionError> =
+ (&provider.logical_codec.task_ctx_provider).try_into();
+ let runtime = provider.runtime().clone();
+ let logical_codec: Arc<dyn LogicalExtensionCodec> =
(&provider.logical_codec).into();
+ let internal_provider = Arc::clone(provider.inner());
+
+ async move {
+ let mut foreign_session = None;
+ let session = sresult_return!(
+ session
+ .as_local()
+ .map(Ok::<&(dyn Session + Send + Sync), DataFusionError>)
+ .unwrap_or_else(|| {
+ foreign_session =
Some(ForeignSession::try_from(&session)?);
+ Ok(foreign_session.as_ref().unwrap())
+ })
+ );
+
+ let task_ctx = sresult_return!(task_ctx);
+ let assignments = sresult_return!(
+ assignments
+ .into_iter()
+ .map(|assignment| {
+ let mut exprs = parse_serialized_exprs(
+ &assignment.expr_serialized,
+ &task_ctx,
+ logical_codec.as_ref(),
+ )?;
+ let expr = match exprs.len() {
+ 1 => exprs.remove(0),
+ _ => {
+ return Err(DataFusionError::Plan(
+ "Expected exactly one expression for update
assignment"
+ .to_string(),
+ ));
+ }
+ };
Review Comment:
The error message for malformed update assignments is missing actionable
context (e.g., which column failed and how many expressions were received).
Recommendation (optional but useful for debugging across FFI): include
`assignment.column` and the actual count (and possibly whether the payload was
empty vs. multi-expr) in the error to speed diagnosis of foreign library issues.
##########
datafusion/ffi/src/table_provider.rs:
##########
@@ -577,11 +744,79 @@ impl TableProvider for ForeignTableProvider {
Ok(plan)
}
+
+ async fn delete_from(
+ &self,
+ session: &dyn Session,
+ filters: Vec<Expr>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ let session = FFI_SessionRef::new(session, None,
self.0.logical_codec.clone());
+ let codec: Arc<dyn LogicalExtensionCodec> =
(&self.0.logical_codec).into();
+ let filters_serialized = serialize_expr_list(filters.iter(),
codec.as_ref())?;
+
+ let plan = unsafe {
+ let maybe_plan =
+ (self.0.delete_from)(&self.0, session,
filters_serialized).await;
+
+ <Arc<dyn ExecutionPlan>>::try_from(&df_result!(maybe_plan)?)?
+ };
+
+ Ok(plan)
+ }
+
+ async fn update(
+ &self,
+ session: &dyn Session,
+ assignments: Vec<(String, Expr)>,
+ filters: Vec<Expr>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ let session = FFI_SessionRef::new(session, None,
self.0.logical_codec.clone());
+ let codec: Arc<dyn LogicalExtensionCodec> =
(&self.0.logical_codec).into();
+
+ let assignments: SVec<_> = assignments
+ .iter()
+ .map(|(column, expr)| {
+ Ok(FFI_TableProviderUpdateAssignment {
+ column: SString::from(column.as_str()),
+ expr_serialized: serialize_expr_list(
+ std::iter::once(expr),
+ codec.as_ref(),
+ )?,
+ })
+ })
+ .collect::<Result<Vec<_>>>()?
+ .into_iter()
+ .collect();
+ let filters_serialized = serialize_expr_list(filters.iter(),
codec.as_ref())?;
+
+ let plan = unsafe {
+ let maybe_plan =
+ (self.0.update)(&self.0, session, assignments,
filters_serialized).await;
+
+ <Arc<dyn ExecutionPlan>>::try_from(&df_result!(maybe_plan)?)?
+ };
+
+ Ok(plan)
+ }
+
+ async fn truncate(&self, session: &dyn Session) -> Result<Arc<dyn
ExecutionPlan>> {
+ let session = FFI_SessionRef::new(session, None,
self.0.logical_codec.clone());
+
+ let plan = unsafe {
+ let maybe_plan = (self.0.truncate)(&self.0, session).await;
+
+ <Arc<dyn ExecutionPlan>>::try_from(&df_result!(maybe_plan)?)?
+ };
+
+ Ok(plan)
+ }
}
#[cfg(test)]
mod tests {
- use arrow::datatypes::Schema;
+ use std::sync::Mutex;
+
+ use arrow::datatypes::{DataType, Field, Schema};
Review Comment:
`std::sync::Mutex` is used from within async `TableProvider` methods
(`delete_from`/`update`/`truncate`). Locking a std mutex inside async code can
block a Tokio worker thread. Recommendation (mandatory): switch to
`tokio::sync::Mutex` (or another async-aware lock) for `calls`, or collect call
state via channels/atomics appropriate for async contexts.
##########
datafusion/ffi/src/table_provider.rs:
##########
@@ -336,6 +378,133 @@ unsafe extern "C" fn insert_into_fn_wrapper(
.into_ffi()
}
+unsafe extern "C" fn delete_from_fn_wrapper(
+ provider: &FFI_TableProvider,
+ session: FFI_SessionRef,
+ filters_serialized: SVec<u8>,
+) -> FfiFuture<FFI_Result<FFI_ExecutionPlan>> {
+ let task_ctx: Result<Arc<TaskContext>, DataFusionError> =
+ (&provider.logical_codec.task_ctx_provider).try_into();
+ let runtime = provider.runtime().clone();
+ let logical_codec: Arc<dyn LogicalExtensionCodec> =
(&provider.logical_codec).into();
+ let internal_provider = Arc::clone(provider.inner());
+
+ async move {
+ let mut foreign_session = None;
+ let session = sresult_return!(
+ session
+ .as_local()
+ .map(Ok::<&(dyn Session + Send + Sync), DataFusionError>)
+ .unwrap_or_else(|| {
+ foreign_session =
Some(ForeignSession::try_from(&session)?);
+ Ok(foreign_session.as_ref().unwrap())
+ })
+ );
Review Comment:
The local-vs-foreign session conversion boilerplate is duplicated across
`delete_from_fn_wrapper`, `update_fn_wrapper`, and `truncate_fn_wrapper`.
Recommendation (optional): factor this into a shared helper (e.g., a small
function returning a session reference plus the `ForeignSession` guard) to
reduce copy/paste risk and keep future changes consistent across DML wrappers.
--
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]
