Move create_physical_expr to phy-expr-common #3

datafusion/physical-expr-common/Cargo.toml

@@ -36,6 +36,14 @@ name = "datafusion_physical_expr_common"
 path = "src/lib.rs"
 
 [dependencies]
+ahash = { version = "0.8", default-features = false, features = [
+    "runtime-rng",
+] }
 arrow = { workspace = true }
+arrow-schema = { workspace = true }
 datafusion-common = { workspace = true, default-features = true }
 datafusion-expr = { workspace = true }
+half = { workspace = true }
+hashbrown = { version = "0.14", features = ["raw"] }
+itertools = { workspace = true, features = ["use_std"] }
+paste = "^1.0"

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

@@ -23,8 +23,8 @@ use std::{any::Any, sync::Arc};
 use crate::expressions::datum::{apply, apply_cmp};
 use crate::intervals::cp_solver::{propagate_arithmetic, propagate_comparison};
 use crate::physical_expr::down_cast_any_ref;
+use crate::physical_expr::PhysicalExpr;
 use crate::sort_properties::SortProperties;
-use crate::PhysicalExpr;
 
 use arrow::array::*;
 use arrow::compute::kernels::boolean::{and_kleene, not, or_kleene};
@@ -622,8 +622,13 @@ pub fn binary(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::expressions::{col, lit, try_cast, Literal};
-    use datafusion_common::plan_datafusion_err;
+    use crate::expressions::column::col;
+    use crate::expressions::literal::{lit, Literal};
+    use crate::expressions::try_cast::try_cast;
+    use arrow::datatypes::{
+        ArrowNumericType, Decimal128Type, Field, Int32Type, SchemaRef,
+    };
+    use datafusion_common::{plan_datafusion_err, Result};
     use datafusion_expr::type_coercion::binary::get_input_types;
 
     /// Performs a binary operation, applying any type coercion necessary

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

@@ -19,9 +19,10 @@ use std::borrow::Cow;
 use std::hash::{Hash, Hasher};
 use std::{any::Any, sync::Arc};
 
-use crate::expressions::{try_cast, NoOp};
+use crate::expressions::no_op::NoOp;
+use crate::expressions::try_cast::try_cast;
 use crate::physical_expr::down_cast_any_ref;
-use crate::PhysicalExpr;
+use crate::physical_expr::PhysicalExpr;
 
 use arrow::array::*;
 use arrow::compute::kernels::cmp::eq;
@@ -413,7 +414,11 @@ pub fn case(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::expressions::{binary, cast, col, lit};
+
+    use crate::expressions::binary::binary;
+    use crate::expressions::cast::cast;
+    use crate::expressions::column::col;
+    use crate::expressions::literal::{lit, Literal};
 
     use arrow::buffer::Buffer;
     use arrow::datatypes::DataType::Float64;
@@ -957,16 +962,15 @@ mod tests {
         let expr2 = expr
             .clone()
             .transform(|e| {
-                let transformed =
-                    match e.as_any().downcast_ref::<crate::expressions::Literal>() {
-                        Some(lit_value) => match lit_value.value() {
-                            ScalarValue::Utf8(Some(str_value)) => {
-                                Some(lit(str_value.to_uppercase()))
-                            }
-                            _ => None,
-                        },
+                let transformed = match e.as_any().downcast_ref::<Literal>() {
+                    Some(lit_value) => match lit_value.value() {
+                        ScalarValue::Utf8(Some(str_value)) => {
+                            Some(lit(str_value.to_uppercase()))
+                        }
                         _ => None,
-                    };
+                    },
+                    _ => None,
+                };
                 Ok(if let Some(transformed) = transformed {
                     Transformed::yes(transformed)
                 } else {
@@ -979,16 +983,15 @@ mod tests {
         let expr3 = expr
             .clone()
             .transform_down(|e| {
-                let transformed =
-                    match e.as_any().downcast_ref::<crate::expressions::Literal>() {
-                        Some(lit_value) => match lit_value.value() {
-                            ScalarValue::Utf8(Some(str_value)) => {
-                                Some(lit(str_value.to_uppercase()))
-                            }
-                            _ => None,
-                        },
+                let transformed = match e.as_any().downcast_ref::<Literal>() {
+                    Some(lit_value) => match lit_value.value() {
+                        ScalarValue::Utf8(Some(str_value)) => {
+                            Some(lit(str_value.to_uppercase()))
+                        }
                         _ => None,
-                    };
+                    },
+                    _ => None,
+                };
                 Ok(if let Some(transformed) = transformed {
                     Transformed::yes(transformed)
                 } else {

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

@@ -16,8 +16,8 @@
 // under the License.
 
 use crate::physical_expr::down_cast_any_ref;
+use crate::physical_expr::PhysicalExpr;
 use crate::sort_properties::SortProperties;
-use crate::PhysicalExpr;
 use std::any::Any;
 use std::fmt;
 use std::hash::{Hash, Hasher};
@@ -217,7 +217,7 @@ pub fn cast(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::expressions::col;
+    use crate::expressions::column::col;
 
     use arrow::{
         array::{

datafusion/physical-expr/src/expressions/datum.rs → datafusion/physical-expr-common/src/expressions/datum.rs

@@ -15,17 +15,17 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use arrow::array::BooleanArray;
 use arrow::array::{ArrayRef, Datum};
 use arrow::error::ArrowError;
-use arrow_array::BooleanArray;
 use datafusion_common::{Result, ScalarValue};
 use datafusion_expr::ColumnarValue;
 use std::sync::Arc;
 
 /// Applies a binary [`Datum`] kernel `f` to `lhs` and `rhs`
 ///
 /// This maps arrow-rs' [`Datum`] kernels to DataFusion's [`ColumnarValue`] abstraction
-pub(crate) fn apply(
+pub fn apply(
     lhs: &ColumnarValue,
     rhs: &ColumnarValue,
     f: impl Fn(&dyn Datum, &dyn Datum) -> Result<ArrayRef, ArrowError>,
@@ -49,7 +49,7 @@ pub(crate) fn apply(
 }
 
 /// Applies a binary [`Datum`] comparison kernel `f` to `lhs` and `rhs`
-pub(crate) fn apply_cmp(
+pub fn apply_cmp(
     lhs: &ColumnarValue,
     rhs: &ColumnarValue,
     f: impl Fn(&dyn Datum, &dyn Datum) -> Result<BooleanArray, ArrowError>,

datafusion/physical-expr/src/expressions/in_list.rs → datafusion/physical-expr-common/src/expressions/in_list.rs

@@ -22,17 +22,21 @@ use std::fmt::Debug;
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 
+use crate::physical_expr::PhysicalExpr;
 use crate::physical_expr::{down_cast_any_ref, physical_exprs_bag_equal};
-use crate::PhysicalExpr;
 
-use arrow::array::*;
+use arrow::array::downcast_primitive_array;
+use arrow::array::{
+    as_largestring_array, downcast_array, downcast_dictionary_array, Array,
+    ArrayAccessor, ArrayData, ArrayIter, ArrayRef, BooleanArray,
+};
 use arrow::buffer::BooleanBuffer;
 use arrow::compute::kernels::boolean::{not, or_kleene};
 use arrow::compute::kernels::cmp::eq;
 use arrow::compute::take;
-use arrow::datatypes::*;
+use arrow::datatypes::{i256, DataType, Schema};
+use arrow::record_batch::RecordBatch;
 use arrow::util::bit_iterator::BitIndexIterator;
-use arrow::{downcast_dictionary_array, downcast_primitive_array};
 use datafusion_common::cast::{
     as_boolean_array, as_generic_binary_array, as_string_array,
 };
@@ -454,13 +458,29 @@ pub fn in_list(
 
 #[cfg(test)]
 mod tests {
-
     use super::*;
-    use crate::expressions;
-    use crate::expressions::{col, lit, try_cast};
+
+    use crate::expressions::cast::cast;
+    use crate::expressions::column::col;
+    use crate::expressions::literal::lit;
+    use crate::expressions::try_cast::try_cast;
+    use arrow::array::BinaryArray;
+    use arrow::array::Date32Array;
+    use arrow::array::Date64Array;
+    use arrow::array::Decimal128Array;
+    use arrow::array::Float64Array;
+    use arrow::array::Int32Array;
+    use arrow::array::Int64Array;
+    use arrow::array::StringArray;
+    use arrow::array::TimestampMicrosecondArray;
+    use arrow::array::UInt16DictionaryArray;
     use datafusion_common::plan_err;
+    use datafusion_common::Result;
     use datafusion_expr::type_coercion::binary::comparison_coercion;
 
+    use arrow::datatypes::Field;
+    use arrow::datatypes::TimeUnit;
+
     type InListCastResult = (Arc<dyn PhysicalExpr>, Vec<Arc<dyn PhysicalExpr>>);
 
     // Try to do the type coercion for list physical expr.
@@ -1106,8 +1126,8 @@ mod tests {
         // list of phy expr
         let mut phy_exprs = vec![
             lit(1i64),
-            expressions::cast(lit(2i32), &schema, DataType::Int64)?,
-            expressions::try_cast(lit(3.13f32), &schema, DataType::Int64)?,
+            cast(lit(2i32), &schema, DataType::Int64)?,
+            try_cast(lit(3.13f32), &schema, DataType::Int64)?,
         ];
         let result = try_cast_static_filter_to_set(&phy_exprs, &schema).unwrap();
 
@@ -1117,8 +1137,8 @@ mod tests {
 
         try_cast_static_filter_to_set(&phy_exprs, &schema).unwrap();
         // cast(cast(lit())), but the cast to the same data type, one case will be ignored
-        phy_exprs.push(expressions::cast(
-            expressions::cast(lit(2i32), &schema, DataType::Int64)?,
+        phy_exprs.push(cast(
+            cast(lit(2i32), &schema, DataType::Int64)?,
             &schema,
             DataType::Int64,
         )?);
@@ -1127,15 +1147,15 @@ mod tests {
         phy_exprs.clear();
 
         // case(cast(lit())), the cast to the diff data type
-        phy_exprs.push(expressions::cast(
-            expressions::cast(lit(2i32), &schema, DataType::Int64)?,
+        phy_exprs.push(cast(
+            cast(lit(2i32), &schema, DataType::Int64)?,
             &schema,
             DataType::Int32,
         )?);
         try_cast_static_filter_to_set(&phy_exprs, &schema).unwrap();
 
         // column
-        phy_exprs.push(expressions::col("a", &schema)?);
+        phy_exprs.push(col("a", &schema)?);
         assert!(try_cast_static_filter_to_set(&phy_exprs, &schema).is_err());
 
         Ok(())

datafusion/physical-expr/src/expressions/is_not_null.rs → datafusion/physical-expr-common/src/expressions/is_not_null.rs

@@ -21,7 +21,7 @@ use std::hash::{Hash, Hasher};
 use std::{any::Any, sync::Arc};
 
 use crate::physical_expr::down_cast_any_ref;
-use crate::PhysicalExpr;
+use crate::physical_expr::PhysicalExpr;
 use arrow::compute;
 use arrow::{
     datatypes::{DataType, Schema},
@@ -115,7 +115,7 @@ pub fn is_not_null(arg: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>>
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::expressions::col;
+    use crate::expressions::column::col;
     use arrow::{
         array::{BooleanArray, StringArray},
         datatypes::*,

datafusion/physical-expr/src/expressions/is_null.rs → datafusion/physical-expr-common/src/expressions/is_null.rs

@@ -27,7 +27,7 @@ use arrow::{
 };
 
 use crate::physical_expr::down_cast_any_ref;
-use crate::PhysicalExpr;
+use crate::physical_expr::PhysicalExpr;
 use datafusion_common::Result;
 use datafusion_common::ScalarValue;
 use datafusion_expr::ColumnarValue;
@@ -116,7 +116,7 @@ pub fn is_null(arg: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::expressions::col;
+    use crate::expressions::column::col;
     use arrow::{
         array::{BooleanArray, StringArray},
         datatypes::*,

datafusion/physical-expr/src/expressions/like.rs → datafusion/physical-expr-common/src/expressions/like.rs

@@ -18,7 +18,7 @@
 use std::hash::{Hash, Hasher};
 use std::{any::Any, sync::Arc};
 
-use crate::{physical_expr::down_cast_any_ref, PhysicalExpr};
+use crate::physical_expr::{down_cast_any_ref, PhysicalExpr};
 
 use crate::expressions::datum::apply_cmp;
 use arrow::record_batch::RecordBatch;
@@ -174,7 +174,7 @@ pub fn like(
 #[cfg(test)]
 mod test {
     use super::*;
-    use crate::expressions::col;
+    use crate::expressions::column::col;
     use arrow::array::*;
     use arrow_schema::Field;
     use datafusion_common::cast::as_boolean_array;

datafusion/physical-expr/src/expressions/literal.rs → datafusion/physical-expr-common/src/expressions/literal.rs

@@ -22,8 +22,8 @@ use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 
 use crate::physical_expr::down_cast_any_ref;
+use crate::physical_expr::PhysicalExpr;
 use crate::sort_properties::SortProperties;
-use crate::PhysicalExpr;
 
 use arrow::{
     datatypes::{DataType, Schema},

datafusion/physical-expr-common/src/expressions/mod.rs

@@ -15,4 +15,20 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+#[macro_use]
+pub mod binary;
+pub mod case;
+pub mod cast;
 pub mod column;
+pub mod datum;
+pub mod in_list;
+pub mod is_not_null;
+pub mod is_null;
+pub mod like;
+pub mod literal;
+pub mod negative;
+pub mod no_op;
+pub mod not;
+pub mod try_cast;

datafusion/physical-expr/src/expressions/negative.rs → datafusion/physical-expr-common/src/expressions/negative.rs

@@ -22,8 +22,8 @@ use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 
 use crate::physical_expr::down_cast_any_ref;
+use crate::physical_expr::PhysicalExpr;
 use crate::sort_properties::SortProperties;
-use crate::PhysicalExpr;
 
 use arrow::{
     compute::kernels::numeric::neg_wrapping,
@@ -173,7 +173,7 @@ pub fn negative(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::expressions::{col, Column};
+    use crate::expressions::column::{col, Column};
 
     use arrow::array::*;
     use arrow::datatypes::*;

datafusion/physical-expr/src/expressions/no_op.rs → datafusion/physical-expr-common/src/expressions/no_op.rs

@@ -27,7 +27,7 @@ use arrow::{
 };
 
 use crate::physical_expr::down_cast_any_ref;
-use crate::PhysicalExpr;
+use crate::physical_expr::PhysicalExpr;
 use datafusion_common::{internal_err, Result};
 use datafusion_expr::ColumnarValue;