Skip to content

Add streaming shared subplan example - #24780

Draft
nathanb9 wants to merge 8 commits into
apache:mainfrom
nathanb9:streaming-shared-subplan-example
Draft

nathanb9 wants to merge 8 commits into
apache:mainfrom
nathanb9:streaming-shared-subplan-example

Conversation

@nathanb9

@nathanb9 nathanb9 commented Aug 30, 2026 •

Copy link
Copy Markdown
Contributor

Summary

  • Adds an example that streams one shared DataFrame subplan to multiple consumers without materializing it.

motivation

how the example works

  1. Mark the shared logical subplan with an ID.
  2. Plan and rewrite it through DataFusion extension APIs.
  3. StreamingFanoutExec executes the input once and serves consumer 0.
  4. StreamingFanoutReaderExec nodes serve additional consumers through bounded channels.

extension points used

  • UserDefinedLogicalNodeCore
  • QueryPlanner and ExtensionPlanner
  • PhysicalOptimizerRule
  • Custom ExecutionPlan nodes
shared subplan -> StreamingFanoutExec (consumer 0)
                      `-> reader nodes (additional consumers)

@codecov-commenter

codecov-commenter commented Aug 30, 2026 •

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 81.80%. Comparing base (74f4dd6) to head (0d0fe6b).
⚠️ Report is 839 commits behind head on main.

Additional details and impacted files
@@            Coverage Diff             @@
##             main   #24780      +/-   ##
==========================================
+ Coverage   81.53%   81.80%   +0.27%     
==========================================
  Files        1123     1130       +7     
  Lines      406102   417754   +11652     
  Branches   406102   417754   +11652     
==========================================
+ Hits       331111   341748   +10637     
- Misses      55628    55879     +251     
- Partials    19363    20127     +764     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

@alamb

alamb commented Sep 8, 2026

Copy link
Copy Markdown
Contributor

Thanks @nathanb9 -- I will try and give this a look

@github-actions
github-actions Bot marked this pull request as draft October 26, 2026 01:26
@github-actions

Copy link
Copy Markdown

Marked as draft after 14 days without activity: merge conflicts.

Please address these items and mark Ready for review when ready.

alxmrs added a commit to xqlsystems/xarray-sql that referenced this pull request Oct 28, 2026
DataFusion plans a CTE again at every place that reads it, so a model
whose CTEs read each other more than once has a plan that grows
exponentially with its depth (#267). The 1-layer loss took 2.7 s at the
demo's size and 108 s at nanoGPT's CPU size, and the 2-layer loss ran out
of memory while planning.

- The GPT's SQL builders return their CTEs as (name, sql) pairs, and
  every statement starts with the model's and the loss's CTEs, through
  _hosts.with_query.
- Hosts gain query_with(ctes, select) and operators_with. DataFusion's
  runs each CTE on its own into an in-memory table that later CTEs read,
  under TODO(apache/datafusion#26237). It uses collect_partitioned and
  register_record_batches rather than DataFrame.cache(), which einfold
  can't yet plan a join of (xqlsystems/einfold#93).
- The model is no longer the view logits. Creating a DataFusion view
  writes its CTEs out, so the 3-layer view didn't fit in 8 GB; ddx, given
  the CTEs inline, computes each shared one once.

apache/datafusion#24780's streaming fan-out example doesn't fit these
queries: by its own notes it can deadlock when both sides of a hash join
read the shared plan, and doesn't nest. DataFusion 55 parses
`AS MATERIALIZED` only in its PostgreSQL dialect, and ignores it.

With ddxdb 0.4.4, at batch 4, on an Intel i7-8700: the 2-layer loss takes
0.07 s on DataFusion, and a 6-layer SGD step 2.4 s, against 0.30 s and
5.5 s on DuckDB; every step matches JAX.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NyN7idACeDMh94qrnFMu6h
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants