rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-05-23 16:30:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: region storage path (#2404)

Browse Source 
* feat: region storage path

* Update src/common/meta/src/key/datanode_table.rs

Co-authored-by: Weny Xu <wenymedia@gmail.com>

* chore: by cr

* feat: upgrade proto

---------

Co-authored-by: Weny Xu <wenymedia@gmail.com>
This commit is contained in:
JeremyHi
2023-09-15 17:07:54 +08:00
committed by GitHub
parent 43e3c94fd1
commit 9572b1edbb
17 changed files with 144 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/common/meta/src/ddl/create_table.rs
View File

@@ -29,7 +29,7 @@ use strum::AsRefStr;
use table::engine::TableReference;
use table::metadata::{RawTableInfo, TableId};
use crate::ddl::utils::{handle_operate_region_error, handle_retry_error};
use crate::ddl::utils::{handle_operate_region_error, handle_retry_error, region_storage_path};
use crate::ddl::DdlContext;
use crate::error::{self, Result};
use crate::key::table_name::TableNameKey;
@@ -161,8 +161,7 @@ impl CreateTableProcedure {
column_defs,
primary_key,
create_if_not_exists: true,
catalog: String::new(),
schema: String::new(),
path: String::new(),
options: create_table_expr.table_options.clone(),
})
}
@@ -174,6 +173,7 @@ impl CreateTableProcedure {
let create_table_expr = &create_table_data.task.create_table;
let catalog = &create_table_expr.catalog_name;
let schema = &create_table_expr.schema_name;
let storage_path = region_storage_path(catalog, schema);
let request_template = self.create_region_request_template()?;
@@ -191,9 +191,7 @@ impl CreateTableProcedure {
let mut create_region_request = request_template.clone();
create_region_request.region_id = region_id.as_u64();
create_region_request.catalog = catalog.to_string();
create_region_request.schema = schema.to_string();
create_region_request.path = storage_path.clone();
PbRegionRequest::Create(create_region_request)
})
.collect::<Vec<_>>();

5
src/common/meta/src/ddl/utils.rs
View File

@@ -42,3 +42,8 @@ pub fn handle_retry_error(e: Error) -> ProcedureError {
ProcedureError::external(e)
}
}
#[inline]
pub fn region_storage_path(catalog: &str, schema: &str) -> String {
format!("{}/{}", catalog, schema)
}

28
src/common/meta/src/key.rs
View File

@@ -71,6 +71,7 @@ use table_name::{TableNameKey, TableNameManager, TableNameValue};
use self::catalog_name::{CatalogManager, CatalogNameKey, CatalogNameValue};
use self::schema_name::{SchemaManager, SchemaNameKey, SchemaNameValue};
use self::table_route::{TableRouteManager, TableRouteValue};
use crate::ddl::utils::region_storage_path;
use crate::error::{self, Result, SerdeJsonSnafu};
use crate::kv_backend::txn::Txn;
use crate::kv_backend::KvBackendRef;
@@ -242,6 +243,8 @@ impl TableMetadataManager {
table_info.meta.region_numbers = region_numbers;
let table_id = table_info.ident.table_id;
let engine = table_info.meta.engine.clone();
let region_storage_path =
region_storage_path(&table_info.catalog_name, &table_info.schema_name);
// Creates table name.
let table_name = TableNameKey::new(
@@ -261,9 +264,12 @@ impl TableMetadataManager {
// Creates datanode table key value pairs.
let distribution = region_distribution(&region_routes)?;
let create_datanode_table_txn =
self.datanode_table_manager()
.build_create_txn(table_id, &engine, distribution)?;
let create_datanode_table_txn = self.datanode_table_manager().build_create_txn(
table_id,
&engine,
&region_storage_path,
distribution,
)?;
// Creates table route.
let table_route_value = TableRouteValue::new(region_routes);
@@ -441,6 +447,7 @@ impl TableMetadataManager {
&self,
table_id: TableId,
engine: &str,
region_storage_path: &str,
current_table_route_value: TableRouteValue,
new_region_routes: Vec<RegionRoute>,
) -> Result<()> {
@@ -452,6 +459,7 @@ impl TableMetadataManager {
let update_datanode_table_txn = self.datanode_table_manager().build_update_txn(
table_id,
engine,
region_storage_path,
current_region_distribution,
new_region_distribution,
)?;
@@ -554,6 +562,7 @@ mod tests {
use table::metadata::{RawTableInfo, TableInfo, TableInfoBuilder, TableMetaBuilder};
use super::datanode_table::DatanodeTableKey;
use crate::ddl::utils::region_storage_path;
use crate::key::table_info::TableInfoValue;
use crate::key::table_name::TableNameKey;
use crate::key::table_route::TableRouteValue;
@@ -867,6 +876,8 @@ mod tests {
new_test_table_info(region_routes.iter().map(|r| r.region.id.region_number())).into();
let table_id = table_info.ident.table_id;
let engine = table_info.meta.engine.as_str();
let region_storage_path =
region_storage_path(&table_info.catalog_name, &table_info.schema_name);
let current_table_route_value = TableRouteValue::new(region_routes.clone());
// creates metadata.
table_metadata_manager
@@ -884,6 +895,7 @@ mod tests {
.update_table_route(
table_id,
engine,
&region_storage_path,
current_table_route_value.clone(),
new_region_routes.clone(),
)
@@ -896,6 +908,7 @@ mod tests {
.update_table_route(
table_id,
engine,
&region_storage_path,
current_table_route_value.clone(),
new_region_routes.clone(),
)
@@ -909,6 +922,7 @@ mod tests {
.update_table_route(
table_id,
engine,
&region_storage_path,
current_table_route_value.clone(),
new_region_routes.clone(),
)
@@ -925,7 +939,13 @@ mod tests {
new_region_route(4, 4),
]);
assert!(table_metadata_manager
.update_table_route(table_id, engine, wrong_table_route_value, new_region_routes)
.update_table_route(
table_id,
engine,
&region_storage_path,
wrong_table_route_value,
new_region_routes
)
.await
.is_err());
}

40
src/common/meta/src/key/datanode_table.rs
View File

@@ -87,15 +87,23 @@ pub struct DatanodeTableValue {
pub regions: Vec<RegionNumber>,
#[serde(default)]
pub engine: String,
#[serde(default)]
pub region_storage_path: String,
version: u64,
}
impl DatanodeTableValue {
pub fn new(table_id: TableId, regions: Vec<RegionNumber>, engine: String) -> Self {
pub fn new(
table_id: TableId,
regions: Vec<RegionNumber>,
engine: String,
region_storage_path: String,
) -> Self {
Self {
table_id,
regions,
engine,
region_storage_path,
version: 0,
}
}
@@ -147,13 +155,19 @@ impl DatanodeTableManager {
&self,
table_id: TableId,
engine: &str,
region_storage_path: &str,
distribution: RegionDistribution,
) -> Result<Txn> {
let txns = distribution
.into_iter()
.map(|(datanode_id, regions)| {
let key = DatanodeTableKey::new(datanode_id, table_id);
let val = DatanodeTableValue::new(table_id, regions, engine.to_string());
let val = DatanodeTableValue::new(
table_id,
regions,
engine.to_string(),
region_storage_path.to_string(),
);
Ok(TxnOp::Put(key.as_raw_key(), val.try_as_raw_value()?))
})
@@ -169,6 +183,7 @@ impl DatanodeTableManager {
&self,
table_id: TableId,
engine: &str,
region_storage_path: &str,
current_region_distribution: RegionDistribution,
new_region_distribution: RegionDistribution,
) -> Result<Txn> {
@@ -189,16 +204,26 @@ impl DatanodeTableManager {
if *current_region != regions {
let key = DatanodeTableKey::new(datanode, table_id);
let raw_key = key.as_raw_key();
let val = DatanodeTableValue::new(table_id, regions, engine.to_string())
.try_as_raw_value()?;
let val = DatanodeTableValue::new(
table_id,
regions,
engine.to_string(),
region_storage_path.to_string(),
)
.try_as_raw_value()?;
opts.push(TxnOp::Put(raw_key, val));
}
} else {
// New datanodes
let key = DatanodeTableKey::new(datanode, table_id);
let raw_key = key.as_raw_key();
let val = DatanodeTableValue::new(table_id, regions, engine.to_string())
.try_as_raw_value()?;
let val = DatanodeTableValue::new(
table_id,
regions,
engine.to_string(),
region_storage_path.to_string(),
)
.try_as_raw_value()?;
opts.push(TxnOp::Put(raw_key, val));
}
}
@@ -246,9 +271,10 @@ mod tests {
table_id: 42,
regions: vec![1, 2, 3],
engine: Default::default(),
region_storage_path: Default::default(),
version: 1,
};
let literal = br#"{"table_id":42,"regions":[1,2,3],"engine":"","version":1}"#;
let literal = br#"{"table_id":42,"regions":[1,2,3],"engine":"","region_storage_path":"","version":1}"#;
let raw_value = value.try_as_raw_value().unwrap();
assert_eq!(raw_value, literal);