feat: introduce reconcile table procedure

Signed-off-by: WenyXu <wenymedia@gmail.com>

Author: WenyXu

src/api/src/v1/column_def.rs

@@ -24,7 +24,7 @@ use greptime_proto::v1::{
 };
 use snafu::ResultExt;
 
-use crate::error::{self, Result};
+use crate::error::{self, ConvertColumnDefaultConstraintSnafu, Result};
 use crate::helper::ColumnDataTypeWrapper;
 use crate::v1::{ColumnDef, ColumnOptions, SemanticType};
 
@@ -77,6 +77,48 @@ pub fn try_as_column_schema(column_def: &ColumnDef) -> Result<ColumnSchema> {
         })
 }
 
+/// Tries to construct a `ColumnDef` from the given `ColumnSchema`.
+///
+/// TODO(weny): Add tests for this function.
+pub fn try_as_column_def(column_schema: &ColumnSchema, is_primary_key: bool) -> Result<ColumnDef> {
+    let column_datatype =
+        ColumnDataTypeWrapper::try_from(column_schema.data_type.clone()).map(|w| w.to_parts())?;
+
+    let semantic_type = if column_schema.is_time_index() {
+        SemanticType::Timestamp
+    } else if is_primary_key {
+        SemanticType::Tag
+    } else {
+        SemanticType::Field
+    } as i32;
+    let comment = column_schema
+        .metadata()
+        .get(COMMENT_KEY)
+        .cloned()
+        .unwrap_or_default();
+
+    let default_constraint = match column_schema.default_constraint() {
+        None => vec![],
+        Some(v) => v
+            .clone()
+            .try_into()
+            .context(ConvertColumnDefaultConstraintSnafu {
+                column: &column_schema.name,
+            })?,
+    };
+    let options = options_from_column_schema(column_schema);
+    Ok(ColumnDef {
+        name: column_schema.name.clone(),
+        data_type: column_datatype.0 as i32,
+        is_nullable: column_schema.is_nullable(),
+        default_constraint,
+        semantic_type,
+        comment,
+        datatype_extension: column_datatype.1,
+        options,
+    })
+}
+
 /// Constructs a `ColumnOptions` from the given `ColumnSchema`.
 pub fn options_from_column_schema(column_schema: &ColumnSchema) -> Option<ColumnOptions> {
     let mut options = ColumnOptions::default();

src/common/meta/src/ddl/alter_logical_tables/update_metadata.rs

@@ -13,12 +13,12 @@
 // limitations under the License.
 
 use common_grpc_expr::alter_expr_to_request;
-use itertools::Itertools;
 use snafu::ResultExt;
 use table::metadata::{RawTableInfo, TableInfo};
 
 use crate::ddl::alter_logical_tables::executor::AlterLogicalTablesExecutor;
 use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
+use crate::ddl::utils::table_info::batch_update_table_info_values;
 use crate::error;
 use crate::error::{ConvertAlterTableRequestSnafu, Result};
 use crate::key::table_info::TableInfoValue;
@@ -48,25 +48,8 @@ impl AlterLogicalTablesProcedure {
 
     pub(crate) async fn update_logical_tables_metadata(&mut self) -> Result<()> {
         let table_info_values = self.build_update_metadata()?;
-        let manager = &self.context.table_metadata_manager;
-        let chunk_size = manager.batch_update_table_info_value_chunk_size();
-        if table_info_values.len() > chunk_size {
-            let chunks = table_info_values
-                .into_iter()
-                .chunks(chunk_size)
-                .into_iter()
-                .map(|check| check.collect::<Vec<_>>())
-                .collect::<Vec<_>>();
-            for chunk in chunks {
-                manager.batch_update_table_info_values(chunk).await?;
-            }
-        } else {
-            manager
-                .batch_update_table_info_values(table_info_values)
-                .await?;
-        }
-
-        Ok(())
+        batch_update_table_info_values(&self.context.table_metadata_manager, table_info_values)
+            .await
     }
 
     pub(crate) fn build_update_metadata(

src/common/meta/src/ddl/alter_logical_tables/validator.rs

@@ -21,7 +21,9 @@ use store_api::storage::TableId;
 use table::table_reference::TableReference;
 
 use crate::ddl::utils::table_id::get_all_table_ids_by_names;
-use crate::ddl::utils::table_info::get_all_table_info_values_by_table_ids;
+use crate::ddl::utils::table_info::{
+    all_logical_table_routes_have_same_physical_id, get_all_table_info_values_by_table_ids,
+};
 use crate::error::{
     AlterLogicalTablesInvalidArgumentsSnafu, Result, TableInfoNotFoundSnafu,
     TableRouteNotFoundSnafu,
@@ -146,23 +148,16 @@ impl<'a> AlterLogicalTableValidator<'a> {
         table_route_manager: &TableRouteManager,
         table_ids: &[TableId],
     ) -> Result<()> {
-        let table_routes = table_route_manager
-            .table_route_storage()
-            .batch_get(table_ids)
+        let all_logical_table_routes_have_same_physical_id =
+            all_logical_table_routes_have_same_physical_id(
+                table_route_manager,
+                table_ids,
+                self.physical_table_id,
+            )
             .await?;
 
-        let physical_table_id = self.physical_table_id;
-
-        let is_same_physical_table = table_routes.iter().all(|r| {
-            if let Some(TableRouteValue::Logical(r)) = r {
-                r.physical_table_id() == physical_table_id
-            } else {
-                false
-            }
-        });
-
         ensure!(
-            is_same_physical_table,
+            all_logical_table_routes_have_same_physical_id,
             AlterLogicalTablesInvalidArgumentsSnafu {
                 err_msg: "All the tasks should have the same physical table id"
             }

src/common/meta/src/ddl/alter_table/executor.rs

@@ -308,6 +308,5 @@ fn build_new_table_info(
         "Built new table info: {:?} for table {}, table_id: {}",
         new_info.meta, table_name, table_id
     );
-
     Ok(new_info)
 }

src/common/meta/src/ddl/create_table_template.rs

@@ -21,8 +21,7 @@ use store_api::metric_engine_consts::LOGICAL_TABLE_METADATA_KEY;
 use store_api::storage::{RegionId, RegionNumber};
 use table::metadata::TableId;
 
-use crate::error;
-use crate::error::Result;
+use crate::error::{self, Result};
 use crate::wal_options_allocator::prepare_wal_options;
 
 pub(crate) fn build_template(create_table_expr: &CreateTableExpr) -> Result<CreateRequest> {

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

@@ -12,13 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use itertools::Itertools;
 use snafu::OptionExt;
 use store_api::storage::TableId;
+use table::metadata::RawTableInfo;
 use table::table_reference::TableReference;
 
 use crate::error::{Result, TableInfoNotFoundSnafu};
 use crate::key::table_info::{TableInfoManager, TableInfoValue};
-use crate::key::DeserializedValueWithBytes;
+use crate::key::table_route::{TableRouteManager, TableRouteValue};
+use crate::key::{DeserializedValueWithBytes, TableMetadataManager};
 
 /// Get all table info values by table ids.
 ///
@@ -42,3 +45,56 @@ pub(crate) async fn get_all_table_info_values_by_table_ids<'a>(
 
     Ok(table_info_values)
 }
+
+/// Checks if all the logical table routes have the same physical table id.
+pub(crate) async fn all_logical_table_routes_have_same_physical_id(
+    table_route_manager: &TableRouteManager,
+    table_ids: &[TableId],
+    physical_table_id: TableId,
+) -> Result<bool> {
+    let table_routes = table_route_manager
+        .table_route_storage()
+        .batch_get(table_ids)
+        .await?;
+
+    let is_same_physical_table = table_routes.iter().all(|r| {
+        if let Some(TableRouteValue::Logical(r)) = r {
+            r.physical_table_id() == physical_table_id
+        } else {
+            false
+        }
+    });
+
+    Ok(is_same_physical_table)
+}
+
+/// Batch updates the table info values.
+///
+/// The table info values are grouped into chunks, and each chunk is updated in a single transaction.
+///
+/// Returns an error if any table info value fails to update.
+pub(crate) async fn batch_update_table_info_values(
+    table_metadata_manager: &TableMetadataManager,
+    table_info_values: Vec<(DeserializedValueWithBytes<TableInfoValue>, RawTableInfo)>,
+) -> Result<()> {
+    let chunk_size = table_metadata_manager.batch_update_table_info_value_chunk_size();
+    if table_info_values.len() > chunk_size {
+        let chunks = table_info_values
+            .into_iter()
+            .chunks(chunk_size)
+            .into_iter()
+            .map(|check| check.collect::<Vec<_>>())
+            .collect::<Vec<_>>();
+        for chunk in chunks {
+            table_metadata_manager
+                .batch_update_table_info_values(chunk)
+                .await?;
+        }
+    } else {
+        table_metadata_manager
+            .batch_update_table_info_values(table_info_values)
+            .await?;
+    }
+
+    Ok(())
+}

src/common/meta/src/error.rs

@@ -878,6 +878,12 @@ pub enum Error {
         error: object_store::Error,
     },
 
+    #[snafu(display("Missing column ids"))]
+    MissingColumnIds {
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display(
         "Missing column in column metadata: {}, table: {}, table_id: {}",
         column_name,
@@ -907,6 +913,14 @@ pub enum Error {
         table_name: String,
         table_id: TableId,
     },
+
+    #[snafu(display("Failed to convert column def, column: {}", column))]
+    ConvertColumnDef {
+        column: String,
+        #[snafu(implicit)]
+        location: Location,
+        source: api::error::Error,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -928,6 +942,7 @@ impl ErrorExt for Error {
             NoLeader { .. } => StatusCode::TableUnavailable,
             ValueNotExist { .. }
             | ProcedurePoisonConflict { .. }
+            | MissingColumnIds { .. }
             | MissingColumnInColumnMetadata { .. }
             | MismatchColumnId { .. } => StatusCode::Unexpected,
 
@@ -1013,6 +1028,7 @@ impl ErrorExt for Error {
             AbortProcedure { source, .. } => source.status_code(),
             ConvertAlterTableRequest { source, .. } => source.status_code(),
             PutPoison { source, .. } => source.status_code(),
+            ConvertColumnDef { source, .. } => source.status_code(),
 
             ParseProcedureId { .. }
             | InvalidNumTopics { .. }

src/common/meta/src/lib.rs

@@ -32,13 +32,13 @@ pub mod key;
 pub mod kv_backend;
 pub mod leadership_notifier;
 pub mod lock_key;
-pub mod maintenance;
 pub mod metrics;
 pub mod node_expiry_listener;
 pub mod node_manager;
 pub mod peer;
 pub mod poison_key;
 pub mod range_stream;
+pub mod reconciliation;
 pub mod region_keeper;
 pub mod region_registry;
 pub mod rpc;

src/common/meta/src/maintenance.rs

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// TODO(weny): Remove it
+#[allow(dead_code)]
+pub(crate) mod reconcile_table;
 // TODO(weny): Remove it
 #[allow(dead_code)]
 pub(crate) mod utils;