Add a ScalarUDFImpl::simplfy() API, move SimplifyInfo et al to datafusion_expr (#9304)

* first draft

Signed-off-by: jayzhan211 <[email protected]>

* clippy

Signed-off-by: jayzhan211 <[email protected]>

* add comments

Signed-off-by: jayzhan211 <[email protected]>

* move to optimize rule

Signed-off-by: jayzhan211 <[email protected]>

* cleanup

Signed-off-by: jayzhan211 <[email protected]>

* fix explain test

Signed-off-by: jayzhan211 <[email protected]>

* move to simplifier

Signed-off-by: jayzhan211 <[email protected]>

* pass with schema

Signed-off-by: jayzhan211 <[email protected]>

* fix explain

Signed-off-by: jayzhan211 <[email protected]>

* fix doc

Signed-off-by: jayzhan211 <[email protected]>

* move to expr

Signed-off-by: jayzhan211 <[email protected]>

* change simplify signature

Signed-off-by: jayzhan211 <[email protected]>

* cleanup

Signed-off-by: jayzhan211 <[email protected]>

* cleanup

Signed-off-by: jayzhan211 <[email protected]>

* fix doc

Signed-off-by: jayzhan211 <[email protected]>

* fix doc

Signed-off-by: jayzhan211 <[email protected]>

* Update datafusion/expr/src/udf.rs

* Add backwards compatibile uses, inline FunctionSimplifier, rename to ExprSimplifyResult

* Remove DFSchema from SimplifyInfo

* Avoid requiring argument copies

* Improve docs

* fix link

* fix doc test

* Update datafusion/physical-expr/src/lib.rs

* Change example simplify to always simplify its argument

* Clarify comment

---------

Signed-off-by: jayzhan211 <[email protected]>
Co-authored-by: Andrew Lamb <[email protected]>

Author: jayzhan211

datafusion-cli/Cargo.lock

@@ -24,15 +24,16 @@ use arrow::record_batch::RecordBatch;
 use datafusion::arrow::datatypes::{DataType, Field, Schema, TimeUnit};
 use datafusion::common::{DFField, DFSchema};
 use datafusion::error::Result;
-use datafusion::optimizer::simplify_expressions::{ExprSimplifier, SimplifyContext};
-use datafusion::physical_expr::execution_props::ExecutionProps;
+use datafusion::optimizer::simplify_expressions::ExprSimplifier;
 use datafusion::physical_expr::{
     analyze, create_physical_expr, AnalysisContext, ExprBoundaries, PhysicalExpr,
 };
 use datafusion::prelude::*;
 use datafusion_common::{ScalarValue, ToDFSchema};
+use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_expr::expr::BinaryExpr;
 use datafusion_expr::interval_arithmetic::Interval;
+use datafusion_expr::simplify::SimplifyContext;
 use datafusion_expr::{ColumnarValue, ExprSchemable, Operator};
 
 /// This example demonstrates the DataFusion [`Expr`] API.

datafusion-examples/examples/expr_api.rs

@@ -28,8 +28,9 @@ use datafusion::physical_plan::memory::MemoryExec;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
 use datafusion_common::{plan_err, ScalarValue};
+use datafusion_expr::simplify::SimplifyContext;
 use datafusion_expr::{Expr, TableType};
-use datafusion_optimizer::simplify_expressions::{ExprSimplifier, SimplifyContext};
+use datafusion_optimizer::simplify_expressions::ExprSimplifier;
 use std::fs::File;
 use std::io::Seek;
 use std::path::Path;

datafusion-examples/examples/simple_udtf.rs

@@ -33,12 +33,10 @@ use arrow::{
 use arrow_schema::Fields;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
 use datafusion_common::{internal_err, Column, DFField, DFSchema, DataFusionError};
+use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_expr::{Expr, ScalarFunctionDefinition, Volatility};
 use datafusion_physical_expr::create_physical_expr;
-use datafusion_physical_expr::execution_props::ExecutionProps;
-
-use futures::stream::{BoxStream, FuturesUnordered};
-use futures::{StreamExt, TryStreamExt};
+use futures::stream::{BoxStream, FuturesUnordered, StreamExt, TryStreamExt};
 use log::{debug, trace};
 use object_store::path::Path;
 use object_store::{ObjectMeta, ObjectStore};

datafusion/core/src/datasource/listing/helpers.rs

@@ -800,13 +800,14 @@ mod tests {
         ArrayRef, Date64Array, Int32Array, Int64Array, Int8Array, StringArray,
         StructArray,
     };
+
     use arrow::datatypes::{DataType, 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_expr::{col, lit, when, Expr};
     use datafusion_physical_expr::create_physical_expr;
-    use datafusion_physical_expr::execution_props::ExecutionProps;
 
     use chrono::{TimeZone, Utc};
     use futures::StreamExt;

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

@@ -401,9 +401,9 @@ mod test {
     use super::*;
     use arrow::datatypes::Field;
     use datafusion_common::ToDFSchema;
+    use datafusion_expr::execution_props::ExecutionProps;
     use datafusion_expr::{cast, col, lit, Expr};
     use datafusion_physical_expr::create_physical_expr;
-    use datafusion_physical_expr::execution_props::ExecutionProps;
     use parquet::arrow::parquet_to_arrow_schema;
     use parquet::file::reader::{FileReader, SerializedFileReader};
     use rand::prelude::*;

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

@@ -346,8 +346,8 @@ mod tests {
     use arrow::datatypes::Schema;
     use arrow::datatypes::{DataType, Field};
     use datafusion_common::{Result, ToDFSchema};
+    use datafusion_expr::execution_props::ExecutionProps;
     use datafusion_expr::{cast, col, lit, Expr};
-    use datafusion_physical_expr::execution_props::ExecutionProps;
     use datafusion_physical_expr::{create_physical_expr, PhysicalExpr};
     use parquet::arrow::arrow_to_parquet_schema;
     use parquet::arrow::async_reader::ParquetObjectReader;

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

@@ -43,12 +43,12 @@ use datafusion_common::{
     tree_node::{TreeNode, TreeNodeRecursion, TreeNodeVisitor},
 };
 use datafusion_execution::registry::SerializerRegistry;
+pub use datafusion_expr::execution_props::ExecutionProps;
+use datafusion_expr::var_provider::is_system_variables;
 use datafusion_expr::{
     logical_plan::{DdlStatement, Statement},
     Expr, StringifiedPlan, UserDefinedLogicalNode, WindowUDF,
 };
-pub use datafusion_physical_expr::execution_props::ExecutionProps;
-use datafusion_physical_expr::var_provider::is_system_variables;
 use parking_lot::RwLock;
 use std::collections::hash_map::Entry;
 use std::string::String;

datafusion/core/src/execution/context/mod.rs

@@ -1341,10 +1341,10 @@ mod tests {
         datatypes::{DataType, TimeUnit},
     };
     use datafusion_common::{ScalarValue, 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 datafusion_physical_expr::execution_props::ExecutionProps;
     use std::collections::HashMap;
     use std::ops::{Not, Rem};
 

datafusion/core/src/physical_optimizer/pruning.rs

@@ -28,9 +28,10 @@ use crate::datasource::listing::{ListingTableUrl, PartitionedFile};
 use crate::datasource::object_store::ObjectStoreUrl;
 use crate::datasource::physical_plan::{FileScanConfig, ParquetExec};
 use crate::error::Result;
-use crate::optimizer::simplify_expressions::{ExprSimplifier, SimplifyContext};
+use crate::logical_expr::execution_props::ExecutionProps;
+use crate::logical_expr::simplify::SimplifyContext;
+use crate::optimizer::simplify_expressions::ExprSimplifier;
 use crate::physical_expr::create_physical_expr;
-use crate::physical_expr::execution_props::ExecutionProps;
 use crate::physical_plan::filter::FilterExec;
 use crate::physical_plan::metrics::MetricsSet;
 use crate::physical_plan::ExecutionPlan;