sunchao commented on code in PR #25053:
URL: https://github.com/apache/datafusion/pull/25053#discussion_r4209305149
##########
datafusion/physical-expr/src/equivalence/properties/joins.rs:
##########
@@ -107,6 +97,250 @@ pub fn join_equivalence_properties(
Ok(result)
}
+/// Append build orderings to each probe ordering that permits a build suffix.
+fn join_orderings_with_suffix(
+ probe: &EquivalenceProperties,
+ build: &EquivalenceProperties,
+ on: &[(PhysicalExprRef, PhysicalExprRef)],
+ probe_side: JoinSide,
+ preserves_unmatched_probe: bool,
+ has_filter: bool,
+ null_equality: NullEquality,
+) -> Result<OrderingEquivalenceClass> {
+ if (probe.constraints().is_empty() && 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!(),
+ };
+ (Arc::clone(probe_key), Arc::clone(build_key))
+ })
+ .collect::<Vec<_>>();
+ let mut build_orderings = build.oeq_class().clone();
+ if probe_side == JoinSide::Left {
+ build_orderings.add_offset(probe.schema.fields().len() as _)?;
+ }
+ let mut result = OrderingEquivalenceClass::default();
+ let candidates =
+ probe_ordering_candidates(probe, build, &on,
preserves_unmatched_probe)?;
+ for ordering in candidates {
+ if !can_append_build_ordering(
+ &ordering,
+ probe,
+ build,
+ &on,
+ preserves_unmatched_probe,
+ has_filter,
+ null_equality,
+ ) {
+ continue;
+ }
+ // Append before removing redundant prefixes: [a] and [a, b] may both
+ // be valid, but [a, suffix] is not implied by [a, b, suffix].
+ let mut prefix = OrderingEquivalenceClass::new([ordering]);
+ if probe_side == JoinSide::Right {
+ prefix.add_offset(build.schema.fields().len() as _)?;
+ }
+ result.extend(prefix.join_suffix(&build_orderings));
+ }
+ Ok(result)
+}
+
+/// Collect probe ordering prefixes to check before appending build orderings.
+/// Keep the original orderings, their combined ordering, and combinations
selected
+/// by each unique constraint. The caller must prove the suffix for each
candidate.
+fn probe_ordering_candidates(
+ probe: &EquivalenceProperties,
+ build: &EquivalenceProperties,
+ on: &[(PhysicalExprRef, PhysicalExprRef)],
+ preserves_unmatched_probe: bool,
+) -> Result<Vec<LexOrdering>> {
+ let mut orderings = probe.oeq_class().iter().cloned().collect::<Vec<_>>();
+ if orderings.len() > 1 {
+ orderings.extend(probe.oeq_class().output_ordering());
+ }
+
+ // For a unique probe row, look for an ordering of its constrained columns.
+ for constraint in probe.constraints().iter() {
+ let keys = constraint_columns(constraint, &probe.schema);
+ let (ordering, _) = probe.find_longest_permutation(&keys)?;
+ orderings.extend(LexOrdering::new(ordering));
+ }
+
+ // For a unique build match, map the constrained columns to probe join
keys.
+ // This can select [a, b] from [extra], [a], [b] without the unrelated
extra.
+ for constraint in build.constraints().iter() {
+ let mut keys = vec![];
+ for column in constraint_columns(constraint, &build.schema) {
+ let column = build.eq_group().normalize_expr(column);
+ for (probe_key, build_key) in on {
+ let build_key =
build.eq_group().normalize_expr(Arc::clone(build_key));
+ if build_key.eq(&column) {
+ keys.push(Arc::clone(probe_key));
+ }
+ }
+ }
+ // Outer joins also need the additional keys to have a fixed match
status.
+ if preserves_unmatched_probe {
+ keys.extend(on.iter().map(|(key, _)| Arc::clone(key)));
+ }
+ let (ordering, _) = probe.find_longest_permutation(&keys)?;
+ orderings.extend(LexOrdering::new(ordering));
Review Comment:
[P2] Include determining columns when selecting computed-key prefixes
With independent probe orderings [extra], [a], [b], a build PRIMARY
KEY(k1,k2), and join keys a % 2 = k1 and b % 2 = k2, this search considers only
the modulo expressions. It misses the sufficient [a,b] prefix: equal a,b values
determine one build row, making its payload constant. The full combined
candidate includes unrelated extra and cannot establish ORDER BY a,b,payload.
In the native reproducer, LIMIT 1 returns immediately on base a2b8093 but gains
PartialSortExec and waits indefinitely here when the probe remains pending.
Finite controls pass. Include ordered input expressions determining the
computed keys, then apply the existing suffix-safety 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]
