Repository navigation
Avoid recompute CTEs (common table expressions) / share input plans #8777
Description
Is your feature request related to a problem or challenge?
The core usecase is:
with x as (
<expensive full join of two large tables producing small-ish result>
)
select * from x where ...
union all
select * from x where ...
union all
select * from x where ...DataFusion will effectively run the subquery x three times (it will basically copy the LogicalPlan for x wherever it is used.
┌─────────────────────┐
│ UNION ALL │
│ │
└─────────────────────┘
▲ ▲ ▲
│ │ │
┌─────────────────────────────┘ │ └─────────────────────────────┐
│ │ │
│ │ │
┌────────────────┐ ┌────────────────┐ ┌────────────────┐
│ Filter 1 │ │ Filter 2 │ │ Filter 3 │
└────────────────┘ └────────────────┘ └────────────────┘
▲ ▲ ▲
│ │ │
│ │ │
┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐
┌────────────────┐ ┌────────────────┐ ┌────────────────┐
│ │ Expensive Join │ │ │ │ Expensive Join │ │ │ │ Expensive Join │ │
└────────────────┘ └────────────────┘ └────────────────┘
│ ▲ │ │ ▲ │ │ ▲ │
┌──────┴──────┐ ┌──────┴──────┐ ┌──────┴──────┐
│ │ │ │ │ │ │ │ │ │ │ │
.───────. .───────. .───────. .───────. .───────. .───────.
│ ╱ ╲ ╱ ╲ │ │ ╱ ╲ ╱ ╲ │ │ ╱ ╲ ╱ ╲ │
( Input 1 ) ( Input 2 ) ( Input 1 ) ( Input 2 ) ( Input 1 ) ( Input 2 )
│ `. ,' `. ,' │ │ `. ,' `. ,' │ │ `. ,' `. ,' │
`─────' `─────' `─────' `─────' `─────' `─────'
│"x" │ │"x" │ │"x" │
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
This design has certain benefits:
- It is straightforward to implement (probably why DataFusion is like this)
- If the different
UNION ALLarms have different predicates, they could potentially be pushed down in one branch but not the others.
Describe the solution you'd like
However, in many cases it would likely be better to do to the expensive join only once and reuse the results like this:
┌─────────────────────┐
│ UNION ALL │
│ │
└─────────────────────┘
▲ ▲ ▲
│ │ │
┌─────────────────┘ │ └───────────────────┐
│ │ │
│ │ │
│ │ │
┌────────────────┐ ┌────────────────┐ ┌────────────────┐
│ Filter 1 │ │ Filter 2 │ │ Filter 3 │
└────────────────┘ └────────────────┘ └────────────────┘
▲ ▲ ▲
│ │ │
│ │ │
└────────────────────┐ │ ┌─────────────────────┘
│ │ │
│ │ │
│ │ │
┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐
┌────────────────┐
│ │ Expensive Join │ │
└────────────────┘
│ ▲ │
┌──────┴──────┐
│ │ │ │
.───────. .───────.
│ ╱ ╲ ╱ ╲ │
( Input 1 ) ( Input 2 )
│ `. ,' `. ,' │
`─────' `─────'
│"x" │
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
Describe alternatives you've considered
I think there are several considerations for this design, the biggest is that it is a 'diamond' plan where the same stream can be consumed at different rates potentially needing to buffer the entire intermediate result or else the plan will deadlock
For example
┌─────────────────────┐
│ Hash Join │
│ │
└─────────────────────┘
▲ ▲
│ │
┌─────────────┘ └────────────┐
│ │
│ │
│ │
Build Side Probe Side
(read completely (not read at all until Build
before probe side) Side is completely read)
▲ ▲
│ │
│ │
└────────────────────┬────────────────┘
│
│
│
┌ ─ ─ ─ ─ ─ ┴ ─ ─ ─ ─ ─ ─ ─ ┐
┌────────────────┐
│ │ Expensive Join │ │
└────────────────┘
│ ▲ │
┌──────┴──────┐
│ │ │ │
.───────. .───────.
│ ╱ ╲ ╱ ╲ │
( Input 1 ) ( Input 2 )
│ `. ,' `. ,' │
`─────' `─────'
│"x" │
─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
Additional context
This came from a discord thread from @sergiimk
Activity
[-]Do not recompute CTEs (common table expressions)[/-][+]Avoid recompute CTEs (common table expressions)[/+]alamb commentedon Jan 7, 2024
This also came up as potentially part of a solution for #8582
Dandandan commentedon Jan 8, 2024
This is an interesting topic / improvement!
Two suggestions from my side
[NOT] MATERIALIZEDto force / disable CTE materialization. Often it's much faster to disable materialization, as otherwise pushdown / CBO optimizations will not apply (as you suggested). It might be interesting to consider supporting this in DataFusion as well, as it is hard from an optimizer standpoint to decide what to do.https://www.postgresql.org/docs/current/queries-with.html#QUERIES-WITH-CTE-MATERIALIZATION
crepererum commentedon Jan 16, 2024
I think this is very similar to our repartitioning code and the trade-offs and problems we see there. The reason is that a repartition is basically also as single input with mulitiple consumers. Just think of it like the same data but with a column "bool: belongs to this output" added.
[-]Avoid recompute CTEs (common table expressions)[/-][+]Avoid recompute CTEs (common table expressions) / share input plans[/+]suibianwanwank commentedon Jun 28, 2025
Hi, I’d like to take a try at this task.
My plan is to first support
CTEwith the WITH ... AS MATERIALIZED syntax. After that, we can explore broader optimizations in the optimizer phase.As for the implementation of
CTEin the ExecutionPlan, I currently have two ideas:Add a top-level
CTEQueryExecnode to ensure the CTE query is fully executed (blocking), and cache its result in aCTEWorkTableso that subsequent CTEScan nodes can read from it. (It seems DuckDB takes a similar approach.)However, I'm not sure if this is feasible in DataFusion, as it might require blocking the ExecutionPlan and collecting all RecordBatches.
Materialize the CTE before execution by creating a temporary table in the current schema and writing the required data into it. Then, replace all following scan with queries against this temporary table. (I noticed that Databend uses a similar approach.)
I’d like to gather some feedback first to make sure the design is general enough and correct. Thanks!
findepi commentedon Jun 28, 2025
Maintaining streaming processing is very useful in the number of circumstances: a query with LIMIT, a query with TopN over sorted input data, an interactive query.
I don't know how CTE reuse execution v1 should look like, but we should work towards the goal that all plan nodes dependent on the CTE can incrementally consume data passed to the CTE. Effectively, each should maintain an independent iterator over data being produced. Additionally, CTE should be aware of number of upstream consumers, which allows it to free data no longer needed by any consumer.
Implementation wise, it could be done with a separate blocking deque (VecDeque?) for each consumer.
Incoming data is added to all queues (shared Arcs, no data copying).
Consumers pulls from the queue. When it's empty, it waits for new data or end of processing signal.
crepererum commentedon Jul 1, 2025
I think this problem is actually very similar to
RepatitionExecand the distributor channels in there, except thatRepartitionExecsends 1 row to 1 output, the construct required here would just send them to ALL outputs. And we'll likely want a similar buffer limiter, so that the CTE output isn't buffered unlimited when no consumer polls the data (at the same time, multiple consumer may poll at the same time). So I would propose to use the same underlying "distributor channel" primitive (or whatever we replace that with).jayshrivastava commentedon Feb 6, 2026
This is a great solution. We cannot enable it if there's a join. We should either re-execute the shared child in this case or perhaps optionally can buffer all (in memory or disk). Step 1 can be falling back to re-executing the child.
This might apply to nodes other than joins... Thinking out loud, I think the general thing to check for is - if there is any node which is a common ancestor of any two parents of "FanoutExec" that has dependencies between its children, then remove the "FanoutExec". "has dependencies between its children" is hard to check for. Maybe a new property on
ExecutionPlanwould help with that. We might also need some sort of cycle detection to check for common ancestors.alamb commentedon Feb 7, 2026
Why not? Can you provide an example of what you are thinking of here?
Is this like if the join probe input is shared?
jayshrivastava commentedon Feb 14, 2026
Yes. Just restating the deadlock concern from the issue description. I generated this diagram to show the problem - if the probe side buffer is full, the "FanoutExec" is back-pressured. We end up blocking the build side from completing and deadlock.
In this case, we can either spill to disk on the probe side or simply not insert a "FanoutExec".
suibianwanwank commentedon Feb 15, 2026
I agree with your point. I actually tried to implement this approach before, but this issue blocked me.
The fundamental difference from repartition is that the consumers are not stateless — there may be various forms of dependencies between them.
alamb commentedon Aug 6, 2026
You can probably work around this limitation by implementing some sort of "buffer" node that saves the intermediate result (the output to be shared) and then implement "datasource" nodes that read from that intermediate result
So the ExecutionPlan would something like this:
However internally, there would be some connection between Buffer and BufferReader, similar to how dynamic predicates are used to send information between nodes.
It is sort of a hack for sure, but I think it would work
alamb commentedon Aug 7, 2026
This issue came up again at the community showcase (#22963) yesterday
FYI @@milindsrivastava1997 I wrote up a potential way to implement such plans above. If it is interesting to you, perhaps we can work together on coding up an example or something to show how it would work
nathanb9 commentedon Aug 8, 2026
Really liked the sketch talk @milindsrivastava1997
@alamb I can help with this too. One thought: before (or alongside) thinking about the buffer physical nodes, could we try to "fuse" common subplans? Something like these: #23175 (comment). The reason is to try to avoid buffering entirely if possible first
milindsrivastava1997 commentedon Aug 10, 2026
@alamb @nathanb9 Thank you both. I'd love to help with this.
I'll go through the discussion in this issue, and the one Nathan linked. Anything else that might be helpful for me to know?
Full disclosure: I haven't contributed to Datafusion (or any other large open-source project) so might need help in the workflow (apart from what I get from the contribution guide)
alamb commentedon Aug 10, 2026
Thank you @nathanb9 and @milindsrivastava1997 ; I am not sure we need to change the datafusion core for this one -- the initial thing I think we should do is shoot to add an example to the datafusion showing how downstream projects can implement this pattern
https://github.com/apache/datafusion/tree/main/datafusion-examples
The idea would be to cook up an example of wanting to share the output of a dataframe with multiple other data frames, but not wanting to materialize the intermediate result as a table
And then we could illustrate how to use the datafusion extension APIs to implement this sharing node
nathanb9 commentedon Aug 30, 2026
Put together a simple example: #24780. Its a bit of a large example because, in addition to using thje extension points, it requires implementation for broadcasting
RecordBatchs of a single stream to multiple consumers.@alamb is this similar to what you had in mind for an example?
feat: support `WITH ... AS [NOT] MATERIALIZED` CTEs (apache#25580)
feat: support `WITH ... AS [NOT] MATERIALIZED` CTEs (apache#25580)
feat: support `WITH ... AS [NOT] MATERIALIZED` CTEs (apache#25580)