Remove old interpreters

Signed-off-by: Xuanwo <github@xuanwo.io>

Author: Xuanwo

src/binaries/query/main.rs

@@ -25,7 +25,6 @@ use common_tracing::set_panic_hook;
 use databend_query::api::HttpService;
 use databend_query::api::RpcService;
 use databend_query::clusters::ClusterDiscovery;
-use databend_query::interpreters::AsyncInsertManager;
 use databend_query::metrics::MetricService;
 use databend_query::servers::HttpHandler;
 use databend_query::servers::HttpHandlerKind;
@@ -165,13 +164,6 @@ async fn main(_global_tracker: Arc<RuntimeTracker>) -> common_exception::Result<
         );
     }
 
-    // Async Insert Queue
-    {
-        let async_insert_queue = AsyncInsertManager::instance();
-        async_insert_queue.start().await;
-        info!("Databend async insert has been enabled.")
-    }
-
     // Print information to users.
     println!("Databend Query");
     println!();

src/query/service/benches/suites/mod.rs

@@ -14,16 +14,16 @@
 
 use common_base::base::tokio;
 use common_exception::Result;
-use common_legacy_planners::PlanNode;
 use criterion::Criterion;
 use databend_query::interpreters::Interpreter;
-use databend_query::interpreters::SelectInterpreter;
+use databend_query::interpreters::SelectInterpreterV2;
 use databend_query::sessions::SessionManager;
 use databend_query::sessions::SessionType;
-use databend_query::sql::PlanParser;
+use databend_query::sql::plans::Plan;
+use databend_query::sql::Planner;
 use databend_query::Config;
 use databend_query::GlobalServices;
-use futures::StreamExt;
+use futures_util::StreamExt;
 
 pub mod bench_aggregate_query_sql;
 pub mod bench_filter_query_sql;
@@ -35,9 +35,18 @@ pub async fn select_executor(sql: &str) -> Result<()> {
     let sessions = SessionManager::instance();
     let executor_session = sessions.create_session(SessionType::Dummy).await?;
     let ctx = executor_session.create_query_context().await?;
+    let mut planner = Planner::new(ctx.clone());
 
-    if let PlanNode::Select(plan) = PlanParser::parse(ctx.clone(), sql).await? {
-        let executor = SelectInterpreter::try_create(ctx.clone(), plan)?;
+    let (plan, _, _) = planner.plan_sql(sql).await?;
+    if let Plan::Query {
+        s_expr,
+        metadata,
+        bind_context,
+        ..
+    } = plan
+    {
+        let executor =
+            SelectInterpreterV2::try_create(ctx.clone(), *bind_context, *s_expr, metadata)?;
         let mut stream = executor.execute(ctx.clone()).await?;
         while let Some(_block) = stream.next().await {}
     } else {

src/query/service/src/api/rpc/exchange/exchange_manager.rs

@@ -52,7 +52,6 @@ use crate::interpreters::QueryFragmentsActions;
 use crate::pipelines::executor::ExecutorSettings;
 use crate::pipelines::executor::PipelineCompleteExecutor;
 use crate::pipelines::PipelineBuildResult;
-use crate::pipelines::QueryPipelineBuilder;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 use crate::sql::executor::PipelineBuilder as PipelineBuilderV2;
@@ -668,10 +667,6 @@ impl FragmentCoordinator {
             self.initialized = true;
 
             match &self.payload {
-                FragmentPayload::PlanV1(node) => {
-                    let pipeline_builder = QueryPipelineBuilder::create(ctx);
-                    self.pipeline_build_res = Some(pipeline_builder.finalize(node)?);
-                }
                 FragmentPayload::PlanV2(plan) => {
                     let pipeline_builder = PipelineBuilderV2::create(ctx);
                     self.pipeline_build_res = Some(pipeline_builder.finalize(plan)?);

src/query/service/src/api/rpc/packets/packet_fragment.rs

@@ -17,7 +17,6 @@ use std::fmt::Formatter;
 
 use common_datavalues::DataSchemaRef;
 use common_exception::Result;
-use common_legacy_planners::PlanNode;
 
 use crate::api::DataExchange;
 use crate::sql::executor::PhysicalPlan;
@@ -27,14 +26,12 @@ use crate::sql::executor::PhysicalPlan;
 #[allow(clippy::large_enum_variant)]
 #[derive(Clone, serde::Serialize, serde::Deserialize)]
 pub enum FragmentPayload {
-    PlanV1(PlanNode),
     PlanV2(PhysicalPlan),
 }
 
 impl Debug for FragmentPayload {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         match self {
-            FragmentPayload::PlanV1(plan) => write!(f, "PlanNode({})", plan.name()),
             FragmentPayload::PlanV2(plan) => write!(f, "PhysicalPlan({:?})", plan),
         }
     }
@@ -43,7 +40,6 @@ impl Debug for FragmentPayload {
 impl FragmentPayload {
     pub fn schema(&self) -> Result<DataSchemaRef> {
         match self {
-            FragmentPayload::PlanV1(node) => Ok(node.schema()),
             FragmentPayload::PlanV2(plan) => plan.output_schema(),
         }
     }

src/query/service/src/global_services.rs

@@ -31,7 +31,6 @@ use opendal::Operator;
 use crate::api::DataExchangeManager;
 use crate::catalogs::CatalogManagerHelper;
 use crate::clusters::ClusterDiscovery;
-use crate::interpreters::AsyncInsertManager;
 use crate::servers::http::v1::HttpQueryManager;
 use crate::sessions::SessionManager;
 
@@ -40,7 +39,6 @@ pub struct GlobalServices {
     query_logger: UnsafeCell<Option<Arc<QueryLogger>>>,
     cluster_discovery: UnsafeCell<Option<Arc<ClusterDiscovery>>>,
     storage_operator: UnsafeCell<Option<Operator>>,
-    async_insert_manager: UnsafeCell<Option<Arc<AsyncInsertManager>>>,
     cache_manager: UnsafeCell<Option<Arc<CacheManager>>>,
     catalog_manager: UnsafeCell<Option<Arc<CatalogManager>>>,
     http_query_manager: UnsafeCell<Option<Arc<HttpQueryManager>>>,
@@ -60,7 +58,6 @@ impl GlobalServices {
             query_logger: UnsafeCell::new(None),
             cluster_discovery: UnsafeCell::new(None),
             storage_operator: UnsafeCell::new(None),
-            async_insert_manager: UnsafeCell::new(None),
             cache_manager: UnsafeCell::new(None),
             catalog_manager: UnsafeCell::new(None),
             http_query_manager: UnsafeCell::new(None),
@@ -81,7 +78,6 @@ impl GlobalServices {
         ClusterDiscovery::init(config.clone(), global_services.clone()).await?;
 
         StorageOperator::init(&config.storage, global_services.clone()).await?;
-        AsyncInsertManager::init(&config, global_services.clone())?;
         CacheManager::init(&config.query, global_services.clone())?;
         CatalogManager::init(&config, global_services.clone()).await?;
         HttpQueryManager::init(&config, global_services.clone()).await?;
@@ -169,25 +165,6 @@ impl SingletonImpl<Operator> for GlobalServices {
     }
 }
 
-impl SingletonImpl<Arc<AsyncInsertManager>> for GlobalServices {
-    fn get(&self) -> Arc<AsyncInsertManager> {
-        unsafe {
-            match &*self.async_insert_manager.get() {
-                None => panic!("AsyncInsertManager is not init"),
-                Some(async_insert_manager) => async_insert_manager.clone(),
-            }
-        }
-    }
-
-    fn init(&self, value: Arc<AsyncInsertManager>) -> Result<()> {
-        unsafe {
-            *(self.async_insert_manager.get() as *mut Option<Arc<AsyncInsertManager>>) =
-                Some(value);
-            Ok(())
-        }
-    }
-}
-
 impl SingletonImpl<Arc<CacheManager>> for GlobalServices {
     fn get(&self) -> Arc<CacheManager> {
         unsafe {