fix: SanityCheckPlan error when a per-partition constant leads an ordering

datafusion/physical-expr/src/equivalence/properties/mod.rs

     pub fn new_with_orderings( 
 self.add_orderings(new_orderings); 
     pub fn new_with_orderings( 
@@ -379,7 +379,8 @@ impl EquivalenceProperties {
         }
     }
 
-    /// Adds new orderings into the existing ordering equivalence class.
+    /// Adds new orderings into the existing ordering equivalence class, along
+    /// with the orderings they imply through known equalities.
     pub fn add_orderings(
         &mut self,
         orderings: impl IntoIterator<Item = impl IntoIterator<Item = PhysicalSortExpr>>,
@@ -391,14 +392,32 @@ impl EquivalenceProperties {
             .cloned()
             .filter_map(|o| self.normalize_sort_exprs(o))
             .collect();
-        if !normal_orderings.is_empty() {
-            self.oeq_class.extend(orderings);
-            // Normalize given orderings to update the cache:
-            self.oeq_cache.normal_cls.extend(normal_orderings);
-            // TODO: If no ordering is found to be redundant during extension, we
-            //       can use a shortcut algorithm to update the leading map.
-            self.oeq_cache.update_map();
+        if normal_orderings.is_empty() {
+            return;
         }
+        // `add_constants` and `add_equal_conditions` discover implied orderings
+        // too, and whether an ordering is satisfied must not depend on which of
+        // them ran last. Deriving one fails only on an expression that does not
+        // match the schema; that ordering is then not implied, which can cost a
+        // sort but not a wrong result.
+        let mut implied: Vec<LexOrdering> = vec![];
+        for ordering in &normal_orderings {
+            // Two orderings can imply the same one; expand it only once:
+            for ordering in self.implied_by(ordering).unwrap_or_default() {
+                if !implied.contains(&ordering) {
+                    implied.push(ordering);
+                }
+            }
+        }
+        self.oeq_class.extend(orderings);
+        // Normalize given orderings to update the cache:
+        self.oeq_cache.normal_cls.extend(normal_orderings);
+        // TODO: If no ordering is found to be redundant during extension, we
+        //       can use a shortcut algorithm to update the leading map.
+        self.oeq_cache.update_map();
+        // Each implied ordering is shorter than the one implying it, so this
+        // terminates:
+        self.add_orderings(implied);
     }
 
     /// Adds a single ordering to the existing ordering equivalence class.
@@ -568,15 +587,76 @@ impl EquivalenceProperties {
         let mut new_orderings = vec![];
         for idx in ordering_idxs {
             let ordering = &self.oeq_cache.normal_cls[*idx];
-            new_orderings
-                .extend(self.implied_ordering(&ordering[0], ordering[1..].iter())?);
-        }
-        if !new_orderings.is_empty() {
-            self.add_orderings(new_orderings);
+            // As in `add_orderings`, an ordering that cannot be derived is
+            // not implied, so the result does not depend on the call order:
+            new_orderings.extend(self.implied_by(ordering).unwrap_or_default());
         }
+        self.add_orderings(new_orderings);
         Ok(())
     }
 
+    /// Returns the orderings that the normal ordering `ordering` implies
+    /// through known equalities: for each key that is a function of the keys
+    /// after it (see [`Self::implied_ordering`]), `ordering` without that key.
+    ///
+    /// # Example: a key that is a function of the keys after it
+    ///
+    /// * Ordering: `[k ASC, a ASC, c ASC]`
+    /// * Equality: `a = CAST(c AS DATE)`
+    /// * Returns: `[k ASC, c ASC]`
+    ///
+    /// Among the rows that tie on `k`, the rows are sorted by `a`, and the
+    /// rows that also tie on `a` are sorted by `c`. `a` is the date of `c`, so
+    /// a row with an earlier `a` also has an earlier `c`, and the rows that
+    /// tie on `k` are sorted by `c` alone:
+    ///
+    /// ```text
+    /// ┌───┬────────────┬──────────────────┐
+    /// │ k │ a          │ c                │
+    /// ├───┼────────────┼──────────────────┤
+    /// │ x │ 2026-01-01 │ 2026-01-01 09:00 │
+    /// │ x │ 2026-01-02 │ 2026-01-02 01:00 │
+    /// │ x │ 2026-01-02 │ 2026-01-02 05:00 │
+    /// │ y │ 2026-01-01 │ 2026-01-01 03:00 │
+    /// └───┴────────────┴──────────────────┘
+    /// ```
+    ///
+    /// # Example: a constant that differs between partitions
+    ///
+    /// * Ordering: `[k ASC, a ASC, c ASC]`, where `k` is a constant whose value
+    ///   differs between partitions
+    /// * Equality: `a = CAST(c AS DATE)`
+    /// * Returns: `[c ASC]`
+    ///
+    /// A normal ordering can keep such a constant (see
+    /// `EquivalenceGroup::normalize_sort_exprs`). Within a partition `k` has
+    /// one value, so it is left out of the keys and of the result.
+    ///
+    /// # Example: directions that differ
+    ///
+    /// * Ordering: `[a ASC, c DESC]`
+    /// * Equality: `a = CAST(c AS DATE)`
+    /// * Returns: nothing
+    ///
+    /// `c` descends within a date but ascends across dates, so the rows are
+    /// not sorted by `c` in either direction.
+    fn implied_by(&self, ordering: &LexOrdering) -> Result<Vec<LexOrdering>> {
+        let keys: Vec<_> = ordering
+            .iter()
+            .filter(|sort_expr| self.eq_group.is_expr_constant(&sort_expr.expr).is_none())
+            .collect();
+        let mut implied = vec![];
+        for (idx, key) in keys.iter().enumerate() {
+            if let Some(rest) =
+                self.implied_ordering(key, keys[idx + 1..].iter().copied())?
+            {
+                let prefix = keys[..idx].iter().copied().cloned();
+                implied.extend(LexOrdering::new(prefix.chain(rest)));
+            }
+        }
+        Ok(implied)
+    }
+
     /// Returns the ordering that the normal ordering `[leading, rest..]`
     /// implies through an equality on its leading expression, if any.
     ///