zhuqi-lucas commented on code in PR #24445:
URL: https://github.com/apache/datafusion/pull/24445#discussion_r3818560383
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -1169,6 +1169,23 @@ impl EquivalenceProperties {
/// `output_schema`.
pub fn project(&self, mapping: &ProjectionMapping, output_schema:
SchemaRef) -> Self {
let eq_group = self.eq_group.project(mapping);
+ self.project_with_eq_group(mapping, output_schema, eq_group)
+ }
+
+ /// Same as [`Self::project`], but takes an already-projected equivalence
+ /// group instead of computing one.
+ ///
+ /// [`EquivalenceGroup::project`] is a pure function of the group and the
+ /// mapping, so a caller that knows both are unchanged since the last
+ /// projection can hand back the previous result rather than recomputing an
+ /// identical one. Orderings are still derived here: they are precisely
what
+ /// changes when a sort is introduced below this node.
+ pub fn project_with_eq_group(
Review Comment:
Good catch. It cannot be a hard error without performing the projection the
caller is skipping, so it is a `debug_assert!` now. `project` routes around it
through a private unchecked entry point, since it builds the group itself and
satisfies the precondition by construction. Added a `should_panic` test so the
assertion cannot quietly stop being wired up.
##########
datafusion/physical-plan/src/projection.rs:
##########
@@ -2215,4 +2289,281 @@ mod tests {
Ok(())
}
+
+ /// `EmptyExec(a, b, c)` under a filter that equates `lhs` and `rhs`, so
the
+ /// child carries a non-trivial equivalence group.
+ fn filtered_source(lhs: &str, rhs: &str) -> Result<Arc<dyn ExecutionPlan>>
{
+ filtered_source_with_nullability(lhs, rhs, false)
+ }
+
+ /// As [`filtered_source`], but `nullable` varies the schema's nullability
+ /// while leaving field names and order alone.
+ fn filtered_source_with_nullability(
+ lhs: &str,
+ rhs: &str,
+ nullable: bool,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, nullable),
+ Field::new("b", DataType::Int32, nullable),
+ Field::new("c", DataType::Int32, nullable),
+ ]));
+ let input: Arc<dyn ExecutionPlan> =
Arc::new(EmptyExec::new(Arc::clone(&schema)));
+ let predicate = binary(
+ col(lhs, &schema)?,
+ Operator::Eq,
+ col(rhs, &schema)?,
+ &schema,
+ )?;
+ Ok(Arc::new(FilterExec::try_new(predicate, input)?))
+ }
+
+ /// `[a AS x, b AS y, c AS z]` against `filtered_source`'s schema.
+ fn renaming_exprs(schema: &SchemaRef) -> Result<Vec<ProjectionExpr>> {
+ [("a", "x"), ("b", "y"), ("c", "z")]
+ .into_iter()
+ .map(|(source, alias)| {
+ Ok(ProjectionExpr {
+ expr: col(source, schema)?,
+ alias: alias.to_string(),
+ })
+ })
+ .collect()
+ }
+
+ fn assert_same_properties(actual: &dyn ExecutionPlan, expected:
&ProjectionExec) {
+ let actual_props = actual.properties().equivalence_properties();
+ let expected_props = expected.properties().equivalence_properties();
+ assert!(
+ actual_props
+ .eq_group()
+ .has_same_classes(expected_props.eq_group()),
+ "equivalence group: {:?} vs {:?}",
+ actual_props.eq_group(),
+ expected_props.eq_group()
+ );
+ assert_eq!(
+ actual_props.oeq_class(),
+ expected_props.oeq_class(),
+ "orderings"
+ );
+ assert_eq!(
+ actual_props.constraints(),
+ expected_props.constraints(),
+ "constraints"
+ );
+ assert_eq!(actual_props.schema(), expected_props.schema(), "schema");
+ // `Partitioning` has no `PartialEq`, so compare the partition count
and
+ // the explicit `Display` form. Derived `Debug` would change with any
+ // field addition, making this brittle for no gain.
+ let actual_partitioning = actual.properties().output_partitioning();
+ let expected_partitioning =
expected.properties().output_partitioning();
+ assert_eq!(
+ actual_partitioning.partition_count(),
+ expected_partitioning.partition_count(),
+ "partition count"
+ );
+ assert_eq!(
+ actual_partitioning.to_string(),
+ expected_partitioning.to_string(),
+ "partitioning"
+ );
+ }
+
+ #[test]
+ fn test_sort_below_changes_orderings_but_not_the_equivalence_group() ->
Result<()> {
+ // The premise the fast path rests on. If a sort ever starts altering
+ // the equivalence group, reusing the cached group becomes unsound and
+ // this test is the one that should fail first.
+ let child = filtered_source("a", "b")?;
+ let ordering = LexOrdering::new(vec![PhysicalSortExpr::new_default(col(
+ "c",
+ &child.schema(),
+ )?)])
+ .expect("non-empty ordering");
+ let sorted = SortExec::new(ordering, Arc::clone(&child));
+
+ let child_props = child.properties().equivalence_properties();
+ let sorted_props = sorted.properties().equivalence_properties();
+
+ assert!(
+ !child_props.eq_group().is_empty(),
+ "the filter did not produce an equivalence class"
+ );
+ assert!(
+ child_props
+ .eq_group()
+ .has_same_classes(sorted_props.eq_group()),
+ "sorting altered the equivalence group"
+ );
+ assert_ne!(
+ child_props.oeq_class(),
+ sorted_props.oeq_class(),
+ "sorting did not alter the orderings"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_replace_children_reuses_eq_group_when_only_orderings_change() ->
Result<()> {
Review Comment:
You were right, they passed with the fast path deleted. There is a
`cfg(test)` probe now and the tests assert the reuse count: one for the two
cases that should reuse, zero for the case where the child equivalence group
changed. Deleting the fast path fails the first two. The probe is thread local
rather than a global counter, since tests run in parallel and a shared count
would let one test see another's hits.
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -310,6 +310,33 @@ pub struct EquivalenceGroup {
}
impl EquivalenceGroup {
+ /// A cheap, deliberately conservative check that two groups hold the same
+ /// equivalence classes.
+ ///
+ /// This is not `PartialEq`, and the distinction is the point. `classes`
is a
+ /// `Vec` whose order carries no meaning -- `remove_class_at_idx` uses
+ /// `swap_remove` -- so two groups describing exactly the same equalities
can
+ /// hold their classes in different orders and this returns `false` for
them.
+ /// Naming it `PartialEq` would invite callers to read it as semantic
equality,
+ /// which it is not.
+ ///
+ /// The comparison is positional because it runs on a hot path:
+ /// `ProjectionExec` consults it every time a rule replaces its child. Set
+ /// semantics would mean scanning the other group per class, and the
quadratic
+ /// blowup costs far more than the recomputation it is trying to avoid --
+ /// measured at 8x for 32 classes and 15x for 64.
Review Comment:
Agreed, dropped them. The reason the comparison is positional does not
depend on the exact numbers, so the doc just states the shape of the cost now
and the measurements stay in the commit that introduced it.
--
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]
