refactor: add heartbeat to reduce meta requests for history tables (#18594)

* refactor: add heartbeat to reduce meta requests for history tables

* refactor: add heartbeat to reduce meta requests for history tables

* refactor: refine error handling

* refactor: break transform if exit heartbeat loop

* fixup

* fixup

* fix: external tables

Author: dqhl76

src/query/service/src/history_tables/alter_table.rs

@@ -22,6 +22,7 @@ use databend_common_catalog::table::Table;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::TableSchemaRef;
 use databend_common_sql::binder::parse_uri_location;
 use databend_common_sql::plans::Plan;
 use databend_common_sql::Planner;
@@ -30,28 +31,39 @@ use log::info;
 use crate::history_tables::external::ExternalStorageConnection;
 use crate::sessions::QueryContext;
 
-pub async fn get_alter_table_sql(
+pub async fn get_schemas(
     ctx: Arc<QueryContext>,
     new_create_sql: &str,
     table_name: &str,
-) -> Result<Vec<String>> {
-    let mut tracking_payload = ThreadTracker::new_tracking_payload();
-    tracking_payload.capture_log_settings = Some(CaptureLogSettings::capture_off());
-    let _guard = ThreadTracker::tracking(tracking_payload);
+) -> Result<(TableSchemaRef, TableSchemaRef)> {
     let old_table_schema = ThreadTracker::tracking_future(ctx.get_table(
         CATALOG_DEFAULT,
         "system_history",
         table_name,
     ))
     .await?
     .schema();
-    let (create_plan, _) = Planner::new(ctx.clone()).plan_sql(new_create_sql).await?;
+    let mut planner = Planner::new(ctx.clone());
+    let (create_plan, _) = ThreadTracker::tracking_future(planner.plan_sql(new_create_sql)).await?;
     let new_table_schema = match create_plan {
         Plan::CreateTable(plan) => plan.schema,
         _ => {
             unreachable!("logic error: expected CreateTable plan")
         }
     };
+    Ok((old_table_schema, new_table_schema))
+}
+
+pub async fn get_alter_table_sql(
+    ctx: Arc<QueryContext>,
+    new_create_sql: &str,
+    table_name: &str,
+) -> Result<Vec<String>> {
+    let mut tracking_payload = ThreadTracker::new_tracking_payload();
+    tracking_payload.capture_log_settings = Some(CaptureLogSettings::capture_off());
+    let _guard = ThreadTracker::tracking(tracking_payload);
+    let (old_table_schema, new_table_schema) =
+        ThreadTracker::tracking_future(get_schemas(ctx, new_create_sql, table_name)).await?;
     // The table schema change follow "open-closed principle", only accept adding new fields.
     // If the new table schema has less or equal fields than the old one, means older version
     // node restarted, we should not alter the table.

src/query/service/src/history_tables/error_handling.rs

@@ -0,0 +1,99 @@
+// 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.
+
+use std::cmp::min;
+
+use databend_common_exception::ErrorCode;
+
+/// Error counters for tracking persistent and temporary errors during history table operations
+#[derive(Debug, Default)]
+pub struct ErrorCounters {
+    persistent: u32,
+    temporary: u32,
+}
+
+impl ErrorCounters {
+    /// Create a new ErrorCounters instance with zero counts
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Reset both error counters to zero
+    pub fn reset(&mut self) {
+        self.persistent = 0;
+        self.temporary = 0;
+    }
+
+    /// Increment persistent error counter and return the new count
+    pub fn increment_persistent(&mut self) -> u32 {
+        self.persistent += 1;
+        self.persistent
+    }
+
+    /// Increment temporary error counter and return the new count
+    pub fn increment_temporary(&mut self) -> u32 {
+        self.temporary += 1;
+        self.temporary
+    }
+
+    /// Check if persistent error count has exceeded the maximum allowed attempts
+    pub fn persistent_exceeded_limit(&self) -> bool {
+        self.persistent > MAX_PERSISTENT_ERROR_ATTEMPTS
+    }
+
+    /// Calculate backoff duration in seconds for temporary errors using exponential backoff
+    pub fn calculate_temp_backoff(&self) -> u64 {
+        min(
+            2u64.saturating_pow(self.temporary),
+            MAX_TEMP_ERROR_BACKOFF_SECONDS,
+        )
+    }
+
+    /// Get current persistent error count
+    pub fn persistent_count(&self) -> u32 {
+        self.persistent
+    }
+
+    /// Get current temporary error count
+    pub fn temporary_count(&self) -> u32 {
+        self.temporary
+    }
+}
+
+/// Maximum number of persistent error attempts before giving up
+const MAX_PERSISTENT_ERROR_ATTEMPTS: u32 = 3;
+
+/// Maximum backoff time in seconds for temporary errors (10 minutes)
+const MAX_TEMP_ERROR_BACKOFF_SECONDS: u64 = 10 * 60;
+
+/// Check if the error is a temporary error that should be retried
+/// We will use this to determine if we should retry the operation.
+pub fn is_temp_error(e: &ErrorCode) -> bool {
+    let code = e.code();
+
+    // Storage and I/O errors are considered temporary errors
+    let storage = code == ErrorCode::STORAGE_NOT_FOUND
+        || code == ErrorCode::STORAGE_PERMISSION_DENIED
+        || code == ErrorCode::STORAGE_UNAVAILABLE
+        || code == ErrorCode::STORAGE_UNSUPPORTED
+        || code == ErrorCode::STORAGE_INSECURE
+        || code == ErrorCode::INVALID_OPERATION
+        || code == ErrorCode::STORAGE_OTHER;
+
+    // If acquire semaphore failed, we consider it a temporary error
+    let meta = code == ErrorCode::META_SERVICE_ERROR;
+    let transaction = code == ErrorCode::UNRESOLVABLE_CONFLICT;
+
+    storage || transaction || meta
+}