adriangb commented on code in PR #24445:
URL: https://github.com/apache/datafusion/pull/24445#discussion_r3814253573
##########
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:
This is a public function that has an unchecked precondition: a caller
passing a group that isn't `self.eq_group.project(mapping)` silently gets wrong
equivalence properties. Should we make that an error or something?
##########
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:
I worry these measured multipliers will bitrot
##########
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:
I don't think this are hitting the fast path. I.e. if we deleted the fast
path this test would still pass. Could we add a `cfg(test)` counter or
something that *asserts* that we are actually hitting the fast path?
--
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]
