Repository navigation
Conversation
Add support for materializing Common Table Expressions (CTEs) that are referenced more than once in a query. When a CTE ends in an expensive operation (Aggregate, Distinct, Window, or Union), the CTE is computed once and its results are cached in memory for reuse by multiple consumers. This implements a DuckDB-inspired heuristic: only materialize CTEs that end in expensive operations, avoiding regressions where predicate pushdown through the CTE would be more beneficial. The implementation uses Extension nodes (UserDefinedLogicalNode) to avoid modifying the core LogicalPlan enum, and introduces: - MaterializedCteProducer/Reader logical nodes - MaterializedCteExec/ReaderExec physical operators - MaterializedCtePlanner extension planner - Dependency-ordered execution for nested materialized CTEs Benchmarked on TPC-DS SF1 (10 iterations): - Q47: 2.85x speedup (401ms → 141ms) - Q57: 2.67x speedup (112ms → 42ms) - Q2: 1.58x speedup (101ms → 64ms) - Q74: 1.90x speedup (311ms → 164ms) Relates to: apache#17737
Contributor
|
@nathanb9 Cool! Can you at-me when this is ready for review? |
… and return them from partition_statistics
… filters When a CTE ending in aggregate/distinct/window is referenced multiple times but each reference filters on a different literal value of the same column (e.g., d_moy=4 vs d_moy=5), inlining is better because the optimizer can push each filter through the aggregate, specializing each copy to process only a subset of the data. This fixes the TPC-DS Q39 regression (was 1.78x slower with materialization, now 1.01x — within noise). The detection handles: - Column-qualified filters above joins (inv1.d_moy=4, inv2.d_moy=5) - Simple constant arithmetic expressions (4+1 → 5) - Aliased group-by columns (d_year → syear) Also fixes a clippy warning: pass `statistics` by reference in `replace_materialized_cte_readers`. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The disjoint filter detection previously only looked at Filter nodes. When using JOIN ... ON syntax (vs comma-join with WHERE), the equality conditions like `a.d_moy = 1 AND b.d_moy = 2` live inside the Join node's filter field, not as a separate Filter node. This fixes a 4x regression on queries using JOIN ON with disjoint group-key predicates (e.g. benchmark Q4: inventory comparison across months). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…plan to be optimized in physical plan
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
Contributor
Author
|
run benchmarks tpcds |
|
Hi @nathanb9, thanks for the request (#22551 (comment)). Only whitelisted users can trigger benchmarks. Allowed users: Dandandan, Fokko, Jefffrey, Omega359, adriangb, alamb, asubiotto, brunal, buraksenn, cetra3, codephage2020, coderfender, comphead, erenavsarogullari, etseidl, friendlymatthew, gabotechs, geoffreyclaude, grtlr, haohuaijin, jonathanc-n, kevinjqliu, klion26, kosiew, kumarUjjawal, kunalsinghdadhwal, liamzwbao, mbutrovich, mkleen, mzabaluev, neilconway, rluvaton, sdf-jkl, timsaucer, xudong963, zhuqi-lucas. File an issue against this benchmark runner |
This was referenced May 31, 2026
Contributor
Author
|
@neilconway Okay broke down into PR. |
|
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days. |
Theodus
pushed a commit
to Theodus/datafusion
that referenced
this pull request
Oct 15, 2026
## Which issue does this PR close? - Closes apache#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>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Oct 16, 2026
- Closes apache#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? - Closes apache#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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
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.
Materialize CTEs and duplicate subplans — compute once, reuse many.
How It Works
SQL Planner — Detects multi-referenced CTEs and wraps them in
MaterializedCteProducer/Readernodes. RespectsMATERIALIZED/NOT MATERIALIZEDhints. Skips cheap non-volatile CTEs.Optimizer (
CommonSubplanEliminate) — Detects structurally identical subplans anywhere in the plan tree (not just CTEs) via subtree hashing. Wraps duplicates in the same Producer/Reader nodes. Catches repeated views, self-joins, and generated SQL patterns.Optimizer (
InlineCte) — Runs after filter pushdown. Removes materialization where it's not beneficial:force_materialized = true→ keep (explicit hint or CommonSubplan-detected)ref_count <= 1→ inline (single-ref or dead)base_table_refs > 2 AND refs * count > 10→ keep (expensive)Extension Planner (
MaterializedCtePlanner) — Converts logical nodes to physical. Creates one shared cache per materialized subplan and wires all scans to it.Execution (
MaterializeExec) — First partition triggerscollect_partitionedviaOnceAsync. All partitions await the shared future, then execute the continuation. Scans serve cached batches viaMemoryStream.Optimizer (
CteFilterPusher) (follow-up) — For CTEs that survive inlining, OR-combines filters from all readers and pushes the combined predicate into the CTE body. Community feedback welcome on scoping.Structs
Logical Nodes (
datafusion-expr)Physical Operators (
datafusion-physical-plan)General-purpose — reusable for CTEs, views, and duplicate subplans.
Optimizer Rules (
datafusion-optimizer)Extension Planner (
datafusion-core)Syntax / User Configuration
Session config:
Per-CTE SQL hints:
CommonSubplanEliminate (new)
Detects duplicate subplans via structural hashing of
LogicalPlansubtrees. Any subtree appearing 2+ times with sufficient cost (contains TableScan/Aggregate/Join/Window, >= 3 nodes) is wrapped in Producer/Reader for compute-once semantics.This benefits:
Currently does not help TPC-DS queries with:
Benchmark Results
TPC-DS SF1, 10 iterations, vs main:
All gains are from CTE materialization via the SQL planner.
CommonSubplanEliminatedoes not trigger on TPC-DS because repeated-table patterns there have differing filters/projections (not structurally identical subtrees).