Minor Fix for Logical and Physical Expr Conversions (#11142)

* Minor

* Update planner.rs

Author: berkaysynnada

datafusion/core/src/datasource/physical_plan/parquet/mod.rs

@@ -796,17 +796,15 @@ mod tests {
         ArrayRef, Date64Array, Int32Array, Int64Array, Int8Array, StringArray,
         StructArray,
     };
-
     use arrow::datatypes::{Field, Schema, SchemaBuilder};
     use arrow::record_batch::RecordBatch;
     use arrow_schema::Fields;
-    use datafusion_common::{assert_contains, FileType, GetExt, ScalarValue, ToDFSchema};
-    use datafusion_expr::execution_props::ExecutionProps;
+    use datafusion_common::{assert_contains, FileType, GetExt, ScalarValue};
     use datafusion_expr::{col, lit, when, Expr};
-    use datafusion_physical_expr::create_physical_expr;
+    use datafusion_physical_expr::planner::logical2physical;
+    use datafusion_physical_plan::ExecutionPlanProperties;
 
     use chrono::{TimeZone, Utc};
-    use datafusion_physical_plan::ExecutionPlanProperties;
     use futures::StreamExt;
     use object_store::local::LocalFileSystem;
     use object_store::path::Path;
@@ -2061,12 +2059,6 @@ mod tests {
         Ok(())
     }
 
-    fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
-        let df_schema = schema.clone().to_dfschema().unwrap();
-        let execution_props = ExecutionProps::new();
-        create_physical_expr(expr, &df_schema, &execution_props).unwrap()
-    }
-
     #[tokio::test]
     async fn test_struct_filter_parquet() -> Result<()> {
         let tmp_dir = TempDir::new()?;

datafusion/core/src/datasource/physical_plan/parquet/row_filter.rs

@@ -410,23 +410,20 @@ pub fn build_row_filter(
 
 #[cfg(test)]
 mod test {
-    use arrow::datatypes::Field;
-    use arrow_schema::TimeUnit::Nanosecond;
-    use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
-    use parquet::arrow::parquet_to_arrow_schema;
-    use parquet::file::reader::{FileReader, SerializedFileReader};
-    use rand::prelude::*;
-
+    use super::*;
     use crate::datasource::schema_adapter::DefaultSchemaAdapterFactory;
     use crate::datasource::schema_adapter::SchemaAdapterFactory;
 
-    use datafusion_common::ToDFSchema;
-    use datafusion_expr::execution_props::ExecutionProps;
+    use arrow::datatypes::Field;
+    use arrow_schema::TimeUnit::Nanosecond;
     use datafusion_expr::{cast, col, lit, Expr};
-    use datafusion_physical_expr::create_physical_expr;
+    use datafusion_physical_expr::planner::logical2physical;
     use datafusion_physical_plan::metrics::{Count, Time};
 
-    use super::*;
+    use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+    use parquet::arrow::parquet_to_arrow_schema;
+    use parquet::file::reader::{FileReader, SerializedFileReader};
+    use rand::prelude::*;
 
     // We should ignore predicate that read non-primitive columns
     #[test]
@@ -590,10 +587,4 @@ mod test {
             assert_eq!(projection, remapped)
         }
     }
-
-    fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
-        let df_schema = schema.clone().to_dfschema().unwrap();
-        let execution_props = ExecutionProps::new();
-        create_physical_expr(expr, &df_schema, &execution_props).unwrap()
-    }
 }

datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs

@@ -404,15 +404,19 @@ impl<'a> PruningStatistics for RowGroupPruningStatistics<'a> {
 
 #[cfg(test)]
 mod tests {
+    use std::ops::Rem;
+    use std::sync::Arc;
+
     use super::*;
     use crate::datasource::physical_plan::parquet::reader::ParquetFileReader;
     use crate::physical_plan::metrics::ExecutionPlanMetricsSet;
+
     use arrow::datatypes::DataType::Decimal128;
     use arrow::datatypes::{DataType, Field};
-    use datafusion_common::{Result, ToDFSchema};
-    use datafusion_expr::execution_props::ExecutionProps;
+    use datafusion_common::Result;
     use datafusion_expr::{cast, col, lit, Expr};
-    use datafusion_physical_expr::{create_physical_expr, PhysicalExpr};
+    use datafusion_physical_expr::planner::logical2physical;
+
     use parquet::arrow::arrow_to_parquet_schema;
     use parquet::arrow::async_reader::ParquetObjectReader;
     use parquet::basic::LogicalType;
@@ -422,8 +426,6 @@ mod tests {
         basic::Type as PhysicalType, file::statistics::Statistics as ParquetStatistics,
         schema::types::SchemaDescPtr,
     };
-    use std::ops::Rem;
-    use std::sync::Arc;
 
     struct PrimitiveTypeField {
         name: &'static str,
@@ -1111,12 +1113,6 @@ mod tests {
         ParquetFileMetrics::new(0, "file.parquet", &metrics)
     }
 
-    fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
-        let df_schema = schema.clone().to_dfschema().unwrap();
-        let execution_props = ExecutionProps::new();
-        create_physical_expr(expr, &df_schema, &execution_props).unwrap()
-    }
-
     #[tokio::test]
     async fn test_row_group_bloom_filter_pruning_predicate_simple_expr() {
         BloomFilterTest::new_data_index_bloom_encoding_stats()

datafusion/core/src/physical_optimizer/pruning.rs

@@ -1555,22 +1555,22 @@ pub(crate) enum StatisticsType {
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
+    use std::ops::{Not, Rem};
+
     use super::*;
     use crate::assert_batches_eq;
     use crate::logical_expr::{col, lit};
+
     use arrow::array::Decimal128Array;
     use arrow::{
         array::{BinaryArray, Int32Array, Int64Array, StringArray},
         datatypes::TimeUnit,
     };
     use arrow_array::UInt64Array;
-    use datafusion_common::ToDFSchema;
-    use datafusion_expr::execution_props::ExecutionProps;
     use datafusion_expr::expr::InList;
     use datafusion_expr::{cast, is_null, try_cast, Expr};
-    use datafusion_physical_expr::create_physical_expr;
-    use std::collections::HashMap;
-    use std::ops::{Not, Rem};
+    use datafusion_physical_expr::planner::logical2physical;
 
     #[derive(Debug, Default)]
     /// Mock statistic provider for tests
@@ -3876,10 +3876,4 @@ mod tests {
         let expr = logical2physical(expr, schema);
         build_predicate_expression(&expr, schema, required_columns)
     }
-
-    fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
-        let df_schema = schema.clone().to_dfschema().unwrap();
-        let execution_props = ExecutionProps::new();
-        create_physical_expr(expr, &df_schema, &execution_props).unwrap()
-    }
 }

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

@@ -211,6 +211,9 @@ pub trait AggregateExpr: Send + Sync + Debug + PartialEq<dyn Any> {
     /// Rewrites [`AggregateExpr`], with new expressions given. The argument should be consistent
     /// with the return value of the [`AggregateExpr::all_expressions`] method.
     /// Returns `Some(Arc<dyn AggregateExpr>)` if re-write is supported, otherwise returns `None`.
+    /// TODO: This method only rewrites the [`PhysicalExpr`]s and does not handle [`Expr`]s.
+    /// This can cause silent bugs and should be fixed in the future (possibly with physical-to-logical
+    /// conversions).
     fn with_new_expressions(
         &self,
         _args: Vec<Arc<dyn PhysicalExpr>>,

datafusion/physical-expr/src/planner.rs

@@ -17,23 +17,22 @@
 
 use std::sync::Arc;
 
-use arrow::datatypes::Schema;
+use crate::scalar_function;
+use crate::{
+    expressions::{self, binary, like, Column, Literal},
+    PhysicalExpr,
+};
 
+use arrow::datatypes::Schema;
 use datafusion_common::{
-    exec_err, not_impl_err, plan_err, DFSchema, Result, ScalarValue,
+    exec_err, not_impl_err, plan_err, DFSchema, Result, ScalarValue, ToDFSchema,
 };
 use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_expr::expr::{Alias, Cast, InList, ScalarFunction};
 use datafusion_expr::var_provider::is_system_variables;
 use datafusion_expr::var_provider::VarType;
 use datafusion_expr::{binary_expr, Between, BinaryExpr, Expr, Like, Operator, TryCast};
 
-use crate::scalar_function;
-use crate::{
-    expressions::{self, binary, like, Column, Literal},
-    PhysicalExpr,
-};
-
 /// [PhysicalExpr] evaluate DataFusion expressions such as `A + 1`, or `CAST(c1
 /// AS int)`.
 ///
@@ -358,6 +357,13 @@ where
         .collect::<Result<Vec<_>>>()
 }
 
+/// Convert a logical expression to a physical expression (without any simplification, etc)
+pub fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
+    let df_schema = schema.clone().to_dfschema().unwrap();
+    let execution_props = ExecutionProps::new();
+    create_physical_expr(expr, &df_schema, &execution_props).unwrap()
+}
+
 #[cfg(test)]
 mod tests {
     use arrow_array::{ArrayRef, BooleanArray, RecordBatch, StringArray};

datafusion/physical-expr/src/utils/guarantee.rs

@@ -419,15 +419,16 @@ impl<'a> ColOpLit<'a> {
 
 #[cfg(test)]
 mod test {
+    use std::sync::OnceLock;
+
     use super::*;
-    use crate::create_physical_expr;
+    use crate::planner::logical2physical;
+
     use arrow_schema::{DataType, Field, Schema, SchemaRef};
-    use datafusion_common::ToDFSchema;
-    use datafusion_expr::execution_props::ExecutionProps;
     use datafusion_expr::expr_fn::*;
     use datafusion_expr::{lit, Expr};
+
     use itertools::Itertools;
-    use std::sync::OnceLock;
 
     #[test]
     fn test_literal() {
@@ -867,13 +868,6 @@ mod test {
         LiteralGuarantee::try_new(column, Guarantee::NotIn, literals.iter()).unwrap()
     }
 
-    /// Convert a logical expression to a physical expression (without any simplification, etc)
-    fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
-        let df_schema = schema.clone().to_dfschema().unwrap();
-        let execution_props = ExecutionProps::new();
-        create_physical_expr(expr, &df_schema, &execution_props).unwrap()
-    }
-
     // Schema for testing
     fn schema() -> SchemaRef {
         SCHEMA