fix

src/query/sql/src/planner/binder/bind_table_reference/bind_table.rs

@@ -28,7 +28,7 @@ use databend_common_catalog::table_with_options::get_with_opt_max_batch_size;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_storages_view::view_table::QUERY;
-use databend_storages_common_table_meta::table::get_change_type;
+use databend_storages_common_table_meta::table::{get_change_type, is_hilbert_recluster};
 
 use crate::binder::util::TableIdentifier;
 use crate::binder::Binder;
@@ -208,6 +208,10 @@ impl Binder {
             return Ok((s_expr, new_bind_context));
         }
 
+        if is_hilbert_recluster(&table_name_alias) {
+            todo!()
+        }
+
         match table_meta.engine() {
             "VIEW" => {
                 // TODO(leiysky): this check is error-prone,

src/query/sql/src/planner/binder/ddl/table.rs

@@ -1264,7 +1264,7 @@ impl Binder {
             .collect::<Vec<_>>()
             .join(", ");
         let query = format!(
-            "WITH _keys_bound AS MATERIALIZED ( \
+            "WITH _keys_bound AS ( \
                 SELECT \
                     {keys_bounds_str} \
                 FROM {database_name}.{table_name} \

src/query/storages/common/table_meta/src/table/stream_keys.rs

@@ -82,3 +82,19 @@ pub fn get_change_type(table_alias_name: &Option<String>) -> Option<ChangeType>
     }
     change_type
 }
+
+pub fn is_hilbert_recluster(table_alias_name: &Option<String>) -> bool {
+    if let Some(table_alias) = table_alias_name {
+        let alias_param = table_alias.split('$').collect::<Vec<_>>();
+        if alias_param.len() == 2 && alias_param[1].len() == 8 {
+            if let Ok(suffix) = i64::from_str_radix(alias_param[1], 16) {
+                // 2023-01-01 00:00:00.
+                let base_timestamp = 1672502400;
+                if suffix > base_timestamp && alias_param[0] == "_compact"{
+                    return true;
+                }
+            }
+        }
+    }
+    false
+}