Add function volatility to Signature

datafusion-examples/examples/simple_udaf.rs

@@ -22,6 +22,7 @@ use datafusion::arrow::{
     record_batch::RecordBatch,
 };
 
+use datafusion::physical_plan::functions::Volatility;
 use datafusion::{error::Result, logical_plan::create_udaf, physical_plan::Accumulator};
 use datafusion::{prelude::*, scalar::ScalarValue};
 use std::sync::Arc;
@@ -137,6 +138,7 @@ async fn main() -> Result<()> {
         DataType::Float64,
         // the return type; DataFusion expects this to match the type returned by `evaluate`.
         Arc::new(DataType::Float64),
+        Volatility::Immutable,
         // This is the accumulator factory; DataFusion uses it to create new accumulators.
         Arc::new(|| Ok(Box::new(GeometricMean::new()))),
         // This is the description of the state. `state()` must match the types here.

datafusion-examples/examples/simple_udf.rs

@@ -15,10 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use datafusion::arrow::{
-    array::{ArrayRef, Float32Array, Float64Array},
-    datatypes::DataType,
-    record_batch::RecordBatch,
+use datafusion::{
+    arrow::{
+        array::{ArrayRef, Float32Array, Float64Array},
+        datatypes::DataType,
+        record_batch::RecordBatch,
+    },
+    physical_plan::functions::Volatility,
 };
 
 use datafusion::prelude::*;
@@ -112,6 +115,7 @@ async fn main() -> Result<()> {
         vec![DataType::Float64, DataType::Float64],
         // returns f64
         Arc::new(DataType::Float64),
+        Volatility::Immutable,
         pow,
     );
 

datafusion/src/execution/context.rs

@@ -1043,7 +1043,7 @@ impl FunctionRegistry for ExecutionContextState {
 mod tests {
     use super::*;
     use crate::logical_plan::{binary_expr, lit, Operator};
-    use crate::physical_plan::functions::make_scalar_function;
+    use crate::physical_plan::functions::{make_scalar_function, Volatility};
     use crate::physical_plan::{collect, collect_partitioned};
     use crate::test;
     use crate::variable::VarType;
@@ -2727,6 +2727,7 @@ mod tests {
             "MY_FUNC",
             vec![DataType::Int32],
             Arc::new(DataType::Int32),
+            Volatility::Immutable,
             myfunc,
         ));
 
@@ -2805,6 +2806,7 @@ mod tests {
             "MY_AVG",
             DataType::Float64,
             Arc::new(DataType::Float64),
+            Volatility::Immutable,
             Arc::new(|| Ok(Box::new(AvgAccumulator::try_new(&DataType::Float64)?))),
             Arc::new(vec![DataType::UInt64, DataType::Float64]),
         );
@@ -3017,6 +3019,7 @@ mod tests {
             "my_add",
             vec![DataType::Int32, DataType::Int32],
             Arc::new(DataType::Int32),
+            Volatility::Immutable,
             myfunc,
         ));
 
@@ -3144,6 +3147,7 @@ mod tests {
             "my_avg",
             DataType::Float64,
             Arc::new(DataType::Float64),
+            Volatility::Immutable,
             Arc::new(|| Ok(Box::new(AvgAccumulator::try_new(&DataType::Float64)?))),
             Arc::new(vec![DataType::UInt64, DataType::Float64]),
         );

datafusion/src/execution/dataframe_impl.rs

@@ -222,6 +222,7 @@ mod tests {
 
     use super::*;
     use crate::logical_plan::*;
+    use crate::physical_plan::functions::Volatility;
     use crate::{assert_batches_sorted_eq, execution::context::ExecutionContext};
     use crate::{datasource::csv::CsvReadOptions, physical_plan::ColumnarValue};
     use crate::{physical_plan::functions::ScalarFunctionImplementation, test};
@@ -358,6 +359,7 @@ mod tests {
             "my_fn",
             vec![DataType::Float64],
             Arc::new(DataType::Float64),
+            Volatility::Immutable,
             my_fn,
         ));
 

datafusion/src/logical_plan/expr.rs

@@ -21,6 +21,7 @@
 pub use super::Operator;
 use crate::error::{DataFusionError, Result};
 use crate::logical_plan::{window_frames, DFField, DFSchema, LogicalPlan};
+use crate::physical_plan::functions::Volatility;
 use crate::physical_plan::{
     aggregates, expressions::binary_operator_data_type, functions, udf::ScalarUDF,
     window_functions,
@@ -1598,10 +1599,16 @@ pub fn create_udf(
     name: &str,
     input_types: Vec<DataType>,
     return_type: Arc<DataType>,
+    volatility: Volatility,
     fun: ScalarFunctionImplementation,
 ) -> ScalarUDF {
     let return_type: ReturnTypeFunction = Arc::new(move |_| Ok(return_type.clone()));
-    ScalarUDF::new(name, &Signature::Exact(input_types), &return_type, &fun)
+    ScalarUDF::new(
+        name,
+        &Signature::exact(input_types, volatility),
+        &return_type,
+        &fun,
+    )
 }
 
 /// Creates a new UDAF with a specific signature, state type and return type.
@@ -1611,14 +1618,15 @@ pub fn create_udaf(
     name: &str,
     input_type: DataType,
     return_type: Arc<DataType>,
+    volatility: Volatility,
     accumulator: AccumulatorFunctionImplementation,
     state_type: Arc<Vec<DataType>>,
 ) -> AggregateUDF {
     let return_type: ReturnTypeFunction = Arc::new(move |_| Ok(return_type.clone()));
     let state_type: StateTypeFunction = Arc::new(move |_| Ok(state_type.clone()));
     AggregateUDF::new(
         name,
-        &Signature::Exact(vec![input_type]),
+        &Signature::exact(vec![input_type], volatility),
         &return_type,
         &accumulator,
         &state_type,

datafusion/src/physical_plan/aggregates.rs

@@ -27,7 +27,7 @@
 //! * Return type: a function `(arg_types) -> return_type`. E.g. for min, ([f32]) -> f32, ([f64]) -> f64.
 
 use super::{
-    functions::Signature,
+    functions::{Signature, Volatility},
     type_coercion::{coerce, data_types},
     Accumulator, AggregateExpr, PhysicalExpr,
 };
@@ -194,7 +194,7 @@ static DATES: &[DataType] = &[DataType::Date32, DataType::Date64];
 pub fn signature(fun: &AggregateFunction) -> Signature {
     // note: the physical expression must accept the type returned by this function or the execution panics.
     match fun {
-        AggregateFunction::Count => Signature::Any(1),
+        AggregateFunction::Count => Signature::any(1, Volatility::Immutable),
         AggregateFunction::Min | AggregateFunction::Max => {
             let valid = STRINGS
                 .iter()
@@ -203,10 +203,10 @@ pub fn signature(fun: &AggregateFunction) -> Signature {
                 .chain(DATES.iter())
                 .cloned()
                 .collect::<Vec<_>>();
-            Signature::Uniform(1, valid)
+            Signature::uniform(1, valid, Volatility::Immutable)
         }
         AggregateFunction::Avg | AggregateFunction::Sum => {
-            Signature::Uniform(1, NUMERICS.to_vec())
+            Signature::uniform(1, NUMERICS.to_vec(), Volatility::Immutable)
         }
     }
 }