tmp2

Signed-off-by: junxiang Mu <1948535941@qq.com>

s3select/query/src/data_source/data_source.rs

@@ -1,4 +1,5 @@
 use std::any::Any;
+use std::borrow::Cow;
 use std::fmt::Display;
 use std::sync::Arc;
 use std::write;
@@ -10,7 +11,6 @@ use datafusion::datasource::listing::ListingTable;
 use datafusion::datasource::{provider_as_source, TableProvider};
 use datafusion::error::DataFusionError;
 use datafusion::logical_expr::{LogicalPlan, LogicalPlanBuilder, TableProviderFilterPushDown, TableSource};
-use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::Expr;
 use datafusion::sql::TableReference;
 use tracing::debug;
@@ -28,11 +28,9 @@ pub struct TableSourceAdapter {
 impl TableSourceAdapter {
     pub fn try_new(
         table_ref: impl Into<TableReference>,
-        database_name: impl Into<String>,
         table_name: impl Into<String>,
         table_handle: impl Into<TableHandle>,
     ) -> Result<Self, DataFusionError> {
-        let database_name = database_name.into();
         let table_name: String = table_name.into();
 
         let table_handle = table_handle.into();
@@ -46,7 +44,7 @@ impl TableSourceAdapter {
             TableHandle::TableProvider(t) => {
                 let table_source = provider_as_source(t.clone());
                 if let Some(plan) = table_source.get_logical_plan() {
-                    LogicalPlanBuilder::from(plan.clone()).build()?
+                    LogicalPlanBuilder::from(plan.into_owned()).build()?
                 } else {
                     LogicalPlanBuilder::scan(table_ref, table_source, None)?.build()?
                 }
@@ -91,8 +89,8 @@ impl TableSource for TableSourceAdapter {
     }
 
     /// Called by [`InlineTableScan`]
-    fn get_logical_plan(&self) -> Option<&LogicalPlan> {
-        Some(&self.plan)
+    fn get_logical_plan(&self) -> Option<Cow<LogicalPlan>> {
+        Some(Cow::Owned(self.plan.clone()))
     }
 }
 

s3select/query/src/dispatcher/manager.rs

@@ -1,51 +1,62 @@
 use std::{
     collections::HashMap,
+    pin::Pin,
     sync::{Arc, Mutex},
+    task::{Context, Poll},
 };
 
 use api::{
     query::{
+        ast::ExtStatement,
         dispatcher::QueryDispatcher,
         execution::{Output, QueryStateMachine},
         function::FuncMetaManagerRef,
-        logical_planner::Plan,
+        logical_planner::{LogicalPlanner, Plan},
         parser::Parser,
+        session::{SessionCtx, SessionCtxFactory},
         Query,
     },
     QueryError, QueryResult,
 };
 use async_trait::async_trait;
+use datafusion::{
+    arrow::{datatypes::SchemaRef, record_batch::RecordBatch},
+    config::CsvOptions,
+    datasource::{
+        file_format::{csv::CsvFormat, json::JsonFormat, parquet::ParquetFormat},
+        listing::{ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl},
+    },
+    error::Result as DFResult,
+    execution::{RecordBatchStream, SendableRecordBatchStream},
+};
+use futures::{Stream, StreamExt};
+use s3s::dto::SelectObjectContentInput;
 use tokio::task::JoinHandle;
 
-use crate::metadata::TableHandleProviderRef;
+use crate::{
+    execution::factory::QueryExecutionFactoryRef,
+    metadata::{
+        base_table::BaseTableProvider,
+        ContextProviderExtension, MetadataProvider, TableHandleProviderRef,
+    },
+    sql::logical::planner::DefaultLogicalPlanner,
+};
 
 #[derive(Clone)]
 pub struct SimpleQueryDispatcher {
+    input: SelectObjectContentInput,
     // client for default tenant
     default_table_provider: TableHandleProviderRef,
-    // memory pool
-    // memory_pool: MemoryPoolRef,
-    // query tracker
+    session_factory: Arc<SessionCtxFactory>,
     // parser
     parser: Arc<dyn Parser + Send + Sync>,
     // get query execution factory
     query_execution_factory: QueryExecutionFactoryRef,
     func_manager: FuncMetaManagerRef,
-
-    async_task_joinhandle: Arc<Mutex<HashMap<String, JoinHandle<()>>>>,
-    failed_task_joinhandle: Arc<Mutex<HashMap<String, JoinHandle<()>>>>,
 }
 
 #[async_trait]
 impl QueryDispatcher for SimpleQueryDispatcher {
-    async fn start(&self) -> QueryResult<()> {
-        self.execute_persister_query(self.coord.node_id()).await
-    }
-
-    fn stop(&self) {
-        // TODO
-    }
-
     async fn execute_query(&self, query: &Query) -> QueryResult<Output> {
         let query_state_machine = { self.build_query_state_machine(query.clone()).await? };
 
@@ -66,7 +77,6 @@ impl QueryDispatcher for SimpleQueryDispatcher {
 
         let logical_planner = DefaultLogicalPlanner::new(&scheme_provider);
 
-        let span_recorder = session.get_child_span("parse sql");
         let statements = self.parser.parse(query.content())?;
 
         // not allow multi statement
@@ -82,8 +92,6 @@ impl QueryDispatcher for SimpleQueryDispatcher {
             None => return Ok(None),
         };
 
-        drop(span_recorder);
-
         let logical_plan = self
             .statement_to_logical_plan(stmt, &logical_planner, query_state_machine)
             .await?;
@@ -95,9 +103,7 @@ impl QueryDispatcher for SimpleQueryDispatcher {
     }
 
     async fn build_query_state_machine(&self, query: Query) -> QueryResult<Arc<QueryStateMachine>> {
-        let session = self
-            .session_factory
-            .create_session_ctx(query.context(), self.memory_pool.clone(), self.coord.clone())?;
+        let session = self.session_factory.create_session_ctx(query.context())?;
 
         let query_state_machine = Arc::new(QueryStateMachine::begin(query, session));
         Ok(query_state_machine)
@@ -114,7 +120,7 @@ impl SimpleQueryDispatcher {
         // begin analyze
         query_state_machine.begin_analyze();
         let logical_plan = logical_planner
-            .create_logical_plan(stmt, &query_state_machine.session, self.coord.get_config().query.auth_enabled)
+            .create_logical_plan(stmt, &query_state_machine.session)
             .await?;
         query_state_machine.end_analyze();
 
@@ -127,79 +133,85 @@ impl SimpleQueryDispatcher {
             .create_query_execution(logical_plan, query_state_machine.clone())
             .await?;
 
-        // TrackedQuery.drop() is called implicitly when the value goes out of scope,
-        self.query_tracker
-            .try_track_query(query_state_machine.query_id, execution)
-            .await?
-            .start()
-            .await
+        match execution.start().await {
+            Ok(Output::StreamData(stream)) => Ok(Output::StreamData(Box::pin(TrackedRecordBatchStream { inner: stream }))),
+            Ok(nil @ Output::Nil(_)) => Ok(nil),
+            Err(err) => Err(err),
+        }
     }
 
     async fn build_scheme_provider(&self, session: &SessionCtx) -> QueryResult<MetadataProvider> {
-        let meta_client = self.build_current_session_meta_client(session).await?;
-        let current_session_table_provider = self.build_table_handle_provider(meta_client.clone())?;
-        let metadata_provider = MetadataProvider::new(
-            self.coord.clone(),
-            meta_client,
-            current_session_table_provider,
-            self.default_table_provider.clone(),
-            self.func_manager.clone(),
-            self.query_tracker.clone(),
-            session.clone(),
-        );
+        let path = format!("s3://{}/{}", self.input.bucket, self.input.key);
+        let table_path = ListingTableUrl::parse(path)?;
+        let listing_options = if self.input.request.input_serialization.csv.is_some() {
+            let file_format = CsvFormat::default().with_options(CsvOptions::default().with_has_header(false));
+            ListingOptions::new(Arc::new(file_format)).with_file_extension(".csv")
+        } else if self.input.request.input_serialization.parquet.is_some() {
+            let file_format = ParquetFormat::new();
+            ListingOptions::new(Arc::new(file_format)).with_file_extension(".parquet")
+        } else if self.input.request.input_serialization.json.is_some() {
+            let file_format = JsonFormat::default();
+            ListingOptions::new(Arc::new(file_format)).with_file_extension(".json")
+        } else {
+            return Err(QueryError::NotImplemented {
+                err: "not support this file type".to_string(),
+            });
+        };
+
+        let resolve_schema = listing_options.infer_schema(session.inner(), &table_path).await?;
+        let config = ListingTableConfig::new(table_path)
+            .with_listing_options(listing_options)
+            .with_schema(resolve_schema);
+        let provider = Arc::new(ListingTable::try_new(config)?);
+        let current_session_table_provider = self.build_table_handle_provider()?;
+        let metadata_provider =
+            MetadataProvider::new(provider, current_session_table_provider, self.func_manager.clone(), session.clone());
 
         Ok(metadata_provider)
     }
 
-    async fn build_current_session_meta_client(&self, session: &SessionCtx) -> QueryResult<MetaClientRef> {
-        let meta_client = self
-            .coord
-            .tenant_meta(session.tenant())
-            .await
-            .ok_or_else(|| MetaError::TenantNotFound {
-                tenant: session.tenant().to_string(),
-            })
-            .context(MetaSnafu)?;
-
-        Ok(meta_client)
-    }
-
-    fn build_table_handle_provider(&self, meta_client: MetaClientRef) -> QueryResult<TableHandleProviderRef> {
-        let current_session_table_provider: Arc<BaseTableProvider> = Arc::new(BaseTableProvider::new(
-            self.coord.clone(),
-            self.split_manager.clone(),
-            meta_client,
-            self.stream_provider_manager.clone(),
-        ));
+    fn build_table_handle_provider(&self) -> QueryResult<TableHandleProviderRef> {
+        let current_session_table_provider: Arc<BaseTableProvider> = Arc::new(BaseTableProvider::default());
 
         Ok(current_session_table_provider)
     }
 }
 
+pub struct TrackedRecordBatchStream {
+    inner: SendableRecordBatchStream,
+}
+
+impl RecordBatchStream for TrackedRecordBatchStream {
+    fn schema(&self) -> SchemaRef {
+        self.inner.schema()
+    }
+}
+
+impl Stream for TrackedRecordBatchStream {
+    type Item = DFResult<RecordBatch>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.inner.poll_next_unpin(cx)
+    }
+}
+
 #[derive(Default, Clone)]
 pub struct SimpleQueryDispatcherBuilder {
-    coord: Option<CoordinatorRef>,
+    input: Option<SelectObjectContentInput>,
     default_table_provider: Option<TableHandleProviderRef>,
-    split_manager: Option<SplitManagerRef>,
     session_factory: Option<Arc<SessionCtxFactory>>,
     parser: Option<Arc<dyn Parser + Send + Sync>>,
 
     query_execution_factory: Option<QueryExecutionFactoryRef>,
-    query_tracker: Option<Arc<QueryTracker>>,
-    memory_pool: Option<MemoryPoolRef>, // memory
 
     func_manager: Option<FuncMetaManagerRef>,
-    stream_provider_manager: Option<StreamProviderManagerRef>,
-    span_ctx: Option<SpanContext>,
-    auth_cache: Option<Arc<AuthCache<AuthCacheKey, User>>>,
 }
 
 impl SimpleQueryDispatcherBuilder {
-    pub fn with_coord(mut self, coord: CoordinatorRef) -> Self {
-        self.coord = Some(coord);
+    pub fn with_input(mut self, input: SelectObjectContentInput) -> Self {
+        self.input = Some(input);
         self
     }
-
     pub fn with_default_table_provider(mut self, default_table_provider: TableHandleProviderRef) -> Self {
         self.default_table_provider = Some(default_table_provider);
         self
@@ -210,11 +222,6 @@ impl SimpleQueryDispatcherBuilder {
         self
     }
 
-    pub fn with_split_manager(mut self, split_manager: SplitManagerRef) -> Self {
-        self.split_manager = Some(split_manager);
-        self
-    }
-
     pub fn with_parser(mut self, parser: Arc<dyn Parser + Send + Sync>) -> Self {
         self.parser = Some(parser);
         self
@@ -225,43 +232,14 @@ impl SimpleQueryDispatcherBuilder {
         self
     }
 
-    pub fn with_query_tracker(mut self, query_tracker: Arc<QueryTracker>) -> Self {
-        self.query_tracker = Some(query_tracker);
-        self
-    }
-
-    pub fn with_memory_pool(mut self, memory_pool: MemoryPoolRef) -> Self {
-        self.memory_pool = Some(memory_pool);
-        self
-    }
-
     pub fn with_func_manager(mut self, func_manager: FuncMetaManagerRef) -> Self {
         self.func_manager = Some(func_manager);
         self
     }
 
-    pub fn with_stream_provider_manager(mut self, stream_provider_manager: StreamProviderManagerRef) -> Self {
-        self.stream_provider_manager = Some(stream_provider_manager);
-        self
-    }
-
-    pub fn with_span_ctx(mut self, span_ctx: Option<SpanContext>) -> Self {
-        self.span_ctx = span_ctx;
-        self
-    }
-
-    pub fn with_auth_cache(mut self, auth_cache: Arc<AuthCache<AuthCacheKey, User>>) -> Self {
-        self.auth_cache = Some(auth_cache);
-        self
-    }
-
     pub fn build(self) -> QueryResult<Arc<SimpleQueryDispatcher>> {
-        let coord = self.coord.ok_or_else(|| QueryError::BuildQueryDispatcher {
-            err: "lost of coord".to_string(),
-        })?;
-
-        let split_manager = self.split_manager.ok_or_else(|| QueryError::BuildQueryDispatcher {
-            err: "lost of split manager".to_string(),
+        let input = self.input.ok_or_else(|| QueryError::BuildQueryDispatcher {
+            err: "lost of input".to_string(),
         })?;
 
         let session_factory = self.session_factory.ok_or_else(|| QueryError::BuildQueryDispatcher {
@@ -276,56 +254,23 @@ impl SimpleQueryDispatcherBuilder {
             err: "lost of query_execution_factory".to_string(),
         })?;
 
-        let query_tracker = self.query_tracker.ok_or_else(|| QueryError::BuildQueryDispatcher {
-            err: "lost of query_tracker".to_string(),
-        })?;
-
         let func_manager = self.func_manager.ok_or_else(|| QueryError::BuildQueryDispatcher {
             err: "lost of func_manager".to_string(),
         })?;
 
-        let stream_provider_manager = self.stream_provider_manager.ok_or_else(|| QueryError::BuildQueryDispatcher {
-            err: "lost of stream_provider_manager".to_string(),
-        })?;
-
-        let memory_pool = self.memory_pool.ok_or_else(|| QueryError::BuildQueryDispatcher {
-            err: "lost of memory pool".to_string(),
-        })?;
-
         let default_table_provider = self.default_table_provider.ok_or_else(|| QueryError::BuildQueryDispatcher {
             err: "lost of default_table_provider".to_string(),
         })?;
 
-        let span_ctx = self.span_ctx;
-
-        let auth_cache = self.auth_cache.ok_or_else(|| QueryError::BuildQueryDispatcher {
-            err: "lost of auth_cache".to_string(),
-        })?;
-
         let dispatcher = Arc::new(SimpleQueryDispatcher {
-            coord,
+            input,
             default_table_provider,
-            split_manager,
             session_factory,
-            memory_pool,
             parser,
             query_execution_factory,
-            query_tracker,
             func_manager,
-            stream_provider_manager,
-            span_ctx,
-            async_task_joinhandle: Arc::new(Mutex::new(HashMap::new())),
-            failed_task_joinhandle: Arc::new(Mutex::new(HashMap::new())),
-            auth_cache,
         });
 
-        let meta_task_receiver = dispatcher
-            .coord
-            .meta_manager()
-            .take_resourceinfo_rx()
-            .expect("meta resource channel only has one consumer");
-        tokio::spawn(SimpleQueryDispatcher::recv_meta_modify(dispatcher.clone(), meta_task_receiver));
-
         Ok(dispatcher)
     }
 }

s3select/query/src/execution/factory.rs

@@ -1,42 +1,29 @@
 use std::sync::Arc;
 
-use api::query::execution::QueryExecutionFactory;
+use api::{
+    query::{
+        execution::{QueryExecutionFactory, QueryExecutionRef, QueryStateMachineRef},
+        logical_planner::Plan,
+        optimizer::Optimizer,
+        scheduler::SchedulerRef,
+    },
+    QueryError,
+};
+use async_trait::async_trait;
+
+use super::query::SqlQueryExecution;
 
 pub type QueryExecutionFactoryRef = Arc<dyn QueryExecutionFactory + Send + Sync>;
 
 pub struct SqlQueryExecutionFactory {
     optimizer: Arc<dyn Optimizer + Send + Sync>,
     scheduler: SchedulerRef,
-    query_tracker: Arc<QueryTracker>,
-    trigger_executor_factory: TriggerExecutorFactoryRef,
-    runtime: Arc<DedicatedExecutor>,
-    stream_checker_manager: StreamCheckerManagerRef,
 }
 
 impl SqlQueryExecutionFactory {
     #[inline(always)]
-    pub fn new(
-        optimizer: Arc<dyn Optimizer + Send + Sync>,
-        scheduler: SchedulerRef,
-        query_tracker: Arc<QueryTracker>,
-        stream_checker_manager: StreamCheckerManagerRef,
-        config: Arc<QueryOptions>,
-    ) -> Self {
-        // Only do periodic scheduling, no need for many threads
-        let trigger_executor_runtime = DedicatedExecutor::new("stream-trigger", config.stream_trigger_cpu);
-        let trigger_executor_factory = Arc::new(TriggerExecutorFactory::new(Arc::new(trigger_executor_runtime)));
-
-        // perform stream-related preparations, not actual operator execution
-        let runtime = Arc::new(DedicatedExecutor::new("stream-executor", config.stream_executor_cpu));
-
-        Self {
-            optimizer,
-            scheduler,
-            query_tracker,
-            trigger_executor_factory,
-            runtime,
-            stream_checker_manager,
-        }
+    pub fn new(optimizer: Arc<dyn Optimizer + Send + Sync>, scheduler: SchedulerRef) -> Self {
+        Self { optimizer, scheduler }
     }
 }
 
@@ -48,62 +35,12 @@ impl QueryExecutionFactory for SqlQueryExecutionFactory {
         state_machine: QueryStateMachineRef,
     ) -> Result<QueryExecutionRef, QueryError> {
         match plan {
-            Plan::Query(query_plan) => {
-                // 获取执行计划中所有涉及到的stream source
-                let stream_providers = extract_stream_providers(&query_plan);
-
-                // (含有流表, explain, dml)
-                match (!stream_providers.is_empty(), query_plan.is_explain(), is_dml(&query_plan)) {
-                    (false, _, _) | (true, true, _) => Ok(Arc::new(SqlQueryExecution::new(
-                        state_machine,
-                        query_plan,
-                        self.optimizer.clone(),
-                        self.scheduler.clone(),
-                    ))),
-                    (true, false, true) => {
-                        // 流操作
-                        // stream source + dml + !explain
-                        let options = state_machine.session.inner().config().into();
-                        let exec = MicroBatchStreamExecutionBuilder::new(MicroBatchStreamExecutionDesc {
-                            plan: Arc::new(query_plan),
-                            options,
-                        })
-                        .with_stream_providers(stream_providers)
-                        .build(
-                            state_machine,
-                            self.scheduler.clone(),
-                            self.trigger_executor_factory.clone(),
-                            self.runtime.clone(),
-                        )
-                        .await?;
-
-                        Ok(Arc::new(exec))
-                    }
-                    (true, false, false) => {
-                        // stream source + !dml + !explain
-                        Err(QueryError::NotImplemented {
-                            err: "Stream table can only be used as source table in insert select statements.".to_string(),
-                        })
-                    }
-                }
-            }
-            Plan::DDL(ddl_plan) => Ok(Arc::new(DDLExecution::new(state_machine, self.stream_checker_manager.clone(), ddl_plan))),
-            Plan::DML(dml_plan) => Ok(Arc::new(DMLExecution::new(state_machine, dml_plan))),
-            Plan::SYSTEM(sys_plan) => Ok(Arc::new(SystemExecution::new(state_machine, sys_plan, self.query_tracker.clone()))),
+            Plan::Query(query_plan) => Ok(Arc::new(SqlQueryExecution::new(
+                state_machine,
+                query_plan,
+                self.optimizer.clone(),
+                self.scheduler.clone(),
+            ))),
         }
     }
 }
-
-fn is_dml(query_plan: &QueryPlan) -> bool {
-    match &query_plan.df_plan {
-        LogicalPlan::Dml(_) => true,
-        LogicalPlan::Extension(Extension { node }) => downcast_plan_node::<TableWriterMergePlanNode>(node.as_ref()).is_some(),
-        _ => false,
-    }
-}
-
-impl Drop for SqlQueryExecutionFactory {
-    fn drop(&mut self) {
-        self.runtime.shutdown();
-    }
-}

s3select/query/src/execution/mod.rs

@@ -1 +1,3 @@
 pub mod factory;
+pub mod query;
+pub mod scheduler;

s3select/query/src/execution/query.rs

@@ -0,0 +1,92 @@
+use std::sync::Arc;
+
+use api::query::execution::{Output, QueryExecution, QueryStateMachineRef};
+use api::query::logical_planner::QueryPlan;
+use api::query::optimizer::Optimizer;
+use api::query::scheduler::SchedulerRef;
+use api::{QueryError, QueryResult};
+use async_trait::async_trait;
+use futures::stream::AbortHandle;
+use parking_lot::Mutex;
+use tracing::debug;
+
+pub struct SqlQueryExecution {
+    query_state_machine: QueryStateMachineRef,
+    plan: QueryPlan,
+    optimizer: Arc<dyn Optimizer + Send + Sync>,
+    scheduler: SchedulerRef,
+
+    abort_handle: Mutex<Option<AbortHandle>>,
+}
+
+impl SqlQueryExecution {
+    pub fn new(
+        query_state_machine: QueryStateMachineRef,
+        plan: QueryPlan,
+        optimizer: Arc<dyn Optimizer + Send + Sync>,
+        scheduler: SchedulerRef,
+    ) -> Self {
+        Self {
+            query_state_machine,
+            plan,
+            optimizer,
+            scheduler,
+            abort_handle: Mutex::new(None),
+        }
+    }
+
+    async fn start(&self) -> QueryResult<Output> {
+        // begin optimize
+        self.query_state_machine.begin_optimize();
+        let physical_plan = self.optimizer.optimize(&self.plan, &self.query_state_machine.session).await?;
+        self.query_state_machine.end_optimize();
+
+        // begin schedule
+        self.query_state_machine.begin_schedule();
+        let stream = self
+            .scheduler
+            .schedule(physical_plan.clone(), self.query_state_machine.session.inner().task_ctx())
+            .await?
+            .stream();
+
+        debug!("Success build result stream.");
+        self.query_state_machine.end_schedule();
+
+        Ok(Output::StreamData(stream))
+    }
+}
+
+#[async_trait]
+impl QueryExecution for SqlQueryExecution {
+    async fn start(&self) -> QueryResult<Output> {
+        let (task, abort_handle) = futures::future::abortable(self.start());
+
+        {
+            *self.abort_handle.lock() = Some(abort_handle);
+        }
+
+        task.await.map_err(|_| QueryError::Cancel)?
+    }
+
+    fn cancel(&self) -> QueryResult<()> {
+        debug!(
+            "cancel sql query execution: sql: {}, state: {:?}",
+            self.query_state_machine.query.content(),
+            self.query_state_machine.state()
+        );
+
+        // change state
+        self.query_state_machine.cancel();
+        // stop future task
+        if let Some(e) = self.abort_handle.lock().as_ref() {
+            e.abort()
+        };
+
+        debug!(
+            "canceled sql query execution: sql: {}, state: {:?}",
+            self.query_state_machine.query.content(),
+            self.query_state_machine.state()
+        );
+        Ok(())
+    }
+}

s3select/query/src/execution/scheduler/local.rs

@@ -0,0 +1,22 @@
+use std::sync::Arc;
+
+use api::query::scheduler::{ExecutionResults, Scheduler};
+use async_trait::async_trait;
+use datafusion::error::DataFusionError;
+use datafusion::execution::context::TaskContext;
+use datafusion::physical_plan::{execute_stream, ExecutionPlan};
+
+pub struct LocalScheduler {}
+
+#[async_trait]
+impl Scheduler for LocalScheduler {
+    async fn schedule(
+        &self,
+        plan: Arc<dyn ExecutionPlan>,
+        context: Arc<TaskContext>,
+    ) -> Result<ExecutionResults, DataFusionError> {
+        let stream = execute_stream(plan, context)?;
+
+        Ok(ExecutionResults::new(stream))
+    }
+}

s3select/query/src/execution/scheduler/mod.rs

@@ -0,0 +1 @@
+pub mod local;

s3select/query/src/instance.rs

@@ -1,12 +1,22 @@
 use std::sync::Arc;
 
 use api::{
-    query::{dispatcher::QueryDispatcher, execution::QueryStateMachineRef, logical_planner::Plan, Query},
+    query::{
+        dispatcher::QueryDispatcher, execution::QueryStateMachineRef, logical_planner::Plan, session::SessionCtxFactory, Query,
+    },
     server::dbms::{DatabaseManagerSystem, QueryHandle},
     QueryResult,
 };
 use async_trait::async_trait;
 use derive_builder::Builder;
+use s3s::dto::SelectObjectContentInput;
+
+use crate::{
+    dispatcher::manager::SimpleQueryDispatcherBuilder,
+    execution::{factory::SqlQueryExecutionFactory, scheduler::local::LocalScheduler},
+    metadata::base_table::BaseTableProvider,
+    sql::{optimizer::CascadeOptimizerBuilder, parser::DefaultParser},
+};
 
 #[derive(Builder)]
 pub struct RustFSms<D: QueryDispatcher> {
@@ -19,10 +29,6 @@ impl<D> DatabaseManagerSystem for RustFSms<D>
 where
     D: QueryDispatcher,
 {
-    async fn start(&self) -> QueryResult<()> {
-        self.query_dispatcher.start().await
-    }
-
     async fn execute(&self, query: &Query) -> QueryResult<QueryHandle> {
         let result = self.query_dispatcher.execute_query(query).await?;
 
@@ -54,15 +60,31 @@ where
 
         Ok(QueryHandle::new(query.clone(), result))
     }
-
-    fn metrics(&self) -> String {
-        let infos = self.query_dispatcher.running_query_infos();
-        let status = self.query_dispatcher.running_query_status();
-
-        format!(
-            "infos: {}\nstatus: {}\n",
-            infos.iter().map(|e| format!("{:?}", e)).collect::<Vec<_>>().join(","),
-            status.iter().map(|e| format!("{:?}", e)).collect::<Vec<_>>().join(",")
-        )
-    }
+}
+
+pub async fn make_cnosdbms(input: SelectObjectContentInput) -> QueryResult<impl DatabaseManagerSystem> {
+    // TODO session config need load global system config
+    let session_factory = Arc::new(SessionCtxFactory {});
+    let parser = Arc::new(DefaultParser::default());
+    let optimizer = Arc::new(CascadeOptimizerBuilder::default().build());
+    // TODO wrap, and num_threads configurable
+    let scheduler = Arc::new(LocalScheduler {});
+
+    let query_execution_factory = Arc::new(SqlQueryExecutionFactory::new(optimizer, scheduler));
+
+    let default_table_provider = Arc::new(BaseTableProvider::default());
+
+    let query_dispatcher = SimpleQueryDispatcherBuilder::default()
+        .with_input(input)
+        .with_default_table_provider(default_table_provider)
+        .with_session_factory(session_factory)
+        .with_parser(parser)
+        .with_query_execution_factory(query_execution_factory)
+        .build()?;
+
+    let mut builder = RustFSmsBuilder::default();
+
+    let db_server = builder.query_dispatcher(query_dispatcher).build().expect("build db server");
+
+    Ok(db_server)
 }

s3select/query/src/metadata/base_table.rs

@@ -1,63 +1,17 @@
 use std::sync::Arc;
 
 use datafusion::common::Result as DFResult;
-use datafusion::config::{CsvOptions, JsonOptions};
-use datafusion::datasource::file_format::csv::CsvFormat;
-use datafusion::datasource::file_format::json::JsonFormat;
-use datafusion::datasource::file_format::parquet::ParquetFormat;
-use datafusion::datasource::listing::{ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl};
-use datafusion::error::DataFusionError;
-use datafusion::execution::SessionState;
+use datafusion::datasource::listing::ListingTable;
 
 use crate::data_source::data_source::TableHandle;
 
 use super::TableHandleProvider;
 
-pub enum FileType {
-    Csv,
-    Parquet,
-    Json,
-    Unknown,
-}
-
 #[derive(Default)]
-pub struct BaseTableProvider {
-    file_type: FileType,
-}
-
-impl BaseTableProvider {
-    pub fn new(file_type: FileType) -> Self {
-        Self { file_type }
-    }
-}
+pub struct BaseTableProvider {}
 
 impl TableHandleProvider for BaseTableProvider {
-    async fn build_table_handle(&self, session_state: &SessionState, table_name: &str) -> DFResult<TableHandle> {
-        let table_path = ListingTableUrl::parse(table_name)?;
-        let listing_options = match self.file_type {
-            FileType::Csv => {
-                let file_format = CsvFormat::default().with_options(CsvOptions::default().with_has_header(false));
-                ListingOptions::new(Arc::new(file_format)).with_file_extension(".csv")
-            }
-            FileType::Parquet => {
-                let file_format = ParquetFormat::new();
-                ListingOptions::new(Arc::new(file_format)).with_file_extension(".parquet")
-            }
-            FileType::Json => {
-                let file_format = JsonFormat::default();
-                ListingOptions::new(Arc::new(file_format)).with_file_extension(".json")
-            }
-            FileType::Unknown => {
-                return Err(DataFusionError::NotImplemented("not support this file type".to_string()));
-            }
-        };
-
-        let resolve_schema = listing_options.infer_schema(session_state, &table_path).await?;
-        let config = ListingTableConfig::new(table_path)
-            .with_listing_options(listing_options)
-            .with_schema(resolve_schema);
-        let provider = Arc::new(ListingTable::try_new(config)?);
-
+    fn build_table_handle(&self, provider: Arc<ListingTable>) -> DFResult<TableHandle> {
         Ok(TableHandle::External(provider))
     }
 }

s3select/query/src/metadata/mod.rs

@@ -5,10 +5,12 @@ use api::ResolvedTable;
 use async_trait::async_trait;
 use datafusion::arrow::datatypes::DataType;
 use datafusion::common::Result as DFResult;
+use datafusion::datasource::listing::ListingTable;
 use datafusion::error::DataFusionError;
 use datafusion::execution::SessionState;
 use datafusion::logical_expr::var_provider::is_system_variables;
 use datafusion::logical_expr::{AggregateUDF, ScalarUDF, TableSource, WindowUDF};
+use datafusion::sql::ResolvedTableReference;
 use datafusion::variable::VarType;
 use datafusion::{
     config::ConfigOptions,
@@ -21,16 +23,17 @@ pub mod base_table;
 
 #[async_trait]
 pub trait ContextProviderExtension: ContextProvider {
-    fn get_table_source(&self, name: TableReference) -> datafusion::common::Result<Arc<TableSourceAdapter>>;
+    fn get_table_source_(&self, name: TableReference) -> datafusion::common::Result<Arc<TableSourceAdapter>>;
 }
 
 pub type TableHandleProviderRef = Arc<dyn TableHandleProvider + Send + Sync>;
 
 pub trait TableHandleProvider {
-    fn build_table_handle(&self, session_state: &SessionState, table_name: &str) -> DFResult<TableHandle>;
+    fn build_table_handle(&self, provider: Arc<ListingTable>) -> DFResult<TableHandle>;
 }
 
 pub struct MetadataProvider {
+    provider: Arc<ListingTable>,
     session: SessionCtx,
     config_options: ConfigOptions,
     func_manager: FuncMetaManagerRef,
@@ -40,11 +43,13 @@ pub struct MetadataProvider {
 impl MetadataProvider {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
+        provider: Arc<ListingTable>,
         current_session_table_provider: TableHandleProviderRef,
         func_manager: FuncMetaManagerRef,
         session: SessionCtx,
     ) -> Self {
         Self {
+            provider,
             current_session_table_provider,
             config_options: session.inner().config_options().clone(),
             session,
@@ -52,43 +57,22 @@ impl MetadataProvider {
         }
     }
 
-    fn build_table_handle(&self, name: &ResolvedTable) -> datafusion::common::Result<TableHandle> {
-        let table_name = name.table();
-
-        self.current_session_table_provider.build_table_handle(table_name)
+    fn build_table_handle(&self) -> datafusion::common::Result<TableHandle> {
+        self.current_session_table_provider
+            .build_table_handle(self.provider.clone())
     }
+
+    async fn init(&self) {}
 }
 
-#[async_trait::async_trait]
 impl ContextProviderExtension for MetadataProvider {
-    fn get_table_source(&self, table_ref: TableReference) -> datafusion::common::Result<Arc<TableSourceAdapter>> {
-        let name = table_ref
-            .clone()
-            .resolve_object(self.session.tenant(), self.session.default_database())?;
+    fn get_table_source_(&self, table_ref: TableReference) -> datafusion::common::Result<Arc<TableSourceAdapter>> {
+        let name = table_ref.clone().resolve("", "");
+        let table_name = &*name.table;
 
-        let table_name = name.table();
-        let database_name = name.database();
-        let tenant_name = name.tenant();
+        let table_handle = self.build_table_handle()?;
 
-        // Cannot query across tenants
-        if self.session.tenant() != tenant_name {
-            return Err(DataFusionError::Plan(format!(
-                "Tenant conflict, the current connection's tenant is {}",
-                self.session.tenant()
-            )));
-        }
-
-        // save access table
-        self.access_databases.write().push_table(database_name, table_name);
-
-        let table_handle = self.build_table_handle(&name)?;
-
-        Ok(Arc::new(TableSourceAdapter::try_new(
-            table_ref.to_owned_reference(),
-            database_name,
-            table_name,
-            table_handle,
-        )?))
+        Ok(Arc::new(TableSourceAdapter::try_new(table_ref.clone(), table_name, table_handle)?))
     }
 }
 
@@ -132,7 +116,7 @@ impl ContextProvider for MetadataProvider {
     }
 
     fn get_table_source(&self, name: TableReference) -> DFResult<Arc<dyn TableSource>> {
-        Ok(self.get_table_source(name)?)
+        Ok(self.get_table_source_(name)?)
     }
 
     fn udf_names(&self) -> Vec<String> {

s3select/query/src/sql/analyzer.rs

@@ -0,0 +1,33 @@
+use api::query::analyzer::Analyzer;
+use api::query::session::SessionCtx;
+use api::QueryResult;
+use datafusion::logical_expr::LogicalPlan;
+use datafusion::optimizer::analyzer::Analyzer as DFAnalyzer;
+
+pub struct DefaultAnalyzer {
+    inner: DFAnalyzer,
+}
+
+impl DefaultAnalyzer {
+    pub fn new() -> Self {
+        let analyzer = DFAnalyzer::default();
+        // we can add analyzer rule at here
+
+        Self { inner: analyzer }
+    }
+}
+
+impl Default for DefaultAnalyzer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Analyzer for DefaultAnalyzer {
+    fn analyze(&self, plan: &LogicalPlan, session: &SessionCtx) -> QueryResult<LogicalPlan> {
+        let plan = self
+            .inner
+            .execute_and_check(plan.to_owned(), session.inner().config_options(), |_, _| {})?;
+        Ok(plan)
+    }
+}

s3select/query/src/sql/dialect.rs

@@ -0,0 +1,18 @@
+use datafusion::sql::sqlparser::dialect::Dialect;
+
+#[derive(Debug, Default)]
+pub struct RustFsDialect;
+
+impl Dialect for RustFsDialect {
+    fn is_identifier_start(&self, ch: char) -> bool {
+        ch.is_alphabetic() || ch == '_' || ch == '#' || ch == '@'
+    }
+
+    fn is_identifier_part(&self, ch: char) -> bool {
+        ch.is_alphabetic() || ch.is_ascii_digit() || ch == '@' || ch == '$' || ch == '#' || ch == '_'
+    }
+
+    fn supports_group_by_expr(&self) -> bool {
+        true
+    }
+}

s3select/query/src/sql/logical/mod.rs

@@ -1 +1,2 @@
+pub mod optimizer;
 pub mod planner;

s3select/query/src/sql/logical/optimizer.rs

@@ -0,0 +1,115 @@
+use std::sync::Arc;
+
+use api::{
+    query::{analyzer::AnalyzerRef, logical_planner::QueryPlan, session::SessionCtx},
+    QueryResult,
+};
+use datafusion::{
+    execution::SessionStateBuilder,
+    logical_expr::LogicalPlan,
+    optimizer::{
+        common_subexpr_eliminate::CommonSubexprEliminate, decorrelate_predicate_subquery::DecorrelatePredicateSubquery,
+        eliminate_cross_join::EliminateCrossJoin, eliminate_duplicated_expr::EliminateDuplicatedExpr,
+        eliminate_filter::EliminateFilter, eliminate_join::EliminateJoin, eliminate_limit::EliminateLimit,
+        eliminate_outer_join::EliminateOuterJoin, extract_equijoin_predicate::ExtractEquijoinPredicate,
+        filter_null_join_keys::FilterNullJoinKeys, propagate_empty_relation::PropagateEmptyRelation,
+        push_down_filter::PushDownFilter, push_down_limit::PushDownLimit,
+        replace_distinct_aggregate::ReplaceDistinctWithAggregate, scalar_subquery_to_join::ScalarSubqueryToJoin,
+        simplify_expressions::SimplifyExpressions, single_distinct_to_groupby::SingleDistinctToGroupBy,
+        unwrap_cast_in_comparison::UnwrapCastInComparison, OptimizerRule,
+    },
+};
+use tracing::debug;
+
+use crate::sql::analyzer::DefaultAnalyzer;
+
+const PUSH_DOWN_PROJECTION_INDEX: usize = 24;
+
+pub trait LogicalOptimizer: Send + Sync {
+    fn optimize(&self, plan: &QueryPlan, session: &SessionCtx) -> QueryResult<LogicalPlan>;
+
+    fn inject_optimizer_rule(&mut self, optimizer_rule: Arc<dyn OptimizerRule + Send + Sync>);
+}
+
+pub struct DefaultLogicalOptimizer {
+    // fit datafusion
+    // TODO refactor
+    analyzer: AnalyzerRef,
+    rules: Vec<Arc<dyn OptimizerRule + Send + Sync>>,
+}
+
+impl DefaultLogicalOptimizer {
+    #[allow(dead_code)]
+    fn with_optimizer_rules(mut self, rules: Vec<Arc<dyn OptimizerRule + Send + Sync>>) -> Self {
+        self.rules = rules;
+        self
+    }
+}
+
+impl Default for DefaultLogicalOptimizer {
+    fn default() -> Self {
+        let analyzer = Arc::new(DefaultAnalyzer::default());
+
+        // additional optimizer rule
+        let rules: Vec<Arc<dyn OptimizerRule + Send + Sync>> = vec![
+            // df default rules start
+            Arc::new(SimplifyExpressions::new()),
+            Arc::new(UnwrapCastInComparison::new()),
+            Arc::new(ReplaceDistinctWithAggregate::new()),
+            Arc::new(EliminateJoin::new()),
+            Arc::new(DecorrelatePredicateSubquery::new()),
+            Arc::new(ScalarSubqueryToJoin::new()),
+            Arc::new(ExtractEquijoinPredicate::new()),
+            // simplify expressions does not simplify expressions in subqueries, so we
+            // run it again after running the optimizations that potentially converted
+            // subqueries to joins
+            Arc::new(SimplifyExpressions::new()),
+            Arc::new(EliminateDuplicatedExpr::new()),
+            Arc::new(EliminateFilter::new()),
+            Arc::new(EliminateCrossJoin::new()),
+            Arc::new(CommonSubexprEliminate::new()),
+            Arc::new(EliminateLimit::new()),
+            Arc::new(PropagateEmptyRelation::new()),
+            Arc::new(FilterNullJoinKeys::default()),
+            Arc::new(EliminateOuterJoin::new()),
+            // Filters can't be pushed down past Limits, we should do PushDownFilter after PushDownLimit
+            Arc::new(PushDownLimit::new()),
+            Arc::new(PushDownFilter::new()),
+            Arc::new(SingleDistinctToGroupBy::new()),
+            // The previous optimizations added expressions and projections,
+            // that might benefit from the following rules
+            Arc::new(SimplifyExpressions::new()),
+            Arc::new(UnwrapCastInComparison::new()),
+            Arc::new(CommonSubexprEliminate::new()),
+            // PushDownProjection can pushdown Projections through Limits, do PushDownLimit again.
+            Arc::new(PushDownLimit::new()),
+            // df default rules end
+            // custom rules can add at here
+        ];
+
+        Self { analyzer, rules }
+    }
+}
+
+impl LogicalOptimizer for DefaultLogicalOptimizer {
+    fn optimize(&self, plan: &QueryPlan, session: &SessionCtx) -> QueryResult<LogicalPlan> {
+        let analyzed_plan = { self.analyzer.analyze(&plan.df_plan, session).map(|p| p).map_err(|e| e)? };
+
+        debug!("Analyzed logical plan:\n{}\n", plan.df_plan.display_indent_schema(),);
+
+        let optimizeed_plan = {
+            SessionStateBuilder::new_from_existing(session.inner().clone())
+                .with_optimizer_rules(self.rules.clone())
+                .build()
+                .optimize(&analyzed_plan)
+                .map(|p| p)
+                .map_err(|e| e)?
+        };
+
+        Ok(optimizeed_plan)
+    }
+
+    fn inject_optimizer_rule(&mut self, optimizer_rule: Arc<dyn OptimizerRule + Send + Sync>) {
+        self.rules.push(optimizer_rule);
+    }
+}

s3select/query/src/sql/mod.rs

@@ -1,3 +1,7 @@
+pub mod analyzer;
+pub mod dialect;
 pub mod logical;
+pub mod optimizer;
+pub mod parser;
 pub mod physical;
 pub mod planner;

s3select/query/src/sql/optimizer.rs

@@ -0,0 +1,89 @@
+use std::sync::Arc;
+
+use api::{
+    query::{logical_planner::QueryPlan, optimizer::Optimizer, physical_planner::PhysicalPlanner, session::SessionCtx},
+    QueryResult,
+};
+use async_trait::async_trait;
+use datafusion::physical_plan::{displayable, ExecutionPlan};
+use tracing::debug;
+
+use super::{
+    logical::optimizer::{DefaultLogicalOptimizer, LogicalOptimizer},
+    physical::{optimizer::PhysicalOptimizer, planner::DefaultPhysicalPlanner},
+};
+
+pub struct CascadeOptimizer {
+    logical_optimizer: Arc<dyn LogicalOptimizer + Send + Sync>,
+    physical_planner: Arc<dyn PhysicalPlanner + Send + Sync>,
+    physical_optimizer: Arc<dyn PhysicalOptimizer + Send + Sync>,
+}
+
+#[async_trait]
+impl Optimizer for CascadeOptimizer {
+    async fn optimize(&self, plan: &QueryPlan, session: &SessionCtx) -> QueryResult<Arc<dyn ExecutionPlan>> {
+        debug!("Original logical plan:\n{}\n", plan.df_plan.display_indent_schema(),);
+
+        let optimized_logical_plan = self.logical_optimizer.optimize(plan, session)?;
+
+        debug!("Final logical plan:\n{}\n", optimized_logical_plan.display_indent_schema(),);
+
+        let physical_plan = {
+            self.physical_planner
+                .create_physical_plan(&optimized_logical_plan, session)
+                .await
+                .map(|p| p)
+                .map_err(|err| err)?
+        };
+
+        debug!("Original physical plan:\n{}\n", displayable(physical_plan.as_ref()).indent(false));
+
+        let optimized_physical_plan = {
+            self.physical_optimizer
+                .optimize(physical_plan, session)
+                .map(|p| p)
+                .map_err(|err| err)?
+        };
+
+        Ok(optimized_physical_plan)
+    }
+}
+
+#[derive(Default)]
+pub struct CascadeOptimizerBuilder {
+    logical_optimizer: Option<Arc<dyn LogicalOptimizer + Send + Sync>>,
+    physical_planner: Option<Arc<dyn PhysicalPlanner + Send + Sync>>,
+    physical_optimizer: Option<Arc<dyn PhysicalOptimizer + Send + Sync>>,
+}
+
+impl CascadeOptimizerBuilder {
+    pub fn with_logical_optimizer(mut self, logical_optimizer: Arc<dyn LogicalOptimizer + Send + Sync>) -> Self {
+        self.logical_optimizer = Some(logical_optimizer);
+        self
+    }
+
+    pub fn with_physical_planner(mut self, physical_planner: Arc<dyn PhysicalPlanner + Send + Sync>) -> Self {
+        self.physical_planner = Some(physical_planner);
+        self
+    }
+
+    pub fn with_physical_optimizer(mut self, physical_optimizer: Arc<dyn PhysicalOptimizer + Send + Sync>) -> Self {
+        self.physical_optimizer = Some(physical_optimizer);
+        self
+    }
+
+    pub fn build(self) -> CascadeOptimizer {
+        let default_logical_optimizer = Arc::new(DefaultLogicalOptimizer::default());
+        let default_physical_planner = Arc::new(DefaultPhysicalPlanner::default());
+
+        let logical_optimizer = self.logical_optimizer.unwrap_or(default_logical_optimizer);
+        let physical_planner = self.physical_planner.unwrap_or_else(|| default_physical_planner.clone());
+        let physical_optimizer = self.physical_optimizer.unwrap_or(default_physical_planner);
+
+        CascadeOptimizer {
+            logical_optimizer,
+            physical_planner,
+            physical_optimizer,
+        }
+    }
+}

s3select/query/src/sql/parser.rs

@@ -0,0 +1,97 @@
+use std::{collections::VecDeque, fmt::Display};
+
+use api::{
+    query::{ast::ExtStatement, parser::Parser as RustFsParser},
+    ParserSnafu,
+};
+use datafusion::sql::sqlparser::{
+    dialect::Dialect,
+    parser::{Parser, ParserError},
+    tokenizer::{Token, Tokenizer},
+};
+use snafu::ResultExt;
+
+use super::dialect::RustFsDialect;
+
+pub type Result<T, E = ParserError> = std::result::Result<T, E>;
+
+// Use `Parser::expected` instead, if possible
+macro_rules! parser_err {
+    ($MSG:expr) => {
+        Err(ParserError::ParserError($MSG.to_string()))
+    };
+}
+
+#[derive(Default)]
+pub struct DefaultParser {}
+
+impl RustFsParser for DefaultParser {
+    fn parse(&self, sql: &str) -> api::QueryResult<VecDeque<ExtStatement>> {
+        ExtParser::parse_sql(sql).context(ParserSnafu)
+    }
+}
+
+/// SQL Parser
+pub struct ExtParser<'a> {
+    parser: Parser<'a>,
+}
+
+impl<'a> ExtParser<'a> {
+    /// Parse the specified tokens with dialect
+    fn new_with_dialect(sql: &str, dialect: &'a dyn Dialect) -> Result<Self> {
+        let mut tokenizer = Tokenizer::new(dialect, sql);
+        let tokens = tokenizer.tokenize()?;
+        Ok(ExtParser {
+            parser: Parser::new(dialect).with_tokens(tokens),
+        })
+    }
+
+    /// Parse a SQL statement and produce a set of statements
+    pub fn parse_sql(sql: &str) -> Result<VecDeque<ExtStatement>> {
+        let dialect = &RustFsDialect {};
+        ExtParser::parse_sql_with_dialect(sql, dialect)
+    }
+
+    /// Parse a SQL statement and produce a set of statements
+    pub fn parse_sql_with_dialect(sql: &str, dialect: &dyn Dialect) -> Result<VecDeque<ExtStatement>> {
+        let mut parser = ExtParser::new_with_dialect(sql, dialect)?;
+        let mut stmts = VecDeque::new();
+        let mut expecting_statement_delimiter = false;
+        loop {
+            // ignore empty statements (between successive statement delimiters)
+            while parser.parser.consume_token(&Token::SemiColon) {
+                expecting_statement_delimiter = false;
+            }
+
+            if parser.parser.peek_token() == Token::EOF {
+                break;
+            }
+            if expecting_statement_delimiter {
+                return parser.expected("end of statement", parser.parser.peek_token());
+            }
+
+            let statement = parser.parse_statement()?;
+            stmts.push_back(statement);
+            expecting_statement_delimiter = true;
+        }
+
+        // debug!("Parser sql: {}, stmts: {:#?}", sql, stmts);
+
+        Ok(stmts)
+    }
+
+    /// Parse a new expression
+    fn parse_statement(&mut self) -> Result<ExtStatement> {
+        match self.parser.peek_token().token {
+            Token::Word(w) => match w.keyword {
+                _ => Ok(ExtStatement::SqlStatement(Box::new(self.parser.parse_statement()?))),
+            },
+            _ => Ok(ExtStatement::SqlStatement(Box::new(self.parser.parse_statement()?))),
+        }
+    }
+
+    // Report unexpected token
+    fn expected<T>(&self, expected: &str, found: impl Display) -> Result<T> {
+        parser_err!(format!("Expected {}, found: {}", expected, found))
+    }
+}

s3select/query/src/sql/physical/mod.rs

@@ -1 +1,2 @@
-
+pub mod optimizer;
+pub mod planner;

s3select/query/src/sql/physical/optimizer.rs

@@ -0,0 +1,12 @@
+use std::sync::Arc;
+
+use api::query::session::SessionCtx;
+use api::QueryResult;
+use datafusion::physical_optimizer::PhysicalOptimizerRule;
+use datafusion::physical_plan::ExecutionPlan;
+
+pub trait PhysicalOptimizer {
+    fn optimize(&self, plan: Arc<dyn ExecutionPlan>, session: &SessionCtx) -> QueryResult<Arc<dyn ExecutionPlan>>;
+
+    fn inject_optimizer_rule(&mut self, optimizer_rule: Arc<dyn PhysicalOptimizerRule + Send + Sync>);
+}

s3select/query/src/sql/physical/planner.rs

@@ -0,0 +1,104 @@
+use std::sync::Arc;
+
+use api::query::physical_planner::PhysicalPlanner;
+use api::query::session::SessionCtx;
+use api::QueryResult;
+use async_trait::async_trait;
+use datafusion::execution::SessionStateBuilder;
+use datafusion::logical_expr::LogicalPlan;
+use datafusion::physical_optimizer::aggregate_statistics::AggregateStatistics;
+use datafusion::physical_optimizer::coalesce_batches::CoalesceBatches;
+use datafusion::physical_optimizer::join_selection::JoinSelection;
+use datafusion::physical_optimizer::PhysicalOptimizerRule;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion::physical_planner::{
+    DefaultPhysicalPlanner as DFDefaultPhysicalPlanner, ExtensionPlanner, PhysicalPlanner as DFPhysicalPlanner,
+};
+
+use super::optimizer::PhysicalOptimizer;
+
+pub struct DefaultPhysicalPlanner {
+    ext_physical_transform_rules: Vec<Arc<dyn ExtensionPlanner + Send + Sync>>,
+    /// Responsible for optimizing a physical execution plan
+    ext_physical_optimizer_rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>>,
+}
+
+impl DefaultPhysicalPlanner {
+    #[allow(dead_code)]
+    fn with_physical_transform_rules(mut self, rules: Vec<Arc<dyn ExtensionPlanner + Send + Sync>>) -> Self {
+        self.ext_physical_transform_rules = rules;
+        self
+    }
+}
+
+impl DefaultPhysicalPlanner {
+    #[allow(dead_code)]
+    fn with_optimizer_rules(mut self, rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>>) -> Self {
+        self.ext_physical_optimizer_rules = rules;
+        self
+    }
+}
+
+impl Default for DefaultPhysicalPlanner {
+    fn default() -> Self {
+        let ext_physical_transform_rules: Vec<Arc<dyn ExtensionPlanner + Send + Sync>> = vec![
+            // can add rules at here
+        ];
+
+        // We need to take care of the rule ordering. They may influence each other.
+        let ext_physical_optimizer_rules: Vec<Arc<dyn PhysicalOptimizerRule + Sync + Send>> = vec![
+            Arc::new(AggregateStatistics::new()),
+            // Statistics-based join selection will change the Auto mode to a real join implementation,
+            // like collect left, or hash join, or future sort merge join, which will influence the
+            // EnforceDistribution and EnforceSorting rules as they decide whether to add additional
+            // repartitioning and local sorting steps to meet distribution and ordering requirements.
+            // Therefore, it should run before EnforceDistribution and EnforceSorting.
+            Arc::new(JoinSelection::new()),
+            // The CoalesceBatches rule will not influence the distribution and ordering of the
+            // whole plan tree. Therefore, to avoid influencing other rules, it should run last.
+            Arc::new(CoalesceBatches::new()),
+        ];
+
+        Self {
+            ext_physical_transform_rules,
+            ext_physical_optimizer_rules,
+        }
+    }
+}
+
+#[async_trait]
+impl PhysicalPlanner for DefaultPhysicalPlanner {
+    async fn create_physical_plan(
+        &self,
+        logical_plan: &LogicalPlan,
+        session: &SessionCtx,
+    ) -> QueryResult<Arc<dyn ExecutionPlan>> {
+        // 将扩展的物理计划优化规则注入df 的 session state
+        let new_state = SessionStateBuilder::new_from_existing(session.inner().clone())
+            .with_physical_optimizer_rules(self.ext_physical_optimizer_rules.clone())
+            .build();
+
+        // 通过扩展的物理计划转换规则构造df 的 Physical Planner
+        let planner = DFDefaultPhysicalPlanner::with_extension_planners(self.ext_physical_transform_rules.clone());
+
+        // 执行df的物理计划规划及优化
+        planner
+            .create_physical_plan(logical_plan, &new_state)
+            .await
+            .map_err(|e| e.into())
+    }
+
+    fn inject_physical_transform_rule(&mut self, rule: Arc<dyn ExtensionPlanner + Send + Sync>) {
+        self.ext_physical_transform_rules.push(rule)
+    }
+}
+
+impl PhysicalOptimizer for DefaultPhysicalPlanner {
+    fn optimize(&self, plan: Arc<dyn ExecutionPlan>, _session: &SessionCtx) -> QueryResult<Arc<dyn ExecutionPlan>> {
+        Ok(plan)
+    }
+
+    fn inject_optimizer_rule(&mut self, optimizer_rule: Arc<dyn PhysicalOptimizerRule + Send + Sync>) {
+        self.ext_physical_optimizer_rules.push(optimizer_rule);
+    }
+}

s3select/query/src/sql/planner.rs

@@ -1,4 +1,14 @@
-use datafusion::sql::planner::SqlToRel;
+use api::{
+    query::{
+        ast::ExtStatement,
+        logical_planner::{LogicalPlanner, Plan, QueryPlan},
+        session::SessionCtx,
+    },
+    QueryError, QueryResult,
+};
+use async_recursion::async_recursion;
+use async_trait::async_trait;
+use datafusion::sql::{planner::SqlToRel, sqlparser::ast::Statement};
 
 use crate::metadata::ContextProviderExtension;
 
@@ -6,3 +16,45 @@ pub struct SqlPlanner<'a, S: ContextProviderExtension> {
     schema_provider: &'a S,
     df_planner: SqlToRel<'a, S>,
 }
+
+#[async_trait]
+impl<'a, S: ContextProviderExtension + Send + Sync> LogicalPlanner for SqlPlanner<'a, S> {
+    async fn create_logical_plan(&self, statement: ExtStatement, session: &SessionCtx) -> QueryResult<Plan> {
+        let plan = { self.statement_to_plan(statement, session).await.map_err(|err| err)? };
+
+        Ok(plan)
+    }
+}
+
+impl<'a, S: ContextProviderExtension + Send + Sync + 'a> SqlPlanner<'a, S> {
+    /// Create a new query planner
+    pub fn new(schema_provider: &'a S) -> Self {
+        SqlPlanner {
+            schema_provider,
+            df_planner: SqlToRel::new(schema_provider),
+        }
+    }
+
+    /// Generate a logical plan from an  Extent SQL statement
+    #[async_recursion]
+    pub(crate) async fn statement_to_plan(&self, statement: ExtStatement, session: &SessionCtx) -> QueryResult<Plan> {
+        match statement {
+            ExtStatement::SqlStatement(stmt) => self.df_sql_to_plan(*stmt, session).await,
+        }
+    }
+
+    async fn df_sql_to_plan(&self, stmt: Statement, _session: &SessionCtx) -> QueryResult<Plan> {
+        match stmt {
+            Statement::Query(_) => {
+                let df_plan = self.df_planner.sql_statement_to_plan(stmt)?;
+                let plan = Plan::Query(QueryPlan {
+                    df_plan,
+                    is_tag_scan: false,
+                });
+
+                Ok(plan)
+            }
+            _ => Err(QueryError::NotImplemented { err: stmt.to_string() }),
+        }
+    }
+}