From 407f83e15e8a960d8ed07fa8d027acdfc30c29b6 Mon Sep 17 00:00:00 2001 From: Yingwen Date: Tue, 11 Aug 2026 14:17:47 +0800 Subject: [PATCH] fix: backport Prometheus and skip_wal fixes to v1.2 (#8838) * fix: support Utf8View labels in Prometheus response Signed-off-by: evenyag * fix(meta): validate leaders before enabling skip_wal Signed-off-by: jeremyhi Signed-off-by: evenyag * fix(meta): adapt skip_wal test to v1.2 Signed-off-by: evenyag --------- Signed-off-by: evenyag Signed-off-by: jeremyhi Co-authored-by: jeremyhi --- src/common/meta/src/ddl/alter_table.rs | 11 +++- src/common/meta/src/ddl/tests/alter_table.rs | 59 +++++++++++++++++++ .../src/http/result/prometheus_resp.rs | 8 +-- tests-integration/tests/http.rs | 34 +++++++++++ 4 files changed, 105 insertions(+), 7 deletions(-) diff --git a/src/common/meta/src/ddl/alter_table.rs b/src/common/meta/src/ddl/alter_table.rs index 59fd2af298..a19b49b016 100644 --- a/src/common/meta/src/ddl/alter_table.rs +++ b/src/common/meta/src/ddl/alter_table.rs @@ -192,6 +192,12 @@ impl AlterTableProcedure { // DDL manager can submit another procedure with locks for the current route. return Ok(Status::done_with_output(RegionRouteChanged)); } + ensure!( + !find_leaders(&physical_table_route.region_routes).is_empty(), + NoLeaderSnafu { + table_id: physical_table_id + } + ); self.data.region_distribution = Some(region_distribution(&physical_table_route.region_routes)); } @@ -258,9 +264,8 @@ impl AlterTableProcedure { MultipleResults::PartialRetryable(error) => Err(error), MultipleResults::PartialNonRetryable(error) | MultipleResults::AllNonRetryable(error) => { - // The metadata already enables skip-WAL. Keep retrying until every - // replica applies the runtime option instead of leaving the table in - // a permanently inconsistent state. + // The metadata already enables skip-WAL. Retry the idempotent request + // so later attempts can update the remaining replicas. Err(BoxedError::new(error)).context(RetryLaterSnafu { clean_poisons: true, }) diff --git a/src/common/meta/src/ddl/tests/alter_table.rs b/src/common/meta/src/ddl/tests/alter_table.rs index 38fc14e760..b35666e52e 100644 --- a/src/common/meta/src/ddl/tests/alter_table.rs +++ b/src/common/meta/src/ddl/tests/alter_table.rs @@ -816,6 +816,65 @@ async fn test_skip_wal_detects_region_route_change() { assert!(!table_info.meta.options.skip_wal); } +#[tokio::test] +async fn test_skip_wal_rejects_no_leader_before_updating_metadata() { + let ddl_context = new_ddl_context(Arc::new(MockDatanodeManager::new(()))); + let table_name = "foo"; + let table_id = 1024; + let task = test_create_table_task(table_name, table_id); + let mut region_routes = prepare_table_route(table_id) + .region_routes() + .unwrap() + .clone(); + for route in &mut region_routes { + route.leader_peer = None; + } + let region_locks = region_routes.iter().map(|route| route.region.id).collect(); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info, + TableRouteValue::physical(region_routes), + HashMap::new(), + ) + .await + .unwrap(); + + let alter_task = AlterTableTask { + alter_table: AlterTableExpr { + catalog_name: DEFAULT_CATALOG_NAME.to_string(), + schema_name: DEFAULT_SCHEMA_NAME.to_string(), + table_name: table_name.to_string(), + kind: Some(Kind::SetTableOptions(SetTableOptions { + table_options: vec![api::v1::Option { + key: SKIP_WAL_KEY.to_string(), + value: "true".to_string(), + }], + })), + }, + }; + let mut procedure = AlterTableProcedure::new_with_region_locks( + table_id, + alter_task, + region_locks, + ddl_context.clone(), + ) + .unwrap(); + + let error = procedure.on_prepare().await.unwrap_err(); + assert_matches!(error, Error::NoLeader { .. }); + let table_info = ddl_context + .table_metadata_manager + .table_info_manager() + .get(table_id) + .await + .unwrap() + .unwrap() + .into_inner() + .table_info; + assert!(!table_info.meta.options.skip_wal); +} + #[tokio::test] async fn test_skip_wal_updates_metadata_before_all_replicas() { let (tx, mut rx) = mpsc::channel(8); diff --git a/src/servers/src/http/result/prometheus_resp.rs b/src/servers/src/http/result/prometheus_resp.rs index cbe577ec22..368d925c30 100644 --- a/src/servers/src/http/result/prometheus_resp.rs +++ b/src/servers/src/http/result/prometheus_resp.rs @@ -30,6 +30,7 @@ use common_query::native_histogram::{ use common_query::prometheus::{format_prometheus_float, is_prometheus_stale_nan}; use common_query::{Output, OutputData}; use common_recordbatch::RecordBatches; +use datatypes::arrow_array::string_array_value_at_index; use datatypes::prelude::ConcreteDataType; use indexmap::IndexMap; use promql_parser::label::METRIC_NAME; @@ -281,7 +282,7 @@ impl PrometheusJsonResponse { // prepare things... let tag_columns = tag_column_indices .iter() - .map(|i| batch.column(*i).as_string::()) + .map(|i| batch.column(*i)) .collect::>(); let tag_names = tag_column_indices .iter() @@ -344,9 +345,8 @@ impl PrometheusJsonResponse { tags.push((METRIC_NAME, metric_name.as_str())); } for (tag_column, tag_name) in tag_columns.iter().zip(tag_names.iter()) { - // TODO(ruihang): add test for NULL tag - if tag_column.is_valid(row_index) { - tags.push((tag_name, tag_column.value(row_index))); + if let Some(tag_value) = string_array_value_at_index(tag_column, row_index) { + tags.push((tag_name, tag_value)); } } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index ff7b9242a7..dbf085d970 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -115,6 +115,7 @@ macro_rules! http_tests { test_sql_api, test_http_sql_slow_query, test_prometheus_promql_api, + test_prometheus_label_replace_response, test_prom_http_api, test_metrics_api, test_health_api, @@ -830,6 +831,39 @@ pub async fn test_prometheus_promql_api(store_type: StorageType) { guard.remove_all().await; } +pub async fn test_prometheus_label_replace_response(store_type: StorageType) { + let (app, mut guard) = + setup_test_prom_app_with_frontend(store_type, "prometheus_label_replace_response").await; + let client = TestClient::new(app).await; + + let query = encode(r#"label_replace(demo, "host_copy", "$1", "host", "(.*)")"#); + let res = client + .get(&format!("/v1/prometheus/api/v1/query?query={query}&time=0")) + .send() + .await; + + assert_eq!(res.status(), StatusCode::OK); + let body = res.json::().await; + assert_eq!(body.status, "success"); + assert_eq!( + body.data, + serde_json::from_value::(json!({ + "resultType": "vector", + "result": [{ + "metric": { + "__name__": "demo", + "host": "host1", + "host_copy": "host1" + }, + "value": [0.0, "1.1"] + }] + })) + .unwrap() + ); + + guard.remove_all().await; +} + pub async fn test_prom_http_api(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) = setup_test_prom_app_with_frontend(store_type, "promql_api").await;