feat: hilbert clustering (#17045)

* hilbert_clustering * fix test * vacuum temp files after recluster * fix * add test * fix * fix * add hilbert_clustering_information * update * update * add comments

Author: zhyass

Cargo.lock

@@ -184,6 +184,7 @@ databend-enterprise-attach-table = { path = "src/query/ee_features/attach_table"
 databend-enterprise-background-service = { path = "src/query/ee_features/background_service" }
 databend-enterprise-data-mask-feature = { path = "src/query/ee_features/data_mask" }
 databend-enterprise-fail-safe = { path = "src/query/ee_features/fail_safe" }
+databend-enterprise-hilbert-clustering = { path = "src/query/ee_features/hilbert_clustering" }
 databend-enterprise-inverted-index = { path = "src/query/ee_features/inverted_index" }
 databend-enterprise-meta = { path = "src/meta/ee" }
 databend-enterprise-query = { path = "src/query/ee" }

Cargo.toml

@@ -426,8 +426,9 @@ build_exceptions! {
  // recluster error codes
  NoNeedToRecluster(4011),
  NoNeedToCompact(4012),
+ UnsupportedClusterType(4013),
 
- RefreshTableInfoFailure(4012),
+ RefreshTableInfoFailure(4021),
 }
 
 // Service errors [5001,6000].

src/common/exception/src/exception_code.rs

@@ -75,6 +75,8 @@ pub enum Feature {
  AmendTable,
  #[serde(alias = "system_management", alias = "SYSTEM_MANAGEMENT")]
  SystemManagement,
+ #[serde(alias = "hilbert_clustering", alias = "HILBERT_CLUSTERING")]
+ HilbertClustering,
  #[serde(other)]
  Unknown,
 }
@@ -122,6 +124,7 @@ impl fmt::Display for Feature {
  }
  Feature::AmendTable => write!(f, "amend_table"),
  Feature::SystemManagement => write!(f, "system_management"),
+ Feature::HilbertClustering => write!(f, "hilbert_clustering"),
  Feature::Unknown => write!(f, "unknown"),
  }
  }
@@ -169,7 +172,8 @@ impl Feature {
  | (Feature::InvertedIndex, Feature::InvertedIndex)
  | (Feature::VirtualColumn, Feature::VirtualColumn)
  | (Feature::AttacheTable, Feature::AttacheTable)
- | (Feature::StorageEncryption, Feature::StorageEncryption) => Ok(true),
+ | (Feature::StorageEncryption, Feature::StorageEncryption)
+ | (Feature::HilbertClustering, Feature::HilbertClustering) => Ok(true),
  (_, _) => Ok(false),
  }
  }
@@ -337,6 +341,11 @@ mod tests {
  serde_json::from_str::<Feature>("\"amend_table\"").unwrap()
  );
 
+ assert_eq!(
+ Feature::HilbertClustering,
+ serde_json::from_str::<Feature>("\"hilbert_clustering\"").unwrap()
+ );
+
  assert_eq!(
  Feature::Unknown,
  serde_json::from_str::<Feature>("\"ssss\"").unwrap()
@@ -370,11 +379,12 @@ mod tests {
  storage_usage: Some(1),
  }),
  Feature::AmendTable,
+ Feature::HilbertClustering,
  ]),
  };
 
  assert_eq!(
- "LicenseInfo{ type: enterprise, org: databend, tenants: [databend_tenant,foo], features: [aggregate_index,amend_table,attach_table,background_service,compute_quota(threads_num: 1, memory_usage: 1),computed_column,data_mask,inverted_index,license_info,storage_encryption,storage_quota(storage_usage: 1),stream,vacuum,virtual_column] }",
+ "LicenseInfo{ type: enterprise, org: databend, tenants: [databend_tenant,foo], features: [aggregate_index,amend_table,attach_table,background_service,compute_quota(threads_num: 1, memory_usage: 1),computed_column,data_mask,hilbert_clustering,inverted_index,license_info,storage_encryption,storage_quota(storage_usage: 1),stream,vacuum,virtual_column] }",
  license_info.to_string()
  );
  }

src/common/license/src/license.rs

@@ -391,3 +391,10 @@ impl ReclusterParts {
  }
  }
 }
+
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default)]
+pub struct ReclusterInfoSideCar {
+ pub merged_blocks: Vec<Arc<BlockMeta>>,
+ pub removed_segment_indexes: Vec<usize>,
+ pub removed_statistics: Statistics,
+}

src/query/catalog/src/plan/partition.rs

@@ -19,6 +19,9 @@ use std::sync::Arc;
 
 use chrono::DateTime;
 use chrono::Utc;
+use databend_common_ast::ast::Expr;
+use databend_common_ast::parser::parse_comma_separated_exprs;
+use databend_common_ast::parser::tokenize_sql;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::BlockThresholds;
@@ -136,6 +139,28 @@ pub trait Table: Sync + Send {
  Some(cluster_type)
  }
 
+ fn resolve_cluster_keys(&self, ctx: Arc<dyn TableContext>) -> Option<Vec<Expr>> {
+ let Some((_, cluster_key_str)) = &self.cluster_key_meta() else {
+ return None;
+ };
+ let tokens = tokenize_sql(cluster_key_str).unwrap();
+ let sql_dialect = ctx.get_settings().get_sql_dialect().unwrap_or_default();
+ let mut ast_exprs = parse_comma_separated_exprs(&tokens, sql_dialect).unwrap();
+ // unwrap tuple.
+ if ast_exprs.len() == 1 {
+ if let Expr::Tuple { exprs, .. } = &ast_exprs[0] {
+ ast_exprs = exprs.clone();
+ }
+ } else {
+ // Defensive check:
+ // `ast_exprs` should always contain one element which can be one of the following:
+ // 1. A tuple of composite cluster keys
+ // 2. A single cluster key
+ unreachable!("invalid cluster key ast expression, {:?}", ast_exprs);
+ }
+ Some(ast_exprs)
+ }
+
  fn change_tracking_enabled(&self) -> bool {
  false
  }

src/query/catalog/src/table.rs

@@ -279,11 +279,23 @@ pub trait TableContext: Send + Sync {
  max_files: Option<usize>,
  ) -> Result<FilteredCopyFiles>;
 
- fn add_segment_location(&self, segment_loc: Location) -> Result<()>;
+ fn add_written_segment_location(&self, segment_loc: Location) -> Result<()>;
 
- fn clear_segment_locations(&self) -> Result<()>;
+ fn clear_written_segment_locations(&self) -> Result<()>;
 
- fn get_segment_locations(&self) -> Result<Vec<Location>>;
+ fn get_written_segment_locations(&self) -> Result<Vec<Location>>;
+
+ fn add_selected_segment_location(&self, _segment_loc: Location) {
+ unimplemented!()
+ }
+
+ fn get_selected_segment_locations(&self) -> Vec<Location> {
+ unimplemented!()
+ }
+
+ fn clear_selected_segment_locations(&self) {
+ unimplemented!()
+ }
 
  fn add_file_status(&self, file_path: &str, file_status: FileStatus) -> Result<()>;
 

src/query/catalog/src/table_context.rs

@@ -48,6 +48,7 @@ databend-enterprise-attach-table = { workspace = true }
 databend-enterprise-background-service = { workspace = true }
 databend-enterprise-data-mask-feature = { workspace = true }
 databend-enterprise-fail-safe = { workspace = true }
+databend-enterprise-hilbert-clustering = { workspace = true }
 databend-enterprise-inverted-index = { workspace = true }
 databend-enterprise-resources-management = { workspace = true }
 databend-enterprise-storage-encryption = { workspace = true }

src/query/ee/Cargo.toml

@@ -21,6 +21,7 @@ use crate::attach_table::RealAttachTableHandler;
 use crate::background_service::RealBackgroundService;
 use crate::data_mask::RealDatamaskHandler;
 use crate::fail_safe::RealFailSafeHandler;
+use crate::hilbert_clustering::RealHilbertClusteringHandler;
 use crate::inverted_index::RealInvertedIndexHandler;
 use crate::license::license_mgr::RealLicenseManager;
 use crate::resource_management::init_resources_management;
@@ -47,6 +48,7 @@ impl EnterpriseServices {
  RealStorageQuotaHandler::init(&cfg)?;
  RealFailSafeHandler::init()?;
  init_resources_management(&cfg).await?;
+ RealHilbertClusteringHandler::init()?;
  Ok(())
  }
 }