ScalarFunctionExpr Maintaining Order

datafusion/physical-expr/src/functions.rs

@@ -31,6 +31,7 @@
 //! argument is automatically is coerced to f64.
 
 use crate::execution_props::ExecutionProps;
+use crate::sort_properties::SortProperties;
 use crate::{
     array_expressions, conditional_expressions, datetime_expressions,
     expressions::{cast_column, nullif_func},
@@ -47,6 +48,7 @@ use datafusion_common::{internal_err, DataFusionError, Result, ScalarValue};
 use datafusion_expr::{
     BuiltinScalarFunction, ColumnarValue, ScalarFunctionImplementation,
 };
+use std::ops::Neg;
 use std::sync::Arc;
 
 /// Create a physical (function) expression.
@@ -173,11 +175,14 @@ pub fn create_physical_expr(
         _ => create_physical_fun(fun, execution_props)?,
     };
 
+    let monotonicity = get_func_monotonicity(fun);
+
     Ok(Arc::new(ScalarFunctionExpr::new(
         &format!("{fun}"),
         fun_expr,
         input_phy_exprs.to_vec(),
         &data_type,
+        monotonicity,
     )))
 }
 
@@ -879,6 +884,108 @@ pub fn create_physical_fun(
     })
 }
 
+/// Monotonicity of the `ScalarFunctionExpr` with respect to its arguments.
+/// Each element of this vector corresponds to an argument and indicates whether
+/// the function's behavior is monotonic, or non-monotonic/unknown for that argument, namely:
+/// - `None` signifies unknown monotonicity or non-monotonicity.
+/// - `Some(true)` indicates that the function is monotonically increasing w.r.t. the argument in question.
+/// - Some(false) indicates that the function is monotonically decreasing w.r.t. the argument in question.
+pub type FuncMonotonicity = Vec<Option<bool>>;
+
+/// Determines a [`ScalarFunctionExpr`]'s monotonicity for the given arguments
+/// and the function's behavior depending on its arguments.
+pub fn out_ordering(
+    func: &FuncMonotonicity,
+    arg_orderings: &[SortProperties],
+) -> SortProperties {
+    func.iter().zip(arg_orderings).fold(
+        SortProperties::Singleton,
+        |prev_sort, (item, arg)| {
+            let current_sort = func_order_in_one_dimension(item, arg);
+
+            match (prev_sort, current_sort) {
+                (_, SortProperties::Unordered) => SortProperties::Unordered,
+                (SortProperties::Singleton, SortProperties::Ordered(_)) => current_sort,
+                (SortProperties::Ordered(prev), SortProperties::Ordered(current))
+                    if prev.descending != current.descending =>
+                {
+                    SortProperties::Unordered
+                }
+                _ => prev_sort,
+            }
+        },
+    )
+}
+
+/// This function decides the monotonicity property of a [`ScalarFunctionExpr`] for a single argument (i.e. across a single dimension), given that argument's sort properties.
+fn func_order_in_one_dimension(
+    func_monotonicity: &Option<bool>,
+    arg: &SortProperties,
+) -> SortProperties {
+    if *arg == SortProperties::Singleton {
+        SortProperties::Singleton
+    } else {
+        match func_monotonicity {
+            None => SortProperties::Unordered,
+            Some(false) => {
+                if let SortProperties::Ordered(_) = arg {
+                    arg.neg()
+                } else {
+                    SortProperties::Unordered
+                }
+            }
+            Some(true) => {
+                if let SortProperties::Ordered(_) = arg {
+                    *arg
+                } else {
+                    SortProperties::Unordered
+                }
+            }
+        }
+    }
+}
+
+/// This function specifies monotonicity behaviors for built-in scalar functions.
+/// The list can be extended, only mathematical and datetime functions are
+/// considered for the initial implementation of this feature.
+pub fn get_func_monotonicity(fun: &BuiltinScalarFunction) -> Option<FuncMonotonicity> {
+    if matches!(
+        fun,
+        BuiltinScalarFunction::Atan
+            | BuiltinScalarFunction::Acosh
+            | BuiltinScalarFunction::Asinh
+            | BuiltinScalarFunction::Atanh
+            | BuiltinScalarFunction::Ceil
+            | BuiltinScalarFunction::Degrees
+            | BuiltinScalarFunction::Exp
+            | BuiltinScalarFunction::Factorial
+            | BuiltinScalarFunction::Floor
+            | BuiltinScalarFunction::Ln
+            | BuiltinScalarFunction::Log10
+            | BuiltinScalarFunction::Log2
+            | BuiltinScalarFunction::Radians
+            | BuiltinScalarFunction::Round
+            | BuiltinScalarFunction::Signum
+            | BuiltinScalarFunction::Sinh
+            | BuiltinScalarFunction::Sqrt
+            | BuiltinScalarFunction::Cbrt
+            | BuiltinScalarFunction::Tanh
+            | BuiltinScalarFunction::Trunc
+            | BuiltinScalarFunction::Pi
+    ) {
+        Some(vec![Some(true)])
+    } else if matches!(
+        fun,
+        BuiltinScalarFunction::DateTrunc | BuiltinScalarFunction::DateBin
+    ) {
+        Some(vec![None, Some(true)])
+    } else if *fun == BuiltinScalarFunction::Log {
+        Some(vec![Some(true), Some(false)])
+    } else {
+        None
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

datafusion/physical-expr/src/scalar_function.rs

@@ -29,7 +29,10 @@
 //! This module also has a set of coercion rules to improve user experience: if an argument i32 is passed
 //! to a function that supports f64, it is coerced to f64.
 
+use crate::functions::out_ordering;
+use crate::functions::FuncMonotonicity;
 use crate::physical_expr::down_cast_any_ref;
+use crate::sort_properties::SortProperties;
 use crate::utils::expr_list_eq_strict_order;
 use crate::PhysicalExpr;
 use arrow::datatypes::{DataType, Schema};
@@ -51,6 +54,11 @@ pub struct ScalarFunctionExpr {
     name: String,
     args: Vec<Arc<dyn PhysicalExpr>>,
     return_type: DataType,
+    // Keeps monotonicity information of the function.
+    // FuncMonotonicity vector is one to one mapped to `args`,
+    // and it specifies the effect of an increase or decrease in
+    // the corresponding `arg` to the function value.
+    monotonicity: Option<FuncMonotonicity>,
 }
 
 impl Debug for ScalarFunctionExpr {
@@ -71,12 +79,14 @@ impl ScalarFunctionExpr {
         fun: ScalarFunctionImplementation,
         args: Vec<Arc<dyn PhysicalExpr>>,
         return_type: &DataType,
+        monotonicity: Option<FuncMonotonicity>,
     ) -> Self {
         Self {
             fun,
             name: name.to_owned(),
             args,
             return_type: return_type.clone(),
+            monotonicity,
         }
     }
 
@@ -157,6 +167,7 @@ impl PhysicalExpr for ScalarFunctionExpr {
             self.fun.clone(),
             children,
             self.return_type(),
+            self.monotonicity.clone(),
         )))
     }
 
@@ -167,6 +178,13 @@ impl PhysicalExpr for ScalarFunctionExpr {
         self.return_type.hash(&mut s);
         // Add `self.fun` when hash is available
     }
+
+    fn get_ordering(&self, children: &[SortProperties]) -> SortProperties {
+        self.monotonicity
+            .as_ref()
+            .map(|monotonicity| out_ordering(monotonicity, children))
+            .unwrap_or(SortProperties::Unordered)
+    }
 }
 
 impl PartialEq<dyn Any> for ScalarFunctionExpr {

datafusion/physical-expr/src/udf.rs

@@ -39,5 +39,6 @@ pub fn create_physical_expr(
         fun.fun.clone(),
         input_phy_exprs.to_vec(),
         (fun.return_type)(&input_exprs_types)?.as_ref(),
+        None,
     )))
 }

datafusion/proto/src/physical_plan/from_proto.rs

@@ -255,6 +255,7 @@ pub fn parse_physical_expr(
                 fun_expr,
                 args,
                 &convert_required!(e.return_type)?,
+                None,
             ))
         }
         ExprType::ScalarUdf(e) => {
@@ -271,6 +272,7 @@ pub fn parse_physical_expr(
                 scalar_fun,
                 args,
                 &convert_required!(e.return_type)?,
+                None,
             ))
         }
         ExprType::LikeExpr(like_expr) => Arc::new(LikeExpr::new(

datafusion/proto/src/physical_plan/mod.rs

@@ -1813,6 +1813,7 @@ mod roundtrip_tests {
             fun_expr,
             vec![col("a", &schema)?],
             &DataType::Int64,
+            None,
         );
 
         let project =
@@ -1846,6 +1847,7 @@ mod roundtrip_tests {
             scalar_fn,
             vec![col("a", &schema)?],
             &DataType::Int64,
+            None,
         );
 
         let project =

datafusion/sqllogictest/test_files/order.slt

@@ -444,4 +444,137 @@ SortPreservingMergeExec: [result@0 ASC NULLS LAST]
 ------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, b, c], output_ordering=[a@0 ASC NULLS LAST], has_header=true
 
 statement ok
-drop table multiple_ordered_table;
+drop table multiple_ordered_table;
+
+# Create tables having some ordered columns. In the next step, we will expect to observe that scalar
+# functions, such as mathematical functions like atan(), ceil(), sqrt(), or date_time functions 
+# like date_bin() and date_trunc(), will maintain the order of its argument columns.
+statement ok
+CREATE EXTERNAL TABLE csv_with_timestamps (
+  name VARCHAR,
+  ts TIMESTAMP
+)
+STORED AS CSV
+WITH ORDER (ts ASC NULLS LAST)
+LOCATION '../core/tests/data/timestamps.csv';
+
+query TT
+EXPLAIN SELECT DATE_BIN(INTERVAL '15 minutes', ts, TIMESTAMP '2022-08-03 14:40:00Z') as db15
+FROM csv_with_timestamps
+ORDER BY db15;
+----
+logical_plan
+Sort: db15 ASC NULLS LAST
+--Projection: date_bin(IntervalMonthDayNano("900000000000"), csv_with_timestamps.ts, TimestampNanosecond(1659537600000000000, None)) AS db15
+----TableScan: csv_with_timestamps projection=[ts]
+physical_plan
+SortPreservingMergeExec: [db15@0 ASC NULLS LAST]
+--ProjectionExec: expr=[date_bin(900000000000, ts@0, 1659537600000000000) as db15]
+----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/timestamps.csv]]}, projection=[ts], output_ordering=[ts@0 ASC NULLS LAST], has_header=false
+
+query TT
+EXPLAIN SELECT DATE_TRUNC('DAY', ts) as dt_day
+FROM csv_with_timestamps
+ORDER BY dt_day;
+----
+logical_plan
+Sort: dt_day ASC NULLS LAST
+--Projection: date_trunc(Utf8("DAY"), csv_with_timestamps.ts) AS dt_day
+----TableScan: csv_with_timestamps projection=[ts]
+physical_plan
+SortPreservingMergeExec: [dt_day@0 ASC NULLS LAST]
+--ProjectionExec: expr=[date_trunc(DAY, ts@0) as dt_day]
+----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/timestamps.csv]]}, projection=[ts], output_ordering=[ts@0 ASC NULLS LAST], has_header=false
+
+statement ok
+drop table csv_with_timestamps;
+
+statement ok
+drop table aggregate_test_100;
+
+statement ok
+CREATE EXTERNAL TABLE aggregate_test_100 (
+  c1  VARCHAR NOT NULL,
+  c2  TINYINT NOT NULL,
+  c3  SMALLINT NOT NULL,
+  c4  SMALLINT,
+  c5  INT,
+  c6  BIGINT NOT NULL,
+  c7  SMALLINT NOT NULL,
+  c8  INT NOT NULL,
+  c9  BIGINT UNSIGNED NOT NULL,
+  c10 VARCHAR NOT NULL,
+  c11 FLOAT NOT NULL,
+  c12 DOUBLE NOT NULL,
+  c13 VARCHAR NOT NULL
+)
+STORED AS CSV
+WITH HEADER ROW
+WITH ORDER(c11)
+WITH ORDER(c12 DESC)
+LOCATION '../../testing/data/csv/aggregate_test_100.csv'
+
+query TT
+EXPLAIN SELECT ATAN(c11) as atan_c11
+FROM aggregate_test_100
+ORDER BY atan_c11;
+----
+logical_plan
+Sort: atan_c11 ASC NULLS LAST
+--Projection: atan(aggregate_test_100.c11) AS atan_c11
+----TableScan: aggregate_test_100 projection=[c11]
+physical_plan
+SortPreservingMergeExec: [atan_c11@0 ASC NULLS LAST]
+--ProjectionExec: expr=[atan(c11@0) as atan_c11]
+----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c11], output_ordering=[c11@0 ASC NULLS LAST], has_header=true
+
+query TT
+EXPLAIN SELECT CEIL(c11) as ceil_c11
+FROM aggregate_test_100
+ORDER BY ceil_c11;
+----
+logical_plan
+Sort: ceil_c11 ASC NULLS LAST
+--Projection: ceil(aggregate_test_100.c11) AS ceil_c11
+----TableScan: aggregate_test_100 projection=[c11]
+physical_plan
+SortPreservingMergeExec: [ceil_c11@0 ASC NULLS LAST]
+--ProjectionExec: expr=[ceil(c11@0) as ceil_c11]
+----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c11], output_ordering=[c11@0 ASC NULLS LAST], has_header=true
+
+query TT
+  EXPLAIN SELECT LOG(c11, c12) as log_c11_base_c12
+  FROM aggregate_test_100
+  ORDER BY log_c11_base_c12;
+----
+logical_plan
+Sort: log_c11_base_c12 ASC NULLS LAST
+--Projection: log(CAST(aggregate_test_100.c11 AS Float64), aggregate_test_100.c12) AS log_c11_base_c12
+----TableScan: aggregate_test_100 projection=[c11, c12]
+physical_plan
+SortPreservingMergeExec: [log_c11_base_c12@0 ASC NULLS LAST]
+--ProjectionExec: expr=[log(CAST(c11@0 AS Float64), c12@1) as log_c11_base_c12]
+----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c11, c12], output_ordering=[c11@0 ASC NULLS LAST], has_header=true
+
+query TT
+EXPLAIN SELECT LOG(c12, c11) as log_c12_base_c11
+FROM aggregate_test_100
+ORDER BY log_c12_base_c11 DESC;
+----
+logical_plan
+Sort: log_c12_base_c11 DESC NULLS FIRST
+--Projection: log(aggregate_test_100.c12, CAST(aggregate_test_100.c11 AS Float64)) AS log_c12_base_c11
+----TableScan: aggregate_test_100 projection=[c11, c12]
+physical_plan
+SortPreservingMergeExec: [log_c12_base_c11@0 DESC]
+--ProjectionExec: expr=[log(c12@1, CAST(c11@0 AS Float64)) as log_c12_base_c11]
+----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c11, c12], output_ordering=[c11@0 ASC NULLS LAST], has_header=true
+
+statement ok
+drop table aggregate_test_100;