feat: skip hash shuffle for date_bin/date_trunc on Range([timestamp])

datafusion/expr-common/src/interval_arithmetic.rs

@@ -1243,6 +1243,20 @@ fn prev_value(value: ScalarValue) -> ScalarValue {
     value_transition!(MIN, false, value)
 }
 
+/// Returns the previous distinct value of `value`, or `None` if `value` is
+/// null, already at the type minimum, or a type that has no predecessor.
+pub fn checked_predecessor(value: &ScalarValue) -> Option<ScalarValue> {
+    if value.is_null() {
+        return None;
+    }
+    let predecessor = prev_value(value.clone());
+    if predecessor.is_null() || predecessor == *value {
+        None
+    } else {
+        Some(predecessor)
+    }
+}
+
 trait OneTrait: Sized + std::ops::Add + std::ops::Sub {
     fn one() -> Self;
 }
@@ -2261,7 +2275,8 @@ impl NullableInterval {
 mod tests {
     use crate::{
         interval_arithmetic::{
-            Interval, handle_overflow, next_value, prev_value, satisfy_greater,
+            Interval, checked_predecessor, handle_overflow, next_value, prev_value,
+            satisfy_greater,
         },
         operator::Operator,
     };
@@ -2358,6 +2373,29 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_checked_predecessor() {
+        assert_eq!(
+            checked_predecessor(&ScalarValue::Int64(Some(10))),
+            Some(ScalarValue::Int64(Some(9)))
+        );
+        assert_eq!(checked_predecessor(&ScalarValue::Int64(None)), None);
+        assert_eq!(
+            checked_predecessor(&ScalarValue::Int64(Some(i64::MIN))),
+            None
+        );
+        assert_eq!(
+            checked_predecessor(&ScalarValue::TimestampNanosecond(Some(i64::MIN), None)),
+            None
+        );
+        // Types without a discrete predecessor return the same value from
+        // `prev_value`, which `checked_predecessor` treats as absent.
+        assert_eq!(
+            checked_predecessor(&ScalarValue::Utf8(Some("a".into()))),
+            None
+        );
+    }
+
     #[test]
     fn test_new_interval() -> Result<()> {
         use ScalarValue::*;

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

@@ -1319,6 +1319,31 @@ impl EquivalenceProperties {
             .unwrap_or_else(|_| ExprProperties::new_unknown())
     }
 
+    /// Returns true when `expr` is a (possibly non-strict) monotonic function of
+    /// `range_key` plus literals, such as `date_bin(interval, timestamp)` or
+    /// `date_trunc(unit, timestamp)`.
+    ///
+    /// The identity `expr == range_key` returns false so callers can treat "emit
+    /// the key as-is" separately from "emit a function of the key".
+    pub(crate) fn is_monotonic_function_of(
+        &self,
+        expr: &Arc<dyn PhysicalExpr>,
+        range_key: &Arc<dyn PhysicalExpr>,
+    ) -> bool {
+        if expr.eq(range_key) {
+            return false;
+        }
+        let dependencies = Dependencies::new(std::iter::once(PhysicalSortExpr::new(
+            Arc::clone(range_key),
+            Default::default(),
+        )));
+        matches!(
+            get_expr_properties(expr, &dependencies, &self.schema)
+                .map(|properties| properties.sort_properties),
+            Ok(SortProperties::Ordered(_))
+        )
+    }
+
     /// Transforms this `EquivalenceProperties` by mapping columns in the
     /// original schema to columns in the new schema by index.
     pub fn with_new_schema(mut self, schema: SchemaRef) -> Result<Self> {