gabotechs commented on code in PR #26603:
URL: https://github.com/apache/datafusion/pull/26603#discussion_r4413117113
##########
datafusion/substrait/src/logical_plan/consumer/rel/join_rel.rs:
##########
@@ -33,12 +36,20 @@ pub async fn from_join_rel(
return not_impl_err!("JoinRel with post_join_filter is not yet
supported");
}
- let left: LogicalPlanBuilder = LogicalPlanBuilder::from(
- consumer.consume_rel(join.left.as_ref().unwrap()).await?,
- );
- let right = LogicalPlanBuilder::from(
- consumer.consume_rel(join.right.as_ref().unwrap()).await?,
- );
+ let (Some(left), Some(right)) = (join.left.as_deref(),
join.right.as_deref()) else {
+ return substrait_err!("JoinRel requires left and right inputs");
+ };
+ let [left, right] = consumer
+ .consume_rel_inputs(&[left, right])
Review Comment:
It does not feel very clean to have two different APIs for consuming
relations. Why not just consuming them concurrently just with
`futures_util::future::try_join_all`?
##########
datafusion/substrait/src/logical_plan/consumer/substrait_consumer.rs:
##########
@@ -215,6 +215,24 @@ pub trait SubstraitConsumer: Send + Sync + Sized {
from_substrait_rel(self, rel).await
}
+ /// Consumes independent inputs of a JOIN or UNION, returning plans in
input order.
+ ///
+ /// Implementations must return exactly one plan per input, in the same
order.
+ /// On error, implementations must drop any unfinished input futures.
+ ///
+ /// The default preserves sequential consumption. Implementations may
override it
+ /// when they can isolate mutable traversal state between inputs.
+ async fn consume_rel_inputs(
Review Comment:
I don't really understand the rationale for this change, based on the PR
description:
> Custom Substrait consumers cannot overlap independent JOIN/UNION input
consumption while reusing upstream operator assembly. Those helpers await each
child directly. Downstream consumers currently need to override the operators
and wrap already-consumed children just to control scheduling.
"while reusing upstream operator assembly" what does this mean concretely?
"Those helpers await each child directly" a Substrait consumer is not a
helper, it's just a trait, is this referring to something else?
"Downstream consumers" what are downstream consumers? Here there's only one
consumer
"wrap already-consumed children just to control scheduling" what is
scheduling here?
I'd probably just reword the rationale part of the PR description, I think I
understand just 1 out of 5 words there.
--
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]
