example of stateless execution

Author: askalt

datafusion/execution/src/cache/cache_unit.rs

@@ -111,7 +111,9 @@ mod tests {
     use datafusion_common::Statistics;
     use datafusion_common::stats::Precision;
     use datafusion_expr::ColumnarValue;
-    use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
+    use datafusion_physical_expr_common::physical_expr::{
+        ExprExecutionContext, PhysicalExpr,
+    };
     use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
     use object_store::ObjectMeta;
     use object_store::path::Path;
@@ -224,6 +226,13 @@ mod tests {
         fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
             write!(f, "MockExpr")
         }
+
+        fn execute(
+            self: Arc<Self>,
+            _context: &ExprExecutionContext,
+        ) -> datafusion_common::Result<Arc<dyn PhysicalExpr>> {
+            Ok(self)
+        }
     }
 
     fn ordering() -> LexOrdering {

datafusion/ffi/src/physical_expr/mod.rs

@@ -36,7 +36,7 @@ use datafusion_expr::interval_arithmetic::Interval;
 use datafusion_expr::sort_properties::ExprProperties;
 use datafusion_expr::statistics::Distribution;
 use datafusion_physical_expr::PhysicalExpr;
-use datafusion_physical_expr_common::physical_expr::fmt_sql;
+use datafusion_physical_expr_common::physical_expr::{ExprExecutionContext, fmt_sql};
 
 use crate::arrow_wrappers::{WrappedArray, WrappedSchema};
 use crate::expr::columnar_value::FFI_ColumnarValue;
@@ -705,6 +705,13 @@ impl PhysicalExpr for ForeignPhysicalExpr {
     fn is_volatile_node(&self) -> bool {
         unsafe { (self.expr.is_volatile_node)(&self.expr) }
     }
+
+    fn execute(
+        self: Arc<Self>,
+        _context: &ExprExecutionContext,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(self)
+    }
 }
 
 impl Eq for ForeignPhysicalExpr {}

datafusion/physical-expr-common/src/physical_expr.rs

@@ -43,6 +43,21 @@ use itertools::izip;
 /// Shared [`PhysicalExpr`].
 pub type PhysicalExprRef = Arc<dyn PhysicalExpr>;
 
+/// Describes execution context for the particular expression.
+pub struct ExprExecutionContext {
+    /// External parameters.
+    pub external_params: Arc<[ScalarValue]>,
+}
+
+impl ExprExecutionContext {
+    /// Make a new [`ExprExecutionContext`].
+    pub fn new(external_param: impl Into<Arc<[ScalarValue]>>) -> Self {
+        Self {
+            external_params: external_param.into(),
+        }
+    }
+}
+
 /// [`PhysicalExpr`]s represent expressions such as `A + 1` or `CAST(c1 AS int)`.
 ///
 /// `PhysicalExpr` knows its type, nullability and can be evaluated directly on
@@ -430,6 +445,15 @@ pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash {
     fn is_volatile_node(&self) -> bool {
         false
     }
+
+    /// Make this expression executable. The most expressions are executable and do not
+    /// require an additional work so this method could return `self`. However, there are
+    /// expressions that should be transformed prior to execution, e.g. placeholder that
+    /// should be resolved into scalar.
+    fn execute(
+        self: Arc<Self>,
+        context: &ExprExecutionContext,
+    ) -> Result<Arc<dyn PhysicalExpr>>;
 }
 
 #[deprecated(
@@ -662,7 +686,7 @@ pub fn is_volatile(expr: &Arc<dyn PhysicalExpr>) -> bool {
 
 #[cfg(test)]
 mod test {
-    use crate::physical_expr::PhysicalExpr;
+    use crate::physical_expr::{ExprExecutionContext, PhysicalExpr};
     use arrow::array::{Array, BooleanArray, Int64Array, RecordBatch};
     use arrow::datatypes::{DataType, Schema};
     use datafusion_expr_common::columnar_value::ColumnarValue;
@@ -707,6 +731,13 @@ mod test {
         fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
             f.write_str("TestExpr")
         }
+
+        fn execute(
+            self: Arc<Self>,
+            _context: &ExprExecutionContext,
+        ) -> datafusion_common::Result<Arc<dyn PhysicalExpr>> {
+            Ok(self)
+        }
     }
 
     impl Display for TestExpr {

datafusion/physical-expr/src/async_scalar_function.rs

@@ -25,7 +25,9 @@ use datafusion_common::{internal_err, not_impl_err};
 use datafusion_expr::ScalarFunctionArgs;
 use datafusion_expr::async_udf::AsyncScalarUDF;
 use datafusion_expr_common::columnar_value::ColumnarValue;
-use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
+use datafusion_physical_expr_common::physical_expr::{
+    ExprExecutionContext, PhysicalExpr,
+};
 use std::any::Any;
 use std::fmt::Display;
 use std::hash::{Hash, Hasher};
@@ -247,4 +249,11 @@ impl PhysicalExpr for AsyncFuncExpr {
     fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", self.func)
     }
+
+    fn execute(
+        self: Arc<Self>,
+        _context: &ExprExecutionContext,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(self)
+    }
 }

datafusion/physical-expr/src/expressions/binary.rs

@@ -42,6 +42,7 @@ use datafusion_expr::statistics::{
 use datafusion_expr::{ColumnarValue, Operator};
 use datafusion_physical_expr_common::datum::{apply, apply_cmp};
 
+use datafusion_physical_expr_common::physical_expr::ExprExecutionContext;
 use kernels::{
     bitwise_and_dyn, bitwise_and_dyn_scalar, bitwise_or_dyn, bitwise_or_dyn_scalar,
     bitwise_shift_left_dyn, bitwise_shift_left_dyn_scalar, bitwise_shift_right_dyn,
@@ -609,6 +610,13 @@ impl PhysicalExpr for BinaryExpr {
         write!(f, " {} ", self.op)?;
         write_child(f, self.right.as_ref(), precedence)
     }
+
+    fn execute(
+        self: Arc<Self>,
+        _context: &ExprExecutionContext,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(self)
+    }
 }
 
 /// Casts dictionary array to result type for binary numerical operators. Such operators

datafusion/physical-expr/src/expressions/case.rs

@@ -33,6 +33,7 @@ use datafusion_common::{
     internal_datafusion_err, internal_err,
 };
 use datafusion_expr::ColumnarValue;
+use datafusion_physical_expr_common::physical_expr::ExprExecutionContext;
 use indexmap::{IndexMap, IndexSet};
 use std::borrow::Cow;
 use std::hash::Hash;
@@ -1342,6 +1343,13 @@ impl PhysicalExpr for CaseExpr {
         }
         write!(f, "END")
     }
+
+    fn execute(
+        self: Arc<Self>,
+        _context: &ExprExecutionContext,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(self)
+    }
 }
 
 /// Attempts to const evaluate the given `predicate`.

datafusion/physical-expr/src/expressions/cast.rs

@@ -31,6 +31,7 @@ use datafusion_common::{Result, not_impl_err};
 use datafusion_expr_common::columnar_value::ColumnarValue;
 use datafusion_expr_common::interval_arithmetic::Interval;
 use datafusion_expr_common::sort_properties::ExprProperties;
+use datafusion_physical_expr_common::physical_expr::ExprExecutionContext;
 
 const DEFAULT_CAST_OPTIONS: CastOptions<'static> = CastOptions {
     safe: false,
@@ -237,6 +238,13 @@ impl PhysicalExpr for CastExpr {
 
         write!(f, ")")
     }
+
+    fn execute(
+        self: Arc<Self>,
+        _context: &ExprExecutionContext,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(self)
+    }
 }
 
 /// Return a PhysicalExpression representing `expr` casted to

datafusion/physical-expr/src/expressions/cast_column.rs

@@ -27,6 +27,7 @@ use datafusion_common::{
     Result, ScalarValue, format::DEFAULT_CAST_OPTIONS, nested_struct::cast_column,
 };
 use datafusion_expr_common::columnar_value::ColumnarValue;
+use datafusion_physical_expr_common::physical_expr::ExprExecutionContext;
 use std::{
     any::Any,
     fmt::{self, Display},
@@ -180,6 +181,13 @@ impl PhysicalExpr for CastColumnExpr {
     fn fmt_sql(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         Display::fmt(self, f)
     }
+
+    fn execute(
+        self: Arc<Self>,
+        _context: &ExprExecutionContext,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(self)
+    }
 }
 
 #[cfg(test)]

datafusion/physical-expr/src/expressions/column.rs

@@ -30,6 +30,7 @@ use arrow::{
 use datafusion_common::tree_node::{Transformed, TreeNode};
 use datafusion_common::{Result, internal_err, plan_err};
 use datafusion_expr::ColumnarValue;
+use datafusion_physical_expr_common::physical_expr::ExprExecutionContext;
 
 /// Represents the column at a given index in a RecordBatch
 ///
@@ -146,6 +147,13 @@ impl PhysicalExpr for Column {
     fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", self.name)
     }
+
+    fn execute(
+        self: Arc<Self>,
+        _context: &ExprExecutionContext,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(self)
+    }
 }
 
 impl Column {

datafusion/physical-expr/src/expressions/dynamic_filters.rs

@@ -26,7 +26,7 @@ use datafusion_common::{
     tree_node::{Transformed, TransformedResult, TreeNode},
 };
 use datafusion_expr::ColumnarValue;
-use datafusion_physical_expr_common::physical_expr::DynHash;
+use datafusion_physical_expr_common::physical_expr::{DynHash, ExprExecutionContext};
 
 /// State of a dynamic filter, tracking both updates and completion.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -445,6 +445,13 @@ impl PhysicalExpr for DynamicFilterPhysicalExpr {
         // Return the current generation of the expression.
         self.inner.read().generation
     }
+
+    fn execute(
+        self: Arc<Self>,
+        _context: &ExprExecutionContext,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(self)
+    }
 }
 
 #[cfg(test)]