fix: handle remaining buf

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>
chore: add wait_initialized method for frontend client (#7414 )
2025-12-26 16:10:02 +00:00 · 2025-12-17 20:36:54 +08:00 · 2025-12-17 08:13:36 +00:00 · 2025-12-17 07:14:25 +00:00 · 2025-12-17 06:34:29 +00:00 · 2025-12-17 04:18:23 +00:00
11 changed files with 269 additions and 120 deletions
--- a/.github/actions/setup-greptimedb-cluster/action.yml
+++ b/.github/actions/setup-greptimedb-cluster/action.yml
@@ -51,7 +51,7 @@ runs:
    run: |
      helm upgrade \
        --install my-greptimedb \
-        --set meta.backendStorage.etcd.endpoints=${{ inputs.etcd-endpoints }} \
+        --set 'meta.backendStorage.etcd.endpoints[0]=${{ inputs.etcd-endpoints }}' \
        --set meta.enableRegionFailover=${{ inputs.enable-region-failover }} \
        --set image.registry=${{ inputs.image-registry }} \
        --set image.repository=${{ inputs.image-repository }}  \
--- a/.github/scripts/deploy-greptimedb.sh
+++ b/.github/scripts/deploy-greptimedb.sh
@@ -81,7 +81,7 @@ function deploy_greptimedb_cluster() {
    --create-namespace \
    --set image.tag="$GREPTIMEDB_IMAGE_TAG" \
    --set initializer.tag="$GREPTIMEDB_INITIALIZER_IMAGE_TAG" \
-    --set meta.backendStorage.etcd.endpoints="etcd.$install_namespace:2379" \
+    --set "meta.backendStorage.etcd.endpoints[0]=etcd.$install_namespace.svc.cluster.local:2379" \
    --set meta.backendStorage.etcd.storeKeyPrefix="$cluster_name" \
    -n "$install_namespace"

@@ -119,7 +119,7 @@ function deploy_greptimedb_cluster_with_s3_storage() {
    --create-namespace \
    --set image.tag="$GREPTIMEDB_IMAGE_TAG" \
    --set initializer.tag="$GREPTIMEDB_INITIALIZER_IMAGE_TAG" \
-    --set meta.backendStorage.etcd.endpoints="etcd.$install_namespace:2379" \
+    --set "meta.backendStorage.etcd.endpoints[0]=etcd.$install_namespace.svc.cluster.local:2379" \
    --set meta.backendStorage.etcd.storeKeyPrefix="$cluster_name" \
    --set objectStorage.s3.bucket="$AWS_CI_TEST_BUCKET" \
    --set objectStorage.s3.region="$AWS_REGION" \
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -212,7 +212,7 @@ checksum = "d301b3b94cb4b2f23d7917810addbbaff90738e0ca2be692bd027e70d7e0330c"

 [[package]]
 name = "api"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "arrow-schema",
 "common-base",
@@ -733,7 +733,7 @@ dependencies = [

 [[package]]
 name = "auth"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-trait",
@@ -1383,7 +1383,7 @@ dependencies = [

 [[package]]
 name = "cache"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "catalog",
 "common-error",
@@ -1418,7 +1418,7 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"

 [[package]]
 name = "catalog"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "arrow",
@@ -1763,7 +1763,7 @@ checksum = "b94f61472cee1439c0b966b47e3aca9ae07e45d070759512cd390ea2bebc6675"

 [[package]]
 name = "cli"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -1816,7 +1816,7 @@ dependencies = [

 [[package]]
 name = "client"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "arc-swap",
@@ -1849,7 +1849,7 @@ dependencies = [
 "snafu 0.8.6",
 "store-api",
 "substrait 0.37.3",
- "substrait 1.0.0-beta.2",
+ "substrait 1.0.0-beta.3",
 "tokio",
 "tokio-stream",
 "tonic 0.13.1",
@@ -1889,7 +1889,7 @@ dependencies = [

 [[package]]
 name = "cmd"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-trait",
 "auth",
@@ -2023,7 +2023,7 @@ checksum = "55b672471b4e9f9e95499ea597ff64941a309b2cdbffcc46f2cc5e2d971fd335"

 [[package]]
 name = "common-base"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "anymap2",
 "async-trait",
@@ -2047,14 +2047,14 @@ dependencies = [

 [[package]]
 name = "common-catalog"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "const_format",
 ]

 [[package]]
 name = "common-config"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "common-base",
 "common-error",
@@ -2079,7 +2079,7 @@ dependencies = [

 [[package]]
 name = "common-datasource"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "arrow",
 "arrow-schema",
@@ -2114,7 +2114,7 @@ dependencies = [

 [[package]]
 name = "common-decimal"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "bigdecimal 0.4.8",
 "common-error",
@@ -2127,7 +2127,7 @@ dependencies = [

 [[package]]
 name = "common-error"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "common-macro",
 "http 1.3.1",
@@ -2138,7 +2138,7 @@ dependencies = [

 [[package]]
 name = "common-event-recorder"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-trait",
@@ -2160,7 +2160,7 @@ dependencies = [

 [[package]]
 name = "common-frontend"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-trait",
@@ -2182,7 +2182,7 @@ dependencies = [

 [[package]]
 name = "common-function"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "ahash 0.8.12",
 "api",
@@ -2242,7 +2242,7 @@ dependencies = [

 [[package]]
 name = "common-greptimedb-telemetry"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-trait",
 "common-runtime",
@@ -2259,7 +2259,7 @@ dependencies = [

 [[package]]
 name = "common-grpc"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "arrow-flight",
@@ -2294,7 +2294,7 @@ dependencies = [

 [[package]]
 name = "common-grpc-expr"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "common-base",
@@ -2314,7 +2314,7 @@ dependencies = [

 [[package]]
 name = "common-macro"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "greptime-proto",
 "once_cell",
@@ -2325,7 +2325,7 @@ dependencies = [

 [[package]]
 name = "common-mem-prof"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "anyhow",
 "common-error",
@@ -2341,7 +2341,7 @@ dependencies = [

 [[package]]
 name = "common-memory-manager"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "common-error",
 "common-macro",
@@ -2354,7 +2354,7 @@ dependencies = [

 [[package]]
 name = "common-meta"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "anymap2",
 "api",
@@ -2426,7 +2426,7 @@ dependencies = [

 [[package]]
 name = "common-options"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "common-grpc",
 "humantime-serde",
@@ -2435,11 +2435,11 @@ dependencies = [

 [[package]]
 name = "common-plugins"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"

 [[package]]
 name = "common-pprof"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "common-error",
 "common-macro",
@@ -2451,7 +2451,7 @@ dependencies = [

 [[package]]
 name = "common-procedure"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-stream",
@@ -2480,7 +2480,7 @@ dependencies = [

 [[package]]
 name = "common-procedure-test"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-trait",
 "common-procedure",
@@ -2490,7 +2490,7 @@ dependencies = [

 [[package]]
 name = "common-query"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-trait",
@@ -2516,7 +2516,7 @@ dependencies = [

 [[package]]
 name = "common-recordbatch"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "arc-swap",
 "common-base",
@@ -2540,7 +2540,7 @@ dependencies = [

 [[package]]
 name = "common-runtime"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-trait",
 "clap 4.5.40",
@@ -2569,7 +2569,7 @@ dependencies = [

 [[package]]
 name = "common-session"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "serde",
 "strum 0.27.1",
@@ -2577,7 +2577,7 @@ dependencies = [

 [[package]]
 name = "common-sql"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "common-base",
 "common-decimal",
@@ -2595,7 +2595,7 @@ dependencies = [

 [[package]]
 name = "common-stat"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "common-base",
 "common-runtime",
@@ -2610,7 +2610,7 @@ dependencies = [

 [[package]]
 name = "common-telemetry"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "backtrace",
 "common-base",
@@ -2639,7 +2639,7 @@ dependencies = [

 [[package]]
 name = "common-test-util"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "client",
 "common-grpc",
@@ -2652,7 +2652,7 @@ dependencies = [

 [[package]]
 name = "common-time"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "arrow",
 "chrono",
@@ -2670,7 +2670,7 @@ dependencies = [

 [[package]]
 name = "common-version"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "build-data",
 "cargo-manifest",
@@ -2681,7 +2681,7 @@ dependencies = [

 [[package]]
 name = "common-wal"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "common-base",
 "common-error",
@@ -2704,7 +2704,7 @@ dependencies = [

 [[package]]
 name = "common-workload"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "common-telemetry",
 "serde",
@@ -4012,7 +4012,7 @@ dependencies = [

 [[package]]
 name = "datanode"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "arrow-flight",
@@ -4076,7 +4076,7 @@ dependencies = [

 [[package]]
 name = "datatypes"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4750,7 +4750,7 @@ checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be"

 [[package]]
 name = "file-engine"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-trait",
@@ -4882,7 +4882,7 @@ checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8"

 [[package]]
 name = "flow"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "arrow",
@@ -4951,7 +4951,7 @@ dependencies = [
 "sql",
 "store-api",
 "strum 0.27.1",
- "substrait 1.0.0-beta.2",
+ "substrait 1.0.0-beta.3",
 "table",
 "tokio",
 "tonic 0.13.1",
@@ -5012,7 +5012,7 @@ checksum = "28dd6caf6059519a65843af8fe2a3ae298b14b80179855aeb4adc2c1934ee619"

 [[package]]
 name = "frontend"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "arc-swap",
@@ -6227,7 +6227,7 @@ dependencies = [

 [[package]]
 name = "index"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-trait",
 "asynchronous-codec",
@@ -7168,7 +7168,7 @@ checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94"

 [[package]]
 name = "log-query"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "chrono",
 "common-error",
@@ -7180,7 +7180,7 @@ dependencies = [

 [[package]]
 name = "log-store"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -7481,7 +7481,7 @@ dependencies = [

 [[package]]
 name = "meta-client"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-trait",
@@ -7509,7 +7509,7 @@ dependencies = [

 [[package]]
 name = "meta-srv"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-trait",
@@ -7609,7 +7609,7 @@ dependencies = [

 [[package]]
 name = "metric-engine"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "aquamarine",
@@ -7706,7 +7706,7 @@ dependencies = [

 [[package]]
 name = "mito-codec"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "bytes",
@@ -7731,7 +7731,7 @@ dependencies = [

 [[package]]
 name = "mito2"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "aquamarine",
@@ -8471,7 +8471,7 @@ dependencies = [

 [[package]]
 name = "object-store"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "anyhow",
 "bytes",
@@ -8756,7 +8756,7 @@ dependencies = [

 [[package]]
 name = "operator"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "ahash 0.8.12",
 "api",
@@ -8816,7 +8816,7 @@ dependencies = [
 "sql",
 "sqlparser",
 "store-api",
- "substrait 1.0.0-beta.2",
+ "substrait 1.0.0-beta.3",
 "table",
 "tokio",
 "tokio-util",
@@ -9102,7 +9102,7 @@ dependencies = [

 [[package]]
 name = "partition"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-trait",
@@ -9459,7 +9459,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"

 [[package]]
 name = "pipeline"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "ahash 0.8.12",
 "api",
@@ -9615,7 +9615,7 @@ dependencies = [

 [[package]]
 name = "plugins"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "auth",
 "catalog",
@@ -9917,7 +9917,7 @@ dependencies = [

 [[package]]
 name = "promql"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "ahash 0.8.12",
 "async-trait",
@@ -10200,7 +10200,7 @@ dependencies = [

 [[package]]
 name = "puffin"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-compression 0.4.19",
 "async-trait",
@@ -10242,7 +10242,7 @@ dependencies = [

 [[package]]
 name = "query"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "ahash 0.8.12",
 "api",
@@ -10309,7 +10309,7 @@ dependencies = [
 "sql",
 "sqlparser",
 "store-api",
- "substrait 1.0.0-beta.2",
+ "substrait 1.0.0-beta.3",
 "table",
 "tokio",
 "tokio-stream",
@@ -11651,7 +11651,7 @@ dependencies = [

 [[package]]
 name = "servers"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "ahash 0.8.12",
 "api",
@@ -11779,7 +11779,7 @@ dependencies = [

 [[package]]
 name = "session"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "ahash 0.8.12",
 "api",
@@ -12113,7 +12113,7 @@ dependencies = [

 [[package]]
 name = "sql"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "arrow-buffer",
@@ -12173,7 +12173,7 @@ dependencies = [

 [[package]]
 name = "sqlness-runner"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-trait",
 "clap 4.5.40",
@@ -12450,7 +12450,7 @@ dependencies = [

 [[package]]
 name = "standalone"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-trait",
 "catalog",
@@ -12491,7 +12491,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f"

 [[package]]
 name = "store-api"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "aquamarine",
@@ -12704,7 +12704,7 @@ dependencies = [

 [[package]]
 name = "substrait"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "async-trait",
 "bytes",
@@ -12827,7 +12827,7 @@ dependencies = [

 [[package]]
 name = "table"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "async-trait",
@@ -13096,7 +13096,7 @@ checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683"

 [[package]]
 name = "tests-fuzz"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "arbitrary",
 "async-trait",
@@ -13140,7 +13140,7 @@ dependencies = [

 [[package]]
 name = "tests-integration"
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 dependencies = [
 "api",
 "arrow-flight",
@@ -13215,7 +13215,7 @@ dependencies = [
 "sqlx",
 "standalone",
 "store-api",
- "substrait 1.0.0-beta.2",
+ "substrait 1.0.0-beta.3",
 "table",
 "tempfile",
 "time",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -75,7 +75,7 @@ members = [
 resolver = "2"

 [workspace.package]
-version = "1.0.0-beta.2"
+version = "1.0.0-beta.3"
 edition = "2024"
 license = "Apache-2.0"

--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -552,9 +552,8 @@ impl StartCommand {
        let grpc_handler = fe_instance.clone() as Arc<dyn GrpcQueryHandlerWithBoxedError>;
        let weak_grpc_handler = Arc::downgrade(&grpc_handler);
        frontend_instance_handler
-            .lock()
-            .unwrap()
-            .replace(weak_grpc_handler);
+            .set_handler(weak_grpc_handler)
+            .await;

        // set the frontend invoker for flownode
        let flow_streaming_engine = flownode.flow_engine().streaming_engine();
--- a/src/flow/src/batching_mode/frontend_client.rs
+++ b/src/flow/src/batching_mode/frontend_client.rs
@@ -15,7 +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::collections::HashMap;
-use std::sync::{Arc, Weak};
+use std::sync::{Arc, Mutex, Weak};
 use std::time::SystemTime;

 use api::v1::greptime_request::Request;
@@ -38,6 +38,7 @@ use servers::query_handler::grpc::GrpcQueryHandler;
 use session::context::{QueryContextBuilder, QueryContextRef};
 use session::hints::READ_PREFERENCE_HINT;
 use snafu::{OptionExt, ResultExt};
+use tokio::sync::SetOnce;

 use crate::batching_mode::BatchingModeOptions;
 use crate::error::{
@@ -75,7 +76,19 @@ impl<E: ErrorExt + Send + Sync + 'static, T: GrpcQueryHandler<Error = E> + Send
    }
 }

-type HandlerMutable = Arc<std::sync::Mutex<Option<Weak<dyn GrpcQueryHandlerWithBoxedError>>>>;
+#[derive(Debug, Clone)]
+pub struct HandlerMutable {
+    handler: Arc<Mutex<Option<Weak<dyn GrpcQueryHandlerWithBoxedError>>>>,
+    is_initialized: Arc<SetOnce<()>>,
+}
+
+impl HandlerMutable {
+    pub async fn set_handler(&self, handler: Weak<dyn GrpcQueryHandlerWithBoxedError>) {
+        *self.handler.lock().unwrap() = Some(handler);
+        // Ignore the error, as we allow the handler to be set multiple times.
+        let _ = self.is_initialized.set(());
+    }
+}

 /// A simple frontend client able to execute sql using grpc protocol
 ///
@@ -100,7 +113,11 @@ pub enum FrontendClient {
 impl FrontendClient {
    /// Create a new empty frontend client, with a `HandlerMutable` to set the grpc handler later
    pub fn from_empty_grpc_handler(query: QueryOptions) -> (Self, HandlerMutable) {
-        let handler = Arc::new(std::sync::Mutex::new(None));
+        let is_initialized = Arc::new(SetOnce::new());
+        let handler = HandlerMutable {
+            handler: Arc::new(Mutex::new(None)),
+            is_initialized,
+        };
        (
            Self::Standalone {
                database_client: handler.clone(),
@@ -110,23 +127,13 @@ impl FrontendClient {
        )
    }

-    /// Check if the frontend client is initialized.
-    ///
-    /// In distributed mode, it is always initialized.
-    /// In standalone mode, it checks if the database client is set.
-    pub fn is_initialized(&self) -> bool {
-        match self {
-            FrontendClient::Distributed { .. } => true,
-            FrontendClient::Standalone {
-                database_client, ..
-            } => {
-                let guard = database_client.lock();
-                if let Ok(guard) = guard {
-                    guard.is_some()
-                } else {
-                    false
-                }
-            }
+    /// Waits until the frontend client is initialized.
+    pub async fn wait_initialized(&self) {
+        if let FrontendClient::Standalone {
+            database_client, ..
+        } = self
+        {
+            database_client.is_initialized.wait().await;
        }
    }

@@ -158,8 +165,14 @@ impl FrontendClient {
        grpc_handler: Weak<dyn GrpcQueryHandlerWithBoxedError>,
        query: QueryOptions,
    ) -> Self {
+        let is_initialized = Arc::new(SetOnce::new_with(Some(())));
+        let handler = HandlerMutable {
+            handler: Arc::new(Mutex::new(Some(grpc_handler))),
+            is_initialized: is_initialized.clone(),
+        };
+
        Self::Standalone {
-            database_client: Arc::new(std::sync::Mutex::new(Some(grpc_handler))),
+            database_client: handler,
            query,
        }
    }
@@ -341,6 +354,7 @@ impl FrontendClient {
                {
                    let database_client = {
                        database_client
+                            .handler
                            .lock()
                            .map_err(|e| {
                                UnexpectedSnafu {
@@ -418,6 +432,7 @@ impl FrontendClient {
                {
                    let database_client = {
                        database_client
+                            .handler
                            .lock()
                            .map_err(|e| {
                                UnexpectedSnafu {
@@ -480,3 +495,73 @@ impl std::fmt::Display for PeerDesc {
        }
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use common_query::Output;
+    use tokio::time::timeout;
+
+    use super::*;
+
+    #[derive(Debug)]
+    struct NoopHandler;
+
+    #[async_trait::async_trait]
+    impl GrpcQueryHandlerWithBoxedError for NoopHandler {
+        async fn do_query(
+            &self,
+            _query: Request,
+            _ctx: QueryContextRef,
+        ) -> std::result::Result<Output, BoxedError> {
+            Ok(Output::new_with_affected_rows(0))
+        }
+    }
+
+    #[tokio::test]
+    async fn wait_initialized() {
+        let (client, handler_mut) =
+            FrontendClient::from_empty_grpc_handler(QueryOptions::default());
+
+        assert!(
+            timeout(Duration::from_millis(50), client.wait_initialized())
+                .await
+                .is_err()
+        );
+
+        let handler: Arc<dyn GrpcQueryHandlerWithBoxedError> = Arc::new(NoopHandler);
+        handler_mut.set_handler(Arc::downgrade(&handler)).await;
+
+        timeout(Duration::from_secs(1), client.wait_initialized())
+            .await
+            .expect("wait_initialized should complete after handler is set");
+
+        timeout(Duration::from_millis(10), client.wait_initialized())
+            .await
+            .expect("wait_initialized should be a no-op once initialized");
+
+        let handler: Arc<dyn GrpcQueryHandlerWithBoxedError> = Arc::new(NoopHandler);
+        let client =
+            FrontendClient::from_grpc_handler(Arc::downgrade(&handler), QueryOptions::default());
+        assert!(
+            timeout(Duration::from_millis(10), client.wait_initialized())
+                .await
+                .is_ok()
+        );
+
+        let meta_client = Arc::new(MetaClient::default());
+        let client = FrontendClient::from_meta_client(
+            meta_client,
+            None,
+            QueryOptions::default(),
+            BatchingModeOptions::default(),
+        )
+        .unwrap();
+        assert!(
+            timeout(Duration::from_millis(10), client.wait_initialized())
+                .await
+                .is_ok()
+        );
+    }
+}
--- a/src/index/Cargo.toml
+++ b/src/index/Cargo.toml
@@ -7,6 +7,9 @@ license.workspace = true
 [lints]
 workspace = true

+[features]
+vector_index = ["dep:usearch"]
+
 [dependencies]
 async-trait.workspace = true
 asynchronous-codec = "0.7.0"
@@ -41,7 +44,7 @@ tantivy = { version = "0.24", features = ["zstd-compression"] }
 tantivy-jieba = "0.16"
 tokio.workspace = true
 tokio-util.workspace = true
-usearch = { version = "2.21", default-features = false, features = ["fp16lib"] }
+usearch = { version = "2.21", default-features = false, features = ["fp16lib"], optional = true }
 uuid.workspace = true

 [dev-dependencies]
--- a/src/index/src/lib.rs
+++ b/src/index/src/lib.rs
@@ -22,6 +22,7 @@ pub mod external_provider;
 pub mod fulltext_index;
 pub mod inverted_index;
 pub mod target;
+#[cfg(feature = "vector_index")]
 pub mod vector;

 pub type Bytes = Vec<u8>;
--- a/src/object-store/src/config.rs
+++ b/src/object-store/src/config.rs
@@ -16,6 +16,7 @@ use std::time::Duration;

 use common_base::readable_size::ReadableSize;
 use common_base::secrets::{ExposeSecret, SecretString};
+use common_telemetry::tracing::warn;
 use opendal::services::{Azblob, Gcs, Oss, S3};
 use serde::{Deserialize, Serialize};

@@ -123,11 +124,18 @@ impl From<&S3Connection> for S3 {
    fn from(connection: &S3Connection) -> Self {
        let root = util::normalize_dir(&connection.root);

-        let mut builder = S3::default()
-            .root(&root)
-            .bucket(&connection.bucket)
-            .access_key_id(connection.access_key_id.expose_secret())
-            .secret_access_key(connection.secret_access_key.expose_secret());
+        let mut builder = S3::default().root(&root).bucket(&connection.bucket);
+
+        if !connection.access_key_id.expose_secret().is_empty()
+            && !connection.secret_access_key.expose_secret().is_empty()
+        {
+            builder = builder
+                .access_key_id(connection.access_key_id.expose_secret())
+                .secret_access_key(connection.secret_access_key.expose_secret());
+        } else {
+            warn!("No access key id or secret access key provided, using anonymous access");
+            builder = builder.allow_anonymous().disable_ec2_metadata();
+        }

        if let Some(endpoint) = &connection.endpoint {
            builder = builder.endpoint(endpoint);
--- a/src/query/src/part_sort.rs
+++ b/src/query/src/part_sort.rs
@@ -869,8 +869,16 @@ impl PartSortStream {

        // If we've processed all partitions, mark completion.
        if self.cur_part_idx >= self.partition_ranges.len() {
-            debug_assert!(remaining_range.num_rows() == 0);
+            // If there is remaining data here, it means the input data doesn't match the
+            // provided `PartitionRange`s (e.g. out-of-order input or mismatched ranges).
+            // In release builds, the previous `debug_assert!` would silently drop data and
+            // could lead to incorrect empty results. To keep query correctness, fall back
+            // to consuming the remaining data as part of the last range.
+            if remaining_range.num_rows() != 0 {
+                self.push_buffer(remaining_range)?;
+            }
            self.input_complete = true;
+            self.evaluating_batch = None;
            return Ok(());
        }

@@ -937,7 +945,12 @@ impl PartSortStream {
        // If we've processed all partitions, sort and output
        if self.cur_part_idx >= self.partition_ranges.len() {
            // assert there is no data beyond the last partition range (remaining is empty).
-            debug_assert!(remaining_range.num_rows() == 0);
+            // Similar to the TopK path, do not silently drop remaining data in release builds.
+            // If this happens, the input stream doesn't match `PartitionRange`s; include the
+            // remaining data for correctness.
+            if remaining_range.num_rows() != 0 {
+                self.push_buffer(remaining_range)?;
+            }

            // Sort and output the final group
            return self.sorted_buffer_if_non_empty();
@@ -999,11 +1012,11 @@ impl PartSortStream {
            {
                // Check if we've already processed all partitions
                if self.cur_part_idx >= self.partition_ranges.len() {
-                    // All partitions processed, discard remaining data
-                    if let Some(sorted_batch) = self.sorted_buffer_if_non_empty()? {
-                        return Poll::Ready(Some(Ok(sorted_batch)));
-                    }
-                    return Poll::Ready(None);
+                    // All partitions processed but we still have remaining data in-flight.
+                    // Don't discard it, otherwise we may incorrectly return an empty result.
+                    self.push_buffer(evaluating_batch)?;
+                    self.input_complete = true;
+                    continue;
                }

                if let Some(sorted_batch) = self.split_batch(evaluating_batch)? {
@@ -1431,6 +1444,47 @@ mod test {
        }
    }

+    #[tokio::test]
+    async fn topk_does_not_silently_drop_out_of_range_data() {
+        let unit = TimeUnit::Millisecond;
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "ts",
+            DataType::Timestamp(unit, None),
+            false,
+        )]));
+
+        // The input data is outside the provided PartitionRange.
+        // Historically this could lead to an empty result in release builds due to
+        // `debug_assert!`-only checks dropping the remaining batch.
+        let input_ranged_data = vec![(
+            PartitionRange {
+                start: Timestamp::new(0, common_time::timestamp::TimeUnit::from(&unit)),
+                end: Timestamp::new(10, common_time::timestamp::TimeUnit::from(&unit)),
+                num_rows: 1,
+                identifier: 0,
+            },
+            vec![
+                DfRecordBatch::try_new(schema.clone(), vec![new_ts_array(unit, vec![100])])
+                    .unwrap(),
+            ],
+        )];
+
+        let expected_output = Some(
+            DfRecordBatch::try_new(schema.clone(), vec![new_ts_array(unit, vec![100])]).unwrap(),
+        );
+
+        run_test(
+            0,
+            input_ranged_data,
+            schema,
+            SortOptions::default(),
+            Some(10),
+            expected_output,
+            None,
+        )
+        .await;
+    }
+
    #[allow(clippy::print_stdout)]
    async fn run_test(
        case_id: usize,
--- a/tests-integration/src/standalone.rs
+++ b/tests-integration/src/standalone.rs
@@ -259,9 +259,8 @@ impl GreptimeDbStandaloneBuilder {
        let grpc_handler = instance.clone() as Arc<dyn GrpcQueryHandlerWithBoxedError>;
        let weak_grpc_handler = Arc::downgrade(&grpc_handler);
        frontend_instance_handler
-            .lock()
-            .unwrap()
-            .replace(weak_grpc_handler);
+            .set_handler(weak_grpc_handler)
+            .await;

        let flow_streaming_engine = flownode.flow_engine().streaming_engine();
        let invoker = flow::FrontendInvoker::build_from(
Author	SHA1	Message	Date
Ruihang Xia	8ac57368af	fix: handle remaining buf Signed-off-by: Ruihang Xia <waynestxia@gmail.com>	2025-12-17 20:36:54 +08:00
fys	0bc5a305be	chore: add wait_initialized method for frontend client (#7414 ) * chore: add wait_initialized method for frontend client * fix: some * fix: cargo fmt * add comment * add unit test * rename * fix: cargo check * fix: cr by copilot	2025-12-17 08:13:36 +00:00
discord9	1afcddd5a9	chore: feature gate vector_index (#7428 ) Signed-off-by: discord9 <discord9@163.com>	2025-12-17 07:14:25 +00:00
shuiyisong	62808b887b	fix: using anonymous s3 access when ak and sk is not provided (#7425 ) * chore: allow s3 anon Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: disable ec2 metadata Signed-off-by: shuiyisong <xixing.sys@gmail.com> --------- Signed-off-by: shuiyisong <xixing.sys@gmail.com>	2025-12-17 06:34:29 +00:00
discord9	04ddd40e00	chore: bump version to beta.3 (#7423 ) chore: bump to beta.3 Signed-off-by: discord9 <discord9@163.com>	2025-12-17 04:18:23 +00:00
liyang	b4f028be5f	chore: change etcd endpoints to array in the test scripts (#7419 ) chore: change etcd endpoint Signed-off-by: liyang <daviderli614@gmail.com>	2025-12-17 03:14:35 +00:00