alamb commented on code in PR #9080:
URL: https://github.com/apache/arrow-datafusion/pull/9080#discussion_r1478780340
##########
datafusion/physical-plan/src/joins/sort_merge_join.rs:
##########
@@ -1121,16 +1151,138 @@ impl SMJStream {
.collect::<Vec<_>>()
};
+ let streamed_columns_length = streamed_columns.len();
+ let buffered_columns_length = buffered_columns.len();
+
+ // Prepare the columns we apply join filter on later.
+ // Only for joined rows between streamed and buffered.
+ let filter_columns = if chunk.buffered_batch_idx.is_some() {
+ if matches!(self.join_type, JoinType::Right) {
+ get_filter_column(&self.filter, &buffered_columns,
&streamed_columns)
+ } else {
+ get_filter_column(&self.filter, &streamed_columns,
&buffered_columns)
+ }
+ } else {
+ vec![]
+ };
+
let columns = if matches!(self.join_type, JoinType::Right) {
- buffered_columns.extend(streamed_columns);
+ buffered_columns.extend(streamed_columns.clone());
buffered_columns
} else {
streamed_columns.extend(buffered_columns);
streamed_columns
};
- self.output_record_batches
- .push(RecordBatch::try_new(self.schema.clone(), columns)?);
+ let output_batch =
+ RecordBatch::try_new(self.schema.clone(), columns.clone())?;
+
+ // Apply join filter if any
+ if !filter_columns.is_empty() {
Review Comment:
Ah, I see -- this would be for batches that don't have any matches from
equality predicates anyways - which makes sense
--
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]
