[alexanderbianchi]

Problem:

Multi-input nodes need the same scheduling decisions: sequential or concurrent consumption, and how much concurrency to allow. Implementing those choices separately in each node duplicates logic.

Solution:

Route input consumption through one overridable consume_rel_inputs hook. Users can use try_join_all, limit concurrency, or apply their own scheduling policy.

Impact:

One implementation controls input scheduling across all nodes using the hook, without rewriting their conversion logic.

Parallel Consumption

We cannot just try_join_all, because the consumer contains mutable state for CTE's and lambda functions. Blind concurrent consumption would lead to race conditions.

Options

• Fork (#26896): A smaller change today, but consumption still carries mutable scope state that each fork must isolate. Custom consumers can already implement their own forking inside this hook; the upstream fork API is only needed if we want upstream to provide parallel consumption.
• Context (#26890): A broader API change that removes mutable traversal state from the consumer. With a valid session and the appropriate explicit scope context, consumption can start at any point without relying on the consumer's previous traversal history.

Testing

Since this PR is effectively a refactor, existing testing cover's all the relations in substrait.

[geoffreyclaude]

Looks pretty straightforward to me! As commented inline, I'd prefer it this were a bit more general (all multi-children nodes, rather than just join_rel and set_rel, if only for consistency's sake.)

[geoffreyclaude]

For this to be complete, I think makes sense to extend it to other types of multi-children relations. CrossRel for instance, but probably not only.

[gabotechs]

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?

[alexanderbianchi]

We have this lambda code.

consumer.push_lambda_parameters(...);     // Enter the function's scope.
consumer.consume_expression(body, ...).await?;
consumer.pop_lambda_parameters();         // Leave that scope.

It keeps a stack because lambdas can be nested. The consumer carries a record of “where am I in this query?” Running two branches concurrently on one consumer can mix those records. There's a race condition here.

[gabotechs]

🤔 How can that be, consumer is not mutable. Does that mean there's a Mutex somewhere?

[gabotechs]

If that's the case, any chance to fix that instead so that we can consume concurrently?

[alexanderbianchi]

Something like this

let nested = context.with_outer_schema(outer_schema).with_lambda_parameters(parameters);
try_join_all(
    inputs.iter().map(|rel| self.consume_rel(rel, context))
).await

That has some context where we can pass in the context explicitly removes state front the consumer but is a broader API change.

[alexanderbianchi]

Lambda state:

• The RwLock and parameter stack.
• Mutation through &self: acquire the write lock and push parameters.
• The sequence spanning an await: push → consume body → pop.

Correlated-subquery state:

• The outer_schemas lock and push/pop mutations through &self.
• The sequence spanning an await: push → consume subquery → pop.

[alexanderbianchi]

The reason this PR initially just added a multi input consumer was so we could override it and handle the parallelism complexity in the application code that overrides the consume.

[gabotechs]

Then this is still a problem for the new .consume_rel_inputs() API, people implementing it can decide to do try_join_all() inside and shoot themselves in the foot producing incorrect plans because of mutation races to the lambda parameters or the correlated subqueries.

So whether we ship the .consume_rel_inputs() or do the try_join_all() ourselves in this code here seems like a matter of who is responsible for producing incorrect plans, DataFusion or its consumers, but from an API design standpoint, I don't think neither of the two are acceptable, we should not offer APIs that can silently produce incorrect plans.

[alexanderbianchi]

Sounds good, if we want to make the overall structure safer I think we should pursue something like #26890 which I can clean up and push for review. I can close this PR.

[gabotechs]

That sounds more robust definitely, I'd definitely be more in favor of pursuing that