refactor: query::query_engine::datafusion -> query::datafusion

src/query/src/datafusion.rs

@@ -7,9 +7,10 @@ use common_recordbatch::{EmptyRecordBatchStream, SendableRecordBatchStream};
 use snafu::{OptionExt, ResultExt};
 use sql::{dialect::GenericDialect, parser::ParserContext};
 
-use super::{context::QueryContext, state::QueryEngineState};
+use crate::query_engine::{QueryContext, QueryEngineState};
 use crate::{
     catalog::CatalogListRef,
+    datafusion::adapter::PhysicalPlanAdapter,
     error::Result,
     executor::QueryExecutor,
     logical_optimizer::LogicalOptimizer,
@@ -17,7 +18,6 @@ use crate::{
     physical_planner::PhysicalPlanner,
     plan::{LogicalPlan, PhysicalPlan},
     planner::{DfContextProviderAdapter, DfPlanner, Planner},
-    query_engine::datafusion::adapter::PhysicalPlanAdapter,
     Output, QueryEngine,
 };
 
@@ -177,7 +177,6 @@ mod tests {
 
         let plan = engine.sql_to_plan(sql).unwrap();
 
-        println!("{:?}", plan);
         assert_eq!(
             format!("{:?}", plan),
             r#"DfPlan(Limit: 20

src/query/src/datafusion/adapter.rs

@@ -16,10 +16,10 @@ use datatypes::schema::SchemaRef;
 use snafu::ResultExt;
 use table::table::adapter::{DfRecordBatchStreamAdapter, RecordBatchStreamAdapter};
 
+use crate::datafusion::error;
 use crate::error::Result;
 use crate::executor::Runtime;
 use crate::plan::{Partitioning, PhysicalPlan};
-use crate::query_engine::datafusion::error;
 
 /// Datafusion ExecutionPlan -> greptime PhysicalPlan
 pub struct PhysicalPlanAdapter {

src/query/src/lib.rs

@@ -1,5 +1,6 @@
 pub mod catalog;
 pub mod database;
+mod datafusion;
 pub mod error;
 pub mod executor;
 pub mod logical_optimizer;

src/query/src/planner.rs

@@ -13,9 +13,9 @@ use table::table::adapter::DfTableProviderAdapter;
 
 use crate::{
     catalog::{CatalogListRef, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME},
+    datafusion::error,
     error::Result,
     plan::LogicalPlan,
-    query_engine::datafusion::error,
 };
 
 pub trait Planner: Send + Sync {

src/query/src/query_engine.rs

@@ -1,16 +1,16 @@
 mod context;
-pub mod datafusion;
 mod state;
 
 use std::sync::Arc;
 
 use common_recordbatch::SendableRecordBatchStream;
-pub use context::QueryContext;
 
 use crate::catalog::CatalogList;
+use crate::datafusion::DatafusionQueryEngine;
 use crate::error::Result;
 use crate::plan::LogicalPlan;
-use crate::query_engine::datafusion::DatafusionQueryEngine;
+pub use crate::query_engine::context::QueryContext;
+pub use crate::query_engine::state::QueryEngineState;
 
 /// Sql output
 pub enum Output {

src/query/src/query_engine/state.rs

@@ -17,11 +17,14 @@ use table::{
 };
 
 use crate::catalog::{self, schema::SchemaProvider, CatalogListRef, CatalogProvider};
+use crate::datafusion::error;
 use crate::error::Result;
 use crate::executor::Runtime;
-use crate::query_engine::datafusion::error;
 
 /// Query engine global state
+// TODO(yingwen): This QueryEngineState still relies on datafusion, maybe we can define a trait for it,
+// which allows different implementation use different engine state. The state can also be an associated
+// type in QueryEngine trait.
 #[derive(Clone)]
 pub struct QueryEngineState {
     df_context: ExecutionContext,

src/table/src/table/adapter.rs

@@ -30,8 +30,8 @@ use datatypes::schema::{Schema, SchemaRef};
 use futures::Stream;
 use snafu::prelude::*;
 
-use super::{Table, TableProviderFilterPushDown, TableRef, TableType};
 use crate::error::{self, Result};
+use crate::table::{Table, TableProviderFilterPushDown, TableRef, TableType};
 
 /// Greptime SendableRecordBatchStream -> datafusion ExecutionPlan.
 struct ExecutionPlanAdapter {