Skip to content

Commit

Permalink
feat: region storage path (#2404)
Browse files Browse the repository at this point in the history 
* 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>
  • Loading branch information
@fengjiachun @WenyXu
fengjiachun and WenyXu authored Sep 15, 2023
1 parent 43e3c94 commit 9572b1e
Show file tree
Hide file tree
Showing 17 changed files with 143 additions and 48 deletions.
2 changes: 1 addition & 1 deletion Cargo.lock
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Cargo.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git
derive_builder = "0.12"
futures = "0.3"
futures-util = "0.3"
greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "81495b166b2c8909f05b3fcaa09eb299bb43a995" }
greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "e81a60e817a348ee5b7dfbd991f86d35cd068ce5" }
humantime-serde = "1.1"
itertools = "0.10"
lazy_static = "1.4"
Expand Down
12 changes: 11 additions & 1 deletion src/cmd/src/cli/upgrade.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ use std::sync::Arc;
use async_trait::async_trait;
use clap::Parser;
use client::api::v1::meta::TableRouteValue;
use common_meta::ddl::utils::region_storage_path;
use common_meta::error as MetaError;
use common_meta::key::catalog_name::{CatalogNameKey, CatalogNameValue};
use common_meta::key::datanode_table::{DatanodeTableKey, DatanodeTableValue};
Expand Down Expand Up @@ -387,6 +388,10 @@ impl MigrateTableMetadata {
async fn create_datanode_table_keys(&self, value: &TableGlobalValue) {
let table_id = value.table_id();
let engine = value.table_info.meta.engine.as_str();
let region_storage_path = region_storage_path(
&value.table_info.catalog_name,
&value.table_info.schema_name,
);
let region_distribution: RegionDistribution =
value.regions_id_map.clone().into_iter().collect();

Expand All @@ -397,7 +402,12 @@ impl MigrateTableMetadata {
info!("Creating DatanodeTableKey '{k}' => {regions:?}");
(
k,
DatanodeTableValue::new(table_id, regions, engine.to_string()),
DatanodeTableValue::new(
table_id,
regions,
engine.to_string(),
region_storage_path.clone(),
),
)
})
.collect::<Vec<_>>();
Expand Down
10 changes: 4 additions & 6 deletions src/common/meta/src/ddl/create_table.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(),
})
}
Expand All @@ -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()?;

Expand All @@ -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<_>>();
Expand Down
5 changes: 5 additions & 0 deletions src/common/meta/src/ddl/utils.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 changes: 24 additions & 4 deletions src/common/meta/src/key.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand All @@ -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);
Expand Down Expand Up @@ -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<()> {
Expand All @@ -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,
)?;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand All @@ -884,6 +895,7 @@ mod tests {
.update_table_route(
table_id,
engine,
&region_storage_path,
current_table_route_value.clone(),
new_region_routes.clone(),
)
Expand All @@ -896,6 +908,7 @@ mod tests {
.update_table_route(
table_id,
engine,
&region_storage_path,
current_table_route_value.clone(),
new_region_routes.clone(),
)
Expand All @@ -909,6 +922,7 @@ mod tests {
.update_table_route(
table_id,
engine,
&region_storage_path,
current_table_route_value.clone(),
new_region_routes.clone(),
)
Expand All @@ -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());
}
Expand Down
40 changes: 33 additions & 7 deletions src/common/meta/src/key/datanode_table.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}
}
Expand Down Expand Up @@ -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()?))
})
Expand All @@ -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> {
Expand All @@ -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));
}
}
Expand Down Expand Up @@ -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);
Expand Down
10 changes: 6 additions & 4 deletions src/file-table-engine/src/engine/immutable.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ use common_telemetry::{debug, logging};
use datatypes::schema::Schema;
use object_store::ObjectStore;
use snafu::ResultExt;
use store_api::path_utils::table_dir;
use store_api::path_utils::table_dir_with_catalog_and_schema;
use table::engine::{EngineContext, TableEngine, TableEngineProcedure, TableReference};
use table::error::TableOperationSnafu;
use table::metadata::{TableId, TableInfo, TableInfoBuilder, TableMetaBuilder, TableType};
Expand Down Expand Up @@ -245,7 +245,7 @@ impl EngineInner {
let table_schema =
Arc::new(Schema::try_from(request.schema).context(InvalidRawSchemaSnafu)?);

let table_dir = table_dir(&catalog_name, &schema_name, table_id);
let table_dir = table_dir_with_catalog_and_schema(&catalog_name, &schema_name, table_id);

let table_full_name = table_ref.to_string();

Expand Down Expand Up @@ -345,7 +345,8 @@ impl EngineInner {
}

let table_id = request.table_id;
let table_dir = table_dir(&catalog_name, &schema_name, table_id);
let table_dir =
table_dir_with_catalog_and_schema(&catalog_name, &schema_name, table_id);

let (metadata, table_info) = self
.recover_table_manifest_and_info(&table_full_name, &table_dir)
Expand Down Expand Up @@ -388,7 +389,8 @@ impl EngineInner {
let _lock = self.table_mutex.lock().await;
if let Some(table) = self.get_table(req.table_id) {
let table_id = table.table_info().ident.table_id;
let table_dir = table_dir(&req.catalog_name, &req.schema_name, table_id);
let table_dir =
table_dir_with_catalog_and_schema(&req.catalog_name, &req.schema_name, table_id);

delete_table_manifest(
&table_full_name,
Expand Down
4 changes: 2 additions & 2 deletions src/file-table-engine/src/test_util.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ use datatypes::prelude::ConcreteDataType;
use datatypes::schema::{ColumnSchema, RawSchema, Schema, SchemaBuilder, SchemaRef};
use object_store::services::Fs;
use object_store::ObjectStore;
use store_api::path_utils::table_dir;
use store_api::path_utils::table_dir_with_catalog_and_schema;
use table::engine::{EngineContext, TableEngine};
use table::metadata::{RawTableInfo, TableInfo, TableInfoBuilder, TableMetaBuilder, TableType};
use table::requests::{self, CreateTableRequest, TableOptions};
Expand Down Expand Up @@ -143,7 +143,7 @@ pub async fn setup_test_engine_and_table(prefix: &str) -> TestEngineComponents {

let table_info = table_ref.table_info();

let table_dir = table_dir(
let table_dir = table_dir_with_catalog_and_schema(
&table_info.catalog_name,
&table_info.schema_name,
table_info.ident.table_id,
Expand Down
4 changes: 2 additions & 2 deletions src/meta-srv/src/error.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -253,9 +253,9 @@ pub enum Error {
location: Location,
},

#[snafu(display("Table info not found: {}", table_name))]
#[snafu(display("Table info not found: {}", table_id))]
TableInfoNotFound {
table_name: String,
table_id: TableId,
location: Location,
},

Expand Down
Loading

0 comments on commit 9572b1e

Please sign in to comment.