You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
feat: support WITH ... AS [NOT] MATERIALIZED CTEs - #25580
DataFusion inlines every CTE reference. When a query references the same CTE several times, the CTE body is planned and executed once for each reference. Postgres lets users opt out of this with WITH x AS MATERIALIZED (...). sqlparser already parses the directive into Cte.materialized, but the SQL planner ignores it.
Earlier attempts (#22551, #22675) bundled automatic materialization, heuristics and filter pushdown. This PR is limited to the explicit directive, as proposed in #17737 (comment):
No config flag.
No heuristic.
Queries without MATERIALIZED plan exactly as before.
Deciding automatically when to materialize can be discussed separately.
Example:
WITH r AS MATERIALIZED (SELECT random() AS v)
SELECTa.v=b.vFROM r a, r b;
-- true: the body runs once. With NOT MATERIALIZED (or no directive) each reference evaluates random() separately.
What changes are included in this PR?
Planner:datafusion/sql/src/cte.rs and query.rs plan a MATERIALIZED CTE as a MaterializedCte node, whose children are the CTE body and the rest of the query, plus one MaterializedCteScan leaf for each reference. These are UserDefinedLogicalNodes in datafusion-expr.
NOT MATERIALIZED and the default keep today's inlining.
MATERIALIZED also applies in a WITH RECURSIVE block. A recursive CTE is still planned as RecursiveQuery, and its reference to itself stays a work table scan.
head_output_expr, PullUpCorrelatedExpr and PushDownFilter look through MaterializedCte to its continuation, so IN / EXISTS subqueries and filters from an outer query work as with an inlined CTE. The body is never filtered.
Each occurrence of a CTE in a plan gets its own id before physical planning, so a view that is referenced two times runs its CTE two times, as in PostgreSQL.
Physical plan:MaterializedCteExec and MaterializedCteScanExec in datafusion-physical-plan, and a physical planner pass that binds each scan to its producer.
Execution: the body is run to completion once, all body partitions in parallel, on the first poll of any scan. It is started by one shared future and runs in a JoinSet that is aborted when dropped. Every scan then replays the buffered batches.
Memory: buffered batches are tracked by a MemoryReservation created with can_spill(true). When the reservation cannot grow, the rest of that partition is written to a spill file through SpillManager. SpillFile::read_stream can be called by any number of readers at the same time, so every scan replays both the in-memory batches and the spill files. Scans read spill files without read-ahead, and after buffering one decoded batch per concurrent reader is reserved. If the pool cannot grant this, in-memory batches are moved to spill files until it can. Each scan reserves memory for the largest batch on disk, so an in-memory batch that is larger than the batches on disk is written in pieces, and a move is made only when it releases more memory than it adds to the scans. fix: bound spill file batches by bytes so external sorts over wide rows can merge #25910 adds a byte bound to spill files with a better size model; when it merges, this operator can use it and drop its own slicing. A body that reads an unbounded source is rejected at planning time, because it would be buffered forever.
CTE names are not unique: each scan binds to its producer through an id allocated by the planner, not through the CTE name. Sibling subqueries that declare CTEs with the same name are tested.
Alignment with ScalarSubqueryExec:MaterializedCteExec follows the same shape. The rest of the query is its output and passes through input order, partitioning and statistics. A scan inside a scalar subquery can start the body itself (tested).
Why not WorkTable:WorkTable gives its batches to exactly one consumer (take()) and returns an error when memory runs out; RecursiveQueryExec does not spill. A materialized CTE needs N readers and spilling. The scan does follow the same pattern as WorkTableExec: a leaf that reads a handle bound at planning time.
Statistics: the scans report unknown statistics for now. A leaf cannot ask its producer for statistics through StatisticsContext. I would like to do this in a follow-up. This is a known limitation that can make a plan worse than the plan of the inlined CTE, not only leave it the same: JoinSelection does not know that a scan is small, and can put the other, larger input of a hash join on the build side (pinned in cte_materialized.slt).
A streaming fan-out for consumers that are known to be independent.
Are these changes tested?
Yes. The new file datafusion/sqllogictest/test_files/cte_materialized.slt covers:
a volatile body, which returns one value when MATERIALIZED and differs per reference when NOT MATERIALIZED or with no directive;
EXPLAIN, which shows one DataSourceExec for three references;
EXPLAIN ANALYZE, where the body buffers 4 rows once and each of the three scans outputs 4 rows;
a hash join whose two sides both read the CTE, with no deadlock;
a scalar subquery reference;
a CTE that reads an earlier CTE;
sibling CTEs with the same name;
IN, NOT IN and correlated EXISTS / NOT EXISTS / IN subqueries that declare a materialized CTE;
WITH RECURSIVE with recursive and non-recursive materialized CTEs;
a view referenced two times (separate CTEs) and one reference with two scans (one shared CTE);
a filter from an outer query, which is pushed into the continuation;
column aliases;
a spill under memory_limit: 200k rows, about 184k spilled (part of the in-memory prefix moves to disk to make room for the two readers), and both scans still return 200k rows.
a spill where the first batch is larger than the batches on disk: the query finishes under an 18 MB limit, and failed before with an empty pool.
The full sqllogictest suite passes.
Are there any user-facing changes?
Yes. WITH ... AS MATERIALIZED is now honored when the Postgres dialect is used. Queries without the directive are unchanged. There is no new config option and no public API break. The new nodes are additions.
Open questions for reviewers
Should these be a new LogicalPlan variant, like RecursiveQuery, or stay as built-in UserDefinedLogicalNodes, as in this PR, which keeps the change smaller?
Is it acceptable for protobuf serialization of these nodes to fail until a follow-up PR adds it?
❌ Patch coverage is 86.49156% with 144 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.87%. Comparing base (b645de0) to head (1a43b6c).
This rejects an unbounded body even when the CTE is never referenced. The SQL planner wraps every MATERIALIZED declaration, so WITH u AS MATERIALIZED (SELECT * FROM unbounded_t) SELECT 1 reaches this check despite having no scan and despite execution otherwise being lazy; the query should not evaluate u. Eliminate producers with no matching scan (or gate this check on an actual reference) before rejecting unbounded input.
Fails planning for correlated MATERIALIZED CTEs
datafusion/optimizer/src/decorrelate.rs:152
A correlated CTE body is valid and the equivalent inlined CTE can currently be decorrelated, but setting can_pull_up = false makes build_join return None; physical planning then fails with “Physical plan does not support undecorrelated Subquery.” For example, an EXISTS subquery whose materialized CTE body filters on an outer column now errors instead of materializing once per outer row. This needs per-outer-row materialization semantics or a decorrelation that preserves that boundary.
The reason will be displayed to describe this comment to others. Learn more.
I think there's a pre-existing issue that shows up with WITH RECURSIVE ... AS MATERIALIZED, and I cut #25893 for it. If the recursive term widens the type (e.g. n + 0.5), it gets silently cast back to the static term's type, and the query can loop forever. It happens on main with or without MATERIALIZED, so it's not from this PR. Just a heads-up in case you add recursive tests that change types.
The reason will be displayed to describe this comment to others. Learn more.
I think reset_state on MaterializedCteExec might cause problems. It gets called whenever a plan needs to run again from scratch, e.g. by reset_plan_states, which recursive queries use before each iteration. The expectation is that it returns a fresh copy and leaves the original alone.
Here it clears the shared buffer in place, so every copy sees the reset, including one that's still running.
The reason will be displayed to describe this comment to others. Learn more.
Thanks. You are correct that this does not follow the contract. But the in-place reset is necessary for correct results in one case, and the HashJoinExec pattern does not fit here.
RecursiveQueryExec calls reset_plan_states on the recursive term before each iteration. A MATERIALIZED CTE that is declared in the recursive term must be evaluated again in each iteration. If its body reads the recursive CTE and the buffer is not cleared, each iteration replays the first result and the query does not terminate. The new slt tests cover this case.
A new node with a new buffer would also need new scans. reset_state sees only its own node, and a scan in a scalar subquery is not below MaterializedCteExec, because the planner puts ScalarSubqueryExec above the root. So the producer cannot rebind that scan.
A scan that already called execute() holds the old future and continues to work. The in-place reset can change results only if two runs of the same plan overlap and one run starts a scan after the other run resets. Nothing in DataFusion does this today, and I could not write a SQL query that shows it.
ScalarSubqueryExec::reset_state also clears its shared results in place, for the same reason.
A clean fix is to keep the output per run instead of on the node, for example in a registry that RecursiveQueryExec renews for each iteration. That changes TaskContext and RecursiveQueryExec, so I would like to do it in a follow-up. I will add a doc comment on reset_state that describes this, and open an issue. Is that OK for you?
The reason will be displayed to describe this comment to others. Learn more.
FYI: Postgres ignores NOT MATERIALIZED when the CTE has volatile functions if we are trying to follow that closely. But to me it also makes sense to allow this control for a user
The reason will be displayed to describe this comment to others. Learn more.
It's not clear to me we'd want to or need to follow that. I added slt tests pin the current behaviour for all three forms, so a follow-up will show the changes clearly.
The reason will be displayed to describe this comment to others. Learn more.
I think there's a problem when a materialized CTE comes from a table that plans its own saved query in scan(), like a ViewTable that isn't inlined. Each read of that table plans the saved query separately, and the saved plan always has the same CTE id. So reading it twice gives two MaterializedCteExec nodes with the same id, and the final binding pass fails:
-- w is a view over `WITH r AS MATERIALIZED (SELECT a FROM t) SELECT a FROM r`SELECT*FROM w x, w y;
-- Internal error: Two MaterializedCteExec nodes have the id 1
It works with NOT MATERIALIZED. Could each planning run assign fresh ids?
- Plan IN / NOT IN subqueries that declare a MATERIALIZED CTE:
`head_output_expr` forwards to the continuation.
- Decorrelate subqueries whose continuation refers to the outer query:
`PullUpCorrelatedExpr` rewrites the continuation and leaves the shared
body unchanged.
- Honor MATERIALIZED in a WITH RECURSIVE block, for recursive and
non-recursive CTEs.
- Give each occurrence of a CTE its own id before physical planning, so
that a view referenced twice runs its CTE twice, as in PostgreSQL.
Binding now fails on a duplicate id instead of sharing one buffer.
- Push filters above a MaterializedCte into the continuation, in the
logical and the physical plan. The body is never filtered.
- Create one spill file per partition, not one per spilled batch.
- Wrap the scan stream in `cooperative`, as its scheduling type says.
- Remove the unused `MaterializedCteScanExec::is_bound`.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
- Return a planning error when a MATERIALIZED CTE body is unbounded. The
body is fully buffered before a scan yields a row, so the query would
never produce output.
- MaterializedCte and MaterializedCteScan: partial_cmp returns
Some(Equal) only for equal nodes, as dyn_ord requires.
- Scans read spill files without read-ahead. After buffering, reserve one
decoded batch per concurrent reader. If the pool cannot grant it, move
in-memory batches to spill files until it can.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The replay stream now holds the MaterializedOutput, so the memory
reservations of the buffered batches and of the spill reads are kept
until the stream is dropped, also when the plan is dropped first.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Such a body would have to run once per outer row. Before this change the
query failed with an undecorrelated subquery error, which is a regression
from main, where the keyword is ignored.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A table provider can plan a saved logical plan in `scan`, as a `ViewTable`
that is not inlined does. Each read of such a table is a separate planning
run of the same saved plan, so the CTE ids only had to be unique within that
plan. Reading the table twice put two `MaterializedCteExec` nodes with the
same id into the final plan, and the binding pass failed with
"Two MaterializedCteExec nodes have the id ...".
Every occurrence of a materialized CTE now gets a new id in each planning
run, not only the duplicates. The binding pass also skips scans that an
inner planning run has already bound, so their partitions are not counted
twice in the buffer.
…e batch
A scan reserves memory for the largest batch that it reads from disk. To
move an in-memory batch that is larger than every batch on disk to a spill
file raised that reservation for every scan, and could make a query fail
with an empty memory pool.
A large batch is now written in pieces no larger than the batches on disk.
Each step makes the move that gains the most memory, and stops with the
allocation error when no move gains memory.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Type coercion can change the types of the body of a `MaterializedCte`, for example the type of a `NULL` column of a `UNION`.
The scans kept the schema from the SQL planner, so `WITH c AS MATERIALIZED (SELECT NULL a, 12 b UNION ALL SELECT 11 a, 13 b) SELECT * FROM c ORDER BY 1, 2` failed at execution with a schema mismatch.
`TypeCoercion` now coerces the body first, gives its schema to the scans in the continuation, and then coerces the continuation.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@xudong963 would you like me to hold this for you to review?
sorry, missed the PR, I did want to have a look, but it's okay to merge; I can revisit it later. This should be a big improvement for our tpcds/tpch benchmark
@xudong963 would you like me to hold this for you to review?
sorry, missed the PR, I did want to have a look, but it's okay to merge; I can revisit it later. This should be a big improvement for our tpcds/tpch benchmark
No problem. Happy to address any issues in followups (or happy to review a PR with your changes).
It might not help the benchmarks until we make it to #26276. There's still a long road ahead.
- Closesapache#17737.
- Related: apache#8777, apache#22676, apache#10337.
DataFusion inlines every CTE reference. When a query references the same
CTE several times, the CTE body is planned and executed once for each
reference. Postgres lets users opt out of this with `WITH x AS
MATERIALIZED (...)`. `sqlparser` already parses the directive into
`Cte.materialized`, but the SQL planner ignores it.
Earlier attempts (apache#22551, apache#22675) bundled automatic materialization,
heuristics and filter pushdown. This PR is limited to the explicit
directive, as proposed in
apache#17737 (comment):
- No config flag.
- No heuristic.
- Queries without `MATERIALIZED` plan exactly as before.
Deciding automatically when to materialize can be discussed separately.
Example:
```sql
WITH r AS MATERIALIZED (SELECT random() AS v)
SELECT a.v = b.v FROM r a, r b;
-- true: the body runs once. With NOT MATERIALIZED (or no directive) each reference evaluates random() separately.
```
- **Planner:** `datafusion/sql/src/cte.rs` and `query.rs` plan a
`MATERIALIZED` CTE as a `MaterializedCte` node, whose children are the
CTE body and the rest of the query, plus one `MaterializedCteScan` leaf
for each reference. These are `UserDefinedLogicalNode`s in
`datafusion-expr`.
- `NOT MATERIALIZED` and the default keep today's inlining.
- `MATERIALIZED` also applies in a `WITH RECURSIVE` block. A recursive
CTE is still planned as `RecursiveQuery`, and its reference to itself
stays a work table scan.
- `head_output_expr`, `PullUpCorrelatedExpr` and `PushDownFilter` look
through `MaterializedCte` to its continuation, so `IN` / `EXISTS`
subqueries and filters from an outer query work as with an inlined CTE.
The body is never filtered.
- Each occurrence of a CTE in a plan gets its own id before physical
planning, so a view that is referenced two times runs its CTE two times,
as in PostgreSQL.
- **Physical plan:** `MaterializedCteExec` and `MaterializedCteScanExec`
in `datafusion-physical-plan`, and a physical planner pass that binds
each scan to its producer.
- **Execution:** the body is run to completion once, all body partitions
in parallel, on the first poll of any scan. It is started by one shared
future and runs in a `JoinSet` that is aborted when dropped. Every scan
then replays the buffered batches.
- **Why full buffering:** a bounded streaming fan-out deadlocks when,
for example, both sides of one hash join read the same CTE (see apache#8777).
This design cannot deadlock.
- **Memory:** buffered batches are tracked by a `MemoryReservation`
created with `can_spill(true)`. When the reservation cannot grow, the
rest of that partition is written to a spill file through
`SpillManager`. `SpillFile::read_stream` can be called by any number of
readers at the same time, so every scan replays both the in-memory
batches and the spill files. Scans read spill files without read-ahead,
and after buffering one decoded batch per concurrent reader is reserved.
If the pool cannot grant this, in-memory batches are moved to spill
files until it can. Each scan reserves memory for the largest batch on
disk, so an in-memory batch that is larger than the batches on disk is
written in pieces, and a move is made only when it releases more memory
than it adds to the scans.
apache#25910 adds a byte bound to
spill files with a better size model; when it merges, this operator can
use it and drop its own slicing. A body that reads an unbounded source
is rejected at planning time, because it would be buffered forever.
- **CTE names are not unique:** each scan binds to its producer through
an id allocated by the planner, not through the CTE name. Sibling
subqueries that declare CTEs with the same name are tested.
- **Alignment with `ScalarSubqueryExec`:** `MaterializedCteExec` follows
the same shape. The rest of the query is its output and passes through
input order, partitioning and statistics. A scan inside a scalar
subquery can start the body itself (tested).
- **Why not `WorkTable`:** `WorkTable` gives its batches to exactly one
consumer (`take()`) and returns an error when memory runs out;
`RecursiveQueryExec` does not spill. A materialized CTE needs N readers
and spilling. The scan does follow the same pattern as `WorkTableExec`:
a leaf that reads a handle bound at planning time.
- **Statistics:** the scans report unknown statistics for now. A leaf
cannot ask its producer for statistics through `StatisticsContext`. I
would like to do this in a follow-up. This is a known limitation that
can make a plan worse than the plan of the inlined CTE, not only leave
it the same: `JoinSelection` does not know that a scan is small, and can
put the other, larger input of a hash join on the build side (pinned in
`cte_materialized.slt`).
Tracked under the EPIC apache#22676:
- Protobuf serialization. These nodes have no proto encoding yet, so
serializing a plan that contains them fails in the default extension
codec (apache#26231).
- Statistics, ordering and partitioning on the scans (apache#26230).
- Release the buffer when the last scan finishes rather than when the
plan is dropped (apache#26233).
- Byte-bounded spill files (apache#26232).
- Keep the output per run, so that `reset_state` follows its contract
(apache#26229).
- Deciding automatically when to materialize (apache#26237).
- First-class `LogicalPlan` variants (apache#26238).
- Pushing filters that all references share into the CTE body (apache#26239).
- Pruning unused columns from the CTE body (apache#26240).
- `sqlparser` parses `[NOT] MATERIALIZED` only in `PostgreSqlDialect`,
so the tests set that dialect (apache/datafusion-sqlparser-rs#2622).
- A streaming fan-out for consumers that are known to be independent.
Yes. The new file
`datafusion/sqllogictest/test_files/cte_materialized.slt` covers:
- a volatile body, which returns one value when `MATERIALIZED` and
differs per reference when `NOT MATERIALIZED` or with no directive;
- `EXPLAIN`, which shows one `DataSourceExec` for three references;
- `EXPLAIN ANALYZE`, where the body buffers 4 rows once and each of the
three scans outputs 4 rows;
- a hash join whose two sides both read the CTE, with no deadlock;
- a scalar subquery reference;
- a CTE that reads an earlier CTE;
- sibling CTEs with the same name;
- `IN`, `NOT IN` and correlated `EXISTS` / `NOT EXISTS` / `IN`
subqueries that declare a materialized CTE;
- `WITH RECURSIVE` with recursive and non-recursive materialized CTEs;
- a view referenced two times (separate CTEs) and one reference with two
scans (one shared CTE);
- a filter from an outer query, which is pushed into the continuation;
- column aliases;
- a spill under `memory_limit`: 200k rows, about 184k spilled (part of
the in-memory prefix moves to disk to make room for the two readers),
and both scans still return 200k rows.
- a spill where the first batch is larger than the batches on disk: the
query finishes under an 18 MB limit, and failed before with an empty
pool.
The full sqllogictest suite passes.
Yes. `WITH ... AS MATERIALIZED` is now honored when the Postgres dialect
is used. Queries without the directive are unchanged. There is no new
config option and no public API break. The new nodes are additions.
1. Should these be a new `LogicalPlan` variant, like `RecursiveQuery`,
or stay as built-in `UserDefinedLogicalNode`s, as in this PR, which
keeps the change smaller?
2. Is it acceptable for protobuf serialization of these nodes to fail
until a follow-up PR adds it?
---------
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
(cherry picked from commit cdc2fc0)
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Oct 16, 2026
## Which issue does this PR close?
- Closesapache#17737.
- Related: apache#8777, apache#22676, apache#10337.
## Rationale for this change
DataFusion inlines every CTE reference. When a query references the same
CTE several times, the CTE body is planned and executed once for each
reference. Postgres lets users opt out of this with `WITH x AS
MATERIALIZED (...)`. `sqlparser` already parses the directive into
`Cte.materialized`, but the SQL planner ignores it.
Earlier attempts (apache#22551, apache#22675) bundled automatic materialization,
heuristics and filter pushdown. This PR is limited to the explicit
directive, as proposed in
apache#17737 (comment):
- No config flag.
- No heuristic.
- Queries without `MATERIALIZED` plan exactly as before.
Deciding automatically when to materialize can be discussed separately.
Example:
```sql
WITH r AS MATERIALIZED (SELECT random() AS v)
SELECT a.v = b.v FROM r a, r b;
-- true: the body runs once. With NOT MATERIALIZED (or no directive) each reference evaluates random() separately.
```
## What changes are included in this PR?
- **Planner:** `datafusion/sql/src/cte.rs` and `query.rs` plan a
`MATERIALIZED` CTE as a `MaterializedCte` node, whose children are the
CTE body and the rest of the query, plus one `MaterializedCteScan` leaf
for each reference. These are `UserDefinedLogicalNode`s in
`datafusion-expr`.
- `NOT MATERIALIZED` and the default keep today's inlining.
- `MATERIALIZED` also applies in a `WITH RECURSIVE` block. A recursive
CTE is still planned as `RecursiveQuery`, and its reference to itself
stays a work table scan.
- `head_output_expr`, `PullUpCorrelatedExpr` and `PushDownFilter` look
through `MaterializedCte` to its continuation, so `IN` / `EXISTS`
subqueries and filters from an outer query work as with an inlined CTE.
The body is never filtered.
- Each occurrence of a CTE in a plan gets its own id before physical
planning, so a view that is referenced two times runs its CTE two times,
as in PostgreSQL.
- **Physical plan:** `MaterializedCteExec` and `MaterializedCteScanExec`
in `datafusion-physical-plan`, and a physical planner pass that binds
each scan to its producer.
- **Execution:** the body is run to completion once, all body partitions
in parallel, on the first poll of any scan. It is started by one shared
future and runs in a `JoinSet` that is aborted when dropped. Every scan
then replays the buffered batches.
- **Why full buffering:** a bounded streaming fan-out deadlocks when,
for example, both sides of one hash join read the same CTE (see apache#8777).
This design cannot deadlock.
- **Memory:** buffered batches are tracked by a `MemoryReservation`
created with `can_spill(true)`. When the reservation cannot grow, the
rest of that partition is written to a spill file through
`SpillManager`. `SpillFile::read_stream` can be called by any number of
readers at the same time, so every scan replays both the in-memory
batches and the spill files. Scans read spill files without read-ahead,
and after buffering one decoded batch per concurrent reader is reserved.
If the pool cannot grant this, in-memory batches are moved to spill
files until it can. Each scan reserves memory for the largest batch on
disk, so an in-memory batch that is larger than the batches on disk is
written in pieces, and a move is made only when it releases more memory
than it adds to the scans.
apache#25910 adds a byte bound to
spill files with a better size model; when it merges, this operator can
use it and drop its own slicing. A body that reads an unbounded source
is rejected at planning time, because it would be buffered forever.
### Notes on earlier review feedback (apache#22675)
- **CTE names are not unique:** each scan binds to its producer through
an id allocated by the planner, not through the CTE name. Sibling
subqueries that declare CTEs with the same name are tested.
- **Alignment with `ScalarSubqueryExec`:** `MaterializedCteExec` follows
the same shape. The rest of the query is its output and passes through
input order, partitioning and statistics. A scan inside a scalar
subquery can start the body itself (tested).
- **Why not `WorkTable`:** `WorkTable` gives its batches to exactly one
consumer (`take()`) and returns an error when memory runs out;
`RecursiveQueryExec` does not spill. A materialized CTE needs N readers
and spilling. The scan does follow the same pattern as `WorkTableExec`:
a leaf that reads a handle bound at planning time.
- **Statistics:** the scans report unknown statistics for now. A leaf
cannot ask its producer for statistics through `StatisticsContext`. I
would like to do this in a follow-up. This is a known limitation that
can make a plan worse than the plan of the inlined CTE, not only leave
it the same: `JoinSelection` does not know that a scan is small, and can
put the other, larger input of a hash join on the build side (pinned in
`cte_materialized.slt`).
### Out of scope, possible follow-ups
Tracked under the EPIC apache#22676:
- Protobuf serialization. These nodes have no proto encoding yet, so
serializing a plan that contains them fails in the default extension
codec (apache#26231).
- Statistics, ordering and partitioning on the scans (apache#26230).
- Release the buffer when the last scan finishes rather than when the
plan is dropped (apache#26233).
- Byte-bounded spill files (apache#26232).
- Keep the output per run, so that `reset_state` follows its contract
(apache#26229).
- Deciding automatically when to materialize (apache#26237).
- First-class `LogicalPlan` variants (apache#26238).
- Pushing filters that all references share into the CTE body (apache#26239).
- Pruning unused columns from the CTE body (apache#26240).
- `sqlparser` parses `[NOT] MATERIALIZED` only in `PostgreSqlDialect`,
so the tests set that dialect (apache/datafusion-sqlparser-rs#2622).
- A streaming fan-out for consumers that are known to be independent.
## Are these changes tested?
Yes. The new file
`datafusion/sqllogictest/test_files/cte_materialized.slt` covers:
- a volatile body, which returns one value when `MATERIALIZED` and
differs per reference when `NOT MATERIALIZED` or with no directive;
- `EXPLAIN`, which shows one `DataSourceExec` for three references;
- `EXPLAIN ANALYZE`, where the body buffers 4 rows once and each of the
three scans outputs 4 rows;
- a hash join whose two sides both read the CTE, with no deadlock;
- a scalar subquery reference;
- a CTE that reads an earlier CTE;
- sibling CTEs with the same name;
- `IN`, `NOT IN` and correlated `EXISTS` / `NOT EXISTS` / `IN`
subqueries that declare a materialized CTE;
- `WITH RECURSIVE` with recursive and non-recursive materialized CTEs;
- a view referenced two times (separate CTEs) and one reference with two
scans (one shared CTE);
- a filter from an outer query, which is pushed into the continuation;
- column aliases;
- a spill under `memory_limit`: 200k rows, about 184k spilled (part of
the in-memory prefix moves to disk to make room for the two readers),
and both scans still return 200k rows.
- a spill where the first batch is larger than the batches on disk: the
query finishes under an 18 MB limit, and failed before with an empty
pool.
The full sqllogictest suite passes.
## Are there any user-facing changes?
Yes. `WITH ... AS MATERIALIZED` is now honored when the Postgres dialect
is used. Queries without the directive are unchanged. There is no new
config option and no public API break. The new nodes are additions.
## Open questions for reviewers
1. Should these be a new `LogicalPlan` variant, like `RecursiveQuery`,
or stay as built-in `UserDefinedLogicalNode`s, as in this PR, which
keeps the change smaller?
2. Is it acceptable for protobuf serialization of these nodes to fail
until a follow-up PR adds it?
---------
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
DataFusion inlines every CTE reference. When a query references the same CTE several times, the CTE body is planned and executed once for each reference. Postgres lets users opt out of this with
WITH x AS MATERIALIZED (...).sqlparseralready parses the directive intoCte.materialized, but the SQL planner ignores it.Earlier attempts (#22551, #22675) bundled automatic materialization, heuristics and filter pushdown. This PR is limited to the explicit directive, as proposed in #17737 (comment):
MATERIALIZEDplan exactly as before.Deciding automatically when to materialize can be discussed separately.
Example:
What changes are included in this PR?
datafusion/sql/src/cte.rsandquery.rsplan aMATERIALIZEDCTE as aMaterializedCtenode, whose children are the CTE body and the rest of the query, plus oneMaterializedCteScanleaf for each reference. These areUserDefinedLogicalNodes indatafusion-expr.NOT MATERIALIZEDand the default keep today's inlining.MATERIALIZEDalso applies in aWITH RECURSIVEblock. A recursive CTE is still planned asRecursiveQuery, and its reference to itself stays a work table scan.head_output_expr,PullUpCorrelatedExprandPushDownFilterlook throughMaterializedCteto its continuation, soIN/EXISTSsubqueries and filters from an outer query work as with an inlined CTE. The body is never filtered.MaterializedCteExecandMaterializedCteScanExecindatafusion-physical-plan, and a physical planner pass that binds each scan to its producer.JoinSetthat is aborted when dropped. Every scan then replays the buffered batches.MemoryReservationcreated withcan_spill(true). When the reservation cannot grow, the rest of that partition is written to a spill file throughSpillManager.SpillFile::read_streamcan be called by any number of readers at the same time, so every scan replays both the in-memory batches and the spill files. Scans read spill files without read-ahead, and after buffering one decoded batch per concurrent reader is reserved. If the pool cannot grant this, in-memory batches are moved to spill files until it can. Each scan reserves memory for the largest batch on disk, so an in-memory batch that is larger than the batches on disk is written in pieces, and a move is made only when it releases more memory than it adds to the scans. fix: bound spill file batches by bytes so external sorts over wide rows can merge #25910 adds a byte bound to spill files with a better size model; when it merges, this operator can use it and drop its own slicing. A body that reads an unbounded source is rejected at planning time, because it would be buffered forever.Notes on earlier review feedback (#22675)
ScalarSubqueryExec:MaterializedCteExecfollows the same shape. The rest of the query is its output and passes through input order, partitioning and statistics. A scan inside a scalar subquery can start the body itself (tested).WorkTable:WorkTablegives its batches to exactly one consumer (take()) and returns an error when memory runs out;RecursiveQueryExecdoes not spill. A materialized CTE needs N readers and spilling. The scan does follow the same pattern asWorkTableExec: a leaf that reads a handle bound at planning time.StatisticsContext. I would like to do this in a follow-up. This is a known limitation that can make a plan worse than the plan of the inlined CTE, not only leave it the same:JoinSelectiondoes not know that a scan is small, and can put the other, larger input of a hash join on the build side (pinned incte_materialized.slt).Out of scope, possible follow-ups
Tracked under the EPIC #22676:
reset_statefollows its contract (MaterializedCteExec::reset_stateclears its shared buffer in place #26229).LogicalPlanvariants (Make the materialized CTE nodes first-classLogicalPlanvariants #26238).sqlparserparses[NOT] MATERIALIZEDonly inPostgreSqlDialect, so the tests set that dialect (ParseWITH ... AS [NOT] MATERIALIZEDinGenericDialect,DuckDbDialectandSQLiteDialectdatafusion-sqlparser-rs#2622).Are these changes tested?
Yes. The new file
datafusion/sqllogictest/test_files/cte_materialized.sltcovers:MATERIALIZEDand differs per reference whenNOT MATERIALIZEDor with no directive;EXPLAIN, which shows oneDataSourceExecfor three references;EXPLAIN ANALYZE, where the body buffers 4 rows once and each of the three scans outputs 4 rows;IN,NOT INand correlatedEXISTS/NOT EXISTS/INsubqueries that declare a materialized CTE;WITH RECURSIVEwith recursive and non-recursive materialized CTEs;memory_limit: 200k rows, about 184k spilled (part of the in-memory prefix moves to disk to make room for the two readers), and both scans still return 200k rows.The full sqllogictest suite passes.
Are there any user-facing changes?
Yes.
WITH ... AS MATERIALIZEDis now honored when the Postgres dialect is used. Queries without the directive are unchanged. There is no new config option and no public API break. The new nodes are additions.Open questions for reviewers
LogicalPlanvariant, likeRecursiveQuery, or stay as built-inUserDefinedLogicalNodes, as in this PR, which keeps the change smaller?