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

Cargo.toml

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

src/common/exception/src/exception_code.rs

@@ -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/license/src/license.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/query/catalog/src/plan/partition.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/table.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_context.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/ee/Cargo.toml

@@ -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/src/enterprise_services.rs

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