datafusion-contrib / datafusion-contrib/datafusion-distributed

Add forced-spill broadcast NestedLoopJoin regression coverage

Open
#674 0 comments 0 reactions 0 assignees View on GitHub

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:

  • BroadcastExec allocates a fixed set of consumer streams and removes one for each execute(partition) call. An additional call can return Too many consumers for real partition N.
  • NetworkBroadcastExec worker streams are stored as Option<Stream> and taken on first use. An additional call can return WorkerConnections 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

  1. Add a forced-spill integration test with a broadcast build beneath NestedLoopJoinExec.
  2. Constrain the memory pool sufficiently to activate DataFusion memory-limited fallback.
  3. Verify the build child is evaluated once after adopting the DataFusion fix.
  4. Assert correct query results and confirm neither consumer-exhaustion error is produced.
  5. Cover remote and colocated worker execution paths.
  6. 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 partition or already consumed error 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

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.