Amogh-2404 commented on code in PR #23906:
URL: https://github.com/apache/datafusion/pull/23906#discussion_r3766187340
##########
datafusion/ffi/tests/ffi_integration.rs:
##########
@@ -95,6 +96,41 @@ mod tests {
Ok(())
}
+ #[tokio::test]
+ async fn test_ffi_table_provider_dml_cross_library() -> Result<()> {
+ let module = get_module()?;
+ let (ctx, codec) = super::utils::ctx_and_codec();
+
+ let ffi_provider = (module.create_table_with_statistics)(codec);
+ let foreign: Arc<dyn TableProvider> = (&ffi_provider).into();
+ let state = ctx.state();
+
+ let delete_plan = foreign
+ .delete_from(
Review Comment:
Added this. The cross-library test now covers both DELETE and UPDATE with
empty filters, and the producer fixture checks that the empty list reaches the
provider unchanged.
##########
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:
I removed the captured `Mutex` state entirely. The existing test provider
now validates each call directly, with atomics used only to distinguish the
filtered and empty-filter calls.
##########
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:
Removed the `Mutex` as part of the test fixture refactor.
##########
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:
Removed the `Mutex` as part of the test fixture refactor.
--
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]
