You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "jayzhan211 (via GitHub)" <gi...@apache.org> on 2026/10/01 13:52:57 UTC
Re: [PR] perf(pwmj):settle never-matching streamed rows against the buffered extreme… [datafusion]
jayzhan211 commented on code in PR #25840:
URL: https://github.com/apache/datafusion/pull/25840#discussion_r4156141966
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs:
##########
@@ -456,126 +494,49 @@ fn resolve_classic_join(
)?;
let mut buffer_idx = batch_process_state.start_buffer_idx;
- let mut stream_idx = batch_process_state.start_stream_idx;
+ let stream_idx = batch_process_state.start_stream_idx;
+ // NULL keys never match and sort first, so the scan starts past the buffered ones.
+ // Streamed NULL keys never get here: `matchable_rows` settled them before the sort.
+ debug_assert_eq!(stream_values[0].null_count(), 0);
if !batch_process_state.processed_null_count {
- let buffered_null_idx = buffered_side.buffered_data.values().null_count();
- let stream_null_idx = stream_values[0].null_count();
- buffer_idx = buffered_null_idx;
- stream_idx = stream_null_idx;
+ buffer_idx = buffered_side.buffered_data.values().null_count();
batch_process_state.processed_null_count = true;
-
- // The scan below starts past the streamed side's NULL-keyed rows, which
- // sit at the front (`nulls_first`). A NULL join key never matches under
- // `NullEqualsNothing`, so for `Right`/`Full` those rows are unmatched and
- // must still be emitted; record them here since the scan will skip them.
- if matches!(join_type, JoinType::Right | JoinType::Full) {
- for row_idx in 0..stream_null_idx as u32 {
- batch_process_state.unmatched_indices.append_value(row_idx);
- }
- }
}
- // Our buffer_idx variable allows us to start probing on the buffered side where we last matched
- // in the previous stream row.
- for row_idx in stream_idx..stream_batch.batch.num_rows() {
- while buffer_idx < buffered_len {
- let compare = cmp.compare(row_idx, buffer_idx);
-
- // If we find a match we append all indices and move to the next stream row index
- match operator {
- Operator::Gt | Operator::Lt => {
- if compare == Ordering::Less {
- batch_process_state.found = true;
- let count = buffered_len - buffer_idx;
-
- let batch = build_matched_indices_and_mark_buffered(
- (buffer_idx, count),
- (row_idx, count),
- buffered_side,
- stream_batch,
- join_type,
- join_schema,
- &mut batch_process_state.batch_min_marked,
- )?;
-
- batch_process_state.output_batches.push_batch(batch)?;
-
- // Flush batch and update pointers if we have a completed batch
- if let Some(batch) =
- batch_process_state.output_batches.next_completed_batch()
- {
- batch_process_state.found = false;
- batch_process_state.start_buffer_idx = buffer_idx;
- batch_process_state.start_stream_idx = row_idx + 1;
- return Ok(batch);
- }
+ let match_on_equal = matches_on_equal(operator)?;
- break;
- }
- }
- Operator::GtEq | Operator::LtEq => {
- if matches!(compare, Ordering::Equal | Ordering::Less) {
- batch_process_state.found = true;
- let count = buffered_len - buffer_idx;
- let batch = build_matched_indices_and_mark_buffered(
- (buffer_idx, count),
- (row_idx, count),
- buffered_side,
- stream_batch,
- join_type,
- join_schema,
- &mut batch_process_state.batch_min_marked,
- )?;
-
- // Flush batch and update pointers if we have a completed batch
- batch_process_state.output_batches.push_batch(batch)?;
- if let Some(batch) =
- batch_process_state.output_batches.next_completed_batch()
- {
- batch_process_state.found = false;
- batch_process_state.start_buffer_idx = buffer_idx;
- batch_process_state.start_stream_idx = row_idx + 1;
- return Ok(batch);
- }
-
- break;
- }
- }
- _ => {
- return internal_err!(
- "PiecewiseMergeJoin should not contain operator, {}",
- operator
- );
- }
- }
-
- // Increment buffer_idx after every row
- buffer_idx += 1;
- }
-
- // If a match was not found for the current stream row index the stream indice is appended
- // to the unmatched indices to be flushed later.
- if matches!(join_type, JoinType::Right | JoinType::Full)
- && !batch_process_state.found
- {
- batch_process_state
- .unmatched_indices
- .append_value(row_idx as u32);
- }
-
- batch_process_state.found = false;
- }
-
- // Flushed all unmatched indices on the streamed side
- if matches!(join_type, JoinType::Right | JoinType::Full) {
- let batch = create_unmatched_batch(
- &mut batch_process_state.unmatched_indices,
+ // `buffer_idx` resumes where the previous stream row's first match was: rows are sorted
+ // so that each one's first match is at or after the previous one's, which bounds the
+ // search below to `[buffer_idx, buffered_len)`.
+ //
+ // Every row here matches: `matchable_rows` kept only rows that match the last buffered
+ // row, and settled the rest (as unmatched for `Right`/`Full`) before the sort.
+ for row_idx in stream_idx..stream_batch.batch.num_rows() {
+ buffer_idx = first_match(buffer_idx, buffered_len, |idx| {
+ is_match(cmp.compare(row_idx, idx), match_on_equal)
+ });
+ debug_assert!(buffer_idx < buffered_len);
Review Comment:
Maybe internal error here?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org