fix: support Utf8View labels in Prometheus response (#8754)

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

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;