feat: refactor Resolver in dist sql query (#1186)

## Rationale
Part of #1112 
The old resolver has some problems, need to refactor before adding it
into the main query process.

## Detailed Changes
+ Refactor `Resolver` in dist sql query.

## Test Plan
Test by richer unit tests.

df_engine_extensions/Cargo.toml

@@ -26,17 +26,20 @@ workspace = true
 
 [dependencies]
 arrow = { workspace = true }
+async-recursion = "1.0.4"
 async-trait = { workspace = true }
 catalog = { workspace = true, features = ["test"] }
 ceresdbproto = { workspace = true }
+common_types = { workspace = true, features = ["test"] }
 datafusion = { workspace = true }
 datafusion-proto = { workspace = true }
+futures =   { workspace = true }
 generic_error = { workspace = true }
 prost = { workspace = true }
 snafu = { workspace = true }
 table_engine = { workspace = true }
 
 [dev-dependencies]
-common_types = { workspace = true, features = ["test"] }
 insta = { version = "1.31.0" }
+tokio = { workspace = true }
 trace_metric = { workspace = true }

df_engine_extensions/src/dist_sql_query/mod.rs

@@ -15,10 +15,14 @@
 use std::{fmt, sync::Arc};
 
 use async_trait::async_trait;
+use common_types::schema::RecordSchema;
 use datafusion::{
     error::Result as DfResult,
+    execution::TaskContext,
     physical_plan::{ExecutionPlan, SendableRecordBatchStream},
 };
+use futures::future::BoxFuture;
+use prost::bytes::Bytes;
 use table_engine::{
     remote::model::TableIdentifier,
     table::{ReadRequest, TableRef},
@@ -31,20 +35,33 @@ pub mod resolver;
 pub mod test_util;
 
 /// Remote datafusion physical plan executor
-#[async_trait]
-pub trait RemotePhysicalPlanExecutor: Clone + fmt::Debug + Send + Sync + 'static {
-    async fn execute(
+pub trait RemotePhysicalPlanExecutor: fmt::Debug + Send + Sync + 'static {
+    fn execute(
         &self,
         table: TableIdentifier,
-        physical_plan: Arc<dyn ExecutionPlan>,
-    ) -> DfResult<SendableRecordBatchStream>;
+        task_context: &TaskContext,
+        encoded_plan: EncodedPlan,
+    ) -> DfResult<BoxFuture<'static, DfResult<SendableRecordBatchStream>>>;
+}
+
+pub struct EncodedPlan {
+    pub plan: Bytes,
+    pub schema: RecordSchema,
 }
 
+type RemotePhysicalPlanExecutorRef = Arc<dyn RemotePhysicalPlanExecutor>;
+
 /// Executable scan's builder
 ///
 /// It is not suitable to restrict the detailed implementation of executable
 /// scan, so we define a builder here which return the general `ExecutionPlan`.
+#[async_trait]
 pub trait ExecutableScanBuilder: fmt::Debug + Send + Sync + 'static {
-    fn build(&self, table: TableRef, read_request: ReadRequest)
-        -> DfResult<Arc<dyn ExecutionPlan>>;
+    async fn build(
+        &self,
+        table: TableRef,
+        read_request: ReadRequest,
+    ) -> DfResult<Arc<dyn ExecutionPlan>>;
 }
+
+type ExecutableScanBuilderRef = Box<dyn ExecutableScanBuilder>;