feat: support random partition rule (#1193)

## Rationale Currently, hash/key rule are the only supported rules for partition table. However, it will lead to small write batches if the user tries to use primary key or all tags as the partition key for a uniform distribution. And it is known that the small write batches hurt write performance badly. Actually, for a partition table in `append` mode with primary key or all tags as partition key, most of the queries to this table will fan out to all the partitions because the partition key won't be provided in most cases. So what is the gain from the cost (the massive small write batches) to do such complex partition? It will work much better if a simple partition rule is supported, by which a random partition is chosen for a given write request. ## Detailed Changes - Introduce `RandomPartitionRule` - Add a new method called `validate_create_table` to `TableEngine` trait ## Test Plan Add unit test to apply `RandomPartitionRule` for a partition table in different mode.
apache · Sep 7, 2023 · 1252192 · 1252192
1 parent e5b03d1
commit 1252192
Show file tree

Hide file tree

Showing 36 changed files with 538 additions and 227 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -98,7 +98,7 @@ bytes = "1"
 bytes_ext = { path = "components/bytes_ext" }
 catalog = { path = "catalog" }
 catalog_impls = { path = "catalog_impls" }
-ceresdbproto = "1.0.13"
+ceresdbproto = "1.0.14"
 codec = { path = "components/codec" }
 chrono = "0.4"
 clap = "3.0"

diff --git a/analytic_engine/src/engine.rs b/analytic_engine/src/engine.rs
@@ -22,9 +22,9 @@ use log::{error, info};
 use snafu::{OptionExt, ResultExt};
 use table_engine::{
     engine::{
-        Close, CloseShardRequest, CloseTableRequest, CreateTableRequest, DropTableRequest,
-        OpenShard, OpenShardRequest, OpenShardResult, OpenTableNoCause, OpenTableRequest,
-        OpenTableWithCause, Result, TableDef, TableEngine,
+        Close, CloseShardRequest, CloseTableRequest, CreateTableParams, CreateTableRequest,
+        DropTableRequest, OpenShard, OpenShardRequest, OpenShardResult, OpenTableNoCause,
+        OpenTableRequest, OpenTableWithCause, Result, TableDef, TableEngine,
     },
     table::{SchemaId, TableRef},
     ANALYTIC_ENGINE_TYPE,
@@ -100,6 +100,12 @@ impl TableEngine for TableEngineImpl {
         Ok(())
     }
 
+    async fn validate_create_table(&self, params: &CreateTableParams) -> Result<()> {
+        self.instance.validate_create_table(params)?;
+
+        Ok(())
+    }
+
     async fn create_table(&self, request: CreateTableRequest) -> Result<TableRef> {
         let space_id = build_space_id(request.schema_id);
 

diff --git a/analytic_engine/src/instance/alter.rs b/analytic_engine/src/instance/alter.rs
@@ -212,15 +212,16 @@ impl<'a> Alterer<'a> {
             "Instance alter options, space_id:{}, tables:{:?}, old_table_opts:{:?}, options:{:?}",
             self.table_data.space_id, self.table_data.name, current_table_options, options
         );
-        let mut table_opts =
-            table_options::merge_table_options_for_alter(&options, &current_table_options)
-                .box_err()
-                .context(InvalidOptions {
-                    space_id: self.table_data.space_id,
-                    table: &self.table_data.name,
-                    table_id: self.table_data.id,
-                })?;
-        table_opts.sanitize();
+        let table_opts = {
+            let mut opts =
+                table_options::merge_table_options_for_alter(&options, &current_table_options)
+                    .box_err()
+                    .context(InvalidOptions {
+                        table: &self.table_data.name,
+                    })?;
+            opts.sanitize();
+            opts
+        };
         let manifest_update = AlterOptionsMeta {
             space_id: self.table_data.space_id,
             table_id: self.table_data.id,

diff --git a/analytic_engine/src/instance/create.rs b/analytic_engine/src/instance/create.rs
@@ -16,21 +16,51 @@
 
 use generic_error::BoxError;
 use log::info;
-use snafu::{OptionExt, ResultExt};
-use table_engine::engine::CreateTableRequest;
+use snafu::{ensure, OptionExt, ResultExt};
+use table_engine::{
+    engine::{CreateTableParams, CreateTableRequest},
+    partition::PartitionInfo,
+};
 
 use crate::{
     instance::{
-        engine::{CreateOpenFailedTable, InvalidOptions, Result, TableNotExist, WriteManifest},
+        engine::{
+            CreateOpenFailedTable, InvalidOptions, Result, TableNotExist,
+            TryCreateRandomPartitionTableInOverwriteMode, WriteManifest,
+        },
         Instance,
     },
     manifest::meta_edit::{AddTableMeta, MetaEdit, MetaEditRequest, MetaUpdate},
     space::SpaceRef,
     table::data::{TableDataRef, TableShardInfo},
-    table_options,
+    table_options, TableOptions,
 };
 
 impl Instance {
+    /// Validate the request of creating table.
+    pub fn validate_create_table(&self, params: &CreateTableParams) -> Result<TableOptions> {
+        let table_opts =
+            table_options::merge_table_options_for_create(&params.table_options, &self.table_opts)
+                .box_err()
+                .context(InvalidOptions {
+                    table: &params.table_name,
+                })?;
+
+        if let Some(partition_info) = &params.partition_info {
+            let dedup_on_random_partition =
+                table_opts.need_dedup() && matches!(partition_info, PartitionInfo::Random(_));
+
+            ensure!(
+                !dedup_on_random_partition,
+                TryCreateRandomPartitionTableInOverwriteMode {
+                    table: &params.table_name,
+                }
+            );
+        }
+
+        Ok(table_opts)
+    }
+
     /// Create table need to be handled by write worker.
     pub async fn do_create_table(
         &self,
@@ -39,21 +69,14 @@ impl Instance {
     ) -> Result<TableDataRef> {
         info!("Instance create table, request:{:?}", request);
 
-        if space.is_open_failed_table(&request.table_name) {
+        if space.is_open_failed_table(&request.params.table_name) {
             return CreateOpenFailedTable {
-                table: request.table_name,
+                table: request.params.table_name,
             }
             .fail();
         }
 
-        let mut table_opts =
-            table_options::merge_table_options_for_create(&request.options, &self.table_opts)
-                .box_err()
-                .context(InvalidOptions {
-                    space_id: space.id,
-                    table: &request.table_name,
-                    table_id: request.table_id,
-                })?;
+        let mut table_opts = self.validate_create_table(&request.params)?;
         // Sanitize options before creating table.
         table_opts.sanitize();
 
@@ -66,8 +89,8 @@ impl Instance {
             let meta_update = MetaUpdate::AddTable(AddTableMeta {
                 space_id: space.id,
                 table_id: request.table_id,
-                table_name: request.table_name.clone(),
-                schema: request.table_schema,
+                table_name: request.params.table_name.clone(),
+                schema: request.params.table_schema,
                 opts: table_opts,
             });
             MetaEditRequest {
@@ -79,9 +102,9 @@ impl Instance {
             .manifest
             .apply_edit(edit_req)
             .await
-            .with_context(|| WriteManifest {
+            .context(WriteManifest {
                 space_id: space.id,
-                table: request.table_name.clone(),
+                table: &request.params.table_name,
                 table_id: request.table_id,
             })?;
 
@@ -91,7 +114,7 @@ impl Instance {
             .with_context(|| TableNotExist {
                 msg: format!(
                     "table not exist, space_id:{}, table_id:{}, table_name:{}",
-                    space.id, request.table_id, request.table_name
+                    space.id, request.table_id, request.params.table_name
                 ),
             })
     }

diff --git a/analytic_engine/src/instance/engine.rs b/analytic_engine/src/instance/engine.rs
@@ -124,19 +124,8 @@ pub enum Error {
         source: GenericError,
     },
 
-    #[snafu(display(
-        "Invalid options, space_id:{}, table:{}, table_id:{}, err:{}",
-        space_id,
-        table,
-        table_id,
-        source
-    ))]
-    InvalidOptions {
-        space_id: SpaceId,
-        table: String,
-        table_id: TableId,
-        source: GenericError,
-    },
+    #[snafu(display("Invalid options, table:{table}, err:{source}",))]
+    InvalidOptions { table: String, source: GenericError },
 
     #[snafu(display(
         "Failed to create table data, space_id:{}, table:{}, table_id:{}, err:{}",
@@ -243,14 +232,21 @@ pub enum Error {
         msg: Option<String>,
         backtrace: Backtrace,
     },
+
+    #[snafu(display(
+        "Try to create a random partition table in overwrite mode, table:{table}.\nBacktrace:\n{backtrace}",
+    ))]
+    TryCreateRandomPartitionTableInOverwriteMode { table: String, backtrace: Backtrace },
 }
 
 define_result!(Error);
 
 impl From<Error> for table_engine::engine::Error {
     fn from(err: Error) -> Self {
         match &err {
-            Error::InvalidOptions { table, .. } | Error::SpaceNotExist { table, .. } => {
+            Error::InvalidOptions { table, .. }
+            | Error::SpaceNotExist { table, .. }
+            | Error::TryCreateRandomPartitionTableInOverwriteMode { table, .. } => {
                 Self::InvalidArguments {
                     table: table.clone(),
                     source: Box::new(err),
@@ -330,8 +326,8 @@ impl Instance {
         request: CreateTableRequest,
     ) -> Result<SpaceAndTable> {
         let context = SpaceContext {
-            catalog_name: request.catalog_name.clone(),
-            schema_name: request.schema_name.clone(),
+            catalog_name: request.params.catalog_name.clone(),
+            schema_name: request.params.schema_name.clone(),
         };
         let space = self.find_or_create_space(space_id, context).await?;
         let table_data = self.do_create_table(space.clone(), request).await?;

diff --git a/analytic_engine/src/table/data.rs b/analytic_engine/src/table/data.rs
@@ -705,7 +705,7 @@ pub mod tests {
     use arena::NoopCollector;
     use common_types::{datum::DatumKind, table::DEFAULT_SHARD_ID};
     use table_engine::{
-        engine::{CreateTableRequest, TableState},
+        engine::{CreateTableParams, CreateTableRequest, TableState},
         table::SchemaId,
     };
     use time_ext::ReadableDuration;
@@ -793,18 +793,21 @@ pub mod tests {
         pub fn build(self) -> TableData {
             let space_id = DEFAULT_SPACE_ID;
             let table_schema = default_schema();
-            let create_request = CreateTableRequest {
+            let params = CreateTableParams {
                 catalog_name: "test_catalog".to_string(),
                 schema_name: "public".to_string(),
-                schema_id: SchemaId::from_u32(DEFAULT_SPACE_ID),
-                table_id: self.table_id,
                 table_name: self.table_name,
                 table_schema,
                 engine: table_engine::ANALYTIC_ENGINE_TYPE.to_string(),
-                options: HashMap::new(),
+                table_options: HashMap::new(),
+                partition_info: None,
+            };
+            let create_request = CreateTableRequest {
+                params,
+                schema_id: SchemaId::from_u32(DEFAULT_SPACE_ID),
+                table_id: self.table_id,
                 state: TableState::Stable,
                 shard_id: self.shard_id,
-                partition_info: None,
             };
 
             let table_opts = TableOptions::default();
@@ -814,8 +817,8 @@ pub mod tests {
             TableData::new(
                 space_id,
                 create_request.table_id,
-                create_request.table_name,
-                create_request.table_schema,
+                create_request.params.table_name,
+                create_request.params.table_schema,
                 create_request.shard_id,
                 table_opts,
                 &purger,

diff --git a/analytic_engine/src/tests/table.rs b/analytic_engine/src/tests/table.rs
@@ -29,7 +29,7 @@ use common_types::{
 };
 use table_engine::{
     self,
-    engine::{CreateTableRequest, TableState},
+    engine::{CreateTableParams, CreateTableRequest, TableState},
     predicate::Predicate,
     table::{GetRequest, ReadOptions, ReadRequest, SchemaId, TableId, TableSeq},
 };
@@ -122,7 +122,7 @@ impl FixedSchemaTable {
     }
 
     fn new_row_group(&self, rows: Vec<Row>) -> RowGroup {
-        RowGroupBuilder::with_rows(self.create_request.table_schema.clone(), rows)
+        RowGroupBuilder::with_rows(self.create_request.params.table_schema.clone(), rows)
             .unwrap()
             .build()
     }
@@ -132,7 +132,7 @@ impl FixedSchemaTable {
     }
 
     pub fn new_read_all_request(&self, opts: ReadOptions) -> ReadRequest {
-        new_read_all_request_with_order(self.create_request.table_schema.clone(), opts)
+        new_read_all_request_with_order(self.create_request.params.table_schema.clone(), opts)
     }
 
     pub fn new_get_request(&self, key: KeyTuple) -> GetRequest {
@@ -141,7 +141,7 @@ impl FixedSchemaTable {
         GetRequest {
             request_id: RequestId::next_id(),
             projected_schema: ProjectedSchema::no_projection(
-                self.create_request.table_schema.clone(),
+                self.create_request.params.table_schema.clone(),
             ),
             primary_key,
         }
@@ -272,7 +272,7 @@ impl Builder {
     }
 
     pub fn table_name(mut self, table_name: String) -> Self {
-        self.create_request.table_name = table_name;
+        self.create_request.params.table_name = table_name;
         self
     }
 
@@ -282,7 +282,7 @@ impl Builder {
     }
 
     pub fn enable_ttl(mut self, enable_ttl: bool) -> Self {
-        self.create_request.options.insert(
+        self.create_request.params.table_options.insert(
             common_types::OPTION_KEY_ENABLE_TTL.to_string(),
             enable_ttl.to_string(),
         );
@@ -291,7 +291,8 @@ impl Builder {
 
     pub fn ttl(mut self, duration: ReadableDuration) -> Self {
         self.create_request
-            .options
+            .params
+            .table_options
             .insert(common_types::TTL.to_string(), duration.to_string());
         self
     }
@@ -305,17 +306,21 @@ impl Builder {
 
 impl Default for Builder {
     fn default() -> Self {
+        let params = CreateTableParams {
+            catalog_name: "ceresdb".to_string(),
+            schema_name: "public".to_string(),
+            table_name: "test_table".to_string(),
+            table_schema: FixedSchemaTable::default_schema(),
+            partition_info: None,
+            engine: table_engine::ANALYTIC_ENGINE_TYPE.to_string(),
+            table_options: HashMap::new(),
+        };
+
         Self {
             create_request: CreateTableRequest {
-                catalog_name: "ceresdb".to_string(),
-                schema_name: "public".to_string(),
+                params,
                 schema_id: SchemaId::from_u32(2),
                 table_id: new_table_id(2, 1),
-                table_name: "test_table".to_string(),
-                table_schema: FixedSchemaTable::default_schema(),
-                partition_info: None,
-                engine: table_engine::ANALYTIC_ENGINE_TYPE.to_string(),
-                options: HashMap::new(),
                 state: TableState::Stable,
                 shard_id: DEFAULT_SHARD_ID,
             },

diff --git a/analytic_engine/src/tests/util.rs b/analytic_engine/src/tests/util.rs
@@ -311,7 +311,7 @@ impl<T: WalsOpener> TestContext<T> {
     }
 
     async fn create_table(&mut self, create_request: CreateTableRequest) {
-        let table_name = create_request.table_name.clone();
+        let table_name = create_request.params.table_name.clone();
         let table = self.engine().create_table(create_request).await.unwrap();
 
         self.name_to_tables.insert(table_name.to_string(), table);