feat: query partition table with proxy in grpc service

Cargo.toml

@@ -38,6 +38,7 @@ members = [
     "integration_tests/sdk/rust",
     "interpreters",
     "meta_client",
+    "partition_table_engine",
     "query_engine",
     "remote_engine_client",
     "router",
@@ -91,6 +92,7 @@ interpreters = { path = "interpreters" }
 itertools = "0.10.5"
 meta_client = { path = "meta_client" }
 object_store = { path = "components/object_store" }
+partition_table_engine = { path = "partition_table_engine" }
 parquet_ext = { path = "components/parquet_ext" }
 parquet = { version = "36.0.0" }
 paste = "1.0"

analytic_engine/src/engine.rs

@@ -1,4 +1,4 @@
-// Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
+// Copyright 2022-2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
 //! Implements the TableEngine trait
 
@@ -7,21 +7,17 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use common_util::error::BoxError;
 use log::info;
-use snafu::{OptionExt, ResultExt};
+use snafu::ResultExt;
 use table_engine::{
     engine::{
         Close, CloseTableRequest, CreateTableRequest, DropTableRequest, OpenTableRequest, Result,
-        TableEngine, Unexpected, UnexpectedNoCause,
+        TableEngine,
     },
     table::{SchemaId, TableRef},
     ANALYTIC_ENGINE_TYPE,
 };
 
-use crate::{
-    instance::InstanceRef,
-    space::SpaceId,
-    table::{partition::PartitionTableImpl, TableImpl},
-};
+use crate::{instance::InstanceRef, space::SpaceId, table::TableImpl};
 
 /// TableEngine implementation
 pub struct TableEngineImpl {
@@ -76,30 +72,14 @@ impl TableEngine for TableEngineImpl {
 
         let space_table = self.instance.create_table(space_id, request).await?;
 
-        let table_impl: TableRef = match &space_table.table_data().partition_info {
-            None => Arc::new(TableImpl::new(
-                self.instance.clone(),
-                ANALYTIC_ENGINE_TYPE.to_string(),
-                space_id,
-                space_table.table_data().id,
-                space_table.table_data().clone(),
-                space_table,
-            )),
-            Some(_v) => Arc::new(
-                PartitionTableImpl::new(
-                    self.instance
-                        .remote_engine
-                        .clone()
-                        .context(UnexpectedNoCause {
-                            msg: "remote engine not found",
-                        })?,
-                    ANALYTIC_ENGINE_TYPE.to_string(),
-                    space_table,
-                )
-                .box_err()
-                .context(Unexpected)?,
-            ),
-        };
+        let table_impl: TableRef = Arc::new(TableImpl::new(
+            self.instance.clone(),
+            ANALYTIC_ENGINE_TYPE.to_string(),
+            space_id,
+            space_table.table_data().id,
+            space_table.table_data().clone(),
+            space_table,
+        ));
 
         Ok(table_impl)
     }
@@ -128,30 +108,14 @@ impl TableEngine for TableEngineImpl {
             None => return Ok(None),
         };
 
-        let table_impl: TableRef = match &space_table.table_data().partition_info {
-            None => Arc::new(TableImpl::new(
-                self.instance.clone(),
-                ANALYTIC_ENGINE_TYPE.to_string(),
-                space_id,
-                space_table.table_data().id,
-                space_table.table_data().clone(),
-                space_table,
-            )),
-            Some(_v) => Arc::new(
-                PartitionTableImpl::new(
-                    self.instance
-                        .remote_engine
-                        .clone()
-                        .context(UnexpectedNoCause {
-                            msg: "remote engine is empty",
-                        })?,
-                    ANALYTIC_ENGINE_TYPE.to_string(),
-                    space_table,
-                )
-                .box_err()
-                .context(Unexpected)?,
-            ),
-        };
+        let table_impl = Arc::new(TableImpl::new(
+            self.instance.clone(),
+            ANALYTIC_ENGINE_TYPE.to_string(),
+            space_id,
+            space_table.table_data().id,
+            space_table.table_data().clone(),
+            space_table,
+        ));
 
         Ok(Some(table_impl))
     }

analytic_engine/src/instance/create.rs

@@ -1,4 +1,4 @@
-// Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
+// Copyright 2022-2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
 //! Create table logic of instance
 
@@ -113,7 +113,6 @@ impl Instance {
                 table_name: table_data.name.clone(),
                 schema: table_data.schema(),
                 opts: table_data.table_options().as_ref().clone(),
-                partition_info: table_data.partition_info.clone(),
             });
             MetaUpdateRequest {
                 shard_info: table_data.shard_info,

analytic_engine/src/instance/mod.rs

@@ -1,4 +1,4 @@
-// Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
+// Copyright 2022-2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
 //! A table engine instance
 //!
@@ -27,7 +27,7 @@ use common_util::{define_result, runtime::Runtime};
 use log::info;
 use mem_collector::MemUsageCollector;
 use snafu::{ResultExt, Snafu};
-use table_engine::{engine::EngineRuntimes, remote::RemoteEngineRef};
+use table_engine::engine::EngineRuntimes;
 use wal::manager::{WalLocation, WalManagerRef};
 
 use crate::{
@@ -177,7 +177,6 @@ pub struct Instance {
     /// Options for scanning sst
     pub(crate) scan_options: ScanOptions,
     pub(crate) iter_options: Option<IterOptions>,
-    pub(crate) remote_engine: Option<RemoteEngineRef>,
 }
 
 impl Instance {

analytic_engine/src/instance/open.rs

@@ -10,7 +10,7 @@ use std::{
 use common_types::schema::IndexInWriterSchema;
 use log::{debug, error, info, trace, warn};
 use snafu::ResultExt;
-use table_engine::{engine::OpenTableRequest, remote::RemoteEngineRef};
+use table_engine::engine::OpenTableRequest;
 use tokio::sync::oneshot;
 use wal::{
     log_batch::LogEntry,
@@ -54,7 +54,6 @@ impl Instance {
         wal_manager: WalManagerRef,
         store_picker: ObjectStorePickerRef,
         sst_factory: SstFactoryRef,
-        remote_engine_ref: Option<RemoteEngineRef>,
     ) -> Result<Arc<Self>> {
         let space_store = Arc::new(SpaceStore {
             spaces: RwLock::new(Spaces::default()),
@@ -111,7 +110,6 @@ impl Instance {
                 .map(|v| v.as_byte() as usize),
             iter_options,
             scan_options,
-            remote_engine: remote_engine_ref,
         });
 
         Ok(instance)

analytic_engine/src/manifest/details.rs

@@ -689,17 +689,13 @@ where
 mod tests {
     use std::{path::PathBuf, sync::Arc, vec};
 
-    use bytes::Bytes;
     use common_types::{
         column_schema, datum::DatumKind, schema, schema::Schema, table::DEFAULT_SHARD_ID,
     };
     use common_util::{runtime, runtime::Runtime, tests::init_log_for_test};
     use futures::future::BoxFuture;
     use object_store::LocalFileSystem;
-    use table_engine::{
-        partition::{HashPartitionInfo, PartitionDefinition, PartitionInfo},
-        table::{SchemaId, TableId, TableSeqGenerator},
-    };
+    use table_engine::table::{SchemaId, TableId, TableSeqGenerator};
     use wal::rocks_impl::manager::Builder as WalBuilder;
 
     use super::*;
@@ -831,23 +827,6 @@ mod tests {
                 table_name,
                 schema: common_types::tests::build_schema(),
                 opts: TableOptions::default(),
-                partition_info: None,
-            })
-        }
-
-        fn meta_update_add_table_with_partition_info(
-            &self,
-            table_id: TableId,
-            partition_info: Option<PartitionInfo>,
-        ) -> MetaUpdate {
-            let table_name = Self::table_name_from_id(table_id);
-            MetaUpdate::AddTable(AddTableMeta {
-                space_id: self.schema_id.as_u32(),
-                table_id,
-                table_name,
-                schema: common_types::tests::build_schema(),
-                opts: TableOptions::default(),
-                partition_info,
             })
         }
 
@@ -897,16 +876,14 @@ mod tests {
         async fn add_table_with_manifest(
             &self,
             table_id: TableId,
-            partition_info: Option<PartitionInfo>,
             manifest_data_builder: &mut TableManifestDataBuilder,
             manifest: &ManifestImpl,
         ) {
             let shard_info = TableShardInfo {
                 shard_id: DEFAULT_SHARD_ID,
             };
 
-            let add_table =
-                self.meta_update_add_table_with_partition_info(table_id, partition_info);
+            let add_table = self.meta_update_add_table(table_id);
             let update_req = {
                 MetaUpdateRequest {
                     shard_info,
@@ -967,7 +944,7 @@ mod tests {
             manifest_data_builder: &mut TableManifestDataBuilder,
         ) {
             let manifest = self.open_manifest().await;
-            self.add_table_with_manifest(table_id, None, manifest_data_builder, &manifest)
+            self.add_table_with_manifest(table_id, manifest_data_builder, &manifest)
                 .await;
         }
 
@@ -1128,26 +1105,11 @@ mod tests {
 
         runtime.block_on(async move {
             let table_id = ctx.alloc_table_id();
-            let default_version = 0;
-            let partition_info = Some(PartitionInfo::Hash(HashPartitionInfo {
-                version: default_version,
-                definitions: vec![PartitionDefinition {
-                    name: "p0".to_string(),
-                    origin_name: Some("region0".to_string()),
-                }],
-                expr: Bytes::from("test"),
-                linear: false,
-            }));
             let location = WalLocation::new(DEFAULT_SHARD_ID as u64, table_id.as_u64());
             let mut manifest_data_builder = TableManifestDataBuilder::default();
             let manifest = ctx.open_manifest().await;
-            ctx.add_table_with_manifest(
-                table_id,
-                partition_info,
-                &mut manifest_data_builder,
-                &manifest,
-            )
-            .await;
+            ctx.add_table_with_manifest(table_id, &mut manifest_data_builder, &manifest)
+                .await;
 
             manifest
                 .maybe_do_snapshot(ctx.schema_id.as_u32(), table_id, location, true)
@@ -1188,7 +1150,7 @@ mod tests {
             };
             let mut manifest_data_builder = TableManifestDataBuilder::default();
             let manifest = ctx.open_manifest().await;
-            ctx.add_table_with_manifest(table_id, None, &mut manifest_data_builder, &manifest)
+            ctx.add_table_with_manifest(table_id, &mut manifest_data_builder, &manifest)
                 .await;
 
             for i in 0..500 {