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: [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]
