Add streaming fanout execution for materialized CTEs by kryonix · Pull Request #23750 · duckdb/duckdb

kryonix · 2026-07-10T06:59:49Z

This PR adds streaming fanout execution for materialized CTEs.

Previously, a materialized CTE was always a full pipeline breaker. DuckDB first wrote the complete CTE result into a ColumnDataCollection; only after that collection was finalized could any CTE consumer start. This PR preserves the logical semantics of a materialized CTE, but allows its physical producer to feed eligible consumer pipelines while the CTE body is still running.

The core addition is a general pipeline-level broadcast exchange. For its first use, a physical materialized CTE can now serve consumers in three ways:

Push producer chunks directly into externally-fed consumer pipelines.
Let consumers scan chunks retained by the exchange.
Materialize the result into the existing ColumnDataCollection and scan it after completion.

A single CTE can combine these strategies. The CTE body is still executed once, every consumer observes the same CTE result, and CTEs with side effects still run to completion. The change is entirely in physical planning and execution; it adds no SQL syntax and does not change the meaning of MATERIALIZED or NOT MATERIALIZED.

Motivation

Full materialization introduces two costs even when all consumers could process the rows immediately:

The producer and consumers cannot overlap.
Every row must be written to and read from intermediate storage.

This is especially relevant for rewrites that deliberately introduce a shared CTE. A grouped distinct aggregate rewrite, for example, can scan an input once and fan it out into multiple independent hash aggregations. Without streaming fanout, the rewrite first materializes the entire input and then scans it once per aggregate. With direct fanout, the input scan drives all aggregate pipelines in parallel without exchange chunk storage.

Streaming can also avoid work. If all relevant consumers stop early, such as through a streaming limit, a side-effect-free CTE producer can stop before exhausting its source.

Physical CTE Modes

The final strategy is pipeline-dependent. Consumer pipelines are classified while pipelines are built, after logical optimization and physical planning. This is intentional: eligibility depends on the complete physical consumer pipeline, not on properties attached to the CTE when it was parsed or created.

This also means the feature applies to materialized CTEs introduced or retained by optimization. A CTE does not need to appear explicitly in the original SQL. Conversely, a CTE that was inlined before physical planning has no physical CTE operator and therefore does not use the exchange.

`PIPELINE_DEPENDENT`

A regular EXPLAIN can display Execution Mode: PIPELINE_DEPENDENT. At that point the physical operator supports exchange execution, but pipeline construction has not yet classified every consumer. The final mode cannot be reported truthfully yet. EXPLAIN ANALYZE builds and executes the pipelines, so its profile reports one of the resolved modes below.

`MATERIALIZED`

No consumer uses the exchange. The producer writes the complete result into the CTE ColumnDataCollection, and consumers scan it after producer completion. This is the fallback when exchange execution is unavailable. It is also selected deliberately when materialization is cheaper than streaming, notably for multi-consumer fanout where every consumer needs the complete result and no consumer pipeline is eligible for direct input.

`STREAMING_FANOUT`

Every CTE consumer is served by the exchange. Consumers can process rows while the CTE body is still producing them. The fanout can be direct, buffered, or a mixture of both. The materialized working table is not populated in this mode.

`HYBRID_FANOUT`

At least one consumer uses the exchange and at least one consumer scans the materialized working table. The producer both feeds exchange consumers and appends its output to the ColumnDataCollection required by materialized consumers. This allows streaming where it is useful without making it a requirement for every consumer. For example, a direct aggregate consumer and an early-stopping consumer can run through the exchange while a consumer with an unsupported pipeline shape uses the normal materialized scan.

Consumer Modes

Each CTE scan is registered with the exchange and resolved to exactly one consumer mode during pipeline construction. The exchange is the single source of truth for these assignments and their counts, including across prepared statement pipeline rebuilds.

`DIRECT`

The consumer's source pipeline is converted into an externally-fed pipeline. The CTE producer creates a PipelineExecutor for that pipeline and calls PushExternal for every producer chunk. Finalization is driven through FinishExternal after producer input ends.

Direct fanout avoids copying and intermediate chunk storage. It is only selected when the complete consumer pipeline supports external input:

The pipeline has a parallel, order-independent sink.
The sink explicitly supports external input.
The sink does not require partitioned input.
Every intermediate operator explicitly supports external input.
Every intermediate operator is parallel and does not require fixed order.

These checks are properties of the physical pipeline. There are no query-name, CTE-name, or optimizer-rewrite-specific exceptions.

`BUFFERED`

The CTE scan remains a source, but reads from the exchange instead of waiting for the materialized table. The producer copies chunks into exchange-owned storage and wakes blocked readers. Chunks are retired after every active buffered consumer has advanced past them.

A single exchange consumer uses buffered execution when direct input is unavailable. With multiple consumers, buffered execution is kept only for a pipeline that may stop before source exhaustion. Multi-consumer pipelines that require all rows use materialization unless they qualify for direct input. This avoids paying broadcast-buffering overhead when materialized scans are the simpler and faster execution strategy.

`MATERIALIZED`

The consumer does not participate in exchange flow control. It scans the existing CTE ColumnDataCollection after the producer has completed. In hybrid mode, these consumers coexist with direct and buffered exchange consumers.

Fanout Fields In `EXPLAIN ANALYZE`

For resolved streaming modes, the CTE operator reports:

Fanout Mode: DIRECT: every exchange consumer is direct.
Fanout Mode: BUFFERED: every exchange consumer reads exchange storage.
Fanout Mode: MIXED: both direct and buffered exchange consumers exist.
Consumers: the number of direct plus buffered consumers.
Direct Consumers: externally-fed pipelines.
Buffered Consumers: exchange scan sources.
Materialized Consumers: consumers scanning the CTE working table.
Chunk Storage: NONE: no buffered consumer exists, so producer chunks are not copied by the exchange.
Chunk Storage: POOLED/COLUMN_DATA: buffered consumers start with pooled in-memory chunks and can transition to shared ColumnDataCollection spool storage.
Run To Completion: true: producer execution cannot stop when exchange consumers detach or finish early.

Example:

Execution Mode: HYBRID_FANOUT
Fanout Mode: DIRECT
Consumers: 1
Direct Consumers: 1
Buffered Consumers: 0
Chunk Storage: NONE
Materialized Consumers: 1

Individual CTE scan profiles report their resolved CTE Mode and consumer index.

Broadcast Exchange

PipelineBroadcastExchange coordinates registration, producer lifecycle, consumer lifecycle, backpressure, cancellation, progress, and transition logging. Direct pipeline execution is owned by the per-thread exchange local state. Buffered storage and its positional bookkeeping are isolated in BufferState. The exchange remains the owner of the existing annotated mutex, so the separation does not introduce another lock or a new lock ordering.

Producer Path

For each input chunk, a producer local state can perform the following work:

Resume or push the chunk through every direct consumer executor.
Reserve an append for buffered consumers.
Copy the chunk outside the exchange lock.
Append it to pooled memory or the shared spool.
Publish the completed append and wake readers.
Append it to the materialized working table when materialized consumers exist.

Only one buffered append reservation can be published at a time. Chunk copying and spool I/O do not hold the exchange mutex. Admission is checked again after copying so cancellation, consumer retirement, and backpressure changes are observed before publication.

In-Memory Buffering And Backpressure

Buffered positions are monotonically increasing chunk indexes. Each buffered consumer tracks its next position. A pooled chunk can be retired once no active consumer or in-progress reader can still reference it. For a single buffered consumer, the exchange uses chunk-count watermarks:

At 32 buffered chunks, the producer blocks.
Once retirement reaches 16 buffered chunks, blocked producers are woken.

Chunk counts are intentionally used instead of byte estimates. This keeps admission and retirement deterministic and avoids maintaining a second approximate memory-accounting model beside DuckDB's buffer management.

Shared Spool

When broadcast fanout would otherwise keep many pooled chunks alive, the exchange promotes retained chunks into one shared ColumnDataCollection-backed spool. This applies when there are multiple exchange consumers, or direct and buffered consumers coexist, and the pooled buffer reaches its high watermark.

The spool provides a shared positional stream. Every buffered consumer has its own reader state, while stored data is shared. Spool segments are retired according to the slowest active or in-progress reader. The spool uses DuckDB's existing buffer management and can spill under memory pressure.

Blocking And Wakeups

The exchange has distinct wait lists for readers, watermark-blocked writers, and producers waiting for the single append reservation. Callbacks are always collected while holding the exchange lock and invoked after releasing it.

Important transitions are logged, rather than every chunk operation: spool creation, high-watermark blocking, low-watermark wakeup, and consumer unregistration.

Early Completion And Cancellation

When an exchange consumer finishes or stops reading, it unregisters. Retained chunks are retired against the remaining consumers. If no exchange consumer remains, a side-effect-free producer can return FINISHED early and release buffered storage.

CTE bodies with side effects use RUN_TO_COMPLETION. Their exchange consumers may finish early, but the producer continues so all DML side effects occur. Cancellation changes the producer state, deactivates consumers, releases storage when no in-progress read or append still owns it, and wakes every class of blocked task.

Pipeline Dependencies

The exchange introduces dataflow dependencies rather than completion dependencies for streaming consumers:

Buffered readers can run while the producer is active and block when they catch up.
Direct consumers are driven by the producer and have an external-finish dependency so their events complete after external input is finalized.
Materialized consumers retain the normal dependency on CTE producer completion.

Prepared plans can rebuild their pipelines. Before rebuilding, the exchange ends the previous buffer execution, clears direct pipeline registrations and blocked task state, and resets every consumer to UNRESOLVED. Pipeline construction then assigns all consumers again. Runtime reset preserves the resolved assignments while resetting positions, lifecycle, progress, and storage.

Correctness And Safety

The implementation preserves the following invariants:

A materialized CTE body is executed once.
Every active buffered consumer observes every published chunk exactly once.
Direct consumers receive each producer chunk exactly once, including across blocked/resumed pushes.
Materialized consumers observe the complete working table.
DML CTE bodies complete even when consumers stop early.
Pooled chunks and spool segments are not retired while an active or in-progress reader can reference them.
Append publication is serialized, while copying and spool I/O occur outside the exchange lock.
Blocked tasks are woken on data publication, retirement, producer completion, cancellation, and pipeline reset.
Consumer selection has one owner and is rebuilt rather than inferred from stale physical state.

The synchronization paths use annotated_mutex and explicit state enums for producer state, consumer lifecycle, read state, append reservation state, watermarks, and external-input event state. This makes illegal or unresolved transitions assertable and keeps TSAN-visible ownership explicit.

Supporting Changes

Validation of streaming CTE execution exposed runtime-filter cast behavior that was previously hidden by full materialization and different pipeline timing. This PR therefore also makes join runtime-filter reconstruction preserve cast failure semantics. Reconstructed narrowing cast chains use TRY_CAST only for the optional runtime filter, while the original join expression remains responsible for query-visible conversion errors. Runtime-filter pushdown remains general and is not keyed to CTEs or query names.

The PR also includes focused fixes and coverage for external joins with many duplicate keys, externally-fed pipeline event lifetime, prepared CTE pipeline rebuilds, correlated CTE output ordering, and extension-created or optimizer-created CTE plans.

lnkuiper · 2026-07-10T08:48:51Z

I saw this in CI:

benchmark/tpch/sf10/q07.benchmark
Old timing: 0.0443185
New timing: 0.10119249999999999

But I built this branch locally and could not reproduce

kryonix · 2026-07-14T13:21:50Z

The q07 regression was real, but is resolved now.

lnkuiper · 2026-07-14T13:28:17Z

Still some differences (improvements and regressions) but overall geomean is only 1% slower so that's OK from me.

benchmark/tpch/sf10/q03.benchmark
Old timing: 0.098633
New timing: 0.055284

benchmark/tpch/sf10/q07.benchmark
Old timing: 0.042372
New timing: 0.0524735

kryonix · 2026-07-14T14:05:01Z

That's a much bigger diff. than I measured locally. I will rebase and check q07 again.

kryonix · 2026-07-15T15:13:25Z

Removing the CTE statistics again (for now?) was the right call:

TPC-DS:

Old timing geometric mean: 0.01782409098689881
New timing geometric mean: 0.016357362451187598, roughly 8% faster

This did not change TPC-H Q07, there's something else going on.

lnkuiper

Some minor comments below, looking nice now

lnkuiper · 2026-07-17T07:20:01Z

+		return true;
+	}
+	return source_size >= target_size;
+}


This would be so much cleaner if casts were scalar functions that have SetFallible(true). That's a separate issue, just another reason for the rework!

lnkuiper · 2026-07-17T07:23:29Z

+		HIGH_WATERMARK_BLOCKED,
+		LOW_WATERMARK_WAKE,
+		CONSUMER_UNREGISTERED
+	};


The state machine cleanup really nice with all the enums now. I do worry that this class is a bit of a "god mode" class that handles everything (external execution, in-memory buffering, spilled buffering, backpressure, cancellation, registration, progress, and logging). Maybe it would be better to separate concerns into classes so it's easier to reason about (chunks, shared_spool, positions, retirement). What do you think?

Yeah, it's doing a lot. I will try a split.

…ed CTE consumers

Remove a bunch of dead code etc

Mytherin · 2026-07-20T14:55:35Z

Thanks!

…#24031) After duckdb#22211 was merged, I went back through recursive CTE execution to audit the assumptions behind retained state, epoch scheduling, recursive sinks, and `USING KEY`. Retaining and resetting the physical recursive plan remains the right foundation, but I found several places where state lifetime was too optimistic or where each epoch still did substantially more work than necessary. In short, this PR: - Makes retained invariant builds safe for volatile and side-effecting operators. - Reuses recursion-independent, physically materialized CTE producers across epochs. - Preserves early termination through streaming `LIMIT`. - Reuses cached recursive projections of the generic pipeline schedule and selects workers from the current epoch's work. - Parallelizes broad `UNION ALL` and `UNION DISTINCT` sinks. - Adds fixed-point column pruning for `UNION ALL` recursive CTEs. - Freezes `USING KEY` state at epoch boundaries and reads it without rematerializing recurring state. - Adds `PhysicalRecursiveCTEKeyJoin` for complete keys and secondary indexes for partial composite keys. - Makes concurrent fallback scans use task-local hash-table probes and synchronized aggregate finalization. - Adds conservative row-equality normalization, recurring-state join ordering, structured runtime metrics, and deterministic benchmarks. - Adds generated Flummi programs covering control-flow-heavy graph, cellular-automaton, clustering, and ray-tracing recursion. - Adds 16 adapted AoC 2022 workloads plus keyed AoC 2024 shortest-path and flood variants. - Runs the recursive research, Flummi, AoC22, and AoC24 suites in CI's performance-regression workflow. ## Recursive execution The most important correctness issue was invariant-build classification. A build is safe to retain only when it is independent of recursive state and repeatable across epochs. The previous check established independence but not repeatability, so a volatile build containing `nextval()` could produce different results when caching was enabled. A second unsafe assumption treated every unfamiliar logical operator as repeatable; this could freeze one unseeded Bernoulli sample across all recursive epochs. There is now one conservative repeatability classifier shared by physical planning and recursive column pruning. It explicitly classifies the current logical operators, checks scalar, aggregate, and window function stability, recognizes both ordinary and pushed-down sampling, and defaults unknown operators and opaque table functions to rebuilding. Deterministic table functions opt in through a bind-data-aware capability. Seeded samples remain eligible, while unseeded samples, DML, side effects, dynamic filters, and unknown extensions do not. The same dependency analysis now retains recursion-independent `PhysicalCTE` producers when their selected execution mode is fully materialized. The producer collection survives after the first epoch while each CTE scan is reset normally, so multiple consumers can reread it without rebuilding its source. Streaming and hybrid fan-out modes remain ineligible. This is entirely physical and adds no logical or serialization field. Caching operators could also defeat partial demand: below a streaming `LIMIT`, a selective pipeline could withhold output until it had a complete vector, preventing early termination. Caching now respects the existing physical `CanStopSourceEarly()` property. This fixes duckdb#20838 without a recursive-specific setting or coupling to the materialized-CTE exchange from duckdb#23750; the reduced case improved from 0.36 to 0.01 seconds. Inline and parallel recursion now consume projections of the generic pipeline schedule, while executors and lookup structures are reused on the warm path. These immutable projections are built once during physical pipeline construction, including the variant that omits retained invariant builds, rather than once per recursive state. This matters for nested queries such as AoC 2024 day 15, which creates 1,405 short-lived recursive states. Worker selection uses the current frontier, physical work units, recursive-reference count, configured threads, and the number of execute pipelines it would schedule. Materialized CTE scans count as fan-out work rather than fixed overhead; ordinary `UNION ALL` recursion stays inline unless each worker can amortize the remaining pipelines. Broad `UNION ALL` epochs concatenate worker-local output at the boundary; small epochs retain locked append because private collections cost more there. `UNION DISTINCT` promotes broad parallel epochs to four persistent hash partitions, reducing the representative 1M-row workload from 0.88 to 0.68 seconds. The CI regression on AoC day 25 exposed a related fixed-cost problem in narrow epochs. The runtime now avoids parallel-only work-unit discovery, scheduler lookups, worker-local resets, metrics branches, and settings lookups when the frontier already proves that execution is serial. While reducing that overhead, I found a more serious allocator-lifetime bug: after a worker-local collection was combined into the recursive output, its retained segment still owned the worker allocator, but `ResetForReuse()` reset only the destination collection's allocator. Repeated narrow epochs therefore accumulated blocks, spilled, and could fail with spilling disabled. `ResetForReuse()` now resets the retained segment's allocator as well. A 500,000-epoch reproducer that previously exhausted a 64 MB limit now completes without spilling. Recursive `UNION ALL` column liveness is computed to a fixed point without copying or serializing the recursive subplan during analysis. This removes unused recursive expressions from both the anchor and recursive member, including the wide payload in the public reproducer. `UNION DISTINCT` retains every identity column, and `USING KEY` remains unpruned until key/update liveness can be proven. Fixes duckdb#24109. The AoC protection cases confirmed that the final scheduler does not materially regress ordinary recursive CTEs. In the final two-thread regression run, all 25 workloads passed and the geometric mean was effectively flat at 0.064 seconds on both builds (`+0.3%`). Day 15 improved from 0.251 to 0.244 seconds after caching the schedule projections, and day 25 remained effectively flat at 0.074 versus 0.073 seconds rather than reproducing the 20% regression from an earlier CI run. ## Frozen `USING KEY` state and indexed probes The largest remaining `USING KEY` cost was access to `recurring.T`. The sink previously updated keyed state during an epoch and copied the complete aggregate hash table into a recurring collection for later reads. It now buffers candidates and commits them during prepare-finalize, after all execute stages have completed. The recursive term reads that frozen epoch snapshot directly from the aggregate hash table, and the source explicitly drains the final keyed state once recursion finishes. For complete-key equality joins, I added read-only, missing-key-safe aggregate lookup APIs plus `PhysicalRecursiveCTEKeyJoin`. Planning selects it only for an inner join against a direct recurring-state scan with direct, exactly typed scalar key equalities. It handles either orientation, reordered composite keys, projections, duplicates, NULLs, and missing groups without mutating state. Proper subsets of composite keys use an indexed lookup rather than a filtered scan. The state builds one epoch-stable secondary hash index per normalized subset, storing aggregate row addresses and collision links and checking full typed equality during probing. The row collection uses `KEEP_EVERYTHING_PINNED`, outlives the indexes, and is frozen during probes, so hash-directory growth cannot invalidate addresses. Only new complete-key groups extend an index; payload updates require no maintenance. The sparse partial-key workload improved from 0.788 to 0.059 seconds median, or 13.3x. Eligibility is deliberately strict. Residual predicates, wrapped scans, nested or mismatched keys, and non-inner joins use the ordinary hash-join/full-scan path. Direct LEFT, SEMI, and ANTI prototypes were slower on dense access, so I did not add a workload-specific threshold. The operator cannot replace ordinary recursive references: those read the current frontier, whereas accumulated state contains older rows and does not exist for `UNION ALL`. Substitution would change results and potentially termination. The final hardening binds lookup scratch to its originating hash table and replaces assertion-only executor-cache and schedule failures with checked internal errors. Fallback scans needed a separate concurrency fix. Each scan had its own cursor lock, but all of them called the aggregate hash table's mutable scan/fetch path and finalized aggregate payloads from the same state. Two recurring references with an `approx_quantile` payload could therefore crash at eight threads. Fallback scans now scan only group columns, resolve row addresses through task-local read-only probe state, and share the CTE-wide aggregate-finalization lock with direct key joins. Append and lookup paths also compose the same owner-checked probe scratch layout. ## Runtime evidence and benchmarks Runtime metrics use DuckDB's existing `PhysicalOperator` structured log type. Query state checks once whether logging is enabled; the disabled path avoids timers, byte scans, and counter updates. The metrics showed, for example, that the sparse complete-key case replaces roughly 20M full-state scan rows with 20,000 direct probes. The metrics state is now encapsulated and also reports retained materialized-CTE production and reuse. The existing benchmark definitions are used directly by a dedicated CI regression job. It runs the 57-query recursive suite, covering the research, Flummi, asql-ws2526, and AoC 2022 workloads, plus all 25 AoC 2024 queries against the merge-base and candidate runners with two threads, seven timed runs, answer validation, and the standard confirmed-regression logic with a 10 ms absolute allowance. I also added a 50,000-epoch, one-row-frontier `UNION ALL` benchmark to protect fixed recursive runtime and allocator-reuse costs. In the final regression run it was flat at 0.110 seconds on `main` and 0.108 seconds on the branch. Across all 57 recursive benchmarks, the geometric mean improved from 0.093 to 0.064 seconds, or 31.3%; every individual workload passed the regression gate. I added a deterministic benchmark suite based on the queries accompanying [How DuckDB is `USING KEY` to Unlock Recursive Query Performance](https://db.cs.uni-tuebingen.de/publications/2025/using-key/). It covers eight algorithms plus isolated key probes. The Game of Life variant uses an equivalent equality-join formulation rather than another specialized operator. Two workloads in the research suite exposed general optimizer opportunities. Connected components exposed row equality that hid scalar key predicates. It is now decomposed only at a filter root for non-empty, matching scalar column references, preserving NULL semantics elsewhere. BFS exposed a separate join-order problem: the full edge relation could reach recurring state before frontier restriction. LEFT, SEMI, and ANTI joins whose right subtree contains recurring state are now kept above candidate generation. I measured the merge base against the PR. Both were Release builds with the same linked extensions on the same macOS arm64 machine. I ran every workload by its individual benchmark name with two threads, one runner warmup, 15 timed runs, and answer validation. This is the same protocol used for the Flummi table below and matches the recursive CI regression job. p95 uses the inclusive nearest-rank percentile. | Workload | `main` median / p95 | PR median / p95 | Median speedup | | --- | ---: | ---: | ---: | | DVR | 0.6512 / 0.6944 s | 0.5302 / 0.5661 s | 1.23x | | PageRank | 0.6544 / 0.6864 s | 0.3332 / 0.3628 s | 1.96x | | k-means | 0.0944 / 0.0953 s | 0.0805 / 0.0820 s | 1.17x | | Sparse complete-key probe | 0.3185 / 0.3696 s | 0.0477 / 0.0778 s | 6.67x | | Dense complete-key probe | 0.2588 / 0.2958 s | 0.1659 / 0.1703 s | 1.56x | | Sparse partial-key probe | 0.7877 / 0.9389 s | 0.0591 / 0.0609 s | 13.33x | | BFS | 1.1782 / 1.2359 s | 0.0327 / 0.0616 s | 36.04x | | Connected components | 4.5046 / 4.5654 s | 1.1273 / 1.1931 s | 4.00x | The algorithms do not naturally probe a proper subset of a composite `USING KEY`; the partial-key cases are intentionally isolated operator benchmarks. The metrics also explain why I stopped adding specializations: DVR performs far more probes than state scans, PageRank already benefits from retained invariant joins, and k-means consumes complete state for grouping. I also added seven answer-validated workloads generated by [Flummi](https://github.com/DBatUTuebingen/flummi). They exercise a very different shape: an imperative program counter, control-flow label, emitted values, and live variables all travel through one recursive CTE. The 384 by 240 ray trace retains shadows and reflections but is small enough for the Release regression job. Flummi's Generations program exposed two successive worker-classifier mistakes. Its 52 small epochs initially launched too many tasks for its pipeline fan-out, while the post-duckdb#23750 classifier then counted materialized CTE scan pipelines both as useful fan-out work and as fixed scheduling overhead. A matched profile showed that this made the branch stay inline at two threads even though two workers were beneficial. CTE scan pipelines now contribute work without also inflating fixed overhead. In the 30-run confirmation, two-thread execution moved from 0.1494 to 0.1528 seconds (`+2.3%`, `+3.4 ms`), while the four-thread protection still avoids `main`'s over-parallelization and moves from 0.2974 to 0.1692 seconds. Community detection exposed a separate planning regression in the first column-pruning implementation. Every fixed-point step copied the complete generated recursive subplan, and `LogicalOperator::Copy()` performs that copy through serialization. I replaced it with a non-mutating dependency analysis over the original plan. In the confirming profile, recursive dependency analysis fell from 796 to 7 samples; the one remaining `LogicalOperator::Copy()` sample came from unrelated CTE inlining. The Flummi comparison uses the same matched builds and two-thread protocol. | Flummi workload | `main` median / p95 | PR median / p95 | Median speedup | | --- | ---: | ---: | ---: | | Community detection | 0.0342 / 0.0362 s | 0.0330 / 0.0361 s | 1.03x | | Connected components | 0.0369 / 0.0374 s | 0.0363 / 0.0584 s | 1.02x | | Generations | 0.1494 / 0.1548 s | 0.1528 / 0.1600 s | 0.98x | | k-core `USING KEY` | 0.4864 / 0.4901 s | 0.0016 / 0.0020 s | 305.91x | | k-means | 0.1666 / 0.1690 s | 0.1670 / 0.1703 s | 1.00x | | Game of Life | 0.0751 / 0.0772 s | 0.0746 / 0.0764 s | 1.01x | | Ray tracing | 0.8507 / 0.8803 s | 0.8738 / 0.9176 s | 0.97x | At two threads the seven-workload geometric mean improves by approximately 55.9%. This aggregate is dominated by keyed k-core. The Generations row uses the focused 30-run confirmation after the classifier fix; its remaining 2.3% delta and ray tracing's 2.7% delta both remain below the CI regression gate. I also adapted 16 recursive SQL solutions from [DBatUTuebingen/Advent_of_Code](https://github.com/DBatUTuebingen/Advent_of_Code) at `10ead5d33412cb6484bf8b9bc4ed91f454c1fd88`. I did not copy the Advent of Code inputs; the load scripts generate deterministic data that retains the recursive plan shapes. The second group adds nested-list stack and sand state, hierarchical ascent, bit-packed falling rocks, expression-tree self-joins, alternating map branches, moving hazards, and balanced-base conversion. Four queries have answer-equivalent `USING KEY` alternatives. The reverse BFS and directed-graph traversal benefit because their keys remove redundant historical states. The permutation and moving-frontier variants overwrite almost every row in every iteration, so they deliberately remain negative controls. I measured current local main against this branch with the same matched Release builds, two threads, one runner warmup, 15 timed runs, and answer validation used for the Flummi table. | AoC 2022 workload | `main` median / p95 | PR median / p95 | Median speedup | | --- | ---: | ---: | ---: | | Day 9: chained rope recursion | 0.3207 / 0.3279 s | 0.3178 / 0.3217 s | 1.01x | | Day 11: state machine | 0.6572 / 0.6679 s | 0.6518 / 0.6631 s | 1.01x | | Day 12: reverse BFS | 0.2961 / 0.2993 s | 0.2936 / 0.3021 s | 1.01x | | Day 12: `USING KEY` | 0.0283 / 0.0880 s | 0.0270 / 0.1093 s | 1.05x | | Day 16: directed traversal and search | 0.6568 / 0.6842 s | 0.6296 / 0.6606 s | 1.04x | | Day 16: `USING KEY` | 0.0241 / 0.0247 s | 0.0247 / 0.0252 s | 0.98x | | Day 18: `UNION` flood fill | 1.3416 / 1.3719 s | 1.3666 / 1.3833 s | 0.98x | | Day 19: materialized branching search | 0.3763 / 0.4070 s | 0.3776 / 0.4069 s | 1.00x | | Day 20: recursive permutation | 0.1477 / 0.1512 s | 0.1482 / 0.1530 s | 1.00x | | Day 20: `USING KEY` | 0.1493 / 0.1518 s | 0.1503 / 0.1541 s | 0.99x | | Day 23: correlated frontier | 0.3335 / 0.3351 s | 0.3316 / 0.3334 s | 1.01x | | Day 23: `USING KEY` | 0.3365 / 0.3374 s | 0.3367 / 0.3402 s | 1.00x | The more useful keyed comparison is within the PR: Day 12 moves from 0.2936 to 0.0270 seconds (10.86x), and Day 16 from 0.6296 to 0.0247 seconds (25.54x). Days 20 and 23 are 1.4% and 1.5% slower, respectively. Day 12 has a visibly multimodal two-thread distribution in both builds, with occasional roughly 30 ms scheduler stalls, so its median and p95 are both reported. Across the original 12 variants, the branch geometric-mean runtime improves by 0.8% with one thread, 0.6% with two threads, and 9.7% with four threads. The four-thread aggregate is strongly influenced by Day 12's scheduler stalls, so I do not use it as a general throughput claim. Day 19 originally exposed a real 56% two-thread regression in the first worker classifier: it treated all six scans of a materialized recursive state as scheduler overhead and ran every epoch inline. Counting those scans as work restored parallelism. Keeping one private recursive-output collection per worker and per scan then added another fan-out-specific combine cost; using the synchronized global output for this shape made the final one-, two-, and four-thread comparisons flat. I also reformulated two recursive AoC 2024 searches with semantic keys. Day 16 retains the cheapest `(x, y, direction)` state, while Day 20 retains the first distance for each `(x, y)`. In the current two-thread regression run, Day 16 is approximately 2.9x faster with the key and the synthetic Day 20 race-track pair approximately 33x faster. Both produce the same answers and distance aggregates. The keyed terms use anti/improvement joins against recurring state, so they do not use the inner-join-only `PhysicalRecursiveCTEKeyJoin`. All new AoC 2022 and AoC 2024 entries pass the matched-main regression gate. The recursive state and key-join machinery is physical-only. The optimizer changes reuse existing logical operators.

github-actions Bot added the Member label Jul 10, 2026

kryonix requested review from Mytherin and lnkuiper July 10, 2026 08:12

kryonix force-pushed the streaming_ctes branch from 6609468 to 79a02c7 Compare July 14, 2026 14:06

kryonix added the Ready For Review label Jul 16, 2026

lnkuiper reviewed Jul 17, 2026

View reviewed changes

kryonix and others added 19 commits July 20, 2026 10:51

First step towards CTE fan out execution

53854d2

Fix more issues with streaming CTE fanout

e2b6ab0

Fix streaming CTE bugs

49997ed

Migrate external input to pipeline level

4ed94fd

Move pipeline exchange from CTE to pipeline level

0207f75

Optimize pipeline broadcast exchange more

ccfcc82

Add ColumnDataCollection-backed spools for shared and detached buffer…

70aec70

…ed CTE consumers

Cleanup pipeline broadcast

c0d8d9a

Remove a bunch of dead code etc

Fix runtime filters through compressed casts

05357d7

Simplify pipeline broadcast exchange

214ae92

Materialize non-streaming CTE fanout

5e3184f

Run generate files

17b49a9

Document CTE fanout selection

0113a63

Use try casts for compressed runtime filters

e38e505

Reset CTE exchange consumers on rebuild

6aa1571

Fix pipeline exchange TSAN races

0ea8e16

Fix pipeline source state lock ordering

41eb589

Fix materialized CTE test

e82fad0

Stabilize correlated CTE subquery test

b46df82

kryonix added 16 commits July 20, 2026 10:51

Simplify pipeline exchange reset state

daf01bd

Use chunk counts for pipeline exchange pressure

fc32824

Fix external input event parameter passing

c848368

Simplify runtime filter cast reconstruction

04646e3

Simplify pipeline broadcast exchange state

3374bfe

Fix pipeline exchange tidy checks

6763af8

Fix runtime filter cast regressions

9bc5f65

Fix CTE pipeline build regressions

b652c5c

Fix runtime filter cast reconstruction

446c9f9

Fix Windows pipeline event enum names

5ecd75a

Fix runtime filter performance regression

2cc9d62

Make format-fix

8c6acc2

Harden pipeline broadcast exchange

b22d767

Fix runtime filter cast chain reconstruction

15fe4e7

Remove CTE statistics propagation

f346513

Simplify pipeline broadcast exchange ownership

e541455

kryonix force-pushed the streaming_ctes branch from b54b425 to e541455 Compare July 20, 2026 08:56

Mytherin added this pull request to the merge queue Jul 20, 2026

Merged via the queue into duckdb:main with commit b982635 Jul 20, 2026
56 checks passed

kryonix mentioned this pull request Jul 21, 2026

Harden recursive CTE execution and optimize USING KEY joins #24031

Merged

kryonix mentioned this pull request Aug 8, 2026

Propagate stats in materialized CTEs and array distance/similarity functions #24600

Merged

This was referenced Oct 16, 2026

feat: release the output of a materialized CTE when the last scan finishes apache/datafusion#26272

Closed

feat: stream the output of a materialized CTE to its scans apache/datafusion#26456

Draft

Navigation Menu

Repository navigation

Add streaming fanout execution for materialized CTEs - #23750

Add streaming fanout execution for materialized CTEs#23750
Mytherin merged 35 commits into
duckdb:mainfrom
kryonix:streaming_ctes

kryonix commented Jul 10, 2026 •

edited

Loading

Uh oh!

lnkuiper commented Jul 10, 2026

Uh oh!

kryonix commented Jul 14, 2026

Uh oh!

lnkuiper commented Jul 14, 2026 •

edited

Loading

Uh oh!

kryonix commented Jul 14, 2026

Uh oh!

kryonix commented Jul 15, 2026

Uh oh!

lnkuiper left a comment

Uh oh!

Uh oh!

Uh oh!

lnkuiper Jul 17, 2026

Uh oh!

lnkuiper Jul 17, 2026

Uh oh!

kryonix Jul 17, 2026

Uh oh!

Mytherin commented Jul 20, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

kryonix commented Jul 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Physical CTE Modes

PIPELINE_DEPENDENT

MATERIALIZED

STREAMING_FANOUT

HYBRID_FANOUT

Consumer Modes

DIRECT

BUFFERED

MATERIALIZED

Fanout Fields In EXPLAIN ANALYZE

Broadcast Exchange

Producer Path

In-Memory Buffering And Backpressure

Shared Spool

Blocking And Wakeups

Early Completion And Cancellation

Pipeline Dependencies

Correctness And Safety

Supporting Changes

Uh oh!

lnkuiper commented Jul 10, 2026

Uh oh!

kryonix commented Jul 14, 2026

Uh oh!

lnkuiper commented Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kryonix commented Jul 14, 2026

Uh oh!

kryonix commented Jul 15, 2026

Uh oh!

lnkuiper left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

lnkuiper Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

lnkuiper Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

kryonix Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

Mytherin commented Jul 20, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

kryonix commented Jul 10, 2026 •

edited

Loading

`PIPELINE_DEPENDENT`

`MATERIALIZED`

`STREAMING_FANOUT`

`HYBRID_FANOUT`

`DIRECT`

`BUFFERED`

`MATERIALIZED`

Fanout Fields In `EXPLAIN ANALYZE`

lnkuiper commented Jul 14, 2026 •

edited

Loading