chore: Move remaining expressions to spark-expr crate + some minor refactoring (apache#1165)

* move CheckOverflow to spark-expr crate

* move NegativeExpr to spark-expr crate

* move UnboundColumn to spark-expr crate

* move ExpandExec from execution::datafusion::operators to execution::operators

* refactoring to remove datafusion subpackage

* update imports in benches

* fix

* fix

Author: andygrove

native/Cargo.lock

@@ -19,7 +19,7 @@ use arrow::datatypes::{DataType, Field, Schema};
 use arrow_array::builder::Int64Builder;
 use arrow_array::{ArrayRef, RecordBatch};
 use arrow_schema::SchemaRef;
-use comet::execution::datafusion::expressions::bloom_filter_agg::BloomFilterAgg;
+use comet::execution::expressions::bloom_filter_agg::BloomFilterAgg;
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_plan::aggregates::{AggregateExec, AggregateMode, PhysicalGroupBy};

native/core/benches/bloom_filter_agg.rs

@@ -17,7 +17,7 @@
 
 use arrow_array::{builder::StringBuilder, RecordBatch};
 use arrow_schema::{DataType, Field, Schema};
-use comet::execution::datafusion::shuffle_writer::ShuffleWriterExec;
+use comet::execution::shuffle::ShuffleWriterExec;
 use criterion::{criterion_group, criterion_main, Criterion};
 use datafusion::{
     physical_plan::{common::collect, memory::MemoryExec, ExecutionPlan},

native/core/benches/shuffle_writer.rs

@@ -19,8 +19,8 @@ use arrow_schema::Field;
 use datafusion::{arrow::datatypes::DataType, logical_expr::Volatility};
 use std::{any::Any, sync::Arc};
 
-use crate::execution::datafusion::util::spark_bloom_filter;
-use crate::execution::datafusion::util::spark_bloom_filter::SparkBloomFilter;
+use crate::execution::util::spark_bloom_filter;
+use crate::execution::util::spark_bloom_filter::SparkBloomFilter;
 use arrow::array::ArrayRef;
 use arrow_array::BinaryArray;
 use datafusion::error::Result;

native/core/src/execution/datafusion/expressions/mod.rs

@@ -15,9 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::{
-    execution::datafusion::util::spark_bloom_filter::SparkBloomFilter, parquet::data_type::AsBytes,
-};
+use crate::{execution::util::spark_bloom_filter::SparkBloomFilter, parquet::data_type::AsBytes};
 use arrow::record_batch::RecordBatch;
 use arrow_array::cast::as_primitive_array;
 use arrow_schema::{DataType, Schema};

native/core/src/execution/datafusion/operators/mod.rs

@@ -15,11 +15,10 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Native execution through DataFusion
+//! Native DataFusion expressions
 
-pub mod expressions;
-mod operators;
-pub mod planner;
-pub mod shuffle_writer;
-pub(crate) mod spark_plan;
-mod util;
+pub mod bloom_filter_agg;
+pub mod bloom_filter_might_contain;
+pub mod subquery;
+
+pub use datafusion_comet_spark_expr::EvalMode;

…tafusion/expressions/bloom_filter_agg.rs …xecution/expressions/bloom_filter_agg.rsnative/core/src/execution/datafusion/expressions/bloom_filter_agg.rs renamed to native/core/src/execution/expressions/bloom_filter_agg.rs native/core/src/execution/datafusion/expressions/bloom_filter_agg.rs renamed to native/core/src/execution/expressions/bloom_filter_agg.rs

@@ -45,8 +45,8 @@ use super::{serde, utils::SparkArrowConvert, CometMemoryPool};
 use crate::{
     errors::{try_unwrap_or_throw, CometError, CometResult},
     execution::{
-        datafusion::planner::PhysicalPlanner, metrics::utils::update_comet_metric,
-        serde::to_arrow_datatype, shuffle::row::process_sorted_row_partition, sort::RdxSort,
+        metrics::utils::update_comet_metric, planner::PhysicalPlanner, serde::to_arrow_datatype,
+        shuffle::row::process_sorted_row_partition, sort::RdxSort,
     },
     jvm_bridge::{jni_new_global_ref, JVMClasses},
 };
@@ -59,8 +59,8 @@ use jni::{
 };
 use tokio::runtime::Runtime;
 
-use crate::execution::datafusion::spark_plan::SparkPlan;
 use crate::execution::operators::ScanExec;
+use crate::execution::spark_plan::SparkPlan;
 use log::info;
 
 /// Comet native execution context. Kept alive across JNI calls.