sunchao commented on code in PR #25053:
URL: https://github.com/apache/datafusion/pull/25053#discussion_r4027940670
##########
datafusion/physical-expr/src/equivalence/properties/joins.rs:
##########
@@ -107,6 +95,112 @@ pub fn join_equivalence_properties(
Ok(result)
}
+/// Append build orderings only when equal probe ordering values identify at
most
+/// one build row. The suffix is then constant within each probe ordering
group,
+/// even if probe rows repeat. For an outer join preserving the probe side, all
+/// join keys must be fixed within the group, and the caller must rule out
filters,
+/// so the group cannot mix matched build rows with NULL-extended rows.
+fn unique_build_join_orderings(
+ probe: &EquivalenceProperties,
+ build: &EquivalenceProperties,
+ on: &[(PhysicalExprRef, PhysicalExprRef)],
+ probe_side: JoinSide,
+ preserves_unmatched_probe: bool,
+ null_equality: NullEquality,
+) -> Result<OrderingEquivalenceClass> {
+ if build.constraints().is_empty() || build.oeq_class().is_empty() {
+ return Ok(OrderingEquivalenceClass::default());
+ }
+ let on = on
+ .iter()
+ .map(|(left, right)| {
+ let (probe_key, build_key) = match probe_side {
+ JoinSide::Left => (left, right),
+ JoinSide::Right => (right, left),
+ JoinSide::None => unreachable!(),
+ };
+ (
+ probe.eq_group().normalize_expr(Arc::clone(probe_key)),
+ build.eq_group().normalize_expr(Arc::clone(build_key)),
+ )
+ })
+ .collect::<Vec<_>>();
+ let mut valid_orderings = Vec::new();
+ for ordering in probe.oeq_class().iter() {
+ // Within a group of equal ordering values, these expressions are
+ // constant. Keep this assumption local to the ordering proof.
+ let mut group = probe.eq_group().clone();
+ for sort in ordering {
+ let expr = probe.eq_group().normalize_expr(Arc::clone(&sort.expr));
+ group.add_constant(ConstExpr::from(expr));
+ }
+ let probe_key_is_fixed = |key: &PhysicalExprRef| {
+ !is_volatile(key) && group.is_expr_constant(key).is_some()
Review Comment:
[P2] Check volatility before normalizing join keys
A filter such as p = volatile_flip(p) records an equivalence that normalizes
the join key p + volatile_flip(p) into p + p. This volatility check then misses
the function and incorrectly accepts the ordering suffix. The native reproducer
returns [10,20] for LIMIT 2 instead of [10,10]. The previous PR revision passed
this case; the current exception reintroduces the original failure. Preserve
the original key's volatility before normalization.
##########
datafusion/physical-expr/src/equivalence/properties/joins.rs:
##########
@@ -107,6 +95,112 @@ pub fn join_equivalence_properties(
Ok(result)
}
+/// Append build orderings only when equal probe ordering values identify at
most
+/// one build row. The suffix is then constant within each probe ordering
group,
+/// even if probe rows repeat. For an outer join preserving the probe side, all
+/// join keys must be fixed within the group, and the caller must rule out
filters,
+/// so the group cannot mix matched build rows with NULL-extended rows.
+fn unique_build_join_orderings(
+ probe: &EquivalenceProperties,
+ build: &EquivalenceProperties,
+ on: &[(PhysicalExprRef, PhysicalExprRef)],
+ probe_side: JoinSide,
+ preserves_unmatched_probe: bool,
+ null_equality: NullEquality,
+) -> Result<OrderingEquivalenceClass> {
+ if build.constraints().is_empty() || build.oeq_class().is_empty() {
+ return Ok(OrderingEquivalenceClass::default());
+ }
+ let on = on
+ .iter()
+ .map(|(left, right)| {
+ let (probe_key, build_key) = match probe_side {
+ JoinSide::Left => (left, right),
+ JoinSide::Right => (right, left),
+ JoinSide::None => unreachable!(),
+ };
+ (
+ probe.eq_group().normalize_expr(Arc::clone(probe_key)),
+ build.eq_group().normalize_expr(Arc::clone(build_key)),
+ )
+ })
+ .collect::<Vec<_>>();
+ let mut valid_orderings = Vec::new();
+ for ordering in probe.oeq_class().iter() {
+ // Within a group of equal ordering values, these expressions are
+ // constant. Keep this assumption local to the ordering proof.
+ let mut group = probe.eq_group().clone();
+ for sort in ordering {
+ let expr = probe.eq_group().normalize_expr(Arc::clone(&sort.expr));
+ group.add_constant(ConstExpr::from(expr));
+ }
+ let probe_key_is_fixed = |key: &PhysicalExprRef| {
+ !is_volatile(key) && group.is_expr_constant(key).is_some()
+ };
+
+ // Outer joins must have the same match status throughout the group.
+ if preserves_unmatched_probe
+ && !on
+ .iter()
+ .all(|(probe_key, _)| probe_key_is_fixed(probe_key))
+ {
+ continue;
+ }
+ if !ordering_covers_unique_build_key(
+ build,
+ &on,
+ probe_key_is_fixed,
+ null_equality,
+ ) {
+ continue;
+ }
+ valid_orderings.push(ordering.clone());
+ }
+ let mut probe_orderings = OrderingEquivalenceClass::new(valid_orderings);
+ if probe_orderings.is_empty() {
+ return Ok(probe_orderings);
+ }
+ let mut build_orderings = build.oeq_class().clone();
+ match probe_side {
+ JoinSide::Left =>
build_orderings.add_offset(probe.schema.fields().len() as _)?,
+ JoinSide::Right =>
probe_orderings.add_offset(build.schema.fields().len() as _)?,
+ JoinSide::None => unreachable!(),
+ }
+ Ok(probe_orderings.join_suffix(&build_orderings))
+}
+
+/// Check whether the probe ordering determines a unique build key.
+/// Join keys must already be normalized against their input equivalence
groups.
+fn ordering_covers_unique_build_key(
+ build: &EquivalenceProperties,
+ on: &[(PhysicalExprRef, PhysicalExprRef)],
+ probe_key_is_fixed: impl Fn(&PhysicalExprRef) -> bool,
+ null_equality: NullEquality,
+) -> bool {
+ build.constraints().iter().any(|constraint| {
+ let (Constraint::PrimaryKey(indices) | Constraint::Unique(indices)) =
constraint;
+ !indices.is_empty()
+ && indices.iter().all(|&index| {
+ let Some(field) = build.schema.fields().get(index) else {
+ return false;
+ };
+ // UNIQUE permits repeated NULLs, which only invalidate the
+ // uniqueness proof when NULL join keys can match each other.
+ if matches!(constraint, Constraint::Unique(_))
+ && field.is_nullable()
+ && null_equality == NullEquality::NullEqualsNull
+ {
+ return false;
Review Comment:
[P2] Account for nonnullable probe keys under null-safe equality
A nullable UNIQUE build key still guarantees at most one match when the
corresponding probe key cannot be NULL, even with NullEqualsNull. This guard
rejects that safe case. For an ordered unbounded probe, ORDER BY probe.k,
build.v LIMIT 1 returns immediately on base 5747e87 but gains PartialSortExec
on this head and waits indefinitely if the key group never finishes. The finite
control passes. Include probe-key nullability in the uniqueness proof.
--
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]
