timsaucer commented on code in PR #23906:
URL: https://github.com/apache/datafusion/pull/23906#discussion_r3707366788
##########
datafusion/ffi/src/table_provider.rs:
##########
@@ -202,21 +228,45 @@ unsafe extern "C" fn table_type_fn_wrapper(
provider.inner().table_type().into()
}
-fn supports_filters_pushdown_internal(
- provider: &Arc<dyn TableProvider>,
- filters_serialized: &[u8],
+fn parse_serialized_exprs(
+ exprs_serialized: &[u8],
task_ctx: &Arc<TaskContext>,
codec: &dyn LogicalExtensionCodec,
-) -> Result<SVec<FFI_TableProviderFilterPushDown>> {
- let filters = match filters_serialized.is_empty() {
- true => vec![],
+) -> Result<Vec<Expr>> {
Review Comment:
Same as my other comment, it feels like this is a helper function that
should live in `datafusion-proto` crate.
##########
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(),
+ ))
+ }
Review Comment:
I'm starting to notice bloat in the FFI crate where as we're adding in
coverage of the trait functions we're adding in new implementations instead of
expanding existing implementations to cover the new functions. I think for long
term maintainability it's better to have fewer implementations that are more
full featured rather than adding in a new impl for every PR.
##########
datafusion/ffi/src/table_provider.rs:
##########
@@ -202,21 +228,45 @@ unsafe extern "C" fn table_type_fn_wrapper(
provider.inner().table_type().into()
}
-fn supports_filters_pushdown_internal(
- provider: &Arc<dyn TableProvider>,
- filters_serialized: &[u8],
+fn parse_serialized_exprs(
+ exprs_serialized: &[u8],
task_ctx: &Arc<TaskContext>,
codec: &dyn LogicalExtensionCodec,
-) -> Result<SVec<FFI_TableProviderFilterPushDown>> {
- let filters = match filters_serialized.is_empty() {
- true => vec![],
+) -> Result<Vec<Expr>> {
+ match exprs_serialized.is_empty() {
+ true => Ok(vec![]),
false => {
- let proto_filters = LogicalExprList::decode(filters_serialized)
+ let proto_exprs = LogicalExprList::decode(exprs_serialized)
.map_err(|e| DataFusionError::Plan(e.to_string()))?;
- parse_exprs(proto_filters.expr.iter(), task_ctx.as_ref(), codec)?
+ Ok(parse_exprs(
+ proto_exprs.expr.iter(),
+ task_ctx.as_ref(),
+ codec,
+ )?)
}
- };
+ }
+}
+
+fn serialize_expr_list<'a>(
+ exprs: impl IntoIterator<Item = &'a Expr>,
+ codec: &dyn LogicalExtensionCodec,
+) -> Result<SVec<u8>> {
+ Ok(LogicalExprList {
+ expr: serialize_exprs(exprs, codec)?,
+ }
+ .encode_to_vec()
+ .into_iter()
+ .collect())
+}
Review Comment:
It feels like helper functions like this should go in the `datafusion-proto`
crate instead.
##########
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'd say this is optional since it's only used in tests, but it may be good
form to switch to the tokio mutex.
--
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]
