refactor(meta): extract utilities from monolithic util.rs (#18678)

Transform the large util.rs file (561 lines) into 8 focused utility modules
to improve code organization, maintainability, and developer experience.

PROBLEM:
- Monolithic util.rs with 561 lines and 16 scattered functions
- Inconsistent import patterns across 39 API files
- 165+ occurrences of duplicated transaction utilities
- Poor discoverability and maintainability

SOLUTION:
Applied incremental refactoring with backward compatibility:

EXTRACTED MODULES:
- txn_condition_util.rs - Transaction condition builders (txn_cond_eq_seq, txn_cond_seq)
- txn_retry_util.rs - Transaction retry patterns (txn_backoff re-export)
- serialization_util.rs - Data serialization (serialize_struct, deserialize_struct, etc.)
- error_util.rs - Error construction (6 functions from util.rs, database_util.rs, schema_api.rs)
- kv_fetch_util.rs - KV operations (get_u64_value, fetch_id, 4 others)
- txn_op_builder_util.rs - Transaction builders (txn_put_pb, txn_op_del, 3 others)
- txn_core_util.rs - Core transactions (send_txn, txn_delete_exact, txn_replace_exact)
- data_retention_util.rs - Retention logic (is_drop_time_retainable, get_retention_boundary)

Author: drmingdrmer

src/meta/api/src/data_mask_api_impl.rs

@@ -36,12 +36,12 @@ use crate::data_mask_api::DatamaskApi;
 use crate::fetch_id;
 use crate::kv_app_error::KVAppError;
 use crate::kv_pb_api::KVPbApi;
-use crate::send_txn;
 use crate::txn_backoff::txn_backoff;
-use crate::txn_cond_eq_seq;
-use crate::util::txn_delete_exact;
-use crate::util::txn_op_put_pb;
-use crate::util::txn_replace_exact;
+use crate::txn_condition_util::txn_cond_eq_seq;
+use crate::txn_core_util::send_txn;
+use crate::txn_core_util::txn_delete_exact;
+use crate::txn_core_util::txn_replace_exact;
+use crate::txn_op_builder_util::txn_op_put_pb;
 
 /// DatamaskApi is implemented upon kvapi::KVApi.
 /// Thus every type that impl kvapi::KVApi impls DatamaskApi.

src/meta/api/src/data_retention_util.rs

@@ -0,0 +1,43 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Data retention utilities extracted from util.rs
+
+use std::time::Duration;
+
+use chrono::DateTime;
+use chrono::Utc;
+
+const DEFAULT_DATA_RETENTION_SECONDS: i64 = 24 * 60 * 60;
+
+/// Determines if an item is within the retention period based on its drop time.
+///
+/// # Arguments
+/// * `drop_on` - The optional timestamp when the item was marked for deletion.
+/// * `now` - The current timestamp used as a reference point.
+///
+/// Items without a drop time (`None`) are always considered retainable.
+/// The retention period is defined by `DEFAULT_DATA_RETENTION_SECONDS`.
+pub fn is_drop_time_retainable(drop_on: Option<DateTime<Utc>>, now: DateTime<Utc>) -> bool {
+    let retention_boundary = get_retention_boundary(now);
+
+    // If it is None, fill it with a very big time.
+    let drop_on = drop_on.unwrap_or(DateTime::<Utc>::MAX_UTC);
+    drop_on > retention_boundary
+}
+
+/// Get the retention boundary time before which the data can be permanently removed.
+pub fn get_retention_boundary(now: DateTime<Utc>) -> DateTime<Utc> {
+    now - Duration::from_secs(DEFAULT_DATA_RETENTION_SECONDS as u64)
+}

src/meta/api/src/database_api.rs

@@ -55,21 +55,21 @@ use log::error;
 use log::warn;
 use seq_marked::SeqValue;
 
-use crate::database_util::db_has_to_not_exist;
+use crate::data_retention_util::is_drop_time_retainable;
 use crate::database_util::drop_database_meta;
 use crate::database_util::get_db_or_err;
 use crate::db_has_to_exist;
+use crate::error_util::db_has_to_not_exist;
 use crate::fetch_id;
 use crate::kv_app_error::KVAppError;
 use crate::kv_pb_api::KVPbApi;
-use crate::send_txn;
 use crate::serialize_struct;
 use crate::serialize_u64;
 use crate::txn_backoff::txn_backoff;
-use crate::txn_cond_seq;
+use crate::txn_condition_util::txn_cond_seq;
+use crate::txn_core_util::send_txn;
 use crate::txn_op_del;
 use crate::txn_op_put;
-use crate::util::is_drop_time_retainable;
 
 impl<KV> DatabaseApi for KV
 where

src/meta/api/src/database_util.rs

@@ -16,7 +16,6 @@ use std::fmt::Display;
 
 use chrono::Utc;
 use databend_common_meta_app::app_error::AppError;
-use databend_common_meta_app::app_error::DatabaseAlreadyExists;
 use databend_common_meta_app::app_error::DropDbWithDropTime;
 use databend_common_meta_app::schema::database_name_ident::DatabaseNameIdent;
 use databend_common_meta_app::schema::DatabaseId;
@@ -32,35 +31,13 @@ use databend_common_meta_types::TxnRequest;
 use log::debug;
 use log::warn;
 
+use crate::error_util::unknown_database_error;
 use crate::kv_app_error::KVAppError;
 use crate::kv_pb_api::KVPbApi;
 use crate::serialize_struct;
-use crate::txn_cond_seq;
+use crate::txn_condition_util::txn_cond_seq;
 use crate::txn_op_del;
 use crate::txn_op_put;
-use crate::util::unknown_database_error;
-
-/// Return OK if a db_id or db_meta does not exist by checking the seq.
-///
-/// Otherwise returns DatabaseAlreadyExists error
-pub(crate) fn db_has_to_not_exist(
-    seq: u64,
-    name_ident: &DatabaseNameIdent,
-    ctx: impl Display,
-) -> Result<(), KVAppError> {
-    if seq == 0 {
-        Ok(())
-    } else {
-        debug!(seq = seq, name_ident :? =(name_ident); "exist");
-
-        Err(KVAppError::AppError(AppError::DatabaseAlreadyExists(
-            DatabaseAlreadyExists::new(
-                name_ident.database_name(),
-                format!("{}: {}", ctx, name_ident.display()),
-            ),
-        )))
-    }
-}
 
 pub(crate) async fn drop_database_meta(
     kv_api: &(impl kvapi::KVApi<Error = MetaError> + ?Sized),

src/meta/api/src/dictionary_api.rs

@@ -38,11 +38,11 @@ use crate::kv_app_error::KVAppError;
 use crate::kv_pb_api::KVPbApi;
 use crate::meta_txn_error::MetaTxnError;
 use crate::name_id_value_api::NameIdValueApi;
-use crate::send_txn;
 use crate::txn_backoff::txn_backoff;
-use crate::txn_cond_seq;
+use crate::txn_condition_util::txn_cond_seq;
+use crate::txn_core_util::send_txn;
+use crate::txn_op_builder_util::txn_op_put_pb;
 use crate::txn_op_del;
-use crate::util::txn_op_put_pb;
 
 /// DictionaryApi defines APIs for dictionary management.
 ///

src/meta/api/src/error_util.rs

@@ -0,0 +1,108 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Error construction utilities extracted from util.rs and other modules
+
+use std::fmt::Display;
+
+use databend_common_meta_app::app_error::AppError;
+use databend_common_meta_app::app_error::DatabaseAlreadyExists;
+use databend_common_meta_app::app_error::TableAlreadyExists;
+use databend_common_meta_app::app_error::UnknownDatabase;
+use databend_common_meta_app::app_error::UnknownDatabaseId;
+use databend_common_meta_app::app_error::UnknownTable;
+use databend_common_meta_app::schema::database_name_ident::DatabaseNameIdent;
+use databend_common_meta_app::schema::TableNameIdent;
+
+use crate::kv_app_error::KVAppError;
+
+pub fn unknown_database_error(db_name_ident: &DatabaseNameIdent, msg: impl Display) -> AppError {
+    let e = UnknownDatabase::new(
+        db_name_ident.database_name(),
+        format!("{}: {}", msg, db_name_ident.display()),
+    );
+    AppError::UnknownDatabase(e)
+}
+
+pub fn db_has_to_exist(
+    seq: u64,
+    db_name_ident: &DatabaseNameIdent,
+    msg: impl Display,
+) -> Result<(), KVAppError> {
+    if seq == 0 {
+        Err(KVAppError::AppError(unknown_database_error(
+            db_name_ident,
+            msg,
+        )))
+    } else {
+        Ok(())
+    }
+}
+
+pub fn db_id_has_to_exist(seq: u64, db_id: u64, msg: impl Display) -> Result<(), KVAppError> {
+    if seq == 0 {
+        let app_err = AppError::UnknownDatabaseId(UnknownDatabaseId::new(
+            db_id,
+            format!("{}: {}", msg, db_id),
+        ));
+        Err(KVAppError::AppError(app_err))
+    } else {
+        Ok(())
+    }
+}
+
+pub fn assert_table_exist(
+    seq: u64,
+    name_ident: &TableNameIdent,
+    ctx: impl Display,
+) -> Result<(), AppError> {
+    if seq > 0 {
+        return Ok(());
+    }
+    Err(AppError::UnknownTable(UnknownTable::new(
+        &name_ident.table_name,
+        format!("{}: {}", ctx, name_ident),
+    )))
+}
+
+pub fn db_has_to_not_exist(
+    seq: u64,
+    name_ident: &DatabaseNameIdent,
+    ctx: impl Display,
+) -> Result<(), KVAppError> {
+    if seq == 0 {
+        Ok(())
+    } else {
+        Err(KVAppError::AppError(AppError::DatabaseAlreadyExists(
+            DatabaseAlreadyExists::new(
+                name_ident.database_name(),
+                format!("{}: {}", ctx, name_ident.display()),
+            ),
+        )))
+    }
+}
+
+pub fn table_has_to_not_exist(
+    seq: u64,
+    name_ident: &TableNameIdent,
+    ctx: impl Display,
+) -> Result<(), KVAppError> {
+    if seq == 0 {
+        Ok(())
+    } else {
+        Err(KVAppError::AppError(AppError::TableAlreadyExists(
+            TableAlreadyExists::new(&name_ident.table_name, format!("{}: {}", ctx, name_ident)),
+        )))
+    }
+}

src/meta/api/src/garbage_collection_api.rs

@@ -54,15 +54,15 @@ use log::info;
 use crate::index_api::IndexApi;
 use crate::kv_app_error::KVAppError;
 use crate::kv_pb_api::KVPbApi;
-use crate::send_txn;
 use crate::txn_backoff::txn_backoff;
-use crate::txn_cond_eq_seq;
+use crate::txn_condition_util::txn_cond_eq_seq;
+use crate::txn_core_util::send_txn;
+use crate::txn_core_util::txn_delete_exact;
+use crate::txn_core_util::txn_replace_exact;
+use crate::txn_op_builder_util::txn_op_put_pb;
+use crate::txn_op_builder_util::txn_put_pb;
 use crate::txn_op_del;
 use crate::txn_op_get;
-use crate::util::txn_delete_exact;
-use crate::util::txn_op_put_pb;
-use crate::util::txn_put_pb;
-use crate::util::txn_replace_exact;
 
 /// GarbageCollectionApi defines APIs for garbage collection operations.
 ///

src/meta/api/src/index_api.rs

@@ -62,15 +62,15 @@ use crate::meta_txn_error::MetaTxnError;
 use crate::name_id_value_api::NameIdValueApi;
 use crate::schema_api::mark_index_as_deleted;
 use crate::schema_api::mark_table_index_as_deleted;
-use crate::send_txn;
 use crate::serialize_struct;
 use crate::txn_backoff::txn_backoff;
-use crate::txn_cond_eq_seq;
-use crate::txn_cond_seq;
+use crate::txn_condition_util::txn_cond_eq_seq;
+use crate::txn_condition_util::txn_cond_seq;
+use crate::txn_core_util::send_txn;
+use crate::txn_core_util::txn_delete_exact;
+use crate::txn_op_builder_util::txn_op_put_pb;
 use crate::txn_op_del;
 use crate::txn_op_put;
-use crate::util::txn_delete_exact;
-use crate::util::txn_op_put_pb;
 
 /// IndexApi defines APIs for index management and metadata.
 ///