Finish removing deprecated datafusion::logical_plan module (#3816)

* remove deprecated module

* fix imports

* infer schema for parquet in benchmarks

* Revert "infer schema for parquet in benchmarks"

This reverts commit 6899e98.

Author: andygrove

benchmarks/src/bin/parquet_filter_pushdown.rs

@@ -22,12 +22,11 @@ use arrow::array::{
 use arrow::datatypes::{DataType, Field, Int32Type, Schema, SchemaRef, TimeUnit};
 use arrow::record_batch::RecordBatch;
 use arrow::util::pretty;
-use datafusion::common::Result;
+use datafusion::common::{Result, ToDFSchema};
 use datafusion::datasource::listing::{ListingTableUrl, PartitionedFile};
 use datafusion::datasource::object_store::ObjectStoreUrl;
 use datafusion::execution::context::ExecutionProps;
 use datafusion::logical_expr::{lit, or, Expr};
-use datafusion::logical_plan::ToDFSchema;
 use datafusion::optimizer::utils::combine_filters_disjunctive;
 use datafusion::physical_expr::create_physical_expr;
 use datafusion::physical_plan::collect;

datafusion-examples/examples/custom_datasource.rs

@@ -20,16 +20,17 @@ use datafusion::arrow::array::{UInt64Builder, UInt8Builder};
 use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::dataframe::DataFrame;
+use datafusion::datasource::provider_as_source;
 use datafusion::datasource::{TableProvider, TableType};
 use datafusion::error::Result;
 use datafusion::execution::context::{SessionState, TaskContext};
-use datafusion::logical_plan::{provider_as_source, Expr, LogicalPlanBuilder};
 use datafusion::physical_plan::expressions::PhysicalSortExpr;
 use datafusion::physical_plan::memory::MemoryStream;
 use datafusion::physical_plan::{
     project_schema, ExecutionPlan, SendableRecordBatchStream, Statistics,
 };
 use datafusion::prelude::*;
+use datafusion_expr::{Expr, LogicalPlanBuilder};
 use std::any::Any;
 use std::collections::{BTreeMap, HashMap};
 use std::fmt::{Debug, Formatter};

datafusion-examples/examples/expr_api.rs

@@ -16,13 +16,12 @@
 // under the License.
 
 use datafusion::arrow::datatypes::{DataType, Field, Schema, TimeUnit};
-
 use datafusion::error::Result;
-use datafusion::logical_plan::ToDFSchema;
 use datafusion::optimizer::expr_simplifier::{ExprSimplifier, SimplifyContext};
 use datafusion::physical_expr::execution_props::ExecutionProps;
 use datafusion::prelude::*;
-use datafusion::{logical_plan::Operator, scalar::ScalarValue};
+use datafusion_common::{ScalarValue, ToDFSchema};
+use datafusion_expr::Operator;
 
 /// This example demonstrates the DataFusion [`Expr`] API.
 ///

datafusion-examples/examples/simple_udaf.rs

@@ -21,11 +21,11 @@ use datafusion::arrow::{
     array::ArrayRef, array::Float32Array, array::Float64Array, datatypes::DataType,
     record_batch::RecordBatch,
 };
-
 use datafusion::from_slice::FromSlice;
 use datafusion::logical_expr::AggregateState;
-use datafusion::{error::Result, logical_plan::create_udaf, physical_plan::Accumulator};
+use datafusion::{error::Result, physical_plan::Accumulator};
 use datafusion::{logical_expr::Volatility, prelude::*, scalar::ScalarValue};
+use datafusion_expr::create_udaf;
 use std::sync::Arc;
 
 // create local session context with an in-memory table

datafusion/core/src/dataframe.rs

@@ -27,9 +27,9 @@ use crate::execution::{
     context::{SessionState, TaskContext},
     FunctionRegistry,
 };
-use crate::logical_expr::{utils::find_window_exprs, TableType};
-use crate::logical_plan::{
-    col, DFSchema, Expr, JoinType, LogicalPlan, LogicalPlanBuilder, Partitioning,
+use crate::logical_expr::{
+    col, utils::find_window_exprs, Expr, JoinType, LogicalPlan, LogicalPlanBuilder,
+    Partitioning, TableType,
 };
 use crate::physical_plan::file_format::{plan_to_csv, plan_to_json, plan_to_parquet};
 use crate::physical_plan::SendableRecordBatchStream;
@@ -38,7 +38,7 @@ use crate::physical_plan::{execute_stream, execute_stream_partitioned, Execution
 use crate::prelude::SessionContext;
 use crate::scalar::ScalarValue;
 use async_trait::async_trait;
-use datafusion_common::Column;
+use datafusion_common::{Column, DFSchema};
 use parking_lot::RwLock;
 use parquet::file::properties::WriterProperties;
 use std::any::Any;

datafusion/core/src/datasource/datasource.rs

@@ -26,7 +26,7 @@ pub use datafusion_expr::{TableProviderFilterPushDown, TableType};
 use crate::arrow::datatypes::SchemaRef;
 use crate::error::Result;
 use crate::execution::context::SessionState;
-use crate::logical_plan::Expr;
+use crate::logical_expr::Expr;
 use crate::physical_plan::ExecutionPlan;
 
 /// Source table

datafusion/core/src/datasource/empty.rs

@@ -26,7 +26,7 @@ use async_trait::async_trait;
 use crate::datasource::{TableProvider, TableType};
 use crate::error::Result;
 use crate::execution::context::SessionState;
-use crate::logical_plan::Expr;
+use crate::logical_expr::Expr;
 use crate::physical_plan::project_schema;
 use crate::physical_plan::{empty::EmptyExec, ExecutionPlan};
 

datafusion/core/src/datasource/file_format/avro.rs

@@ -28,7 +28,7 @@ use object_store::{GetResult, ObjectMeta, ObjectStore};
 use super::FileFormat;
 use crate::avro_to_arrow::read_avro_schema_from_reader;
 use crate::error::Result;
-use crate::logical_plan::Expr;
+use crate::logical_expr::Expr;
 use crate::physical_plan::file_format::{AvroExec, FileScanConfig};
 use crate::physical_plan::ExecutionPlan;
 use crate::physical_plan::Statistics;

datafusion/core/src/datasource/file_format/csv.rs

@@ -35,7 +35,7 @@ use super::FileFormat;
 use crate::datasource::file_format::file_type::FileCompressionType;
 use crate::datasource::file_format::DEFAULT_SCHEMA_INFER_MAX_RECORD;
 use crate::error::Result;
-use crate::logical_plan::Expr;
+use crate::logical_expr::Expr;
 use crate::physical_plan::file_format::{CsvExec, FileScanConfig};
 use crate::physical_plan::ExecutionPlan;
 use crate::physical_plan::Statistics;

datafusion/core/src/datasource/file_format/json.rs

@@ -36,7 +36,7 @@ use super::FileScanConfig;
 use crate::datasource::file_format::file_type::FileCompressionType;
 use crate::datasource::file_format::DEFAULT_SCHEMA_INFER_MAX_RECORD;
 use crate::error::Result;
-use crate::logical_plan::Expr;
+use crate::logical_expr::Expr;
 use crate::physical_plan::file_format::NdJsonExec;
 use crate::physical_plan::ExecutionPlan;
 use crate::physical_plan::Statistics;

datafusion/core/src/datasource/file_format/mod.rs

@@ -32,7 +32,7 @@ use std::sync::Arc;
 
 use crate::arrow::datatypes::SchemaRef;
 use crate::error::Result;
-use crate::logical_plan::Expr;
+use crate::logical_expr::Expr;
 use crate::physical_plan::file_format::FileScanConfig;
 use crate::physical_plan::{ExecutionPlan, Statistics};
 

datafusion/core/src/datasource/file_format/parquet.rs

@@ -41,7 +41,7 @@ use crate::arrow::array::{
 use crate::arrow::datatypes::{DataType, Field};
 use crate::datasource::{create_max_min_accs, get_col_stats};
 use crate::error::Result;
-use crate::logical_plan::Expr;
+use crate::logical_expr::Expr;
 use crate::physical_plan::expressions::{MaxAccumulator, MinAccumulator};
 use crate::physical_plan::file_format::{ParquetExec, SchemaAdapter};
 use crate::physical_plan::{Accumulator, ExecutionPlan, Statistics};

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

@@ -357,7 +357,7 @@ fn parse_partitions_for_path<'a>(
 
 #[cfg(test)]
 mod tests {
-    use crate::logical_plan::{case, col, lit};
+    use crate::logical_expr::{case, col, lit};
     use crate::test::object_store::make_test_store;
     use futures::StreamExt;
 

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

@@ -42,7 +42,7 @@ use crate::logical_expr::TableProviderFilterPushDown;
 use crate::{
     error::{DataFusionError, Result},
     execution::context::SessionState,
-    logical_plan::Expr,
+    logical_expr::Expr,
     physical_plan::{
         empty::EmptyExec,
         file_format::{FileScanConfig, DEFAULT_PARTITION_COLUMN_DATATYPE},
@@ -499,7 +499,7 @@ mod tests {
     use crate::prelude::SessionContext;
     use crate::{
         datasource::file_format::{avro::AvroFormat, parquet::ParquetFormat},
-        logical_plan::{col, lit},
+        logical_expr::{col, lit},
         test::{columns, object_store::register_test_store},
     };
     use arrow::datatypes::DataType;

datafusion/core/src/datasource/memory.rs

@@ -30,7 +30,7 @@ use async_trait::async_trait;
 use crate::datasource::{TableProvider, TableType};
 use crate::error::{DataFusionError, Result};
 use crate::execution::context::{SessionState, TaskContext};
-use crate::logical_plan::Expr;
+use crate::logical_expr::Expr;
 use crate::physical_plan::common;
 use crate::physical_plan::memory::MemoryExec;
 use crate::physical_plan::ExecutionPlan;

datafusion/core/src/datasource/view.rs

@@ -25,7 +25,7 @@ use datafusion_expr::LogicalPlanBuilder;
 
 use crate::{
     error::Result,
-    logical_plan::{Expr, LogicalPlan},
+    logical_expr::{Expr, LogicalPlan},
     physical_plan::ExecutionPlan,
 };
 

datafusion/core/src/execution/context.rs

@@ -29,7 +29,7 @@ use crate::{
         },
         MemTable, ViewTable,
     },
-    logical_plan::{PlanType, ToStringifiedPlan},
+    logical_expr::{PlanType, ToStringifiedPlan},
     optimizer::optimizer::Optimizer,
     physical_optimizer::{
         aggregate_statistics::AggregateStatistics,
@@ -59,13 +59,15 @@ use crate::catalog::{
     schema::{MemorySchemaProvider, SchemaProvider},
 };
 use crate::dataframe::DataFrame;
-use crate::datasource::listing::{ListingTableConfig, ListingTableUrl};
-use crate::datasource::TableProvider;
+use crate::datasource::{
+    listing::{ListingTableConfig, ListingTableUrl},
+    provider_as_source, TableProvider,
+};
 use crate::error::{DataFusionError, Result};
-use crate::logical_plan::{
-    provider_as_source, CreateCatalog, CreateCatalogSchema, CreateExternalTable,
-    CreateMemoryTable, CreateView, DropTable, FunctionRegistry, LogicalPlan,
-    LogicalPlanBuilder, UNNAMED_TABLE,
+use crate::logical_expr::{
+    CreateCatalog, CreateCatalogSchema, CreateExternalTable, CreateMemoryTable,
+    CreateView, DropTable, DropView, Explain, LogicalPlan, LogicalPlanBuilder,
+    TableSource, TableType, UNNAMED_TABLE,
 };
 use crate::optimizer::optimizer::{OptimizerConfig, OptimizerRule};
 use datafusion_sql::{ResolvedTableReference, TableReference};
@@ -80,8 +82,7 @@ use crate::config::{
 };
 use crate::datasource::datasource::TableProviderFactory;
 use crate::datasource::file_format::file_type::{FileCompressionType, FileType};
-use crate::execution::runtime_env::RuntimeEnv;
-use crate::logical_expr::Explain;
+use crate::execution::{runtime_env::RuntimeEnv, FunctionRegistry};
 use crate::physical_plan::file_format::{plan_to_csv, plan_to_json, plan_to_parquet};
 use crate::physical_plan::planner::DefaultPhysicalPlanner;
 use crate::physical_plan::udaf::AggregateUDF;
@@ -92,8 +93,6 @@ use crate::variable::{VarProvider, VarType};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use datafusion_common::ScalarValue;
-use datafusion_expr::logical_plan::DropView;
-use datafusion_expr::{TableSource, TableType};
 use datafusion_sql::{
     parser::DFParser,
     planner::{ContextProvider, SqlToRel},
@@ -1868,20 +1867,17 @@ impl FunctionRegistry for TaskContext {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::assert_batches_eq;
     use crate::execution::context::QueryPlanner;
+    use crate::physical_plan::expressions::AvgAccumulator;
     use crate::test;
     use crate::test_util::parquet_test_data;
     use crate::variable::VarType;
-    use crate::{
-        assert_batches_eq,
-        logical_plan::{create_udf, Expr},
-    };
-    use crate::{logical_plan::create_udaf, physical_plan::expressions::AvgAccumulator};
     use arrow::array::ArrayRef;
     use arrow::datatypes::*;
     use arrow::record_batch::RecordBatch;
     use async_trait::async_trait;
-    use datafusion_expr::Volatility;
+    use datafusion_expr::{create_udaf, create_udf, Expr, Volatility};
     use datafusion_physical_expr::functions::make_scalar_function;
     use std::fs::File;
     use std::sync::Weak;
@@ -2398,7 +2394,7 @@ mod tests {
         fn create_physical_expr(
             &self,
             _expr: &Expr,
-            _input_dfschema: &crate::logical_plan::DFSchema,
+            _input_dfschema: &crate::common::DFSchema,
             _input_schema: &Schema,
             _session_state: &SessionState,
         ) -> Result<Arc<dyn crate::physical_plan::PhysicalExpr>> {

datafusion/core/src/lib.rs

@@ -221,10 +221,6 @@ pub mod dataframe;
 pub mod datasource;
 pub mod error;
 pub mod execution;
-#[deprecated]
-// logical_plan module just contains re-exports and will be removed in a future release
-// https://github.com/apache/arrow-datafusion/issues/2683
-pub mod logical_plan;
 pub mod physical_optimizer;
 pub mod physical_plan;
 pub mod prelude;

datafusion/core/src/logical_plan/expr.rs

@@ -265,7 +265,7 @@ mod tests {
     use datafusion_physical_expr::PhysicalExpr;
 
     use crate::error::Result;
-    use crate::logical_plan::Operator;
+    use crate::logical_expr::Operator;
     use crate::physical_plan::aggregates::{AggregateExec, PhysicalGroupBy};
     use crate::physical_plan::coalesce_partitions::CoalescePartitionsExec;
     use crate::physical_plan::common;