fix: backport Prometheus and skip_wal fixes to v1.2 (#8838)

* fix: support Utf8View labels in Prometheus response

Signed-off-by: evenyag <realevenyag@gmail.com>

* fix(meta): validate leaders before enabling skip_wal

Signed-off-by: jeremyhi <fengjiachun@gmail.com>
Signed-off-by: evenyag <realevenyag@gmail.com>

* fix(meta): adapt skip_wal test to v1.2

Signed-off-by: evenyag <realevenyag@gmail.com>

---------

Signed-off-by: evenyag <realevenyag@gmail.com>
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
Co-authored-by: jeremyhi <fengjiachun@gmail.com>

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,
                     })

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);

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::<i32>())
+                .map(|i| batch.column(*i))
                 .collect::<Vec<_>>();
             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));
                     }
                 }
 

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::<PrometheusJsonResponse>().await;
+    assert_eq!(body.status, "success");
+    assert_eq!(
+        body.data,
+        serde_json::from_value::<PrometheusResponse>(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;