jayzhan211 commented on code in PR #25255:
URL: https://github.com/apache/datafusion/pull/25255#discussion_r4005697490
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -2512,6 +2540,74 @@ mod proto_tests {
}
}
+/// Output column index of a join, mapped to the equivalent join-key expression
+/// on the other side of the join (in that side's input schema).
+type KeyTransferMap = HashMap<usize, PhysicalExprRef>;
+
+fn is_column_at(expr: &PhysicalExprRef, index: usize) -> bool {
+ expr.downcast_ref::<Column>()
+ .is_some_and(|column| column.index() == index)
+}
+
+/// Marks every parent filter whose columns are all join keys in `key_map` as
+/// supported for `child`, rewritten over the other side's key expressions.
+///
+/// Filters `child` already accepts directly are left alone, as are filters
+/// that reference a non-key column or no column at all: the former cannot be
+/// expressed on the other side, the latter were already routed by the plain
+/// column analysis.
+fn transfer_key_filters(
+ parent_filters: &[Arc<dyn PhysicalExpr>],
+ key_map: &KeyTransferMap,
+ child: &mut ChildFilterDescription,
+) -> Result<()> {
+ if key_map.is_empty() {
+ return Ok(());
+ }
+ for (filter, pushed) in
parent_filters.iter().zip(child.parent_filters.iter_mut()) {
+ if matches!(pushed.discriminant, PushedDown::Yes) {
+ continue;
+ }
+ if let Some(transferred) = transfer_filter_across_keys(filter,
key_map)? {
+ *pushed = PushedDownPredicate::supported(transferred);
+ }
+ }
+ Ok(())
+}
+
+/// Rewrites `filter` over the other side's join keys, or returns `None` when
+/// it references a column that is not a transferable key, or no column.
+///
+/// A [`DynamicFilterPhysicalExpr`] comes out as a view sharing the original's
+/// state with its key columns remapped, so it keeps tracking the build side.
+fn transfer_filter_across_keys(
+ filter: &Arc<dyn PhysicalExpr>,
+ key_map: &KeyTransferMap,
+) -> Result<Option<Arc<dyn PhysicalExpr>>> {
+ let mut all_keys = true;
+ let mut any_column = false;
+ let transformed = Arc::clone(filter).transform_down(|expr| {
+ let Some(column) = expr.downcast_ref::<Column>() else {
+ return Ok(Transformed::no(expr));
+ };
+ any_column = true;
+ match key_map.get(&column.index()) {
+ // The replacement is already in the other side's schema: do not
+ // descend into it, its columns are not indices of this join.
+ Some(other_key) => Ok(Transformed::new(
+ Arc::clone(other_key),
+ true,
+ TreeNodeRecursion::Jump,
Review Comment:
test_hashjoin_parent_filter_transfer_cast_key_with_projection added
--
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]
