datafusion-contrib / datafusion-contrib/datafusion-distributed
Add forced-spill broadcast NestedLoopJoin regression coverage
Nobody has claimed this yet.
- Dominant language
- Rust
- Stars
- 139
- Forks
- 67
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 35
Description
Problem
datafusion-distributed inserts network broadcasts beneath eligible NestedLoopJoinExec plans. DataFusion memory-limited NLJ execution currently evaluates the left child once during the initial collection attempt and again when spill fallback starts.
Distributed broadcast streams use exactly-once consumption:
BroadcastExecallocates a fixed set of consumer streams and removes one for eachexecute(partition)call. An additional call can returnToo many consumers for real partition N.NetworkBroadcastExecworker streams are stored asOption<Stream>and taken on first use. An additional call can returnWorkerConnections stream for partition N was already consumed.
Apache DataFusion is tracking removal of the redundant left-side evaluation in apache/datafusion#24661. Once that change is adopted, datafusion-distributed should have regression coverage proving that broadcast NLJs remain compatible with memory-limited execution.
This issue does not propose making the distributed broadcast protocol generally replayable.
Proposed work
- Add a forced-spill integration test with a broadcast build beneath
NestedLoopJoinExec. - Constrain the memory pool sufficiently to activate DataFusion memory-limited fallback.
- Verify the build child is evaluated once after adopting the DataFusion fix.
- Assert correct query results and confirm neither consumer-exhaustion error is produced.
- Cover remote and colocated worker execution paths.
- If releases must support DataFusion versions without the fix, add a selective compatibility guard or configuration option that prevents NLJ broadcast insertion when memory-limited fallback can activate. Do not disable HashJoin or CrossJoin broadcasts.
Suggested test shape
- Use an inner non-equi join that plans as
NestedLoopJoinExec. - Ensure the left/build input is broadcast to multiple consumer tasks.
- Use a counting input or equivalent instrumentation to observe build execution count.
- Configure temporary spill storage and a small bounded memory pool.
- Assert spill occurs and the result matches an unconstrained execution.
- Run through both remote-worker and colocated-worker channels.
The regression should fail against the affected DataFusion behavior and pass after the dependency containing apache/datafusion#24661 is adopted.
Acceptance criteria
- Forced-spill broadcast NLJ coverage exists.
- The broadcast build is evaluated once.
- Results are correct.
- No
Too many consumers for real partitionoralready consumederror occurs. - Remote and colocated paths are covered.
- Existing HashJoin and CrossJoin broadcast tests remain unchanged and passing.
- If older DataFusion versions remain supported, incompatible NLJ broadcast plans are selectively prevented or documented.
References
- DataFusion follow-up: apache/datafusion#24661
- DataFusion independent stream semantics: apache/datafusion#21565
- NLJ and CrossJoin broadcast support: #459 / #484
Contributor guide
No contributing guide indexed for this repository
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start by locating the existing NLJ and CrossJoin broadcast tests referenced by #459 and #484, then inspect how remote and colocated worker paths configure memory pools and spill storage. Add forced-spill coverage with a counting build input for both paths, and compare results with unconstrained execution. Done means one build evaluation, correct results, no consumer-exhaustion errors, and unchanged HashJoin and CrossJoin coverage.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- distributed-systems, testing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 52/100