feat: region storage path (GreptimeTeam#2404)

* 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>

Cargo.toml

@@ -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"

src/cmd/src/cli/upgrade.rs

@@ -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};
@@ -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();
 
@@ -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<_>>();

src/common/meta/src/ddl/create_table.rs

@@ -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<_>>();

src/common/meta/src/ddl/utils.rs

@@ -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)
+}

src/common/meta/src/key.rs

@@ -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());
  }

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

@@ -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);

src/file-table-engine/src/engine/immutable.rs

@@ -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};
@@ -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();
 
@@ -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)
@@ -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,

src/file-table-engine/src/test_util.rs

@@ -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};
@@ -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,

src/meta-srv/src/error.rs

@@ -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,
  },