fix: volatile expressions should not be target of common subexpt elim…

…ination

datafusion/expr/src/expr.rs

@@ -1692,6 +1692,23 @@ fn create_names(exprs: &[Expr]) -> Result<String> {
         .join(", "))
 }
 
+/// Whether the given expression is volatile, i.e. whether it can return different results
+/// when evaluated multiple times with the same input.
+pub fn is_volatile(expr: &Expr) -> bool {
+    match expr {
+        Expr::ScalarFunction(func) => match func.func_def {
+            ScalarFunctionDefinition::BuiltIn(func) => match func {
+                BuiltinScalarFunction::Random => true,
+                _ => false,
+            },
+            // TODO: Add volatile flag to UDFs
+            ScalarFunctionDefinition::UDF(_) => false,
+            ScalarFunctionDefinition::Name(_) => false,
+        },
+        _ => false,
+    }
+}
+
 #[cfg(test)]
 mod test {
     use crate::expr::Cast;

datafusion/optimizer/src/common_subexpr_eliminate.rs

@@ -29,7 +29,7 @@ use datafusion_common::tree_node::{
 use datafusion_common::{
     internal_err, Column, DFField, DFSchema, DFSchemaRef, DataFusionError, Result,
 };
-use datafusion_expr::expr::Alias;
+use datafusion_expr::expr::{is_volatile, Alias};
 use datafusion_expr::logical_plan::{
     Aggregate, Filter, LogicalPlan, Projection, Sort, Window,
 };
@@ -113,6 +113,8 @@ impl CommonSubexprEliminate {
         let Projection { expr, input, .. } = projection;
         let input_schema = Arc::clone(input.schema());
         let mut expr_set = ExprSet::new();
+
+        // Visit expr list and build expr identifier to occuring count map (`expr_set`).
         let arrays = to_arrays(expr, input_schema, &mut expr_set, ExprMask::Normal)?;
 
         let (mut new_expr, new_input) =
@@ -527,11 +529,13 @@ impl ExprMask {
                 | Expr::Wildcard { .. }
         );
 
+        let is_volatile = is_volatile(expr);
+
         let is_aggr = matches!(expr, Expr::AggregateFunction(..));
 
         match self {
-            Self::Normal => is_normal_minus_aggregates || is_aggr,
-            Self::NormalAndAggregates => is_normal_minus_aggregates,
+            Self::Normal => is_volatile || is_normal_minus_aggregates || is_aggr,
+            Self::NormalAndAggregates => is_volatile || is_normal_minus_aggregates,
         }
     }
 }

datafusion/sqllogictest/test_files/functions.slt

@@ -995,3 +995,8 @@ query ?
 SELECT find_in_set(NULL, NULL)
 ----
 NULL
+
+query B
+SELECT r1 == r2 FROM (SELECT random() r1, random() r2)
+----
+false