vbarua commented on code in PR #13803:
URL: https://github.com/apache/datafusion/pull/13803#discussion_r1890823837
##########
datafusion/substrait/src/logical_plan/consumer.rs:
##########
@@ -94,13 +102,400 @@ use substrait::proto::{
join_rel, plan_rel, r#type,
read_rel::ReadType,
rel::RelType,
- rel_common, set_rel,
+ rel_common,
sort_field::{SortDirection, SortKind::*},
- AggregateFunction, Expression, NamedStruct, Plan, Rel, RelCommon, Type,
+ AggregateFunction, AggregateRel, ConsistentPartitionWindowRel, CrossRel,
ExchangeRel,
+ Expression, ExtensionLeafRel, ExtensionMultiRel, ExtensionSingleRel,
FetchRel,
+ FilterRel, JoinRel, NamedStruct, Plan, ProjectRel, ReadRel, Rel,
RelCommon, SetRel,
+ SortRel, Type,
};
use substrait::proto::{ExtendedExpression, FunctionArgument, SortField};
-use super::state::SubstraitPlanningState;
+#[async_trait]
+/// This trait is used to consume Substrait plans, converting them into
DataFusion Logical Plans.
+/// It can be implemented by users to allow for custom handling of relations,
expressions, etc.
+///
+/// # Example Usage
+///
+/// ```
+/// use async_trait::async_trait;
+/// use datafusion::catalog::TableProvider;
+/// use datafusion::common::{not_impl_err, substrait_err, DFSchema,
ScalarValue, TableReference};
+/// use datafusion::error::Result;
+/// use datafusion::execution::SessionState;
+/// use datafusion::logical_expr::{Expr, LogicalPlan, LogicalPlanBuilder};
+/// use std::sync::Arc;
+/// use substrait::proto;
+/// use substrait::proto::{ExtensionLeafRel, FilterRel, ProjectRel};
+/// use datafusion::arrow::datatypes::DataType;
+/// use datafusion::logical_expr::expr::ScalarFunction;
+/// use datafusion_substrait::extensions::Extensions;
+/// use datafusion_substrait::logical_plan::consumer::{
+/// from_project_rel, from_substrait_rel, from_substrait_rex,
SubstraitConsumer
+/// };
+///
+/// use datafusion_substrait::logical_plan::state::SubstraitPlanningState;
+///
+/// struct CustomSubstraitConsumer {
+/// extensions: Arc<Extensions>,
+/// state: Arc<SessionState>,
+/// }
+///
+/// #[async_trait]
+/// impl SubstraitConsumer for CustomSubstraitConsumer {
+/// async fn resolve_table_ref(
+/// &self,
+/// table_ref: &TableReference,
+/// ) -> Result<Option<Arc<dyn TableProvider>>> {
+/// self.state.table(table_ref).await
+/// }
+///
+/// fn get_extensions(&self) -> &Extensions {
+/// self.extensions.as_ref()
+/// }
+///
+/// fn get_state(&self) -> &SessionState {
+/// self.state.as_ref()
+/// }
+///
+/// // You can reuse existing consumer code to assist in handling advanced
extensions
+/// async fn consume_project(&self, rel: &ProjectRel) ->
Result<LogicalPlan> {
+/// let df_plan = from_project_rel(self, rel).await?;
+/// if let Some(advanced_extension) = rel.advanced_extension.as_ref() {
+/// not_impl_err!(
+/// "decode and handle an advanced extension: {:?}",
+/// advanced_extension
+/// )
+/// } else {
+/// Ok(df_plan)
+/// }
+/// }
+///
+/// // You can implement a fully custom consumer method if you need
special handling
+/// async fn consume_filter(&self, rel: &FilterRel) -> Result<LogicalPlan>
{
+/// let input = from_substrait_rel(self,
rel.input.as_ref().unwrap()).await?;
+/// let expression =
+/// from_substrait_rex(self, rel.condition.as_ref().unwrap(),
input.schema())
+/// .await?;
+/// // though this one is quite boring
+/// LogicalPlanBuilder::from(input).filter(expression)?.build()
+/// }
+///
+/// // You can add handlers for extension relations
+/// async fn consume_extension_leaf(
+/// &self,
+/// rel: &ExtensionLeafRel,
+/// ) -> Result<LogicalPlan> {
+/// not_impl_err!(
+/// "handle protobuf Any {} as you need",
+/// rel.detail.as_ref().unwrap().type_url
+/// )
+/// }
+///
+/// // and handlers for user-define types
+/// async fn consume_user_defined_type(&self, typ:
&proto::r#type::UserDefined) -> Result<DataType> {
+/// let type_string =
self.extensions.types.get(&typ.type_reference).unwrap();
+/// match type_string.as_str() {
+/// "u!foo" => not_impl_err!("handle foo conversion"),
+/// "u!bar" => not_impl_err!("handle bar conversion"),
+/// _ => substrait_err!("unexpected type")
+/// }
+/// }
+///
+/// // and user-defined literals
+/// async fn consume_user_defined_literal(&self, literal:
&proto::expression::literal::UserDefined) -> Result<ScalarValue> {
+/// let type_string =
self.extensions.types.get(&literal.type_reference).unwrap();
+/// match type_string.as_str() {
+/// "u!foo" => not_impl_err!("handle foo conversion"),
+/// "u!bar" => not_impl_err!("handle bar conversion"),
+/// _ => substrait_err!("unexpected type")
+/// }
+/// }
+/// }
+/// ```
+///
+pub trait SubstraitConsumer: Send + Sync + Sized {
+ async fn resolve_table_ref(
+ &self,
+ table_ref: &TableReference,
+ ) -> Result<Option<Arc<dyn TableProvider>>>;
+
+ // TODO: Remove these two methods
+ // Ideally, the abstract consumer should not place any constraints on
implementations.
+ // The functionality for which the Extensions and SessionState is needed
should be abstracted
+ // out into methods on the trait. As an example, resolve_table_reference
is such a method.
+ fn get_extensions(&self) -> &Extensions;
+ fn get_state(&self) -> &SessionState;
Review Comment:
Good idea, it's a more precise constraint to have.
--
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: github-unsubscr...@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org
For additional commands, e-mail: github-h...@datafusion.apache.org
