Conversation
pentschev
added
the
improvement
pentschev
added
non-breaking
|
Navigate logical layers of code changes, visualize relationships, and explore their blast radius. No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository: NVIDIA/cudf/.coderabbit.yaml Review profile: CHILL Plan: Enterprise Run ID: 📒 Files selected for processing (2)
Included review availability: Your plan provides up to 12 included reviews per hour; 11 remain after this review. 📝 SummarySummary by CodeRabbit
WalkthroughChangesJoin broadcast selection
Priority: ➖ Normal Estimated code review effort: 2 (Simple) | ~10 minutes Change: Bug fix Merge Risk: ⚪ Minimal · up to No actionable merge-blocking risk is identified from the reviewed change. 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
| # An incomplete all-zero prefix is not evidence that the full input is empty. | ||
| # Treat it as unknown rather than choosing it as the broadcast side. This can | ||
| # happen when an ordered scan's early chunks are eliminated by a predicate. | ||
| left_size_known = left_sample.is_complete or left_total > 0 or left_total_rows > 0 | ||
| right_size_known = ( | ||
| right_sample.is_complete or right_total > 0 or right_total_rows > 0 | ||
| ) | ||
| right_size_ok = right_total < broadcast_threshold and ( | ||
| right_total_rows < MAX_ROWS_PER_PARTITION or right_metadata.duplicated | ||
| left_size_ok = ( | ||
| left_size_known | ||
| and left_total < broadcast_threshold | ||
| and (left_total_rows < MAX_ROWS_PER_PARTITION or left_metadata.duplicated) | ||
| ) | ||
| right_size_ok = ( | ||
| right_size_known | ||
| and right_total < broadcast_threshold | ||
| and (right_total_rows < MAX_ROWS_PER_PARTITION or right_metadata.duplicated) | ||
| ) |
There was a problem hiding this comment.
This feels like a sticking plaster. We don't like to consume too much of an input channel because we don't want to buffer unboundedly. But if the messages are empty frames then that isn't actually a memory problem.
Additionally, sending loads of empty messages is going to reduce performance because we have to schedule and execute this work that produces no output.
So can we somehow address the root cause of this problem? From what I understand, if the input table is ordered then the filter that we apply can produce zero rows in the scan and we send that downstream. Because we basically don't repartition at any point we have this long prefix of empty tables before we get to the part of the scanned channel that has any data in it.
It seems like we should never have sent those empty messages on at all.
We currently rely on channels always sending at least one message (even if it is empty) so that the consumer gets the right schema-d table in various places. But we probably shouldn't send loads.
I think there are only a few places where this can happen, so maybe we can do this differently...
There was a problem hiding this comment.
Hmm, I thought a bit harder and I think squeezing empty messages out of a channel is a no-go for now without reworking the way we match up sides of a shuffled join. Today we require that if we're doing a shuffled join that the sequence numbers of the two sides of the input match up (and we pair them in order). If we were to squeeze out empty messages we would need a different contract for matching message where a gap in sequence numbers is indicative of an empty chunk (which means in a shuffle join the other side would not match).
There was a problem hiding this comment.
Another thing we could do though is to remove the "max sample chunks" from the prefix sampler and sample up to some byte count.
2 participants
Description
Dynamic join planning estimates input sizes from an initial sample of chunks. If every sampled chunk contains zero rows, the current logic treats that input as empty and eligible for broadcasting.
An all-zero sample is conclusive only when sampling consumed the complete input. For an incomplete sample, later chunks may contain substantial data, particularly when reading ordered files after predicate pushdown.
This change treats an incomplete all-zero sample as having an unknown size for broadcast eligibility. An estimate remains eligible when:
Complete empty inputs and incomplete nonzero samples retain their existing behavior.
Motivation
This issue was observed with a TPC-H dataset containing many small, ordered Parquet files.
For SF300 Q3, the first 41 of 90 relevant
lineitemchunks contain no rows after applying thel_shipdate > 1995-03-15predicate. Sampling only the initial prefix therefore produced an incomplete zero-size estimate.Before this change, dynamic planning selected the large
lineiteminput as the broadcast side:broadcast_rightWith this change, the incomplete zero-size estimate is not considered a valid broadcast candidate:
broadcast_leftThe focused regression test reproduces the same strategy-selection difference:
BroadcastJoinStrategy(side="right")BroadcastJoinStrategy(side="left")End-to-end results
The isolated comparison used #24209 at commit
737734dcedc85d8bb8cfe26d7eb67dc1bb937648with the following additional settings:PR #24209 without this change
PR #24209 with this change
Tradeoffs
This makes strategy selection more conservative when an incomplete sampled prefix contains no data.
If the remaining unsampled input is nonempty but genuinely small, the planner may decline to broadcast that side and instead broadcast the other side or select a shuffle join. Such workloads could therefore perform more work than before.
The affected sample currently provides no upper bound for the unsampled data, however, so treating it as broadcastable can result in arbitrarily underestimating its size. This change favors a bounded decision over that potentially unsafe assumption.