2010YOUY01 commented on code in PR #24820:
URL: https://github.com/apache/datafusion/pull/24820#discussion_r3964562981
##########
datafusion/physical-plan/src/joins/nested_loop_join.rs:
##########
@@ -1014,8 +1013,16 @@ impl EmbeddedProjection for NestedLoopJoinExec {
/// Left (build-side) data
pub(crate) struct JoinLeftData {
- /// Build-side data collected to single batch
- batch: RecordBatch,
+ /// Build-side data as bounded chunks, in input order. Kept as chunks
rather than one
+ /// `concat_batches` result so buffering never needs input and output to
coexist, and a
+ /// chunk that already arrived at target size is retained without being
copied at all.
+ chunks: Vec<RecordBatch>,
+ /// Row index of the first row of each chunk, i.e. prefix sums over the
chunk lengths.
+ /// The visited-left bitmap is indexed by these global row numbers.
+ row_offsets: Vec<usize>,
+ total_rows: usize,
+ /// Build-side schema, kept so an empty chunk list still knows its shape
+ schema: SchemaRef,
Review Comment:
Can we wrap it in a module so that callers can use it as if it were a single
concatenated batch?
I feel the current implementation has several leaks -- operator logic has to
understand the internal physical representation of the chunked build-side data.
Some of these could be avoided with such a design.
##########
datafusion/physical-plan/src/joins/nested_loop_join.rs:
##########
@@ -3440,6 +3535,272 @@ pub(crate) mod tests {
Arc::new(TestMemoryExec::update_cache(&source))
}
+ /// A build side that already arrives in target-sized batches is retained
as-is: the chunks
Review Comment:
Can we move all tests to end-to-end tests, like `sqllogictest`s, or remove
them?
The issue with those UTs asserting internal properties is that they have a
lot of correct assertions, but it's very hard to figure out what the end goal
of the test is, and maintaining them is very hard. Today's AI likes to
over-generate them.
For those two tests, I can't easily understand their test goals.
##########
datafusion/physical-plan/src/joins/nested_loop_join.rs:
##########
@@ -1071,10 +1114,14 @@ async fn collect_left_input(
with_visited_left_side: bool,
probe_threads_count: usize,
spill_manager: Option<SpillManager>,
+ target_batch_size: usize,
) -> Result<LeftLoad> {
let schema = stream.schema();
let metrics = join_metrics;
- let mut batches: Vec<RecordBatch> = Vec::new();
+ let mut chunks: Vec<RecordBatch> = Vec::new();
+ // Batches at or above half the target size pass through without being
copied.
+ let mut coalescer = BatchCoalescer::new(Arc::clone(&schema),
target_batch_size)
Review Comment:
This `coalescer` is not necessary, there is a hidden convention: each
operator should promise to output batches coalesced to `batch_size`, so here we
can assume input doesn't contain small batches.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
