chore: bump version to 0.14.2 (#6032)

* chore: only retry when retry-able in flow (#5987)

* chore: only retry when retry-able

* chore: revert dbg change

* refactor: per review

* fix: check for available frontend first

* docs: more explain&longer timeout&feat: more retry at every level&try send select 1

* fix: use `sql` method for "SELECT 1"

* fix: also put recover flows in spawned task and a dead loop

* test: update transient error in flow rebuild test

* chore: sleep after sqlness sleep

* chore: add a warning

* chore: wait even more time after reboot

* fix: sanitize_connection_string (#6012)

* fix: disable recursion limit in prost (#6010)

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* ci: fix the bugs of release-dev-builder-images and add update-dev-builder-image-tag (#6009)

* fix: the dev-builder release job is not triggered by merged event

* ci: add update-dev-builder-image-tag

* fix: always create mito engine (#6018)

* fix: force streaming mode for instant source table (#6031)

* fix: force streaming mode for instant source table

* tests: sqlness test&refactor: get table

* refactor: per review

* chore: bump version to 0.14.2

---------

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>
Co-authored-by: jeremyhi <jiachun_feng@proton.me>
Co-authored-by: Ruihang Xia <waynestxia@gmail.com>
Co-authored-by: zyy17 <zyylsxm@gmail.com>
Co-authored-by: Lei, HUANG <6406592+v0y4g3r@users.noreply.github.com>

src/client/src/database.rs

@@ -36,8 +36,8 @@ use common_grpc::flight::{FlightDecoder, FlightMessage};
 use common_query::Output;
 use common_recordbatch::error::ExternalSnafu;
 use common_recordbatch::RecordBatchStreamWrapper;
-use common_telemetry::error;
 use common_telemetry::tracing_context::W3cTrace;
+use common_telemetry::{error, warn};
 use futures::future;
 use futures_util::{Stream, StreamExt, TryStreamExt};
 use prost::Message;
@@ -192,6 +192,36 @@ impl Database {
         from_grpc_response(response)
     }
 
+    /// Retry if connection fails, max_retries is the max number of retries, so the total wait time
+    /// is `max_retries * GRPC_CONN_TIMEOUT`
+    pub async fn handle_with_retry(&self, request: Request, max_retries: u32) -> Result<u32> {
+        let mut client = make_database_client(&self.client)?.inner;
+        let mut retries = 0;
+        let request = self.to_rpc_request(request);
+        loop {
+            let raw_response = client.handle(request.clone()).await;
+            match (raw_response, retries < max_retries) {
+                (Ok(resp), _) => return from_grpc_response(resp.into_inner()),
+                (Err(err), true) => {
+                    // determine if the error is retryable
+                    if is_grpc_retryable(&err) {
+                        // retry
+                        retries += 1;
+                        warn!("Retrying {} times with error = {:?}", retries, err);
+                        continue;
+                    }
+                }
+                (Err(err), false) => {
+                    error!(
+                        "Failed to send request to grpc handle after {} retries, error = {:?}",
+                        retries, err
+                    );
+                    return Err(err.into());
+                }
+            }
+        }
+    }
+
     #[inline]
     fn to_rpc_request(&self, request: Request) -> GreptimeRequest {
         GreptimeRequest {
@@ -368,6 +398,11 @@ impl Database {
     }
 }
 
+/// by grpc standard, only `Unavailable` is retryable, see: https://github.com/grpc/grpc/blob/master/doc/statuscodes.md#status-codes-and-their-use-in-grpc
+pub fn is_grpc_retryable(err: &tonic::Status) -> bool {
+    matches!(err.code(), tonic::Code::Unavailable)
+}
+
 #[derive(Default, Debug, Clone)]
 struct FlightContext {
     auth_header: Option<AuthHeader>,

src/cmd/src/metasrv.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt;
 use std::time::Duration;
 
 use async_trait::async_trait;
@@ -131,7 +132,7 @@ impl SubCommand {
     }
 }
 
-#[derive(Debug, Default, Parser)]
+#[derive(Default, Parser)]
 pub struct StartCommand {
     /// The address to bind the gRPC server.
     #[clap(long, alias = "bind-addr")]
@@ -171,6 +172,27 @@ pub struct StartCommand {
     backend: Option<BackendImpl>,
 }
 
+impl fmt::Debug for StartCommand {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("StartCommand")
+            .field("rpc_bind_addr", &self.rpc_bind_addr)
+            .field("rpc_server_addr", &self.rpc_server_addr)
+            .field("store_addrs", &self.sanitize_store_addrs())
+            .field("config_file", &self.config_file)
+            .field("selector", &self.selector)
+            .field("use_memory_store", &self.use_memory_store)
+            .field("enable_region_failover", &self.enable_region_failover)
+            .field("http_addr", &self.http_addr)
+            .field("http_timeout", &self.http_timeout)
+            .field("env_prefix", &self.env_prefix)
+            .field("data_home", &self.data_home)
+            .field("store_key_prefix", &self.store_key_prefix)
+            .field("max_txn_ops", &self.max_txn_ops)
+            .field("backend", &self.backend)
+            .finish()
+    }
+}
+
 impl StartCommand {
     pub fn load_options(&self, global_options: &GlobalOptions) -> Result<MetasrvOptions> {
         let mut opts = MetasrvOptions::load_layered_options(
@@ -184,6 +206,15 @@ impl StartCommand {
         Ok(opts)
     }
 
+    fn sanitize_store_addrs(&self) -> Option<Vec<String>> {
+        self.store_addrs.as_ref().map(|addrs| {
+            addrs
+                .iter()
+                .map(|addr| common_meta::kv_backend::util::sanitize_connection_string(addr))
+                .collect()
+        })
+    }
+
     // The precedence order is: cli > config file > environment variables > default values.
     fn merge_with_cli_options(
         &self,

src/common/meta/src/kv_backend.rs

@@ -35,7 +35,7 @@ pub mod memory;
 pub mod rds;
 pub mod test;
 pub mod txn;
-
+pub mod util;
 pub type KvBackendRef<E = Error> = Arc<dyn KvBackend<Error = E> + Send + Sync>;
 
 #[async_trait]

src/common/meta/src/kv_backend/util.rs

@@ -0,0 +1,85 @@
+// Copyright 2023 Greptime Team
+//
+// 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.
+
+/// Removes sensitive information like passwords from connection strings.
+///
+/// This function sanitizes connection strings by removing credentials:
+/// - For URL format (mysql://user:password@host:port/db): Removes everything before '@'
+/// - For parameter format (host=localhost password=secret): Removes the password parameter
+/// - For URL format without credentials (mysql://host:port/db): Removes the protocol prefix
+///
+/// # Arguments
+///
+/// * `conn_str` - The connection string to sanitize
+///
+/// # Returns
+///
+/// A sanitized version of the connection string with sensitive information removed
+pub fn sanitize_connection_string(conn_str: &str) -> String {
+    // Case 1: URL format with credentials (mysql://user:password@host:port/db)
+    // Extract everything after the '@' symbol
+    if let Some(at_pos) = conn_str.find('@') {
+        return conn_str[at_pos + 1..].to_string();
+    }
+
+    // Case 2: Parameter format with password (host=localhost password=secret dbname=mydb)
+    // Filter out any parameter that starts with "password="
+    if conn_str.contains("password=") {
+        return conn_str
+            .split_whitespace()
+            .filter(|param| !param.starts_with("password="))
+            .collect::<Vec<_>>()
+            .join(" ");
+    }
+
+    // Case 3: URL format without credentials (mysql://host:port/db)
+    // Extract everything after the protocol prefix
+    if let Some(host_part) = conn_str.split("://").nth(1) {
+        return host_part.to_string();
+    }
+
+    // Case 4: Already sanitized or unknown format
+    // Return as is
+    conn_str.to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_sanitize_connection_string() {
+        // Test URL format with username/password
+        let conn_str = "mysql://user:password123@localhost:3306/db";
+        assert_eq!(sanitize_connection_string(conn_str), "localhost:3306/db");
+
+        // Test URL format without credentials
+        let conn_str = "mysql://localhost:3306/db";
+        assert_eq!(sanitize_connection_string(conn_str), "localhost:3306/db");
+
+        // Test parameter format with password
+        let conn_str = "host=localhost port=5432 user=postgres password=secret dbname=mydb";
+        assert_eq!(
+            sanitize_connection_string(conn_str),
+            "host=localhost port=5432 user=postgres dbname=mydb"
+        );
+
+        // Test parameter format without password
+        let conn_str = "host=localhost port=5432 user=postgres dbname=mydb";
+        assert_eq!(
+            sanitize_connection_string(conn_str),
+            "host=localhost port=5432 user=postgres dbname=mydb"
+        );
+    }
+}

src/datanode/src/datanode.rs

@@ -398,45 +398,46 @@ impl DatanodeBuilder {
         schema_metadata_manager: SchemaMetadataManagerRef,
         plugins: Plugins,
     ) -> Result<Vec<RegionEngineRef>> {
-        let mut engines = vec![];
-        let mut metric_engine_config = opts.region_engine.iter().find_map(|c| match c {
-            RegionEngineConfig::Metric(config) => Some(config.clone()),
-            _ => None,
-        });
+        let mut metric_engine_config = metric_engine::config::EngineConfig::default();
+        let mut mito_engine_config = MitoConfig::default();
+        let mut file_engine_config = file_engine::config::EngineConfig::default();
 
         for engine in &opts.region_engine {
             match engine {
                 RegionEngineConfig::Mito(config) => {
-                    let mito_engine = Self::build_mito_engine(
-                        opts,
-                        object_store_manager.clone(),
-                        config.clone(),
-                        schema_metadata_manager.clone(),
-                        plugins.clone(),
-                    )
-                    .await?;
-
-                    let metric_engine = MetricEngine::try_new(
-                        mito_engine.clone(),
-                        metric_engine_config.take().unwrap_or_default(),
-                    )
-                    .context(BuildMetricEngineSnafu)?;
-                    engines.push(Arc::new(mito_engine) as _);
-                    engines.push(Arc::new(metric_engine) as _);
+                    mito_engine_config = config.clone();
                 }
                 RegionEngineConfig::File(config) => {
-                    let engine = FileRegionEngine::new(
-                        config.clone(),
-                        object_store_manager.default_object_store().clone(), // TODO: implement custom storage for file engine
-                    );
-                    engines.push(Arc::new(engine) as _);
+                    file_engine_config = config.clone();
                 }
-                RegionEngineConfig::Metric(_) => {
-                    // Already handled in `build_mito_engine`.
+                RegionEngineConfig::Metric(metric_config) => {
+                    metric_engine_config = metric_config.clone();
                 }
             }
         }
-        Ok(engines)
+
+        let mito_engine = Self::build_mito_engine(
+            opts,
+            object_store_manager.clone(),
+            mito_engine_config,
+            schema_metadata_manager.clone(),
+            plugins.clone(),
+        )
+        .await?;
+
+        let metric_engine = MetricEngine::try_new(mito_engine.clone(), metric_engine_config)
+            .context(BuildMetricEngineSnafu)?;
+
+        let file_engine = FileRegionEngine::new(
+            file_engine_config,
+            object_store_manager.default_object_store().clone(), // TODO: implement custom storage for file engine
+        );
+
+        Ok(vec![
+            Arc::new(mito_engine) as _,
+            Arc::new(metric_engine) as _,
+            Arc::new(file_engine) as _,
+        ])
     }
 
     /// Builds [MitoEngine] according to options.

src/flow/src/adapter/flownode_impl.rs

@@ -37,11 +37,12 @@ use tokio::sync::{Mutex, RwLock};
 
 use crate::adapter::{CreateFlowArgs, StreamingEngine};
 use crate::batching_mode::engine::BatchingEngine;
+use crate::batching_mode::{FRONTEND_SCAN_TIMEOUT, MIN_REFRESH_DURATION};
 use crate::engine::FlowEngine;
 use crate::error::{
     CreateFlowSnafu, ExternalSnafu, FlowNotFoundSnafu, IllegalCheckTaskStateSnafu,
-    InsertIntoFlowSnafu, InternalSnafu, JoinTaskSnafu, ListFlowsSnafu, SyncCheckTaskSnafu,
-    UnexpectedSnafu,
+    InsertIntoFlowSnafu, InternalSnafu, JoinTaskSnafu, ListFlowsSnafu, NoAvailableFrontendSnafu,
+    SyncCheckTaskSnafu, UnexpectedSnafu,
 };
 use crate::metrics::METRIC_FLOW_TASK_COUNT;
 use crate::repr::{self, DiffRow};
@@ -81,6 +82,11 @@ impl FlowDualEngine {
         }
     }
 
+    /// Determine if the engine is in distributed mode
+    pub fn is_distributed(&self) -> bool {
+        self.streaming_engine.node_id.is_some()
+    }
+
     pub fn streaming_engine(&self) -> Arc<StreamingEngine> {
         self.streaming_engine.clone()
     }
@@ -89,6 +95,39 @@ impl FlowDualEngine {
         self.batching_engine.clone()
     }
 
+    /// In distributed mode, scan periodically(1s) until available frontend is found, or timeout,
+    /// in standalone mode, return immediately
+    /// notice here if any frontend appear in cluster info this function will return immediately
+    async fn wait_for_available_frontend(&self, timeout: std::time::Duration) -> Result<(), Error> {
+        if !self.is_distributed() {
+            return Ok(());
+        }
+        let frontend_client = self.batching_engine().frontend_client.clone();
+        let sleep_duration = std::time::Duration::from_millis(1_000);
+        let now = std::time::Instant::now();
+        loop {
+            let frontend_list = frontend_client.scan_for_frontend().await?;
+            if !frontend_list.is_empty() {
+                let fe_list = frontend_list
+                    .iter()
+                    .map(|(_, info)| &info.peer.addr)
+                    .collect::<Vec<_>>();
+                info!("Available frontend found: {:?}", fe_list);
+                return Ok(());
+            }
+            let elapsed = now.elapsed();
+            tokio::time::sleep(sleep_duration).await;
+            info!("Waiting for available frontend, elapsed={:?}", elapsed);
+            if elapsed >= timeout {
+                return NoAvailableFrontendSnafu {
+                    timeout,
+                    context: "No available frontend found in cluster info",
+                }
+                .fail();
+            }
+        }
+    }
+
     /// Try to sync with check task, this is only used in drop flow&flush flow, so a flow id is required
     ///
     /// the need to sync is to make sure flush flow actually get called
@@ -338,18 +377,36 @@ struct ConsistentCheckTask {
 
 impl ConsistentCheckTask {
     async fn start_check_task(engine: &Arc<FlowDualEngine>) -> Result<Self, Error> {
-        // first do recover flows
-        engine.check_flow_consistent(true, false).await?;
-
-        let inner = engine.clone();
+        let engine = engine.clone();
         let (tx, mut rx) = tokio::sync::mpsc::channel(1);
         let (trigger_tx, mut trigger_rx) =
             tokio::sync::mpsc::channel::<(bool, bool, tokio::sync::oneshot::Sender<()>)>(10);
         let handle = common_runtime::spawn_global(async move {
+            // first check if available frontend is found
+            if let Err(err) = engine
+                .wait_for_available_frontend(FRONTEND_SCAN_TIMEOUT)
+                .await
+            {
+                warn!("No frontend is available yet:\n {err:?}");
+            }
+
+            // then do recover flows, if failed, always retry
+            let mut recover_retry = 0;
+            while let Err(err) = engine.check_flow_consistent(true, false).await {
+                recover_retry += 1;
+                error!(
+                    "Failed to recover flows:\n {err:?}, retry {} in {}s",
+                    recover_retry,
+                    MIN_REFRESH_DURATION.as_secs()
+                );
+                tokio::time::sleep(MIN_REFRESH_DURATION).await;
+            }
+
+            // then do check flows, with configurable allow_create and allow_drop
             let (mut allow_create, mut allow_drop) = (false, false);
             let mut ret_signal: Option<tokio::sync::oneshot::Sender<()>> = None;
             loop {
-                if let Err(err) = inner.check_flow_consistent(allow_create, allow_drop).await {
+                if let Err(err) = engine.check_flow_consistent(allow_create, allow_drop).await {
                     error!(err; "Failed to check flow consistent");
                 }
                 if let Some(done) = ret_signal.take() {
@@ -534,7 +591,12 @@ impl FlowEngine for FlowDualEngine {
         match flow_type {
             Some(FlowType::Batching) => self.batching_engine.flush_flow(flow_id).await,
             Some(FlowType::Streaming) => self.streaming_engine.flush_flow(flow_id).await,
-            None => Ok(0),
+            None => {
+                warn!(
+                    "Currently flow={flow_id} doesn't exist in flownode, ignore flush_flow request"
+                );
+                Ok(0)
+            }
         }
     }
 

src/flow/src/batching_mode.rs

@@ -31,10 +31,19 @@ pub const DEFAULT_BATCHING_ENGINE_QUERY_TIMEOUT: Duration = Duration::from_secs(
 pub const SLOW_QUERY_THRESHOLD: Duration = Duration::from_secs(60);
 
 /// The minimum duration between two queries execution by batching mode task
-const MIN_REFRESH_DURATION: Duration = Duration::new(5, 0);
+pub const MIN_REFRESH_DURATION: Duration = Duration::new(5, 0);
 
 /// Grpc connection timeout
 const GRPC_CONN_TIMEOUT: Duration = Duration::from_secs(5);
 
 /// Grpc max retry number
 const GRPC_MAX_RETRIES: u32 = 3;
+
+/// Flow wait for available frontend timeout,
+/// if failed to find available frontend after FRONTEND_SCAN_TIMEOUT elapsed, return error
+/// which should prevent flownode from starting
+pub const FRONTEND_SCAN_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Frontend activity timeout
+/// if frontend is down(not sending heartbeat) for more than FRONTEND_ACTIVITY_TIMEOUT, it will be removed from the list that flownode use to connect
+pub const FRONTEND_ACTIVITY_TIMEOUT: Duration = Duration::from_secs(60);

src/flow/src/batching_mode/engine.rs

@@ -49,7 +49,8 @@ use crate::{CreateFlowArgs, Error, FlowId, TableName};
 pub struct BatchingEngine {
     tasks: RwLock<BTreeMap<FlowId, BatchingTask>>,
     shutdown_txs: RwLock<BTreeMap<FlowId, oneshot::Sender<()>>>,
-    frontend_client: Arc<FrontendClient>,
+    /// frontend client for insert request
+    pub(crate) frontend_client: Arc<FrontendClient>,
     flow_metadata_manager: FlowMetadataManagerRef,
     table_meta: TableMetadataManagerRef,
     catalog_manager: CatalogManagerRef,

src/flow/src/batching_mode/frontend_client.rs

@@ -15,6 +15,7 @@
 //! Frontend client to run flow as batching task which is time-window-aware normal query triggered every tick set by user
 
 use std::sync::{Arc, Weak};
+use std::time::SystemTime;
 
 use api::v1::greptime_request::Request;
 use api::v1::CreateTableExpr;
@@ -26,15 +27,17 @@ use common_meta::peer::Peer;
 use common_meta::rpc::store::RangeRequest;
 use common_query::Output;
 use common_telemetry::warn;
+use itertools::Itertools;
 use meta_client::client::MetaClient;
 use servers::query_handler::grpc::GrpcQueryHandler;
 use session::context::{QueryContextBuilder, QueryContextRef};
 use snafu::{OptionExt, ResultExt};
 
 use crate::batching_mode::{
-    DEFAULT_BATCHING_ENGINE_QUERY_TIMEOUT, GRPC_CONN_TIMEOUT, GRPC_MAX_RETRIES,
+    DEFAULT_BATCHING_ENGINE_QUERY_TIMEOUT, FRONTEND_ACTIVITY_TIMEOUT, GRPC_CONN_TIMEOUT,
+    GRPC_MAX_RETRIES,
 };
-use crate::error::{ExternalSnafu, InvalidRequestSnafu, UnexpectedSnafu};
+use crate::error::{ExternalSnafu, InvalidRequestSnafu, NoAvailableFrontendSnafu, UnexpectedSnafu};
 use crate::Error;
 
 /// Just like [`GrpcQueryHandler`] but use BoxedError
@@ -127,10 +130,24 @@ impl DatabaseWithPeer {
     fn new(database: Database, peer: Peer) -> Self {
         Self { database, peer }
     }
+
+    /// Try sending a "SELECT 1" to the database
+    async fn try_select_one(&self) -> Result<(), Error> {
+        // notice here use `sql` for `SELECT 1` return 1 row
+        let _ = self
+            .database
+            .sql("SELECT 1")
+            .await
+            .with_context(|_| InvalidRequestSnafu {
+                context: format!("Failed to handle `SELECT 1` request at {:?}", self.peer),
+            })?;
+        Ok(())
+    }
 }
 
 impl FrontendClient {
-    async fn scan_for_frontend(&self) -> Result<Vec<(NodeInfoKey, NodeInfo)>, Error> {
+    /// scan for available frontend from metadata
+    pub(crate) async fn scan_for_frontend(&self) -> Result<Vec<(NodeInfoKey, NodeInfo)>, Error> {
         let Self::Distributed { meta_client, .. } = self else {
             return Ok(vec![]);
         };
@@ -160,8 +177,8 @@ impl FrontendClient {
         Ok(res)
     }
 
-    /// Get the database with max `last_activity_ts`
-    async fn get_last_active_frontend(
+    /// Get the database with maximum `last_activity_ts`& is able to process query
+    async fn get_latest_active_frontend(
         &self,
         catalog: &str,
         schema: &str,
@@ -177,22 +194,50 @@ impl FrontendClient {
             .fail();
         };
 
-        let frontends = self.scan_for_frontend().await?;
-        let mut peer = None;
+        let mut interval = tokio::time::interval(GRPC_CONN_TIMEOUT);
+        interval.tick().await;
+        for retry in 0..GRPC_MAX_RETRIES {
+            let frontends = self.scan_for_frontend().await?;
+            let now_in_ms = SystemTime::now()
+                .duration_since(SystemTime::UNIX_EPOCH)
+                .unwrap()
+                .as_millis() as i64;
 
-        if let Some((_, val)) = frontends.iter().max_by_key(|(_, val)| val.last_activity_ts) {
-            peer = Some(val.peer.clone());
+            // found node with maximum last_activity_ts
+            for (_, node_info) in frontends
+                .iter()
+                .sorted_by_key(|(_, node_info)| node_info.last_activity_ts)
+                .rev()
+                // filter out frontend that have been down for more than 1 min
+                .filter(|(_, node_info)| {
+                    node_info.last_activity_ts + FRONTEND_ACTIVITY_TIMEOUT.as_millis() as i64
+                        > now_in_ms
+                })
+            {
+                let addr = &node_info.peer.addr;
+                let client = Client::with_manager_and_urls(chnl_mgr.clone(), vec![addr.clone()]);
+                let database = Database::new(catalog, schema, client);
+                let db = DatabaseWithPeer::new(database, node_info.peer.clone());
+                match db.try_select_one().await {
+                    Ok(_) => return Ok(db),
+                    Err(e) => {
+                        warn!(
+                            "Failed to connect to frontend {} on retry={}: \n{e:?}",
+                            addr, retry
+                        );
+                    }
+                }
+            }
+            // no available frontend
+            // sleep and retry
+            interval.tick().await;
         }
 
-        let Some(peer) = peer else {
-            UnexpectedSnafu {
-                reason: format!("No frontend available: {:?}", frontends),
-            }
-            .fail()?
-        };
-        let client = Client::with_manager_and_urls(chnl_mgr.clone(), vec![peer.addr.clone()]);
-        let database = Database::new(catalog, schema, client);
-        Ok(DatabaseWithPeer::new(database, peer))
+        NoAvailableFrontendSnafu {
+            timeout: GRPC_CONN_TIMEOUT,
+            context: "No available frontend found that is able to process query",
+        }
+        .fail()
     }
 
     pub async fn create(
@@ -222,38 +267,18 @@ impl FrontendClient {
     ) -> Result<u32, Error> {
         match self {
             FrontendClient::Distributed { .. } => {
-                let db = self.get_last_active_frontend(catalog, schema).await?;
+                let db = self.get_latest_active_frontend(catalog, schema).await?;
 
                 *peer_desc = Some(PeerDesc::Dist {
                     peer: db.peer.clone(),
                 });
 
-                let mut retry = 0;
-
-                loop {
-                    let ret = db.database.handle(req.clone()).await.with_context(|_| {
-                        InvalidRequestSnafu {
-                            context: format!("Failed to handle request: {:?}", req),
-                        }
-                    });
-                    if let Err(err) = ret {
-                        if retry < GRPC_MAX_RETRIES {
-                            retry += 1;
-                            warn!(
-                                "Failed to send request to grpc handle at Peer={:?}, retry = {}, error = {:?}",
-                                db.peer, retry, err
-                            );
-                            continue;
-                        } else {
-                            common_telemetry::error!(
-                                "Failed to send request to grpc handle at Peer={:?} after {} retries, error = {:?}",
-                                db.peer, retry, err
-                            );
-                            return Err(err);
-                        }
-                    }
-                    return ret;
-                }
+                db.database
+                    .handle_with_retry(req.clone(), GRPC_MAX_RETRIES)
+                    .await
+                    .with_context(|_| InvalidRequestSnafu {
+                        context: format!("Failed to handle request at {:?}: {:?}", db.peer, req),
+                    })
             }
             FrontendClient::Standalone { database_client } => {
                 let ctx = QueryContextBuilder::default()

src/flow/src/error.rs

@@ -61,6 +61,16 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display(
+        "No available frontend found after timeout: {timeout:?}, context: {context}"
+    ))]
+    NoAvailableFrontend {
+        timeout: std::time::Duration,
+        context: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("External error"))]
     External {
         source: BoxedError,
@@ -296,7 +306,8 @@ impl ErrorExt for Error {
             Self::Eval { .. }
             | Self::JoinTask { .. }
             | Self::Datafusion { .. }
-            | Self::InsertIntoFlow { .. } => StatusCode::Internal,
+            | Self::InsertIntoFlow { .. }
+            | Self::NoAvailableFrontend { .. } => StatusCode::Internal,
             Self::FlowAlreadyExist { .. } => StatusCode::TableAlreadyExists,
             Self::TableNotFound { .. }
             | Self::TableNotFoundMeta { .. }

src/flow/src/server.rs

@@ -172,6 +172,8 @@ impl FlownodeServer {
     }
 
     /// Start the background task for streaming computation.
+    ///
+    /// Should be called only after heartbeat is establish, hence can get cluster info
     async fn start_workers(&self) -> Result<(), Error> {
         let manager_ref = self.inner.flow_service.dual_engine.clone();
         let handle = manager_ref

src/meta-srv/src/metasrv.rs

@@ -14,7 +14,7 @@
 
 pub mod builder;
 
-use std::fmt::Display;
+use std::fmt::{self, Display};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::Duration;
@@ -96,7 +96,7 @@ pub enum BackendImpl {
     MysqlStore,
 }
 
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, PartialEq, Serialize, Deserialize)]
 #[serde(default)]
 pub struct MetasrvOptions {
     /// The address the server listens on.
@@ -166,6 +166,47 @@ pub struct MetasrvOptions {
     pub node_max_idle_time: Duration,
 }
 
+impl fmt::Debug for MetasrvOptions {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut debug_struct = f.debug_struct("MetasrvOptions");
+        debug_struct
+            .field("bind_addr", &self.bind_addr)
+            .field("server_addr", &self.server_addr)
+            .field("store_addrs", &self.sanitize_store_addrs())
+            .field("selector", &self.selector)
+            .field("use_memory_store", &self.use_memory_store)
+            .field("enable_region_failover", &self.enable_region_failover)
+            .field(
+                "allow_region_failover_on_local_wal",
+                &self.allow_region_failover_on_local_wal,
+            )
+            .field("http", &self.http)
+            .field("logging", &self.logging)
+            .field("procedure", &self.procedure)
+            .field("failure_detector", &self.failure_detector)
+            .field("datanode", &self.datanode)
+            .field("enable_telemetry", &self.enable_telemetry)
+            .field("data_home", &self.data_home)
+            .field("wal", &self.wal)
+            .field("export_metrics", &self.export_metrics)
+            .field("store_key_prefix", &self.store_key_prefix)
+            .field("max_txn_ops", &self.max_txn_ops)
+            .field("flush_stats_factor", &self.flush_stats_factor)
+            .field("tracing", &self.tracing)
+            .field("backend", &self.backend);
+
+        #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
+        debug_struct.field("meta_table_name", &self.meta_table_name);
+
+        #[cfg(feature = "pg_kvbackend")]
+        debug_struct.field("meta_election_lock_id", &self.meta_election_lock_id);
+
+        debug_struct
+            .field("node_max_idle_time", &self.node_max_idle_time)
+            .finish()
+    }
+}
+
 const DEFAULT_METASRV_ADDR_PORT: &str = "3002";
 
 impl Default for MetasrvOptions {
@@ -249,6 +290,13 @@ impl MetasrvOptions {
             common_telemetry::debug!("detect local IP is not supported on Android");
         }
     }
+
+    fn sanitize_store_addrs(&self) -> Vec<String> {
+        self.store_addrs
+            .iter()
+            .map(|addr| common_meta::kv_backend::util::sanitize_connection_string(addr))
+            .collect()
+    }
 }
 
 pub struct MetasrvInfo {

src/operator/src/statement/ddl.rs

@@ -37,7 +37,7 @@ use common_meta::rpc::ddl::{
 };
 use common_meta::rpc::router::{Partition, Partition as MetaPartition};
 use common_query::Output;
-use common_telemetry::{debug, info, tracing};
+use common_telemetry::{debug, info, tracing, warn};
 use common_time::Timezone;
 use datafusion_common::tree_node::TreeNodeVisitor;
 use datafusion_expr::LogicalPlan;
@@ -369,7 +369,7 @@ impl StatementExecutor {
         query_context: QueryContextRef,
     ) -> Result<SubmitDdlTaskResponse> {
         let flow_type = self
-            .determine_flow_type(&expr.sql, query_context.clone())
+            .determine_flow_type(&expr, query_context.clone())
             .await?;
         info!("determined flow={} type: {:#?}", expr.flow_name, flow_type);
 
@@ -398,9 +398,49 @@ impl StatementExecutor {
     /// Determine the flow type based on the SQL query
     ///
     /// If it contains aggregation or distinct, then it is a batch flow, otherwise it is a streaming flow
-    async fn determine_flow_type(&self, sql: &str, query_ctx: QueryContextRef) -> Result<FlowType> {
+    async fn determine_flow_type(
+        &self,
+        expr: &CreateFlowExpr,
+        query_ctx: QueryContextRef,
+    ) -> Result<FlowType> {
+        // first check if source table's ttl is instant, if it is, force streaming mode
+        for src_table_name in &expr.source_table_names {
+            let table = self
+                .catalog_manager()
+                .table(
+                    &src_table_name.catalog_name,
+                    &src_table_name.schema_name,
+                    &src_table_name.table_name,
+                    Some(&query_ctx),
+                )
+                .await
+                .map_err(BoxedError::new)
+                .context(ExternalSnafu)?
+                .with_context(|| TableNotFoundSnafu {
+                    table_name: format_full_table_name(
+                        &src_table_name.catalog_name,
+                        &src_table_name.schema_name,
+                        &src_table_name.table_name,
+                    ),
+                })?;
+
+            // instant source table can only be handled by streaming mode
+            if table.table_info().meta.options.ttl == Some(common_time::TimeToLive::Instant) {
+                warn!(
+                    "Source table `{}` for flow `{}`'s ttl=instant, fallback to streaming mode",
+                    format_full_table_name(
+                        &src_table_name.catalog_name,
+                        &src_table_name.schema_name,
+                        &src_table_name.table_name
+                    ),
+                    expr.flow_name
+                );
+                return Ok(FlowType::Streaming);
+            }
+        }
+
         let engine = &self.query_engine;
-        let stmt = QueryLanguageParser::parse_sql(sql, &query_ctx)
+        let stmt = QueryLanguageParser::parse_sql(&expr.sql, &query_ctx)
             .map_err(BoxedError::new)
             .context(ExternalSnafu)?;
         let plan = engine