jayzhan211 commented on code in PR #24517:
URL: https://github.com/apache/datafusion/pull/24517#discussion_r3821736654
##########
datafusion/physical-plan/src/joins/sort_merge_join/exec.rs:
##########
@@ -832,14 +927,20 @@ impl SortMergeJoinExec {
})
.collect();
- Ok(Arc::new(Self::try_new(
- left,
- right,
- on,
- filter,
- join_type,
- sort_options,
- null_equality,
- )?))
+ let projection = (!projection.is_empty())
+ .then(|| projection.iter().map(|index| *index as usize).collect());
+
+ Ok(Arc::new(
+ Self::try_new(
+ left,
+ right,
+ on,
+ filter,
+ join_type,
+ sort_options,
+ null_equality,
+ )?
+ .with_projection(projection)?,
Review Comment:
For `swap_inputs`, we need to swap projections like HashJoin as well
##########
datafusion/physical-plan/src/joins/sort_merge_join/exec.rs:
##########
@@ -746,6 +835,11 @@ impl ExecutionPlan for SortMergeJoinExec {
filter,
sort_options,
null_equality: null_equality.into(),
+ projection: projection
+ .iter()
+ .flatten()
+ .map(|index| *index as u32)
+ .collect(),
Review Comment:
We check empty in HashJoin, we probably need this as well
```rs
// Proto3 `repeated` cannot distinguish `None` from
// `Some(vec![])`. `Some(vec![])` (reachable via
// `try_embed_projection` for e.g. `SELECT count(1)
… JOIN …`)
// changes the output schema, so it is encoded with
the
// single-element sentinel `[u32::MAX]` (never a
valid column
// index); every other state is sent as-is. See
// `try_from_proto` for the matching decoder.
projection: match projection.as_ref() {
None => Vec::new(),
Some(v) if v.is_empty() => vec![u32::MAX],
Some(v) => v.iter().map(|x| *x as u32).collect(),
},
```
##########
datafusion/physical-plan/src/joins/sort_merge_join/exec.rs:
##########
@@ -589,7 +659,16 @@ impl ExecutionPlan for SortMergeJoinExec {
spill_manager,
context.runtime_env(),
)
- }
+ }?;
+
+ let Some(projection) = self.projection.clone() else {
+ return Ok(joined);
+ };
Review Comment:
We probably need projection-aware ColumnIndex like Hash Join 🤔 , but we
could add this in another PR
```rs
// update column indices to reflect the projection
let column_indices_after_projection = match self.projection.as_ref()
{
Some(projection) => projection
.iter()
.map(|i| self.column_indices[*i].clone())
.collect(),
None => self.column_indices.clone(),
};
```
--
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]
