Skip to main content

tests_integration/
prom_store.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#[cfg(test)]
16mod tests {
17    use std::sync::Arc;
18
19    use api::prom_store::remote::label_matcher::Type as MatcherType;
20    use api::prom_store::remote::{
21        Label, LabelMatcher, Query, ReadRequest, ReadResponse, Sample, TimeSeries, WriteRequest,
22    };
23    use api::v1::value::ValueData;
24    use api::v1::{
25        ColumnDataType, Row, RowInsertRequest, RowInsertRequests, Rows, SemanticType, Value,
26    };
27    use client::OutputData;
28    use common_catalog::consts::DEFAULT_CATALOG_NAME;
29    use datatypes::arrow::array::AsArray;
30    use datatypes::arrow::datatypes::{
31        DataType, TimeUnit, TimestampMicrosecondType, TimestampMillisecondType,
32        TimestampNanosecondType, TimestampSecondType,
33    };
34    use frontend::instance::Instance;
35    use prost::Message;
36    use servers::http::prom_store::PHYSICAL_TABLE_PARAM;
37    use servers::http::test_helpers::{TestClient, TestResponse};
38    use servers::http::{HttpOptions, HttpServerBuilder};
39    use servers::prom_store;
40    use servers::prom_store::to_grpc_row_insert_requests;
41    use servers::query_handler::PromStoreProtocolHandler;
42    use servers::query_handler::sql::SqlQueryHandler;
43    use session::context::QueryContext;
44
45    use crate::standalone::GreptimeDbStandaloneBuilder;
46    use crate::tests;
47
48    #[tokio::test(flavor = "multi_thread")]
49    async fn test_standalone_prom_store_remote_rw_default_physical_table() {
50        common_telemetry::init_default_ut_logging();
51        let standalone = GreptimeDbStandaloneBuilder::new(
52            "test_standalone_prom_store_remote_rw_default_physical_table",
53        )
54        .build()
55        .await;
56        let instance = standalone.fe_instance();
57
58        test_prom_store_remote_rw(instance, None).await;
59    }
60
61    #[tokio::test(flavor = "multi_thread")]
62    async fn test_distributed_prom_store_remote_rw_default_physical_table() {
63        common_telemetry::init_default_ut_logging();
64        let distributed = tests::create_distributed_instance(
65            "test_distributed_prom_store_remote_rw_default_physical_table",
66        )
67        .await;
68        test_prom_store_remote_rw(&distributed.frontend(), None).await;
69    }
70
71    #[tokio::test(flavor = "multi_thread")]
72    async fn test_standalone_prom_store_remote_rw_custom_physical_table() {
73        common_telemetry::init_default_ut_logging();
74        let standalone = GreptimeDbStandaloneBuilder::new(
75            "test_standalone_prom_store_remote_rw_custom_physical_table",
76        )
77        .build()
78        .await;
79        let instance = standalone.fe_instance();
80
81        test_prom_store_remote_rw(instance, Some("my_custom_physical_table".to_string())).await;
82    }
83
84    #[tokio::test(flavor = "multi_thread")]
85    async fn test_distributed_prom_store_remote_rw_custom_physical_table() {
86        common_telemetry::init_default_ut_logging();
87        let distributed = tests::create_distributed_instance(
88            "test_distributed_prom_store_remote_rw_custom_physical_table",
89        )
90        .await;
91        test_prom_store_remote_rw(
92            &distributed.frontend(),
93            Some("my_custom_physical_table".to_string()),
94        )
95        .await;
96    }
97
98    #[tokio::test(flavor = "multi_thread")]
99    async fn test_standalone_prom_store_remote_rw_microsecond_physical_table() {
100        common_telemetry::init_default_ut_logging();
101        let standalone =
102            GreptimeDbStandaloneBuilder::new("test_prom_store_remote_rw_us_physical_table")
103                .build()
104                .await;
105        let instance = standalone.fe_instance();
106
107        test_prom_store_remote_rw_non_millisecond_physical_table(
108            instance,
109            "TIMESTAMP(6)",
110            TimeUnit::Microsecond,
111            1_000_000,
112            2_000_000,
113        )
114        .await;
115    }
116
117    #[tokio::test(flavor = "multi_thread")]
118    async fn test_distributed_prom_store_remote_rw_microsecond_physical_table() {
119        common_telemetry::init_default_ut_logging();
120        let distributed =
121            tests::create_distributed_instance("test_prom_store_remote_rw_us_physical_table").await;
122        test_prom_store_remote_rw_non_millisecond_physical_table(
123            &distributed.frontend(),
124            "TIMESTAMP(6)",
125            TimeUnit::Microsecond,
126            1_000_000,
127            2_000_000,
128        )
129        .await;
130    }
131
132    #[tokio::test(flavor = "multi_thread")]
133    async fn test_standalone_prom_store_remote_rw_seconds_physical_table() {
134        common_telemetry::init_default_ut_logging();
135        let standalone =
136            GreptimeDbStandaloneBuilder::new("test_prom_store_remote_rw_s_physical_table")
137                .build()
138                .await;
139        let instance = standalone.fe_instance();
140
141        // Narrowing truncates: 1000ms/2000ms -> 1s/2s.
142        test_prom_store_remote_rw_non_millisecond_physical_table(
143            instance,
144            "TIMESTAMP(0)",
145            TimeUnit::Second,
146            1,
147            2,
148        )
149        .await;
150    }
151
152    #[tokio::test(flavor = "multi_thread")]
153    async fn test_standalone_prom_store_remote_rw_nanoseconds_physical_table() {
154        common_telemetry::init_default_ut_logging();
155        let standalone =
156            GreptimeDbStandaloneBuilder::new("test_prom_store_remote_rw_ns_physical_table")
157                .build()
158                .await;
159        let instance = standalone.fe_instance();
160
161        // Lossless widening: 1000ms/2000ms -> 1e9ns/2e9ns.
162        test_prom_store_remote_rw_non_millisecond_physical_table(
163            instance,
164            "TIMESTAMP(9)",
165            TimeUnit::Nanosecond,
166            1_000_000_000,
167            2_000_000_000,
168        )
169        .await;
170    }
171
172    /// Regression test for <https://github.com/GreptimeTeam/greptimedb/issues/9231>:
173    /// prometheus remote write/read against a physical metric table
174    /// pre-created with a non-millisecond time index. Millisecond samples are
175    /// converted to the physical table's unit on write (lossless widening,
176    /// truncating narrowing) and narrowed back to milliseconds on remote read.
177    async fn test_prom_store_remote_rw_non_millisecond_physical_table(
178        instance: &Arc<Instance>,
179        sql_ts_type: &str,
180        expected_unit: TimeUnit,
181        expected_first: i64,
182        expected_second: i64,
183    ) {
184        let db = "prometheus_non_ms";
185        let ctx = Arc::new(QueryContext::with(DEFAULT_CATALOG_NAME, db));
186        assert!(
187            SqlQueryHandler::do_query(
188                instance.as_ref(),
189                &format!("CREATE DATABASE IF NOT EXISTS {db}"),
190                ctx.clone(),
191            )
192            .await
193            .first()
194            .unwrap()
195            .is_ok()
196        );
197
198        let mut output = instance
199            .do_query(
200                &format!(
201                    "CREATE TABLE greptime_physical_table (\
202                 greptime_timestamp {sql_ts_type} NOT NULL, \
203                 greptime_value DOUBLE NULL, \
204                 TIME INDEX (greptime_timestamp)) \
205                 ENGINE = metric WITH ('physical_metric_table' = 'true')"
206                ),
207                ctx.clone(),
208            )
209            .await;
210        assert!(output.remove(0).is_ok());
211
212        let write_request = WriteRequest {
213            timeseries: vec![prom_store::mock_timeseries()[0].clone()],
214            ..Default::default()
215        };
216        let (row_inserts, _) = to_grpc_row_insert_requests(&write_request).unwrap();
217        instance
218            .write(row_inserts, ctx.clone(), true)
219            .await
220            .unwrap();
221
222        let read_request = ReadRequest {
223            queries: vec![Query {
224                start_timestamp_ms: 1000,
225                end_timestamp_ms: 2000,
226                matchers: vec![LabelMatcher {
227                    name: prom_store::METRIC_NAME_LABEL.to_string(),
228                    value: "metric1".to_string(),
229                    r#type: 0,
230                }],
231                ..Default::default()
232            }],
233            ..Default::default()
234        };
235        let resp = instance.read(read_request, ctx.clone()).await.unwrap();
236        let body = prom_store::snappy_decompress(&resp.body).unwrap();
237        let read_response = ReadResponse::decode(&body[..]).unwrap();
238        assert_eq!(1, read_response.results.len());
239        assert_eq!(1, read_response.results[0].timeseries.len());
240        let timeseries = &read_response.results[0].timeseries[0];
241        assert_eq!(
242            timeseries.samples,
243            vec![
244                Sample {
245                    value: 1.0,
246                    timestamp: 1000,
247                },
248                Sample {
249                    value: 2.0,
250                    timestamp: 2000,
251                },
252            ]
253        );
254
255        // The stored timestamps keep the physical table's time index unit.
256        let mut output = instance
257            .do_query("SELECT greptime_timestamp FROM metric1", ctx.clone())
258            .await;
259        let OutputData::Stream(stream) = output.remove(0).unwrap().data else {
260            unreachable!()
261        };
262        let batches = common_recordbatch::RecordBatches::try_collect(stream)
263            .await
264            .unwrap()
265            .take();
266        assert_eq!(batches[0].num_rows(), 2);
267        let ts_column = batches[0].column(0);
268        assert_eq!(
269            ts_column.data_type(),
270            &DataType::Timestamp(expected_unit, None),
271            "unexpected time index type"
272        );
273        let stored = |row: usize| match expected_unit {
274            TimeUnit::Second => ts_column.as_primitive::<TimestampSecondType>().value(row),
275            TimeUnit::Millisecond => ts_column
276                .as_primitive::<TimestampMillisecondType>()
277                .value(row),
278            TimeUnit::Microsecond => ts_column
279                .as_primitive::<TimestampMicrosecondType>()
280                .value(row),
281            TimeUnit::Nanosecond => ts_column
282                .as_primitive::<TimestampNanosecondType>()
283                .value(row),
284        };
285        assert_eq!(stored(0), expected_first);
286        assert_eq!(stored(1), expected_second);
287
288        // Negative, non-aligned timestamps must floor towards negative
289        // infinity on remote read, matching `Timestamp::convert_to` on the
290        // ingestion path: -1001us/-1001000ns -> -2ms, not -1ms.
291        if matches!(expected_unit, TimeUnit::Microsecond | TimeUnit::Nanosecond) {
292            let (ts_datatype, negative_value) = if expected_unit == TimeUnit::Microsecond {
293                (
294                    ColumnDataType::TimestampMicrosecond,
295                    ValueData::TimestampMicrosecondValue(-1001),
296                )
297            } else {
298                (
299                    ColumnDataType::TimestampNanosecond,
300                    ValueData::TimestampNanosecondValue(-1_001_000),
301                )
302            };
303            let negative_request = RowInsertRequests {
304                inserts: vec![RowInsertRequest {
305                    table_name: "metric1".to_string(),
306                    rows: Some(Rows {
307                        schema: vec![
308                            api::v1::ColumnSchema {
309                                column_name: "greptime_timestamp".to_string(),
310                                datatype: ts_datatype as i32,
311                                semantic_type: SemanticType::Timestamp as i32,
312                                datatype_extension: None,
313                                options: None,
314                            },
315                            api::v1::ColumnSchema {
316                                column_name: "greptime_value".to_string(),
317                                datatype: ColumnDataType::Float64 as i32,
318                                semantic_type: SemanticType::Field as i32,
319                                datatype_extension: None,
320                                options: None,
321                            },
322                            api::v1::ColumnSchema {
323                                column_name: "job".to_string(),
324                                datatype: ColumnDataType::String as i32,
325                                semantic_type: SemanticType::Tag as i32,
326                                datatype_extension: None,
327                                options: None,
328                            },
329                        ],
330                        rows: vec![Row {
331                            values: vec![
332                                Value {
333                                    value_data: Some(negative_value),
334                                },
335                                Value {
336                                    value_data: Some(ValueData::F64Value(3.0)),
337                                },
338                                Value {
339                                    value_data: Some(ValueData::StringValue("spark".to_string())),
340                                },
341                            ],
342                        }],
343                    }),
344                }],
345            };
346            instance
347                .write(negative_request, ctx.clone(), true)
348                .await
349                .unwrap();
350
351            let read_request = ReadRequest {
352                queries: vec![Query {
353                    start_timestamp_ms: -2,
354                    end_timestamp_ms: 0,
355                    matchers: vec![LabelMatcher {
356                        name: prom_store::METRIC_NAME_LABEL.to_string(),
357                        value: "metric1".to_string(),
358                        r#type: 0,
359                    }],
360                    ..Default::default()
361                }],
362                ..Default::default()
363            };
364            let resp = instance.read(read_request, ctx.clone()).await.unwrap();
365            let body = prom_store::snappy_decompress(&resp.body).unwrap();
366            let read_response = ReadResponse::decode(&body[..]).unwrap();
367            assert_eq!(1, read_response.results.len());
368            assert_eq!(
369                read_response.results[0].timeseries[0].samples,
370                vec![Sample {
371                    value: 3.0,
372                    timestamp: -2,
373                }],
374                "remote read must floor negative non-aligned timestamps"
375            );
376        }
377    }
378
379    #[tokio::test(flavor = "multi_thread")]
380    async fn test_standalone_prom_store_write_existing_logical_table_of_other_physical_unit() {
381        common_telemetry::init_default_ut_logging();
382        let standalone = GreptimeDbStandaloneBuilder::new(
383            "test_prom_store_write_existing_logical_other_physical_unit",
384        )
385        .build()
386        .await;
387        let instance = standalone.fe_instance();
388
389        // Two physical metric tables with different time index units.
390        let db = "prometheus_mixed_units";
391        let ctx = Arc::new(QueryContext::with(DEFAULT_CATALOG_NAME, db));
392        assert!(
393            SqlQueryHandler::do_query(
394                instance.as_ref(),
395                &format!("CREATE DATABASE IF NOT EXISTS {db}"),
396                ctx.clone(),
397            )
398            .await
399            .first()
400            .unwrap()
401            .is_ok()
402        );
403        let mut output = instance
404            .do_query(
405                "CREATE TABLE phy_us (greptime_timestamp TIMESTAMP(6) NOT NULL, \
406                 greptime_value DOUBLE NULL, TIME INDEX (greptime_timestamp)) \
407                 ENGINE = metric WITH ('physical_metric_table' = 'true')",
408                ctx.clone(),
409            )
410            .await;
411        assert!(output.remove(0).is_ok());
412
413        // First create the logical table `shared_metric` on the default
414        // (millisecond) physical table.
415        let shared_series = TimeSeries {
416            labels: vec![
417                Label {
418                    name: prom_store::METRIC_NAME_LABEL.to_string(),
419                    value: "shared_metric".to_string(),
420                },
421                Label {
422                    name: "job".to_string(),
423                    value: "demo".to_string(),
424                },
425            ],
426            samples: vec![Sample {
427                value: 1.0,
428                timestamp: 1000,
429            }],
430            ..Default::default()
431        };
432        let (row_inserts, _) = to_grpc_row_insert_requests(&WriteRequest {
433            timeseries: vec![shared_series.clone()],
434            ..Default::default()
435        })
436        .unwrap();
437        instance
438            .write(row_inserts, ctx.clone(), true)
439            .await
440            .unwrap();
441
442        // Now select the microsecond physical table while writing to BOTH the
443        // existing millisecond `shared_metric` (bound to the default physical
444        // table) and a new table: the existing table's request must keep the
445        // millisecond unit, and only the new table uses microsecond.
446        let mut hint_ctx = QueryContext::with(DEFAULT_CATALOG_NAME, db);
447        hint_ctx.set_extension(PHYSICAL_TABLE_PARAM, "phy_us".to_string());
448        let hint_ctx = Arc::new(hint_ctx);
449        let (row_inserts, _) = to_grpc_row_insert_requests(&WriteRequest {
450            timeseries: vec![
451                TimeSeries {
452                    samples: vec![Sample {
453                        value: 1.5,
454                        timestamp: 1500,
455                    }],
456                    ..shared_series
457                },
458                TimeSeries {
459                    labels: vec![
460                        Label {
461                            name: prom_store::METRIC_NAME_LABEL.to_string(),
462                            value: "fresh_us_metric".to_string(),
463                        },
464                        Label {
465                            name: "job".to_string(),
466                            value: "demo".to_string(),
467                        },
468                    ],
469                    samples: vec![Sample {
470                        value: 2.5,
471                        timestamp: 1500,
472                    }],
473                    ..Default::default()
474                },
475            ],
476            ..Default::default()
477        })
478        .unwrap();
479        instance
480            .write(row_inserts, hint_ctx.clone(), true)
481            .await
482            .unwrap();
483
484        // The existing table keeps its millisecond unit and precision.
485        let mut output = instance
486            .do_query(
487                "SELECT greptime_timestamp FROM shared_metric ORDER BY greptime_timestamp",
488                ctx.clone(),
489            )
490            .await;
491        let OutputData::Stream(stream) = output.remove(0).unwrap().data else {
492            unreachable!()
493        };
494        let batches = common_recordbatch::RecordBatches::try_collect(stream)
495            .await
496            .unwrap()
497            .take();
498        let shared_ts = batches[0]
499            .column(0)
500            .as_primitive::<TimestampMillisecondType>();
501        assert_eq!((shared_ts.value(0), shared_ts.value(1)), (1000, 1500));
502
503        // The new table is created on the selected physical table with the
504        // microsecond unit.
505        let mut output = instance
506            .do_query(
507                "SELECT greptime_timestamp FROM fresh_us_metric",
508                ctx.clone(),
509            )
510            .await;
511        let OutputData::Stream(stream) = output.remove(0).unwrap().data else {
512            unreachable!()
513        };
514        let batches = common_recordbatch::RecordBatches::try_collect(stream)
515            .await
516            .unwrap()
517            .take();
518        let fresh_ts = batches[0]
519            .column(0)
520            .as_primitive::<TimestampMicrosecondType>();
521        assert_eq!(fresh_ts.value(0), 1_500_000);
522    }
523
524    async fn test_prom_store_remote_rw(instance: &Arc<Instance>, physical_table: Option<String>) {
525        let write_request = WriteRequest {
526            timeseries: prom_store::mock_timeseries(),
527            ..Default::default()
528        };
529
530        let db = "prometheus";
531        let mut ctx = Arc::into_inner(QueryContext::with(DEFAULT_CATALOG_NAME, db).into()).unwrap();
532
533        // set physical table if provided
534        if let Some(physical_table) = &physical_table {
535            ctx.set_extension(PHYSICAL_TABLE_PARAM.to_string(), physical_table.clone());
536        }
537        let ctx = Arc::new(ctx);
538
539        assert!(
540            SqlQueryHandler::do_query(
541                instance.as_ref(),
542                "CREATE DATABASE IF NOT EXISTS prometheus",
543                ctx.clone(),
544            )
545            .await
546            .first()
547            .unwrap()
548            .is_ok()
549        );
550
551        let (row_inserts, _) = to_grpc_row_insert_requests(&write_request).unwrap();
552        instance
553            .write(row_inserts, ctx.clone(), true)
554            .await
555            .unwrap();
556
557        let read_request = ReadRequest {
558            queries: vec![
559                Query {
560                    start_timestamp_ms: 1000,
561                    end_timestamp_ms: 2000,
562                    matchers: vec![LabelMatcher {
563                        name: prom_store::METRIC_NAME_LABEL.to_string(),
564                        value: "metric1".to_string(),
565                        r#type: 0,
566                    }],
567                    ..Default::default()
568                },
569                Query {
570                    start_timestamp_ms: 1000,
571                    end_timestamp_ms: 3000,
572                    matchers: vec![
573                        LabelMatcher {
574                            name: prom_store::METRIC_NAME_LABEL.to_string(),
575                            value: "metric3".to_string(),
576                            r#type: 0,
577                        },
578                        LabelMatcher {
579                            name: "app".to_string(),
580                            value: "biz".to_string(),
581                            r#type: MatcherType::Eq as i32,
582                        },
583                    ],
584                    ..Default::default()
585                },
586            ],
587            ..Default::default()
588        };
589
590        let resp = instance.read(read_request, ctx.clone()).await.unwrap();
591        assert_eq!(resp.content_type, "application/x-protobuf");
592        assert_eq!(resp.content_encoding, "snappy");
593        let body = prom_store::snappy_decompress(&resp.body).unwrap();
594        let read_response = ReadResponse::decode(&body[..]).unwrap();
595        let query_results = read_response.results;
596        assert_eq!(2, query_results.len());
597
598        assert_eq!(1, query_results[0].timeseries.len());
599        let timeseries = &query_results[0].timeseries[0];
600
601        assert_eq!(
602            vec![
603                Label {
604                    name: prom_store::METRIC_NAME_LABEL.to_string(),
605                    value: "metric1".to_string(),
606                },
607                Label {
608                    name: "job".to_string(),
609                    value: "spark".to_string(),
610                },
611            ],
612            timeseries.labels
613        );
614
615        assert_eq!(
616            timeseries.samples,
617            vec![
618                Sample {
619                    value: 1.0,
620                    timestamp: 1000,
621                },
622                Sample {
623                    value: 2.0,
624                    timestamp: 2000,
625                }
626            ]
627        );
628
629        assert_eq!(1, query_results[1].timeseries.len());
630        let timeseries = &query_results[1].timeseries[0];
631
632        assert_eq!(
633            vec![
634                Label {
635                    name: prom_store::METRIC_NAME_LABEL.to_string(),
636                    value: "metric3".to_string(),
637                },
638                Label {
639                    name: "app".to_string(),
640                    value: "biz".to_string(),
641                },
642                Label {
643                    name: "idc".to_string(),
644                    value: "z002".to_string(),
645                },
646            ],
647            timeseries.labels
648        );
649
650        assert_eq!(
651            timeseries.samples,
652            vec![
653                Sample {
654                    value: 5.0,
655                    timestamp: 1000,
656                },
657                Sample {
658                    value: 6.0,
659                    timestamp: 2000,
660                },
661                Sample {
662                    value: 7.0,
663                    timestamp: 3000,
664                }
665            ]
666        );
667
668        instance
669            .do_query(
670                "CREATE TABLE invalid_metric (ts TIMESTAMP TIME INDEX, a DOUBLE, b DOUBLE)",
671                ctx.clone(),
672            )
673            .await[0]
674            .as_ref()
675            .unwrap();
676        let server = HttpServerBuilder::new(HttpOptions::default())
677            .with_prom_handler(instance.clone(), None, true, Default::default(), None)
678            .build();
679        let client = TestClient::new(server.build(server.make_app()).unwrap()).await;
680
681        // Cover missing schemas/catalogs and each database label alias.
682        for (selector, database) in [
683            ("__schema__", "unknown_schema"),
684            ("__database__", "unknown_catalog-prometheus"),
685            ("x_greptime_database", "unknown_catalog-information_schema"),
686        ] {
687            let mut query = remote_query("missing_metric");
688            query.matchers.push(LabelMatcher {
689                name: selector.to_string(),
690                value: database.to_string(),
691                r#type: MatcherType::Eq as i32,
692            });
693            let response = post_remote_read(&client, vec![query]).await;
694            assert_eq!(response.status(), 200, "{selector}={database}");
695            let body = prom_store::snappy_decompress(&response.bytes().await).unwrap();
696            let response = ReadResponse::decode(&body[..]).unwrap();
697            assert_eq!(response.results.len(), 1);
698            assert!(response.results[0].timeseries.is_empty());
699        }
700
701        // Matcher syntax must be checked even when the metric does not exist.
702        for metric in ["metric1", "missing_metric"] {
703            for (matcher_type, value, expected_status) in [
704                (99, "spark", 400),
705                (MatcherType::Re as i32, "[", 400),
706                (MatcherType::Nre as i32, "[", 400),
707                (MatcherType::Eq as i32, "[", 200),
708                (MatcherType::Neq as i32, "[", 200),
709                (MatcherType::Re as i32, "spark", 200),
710                (MatcherType::Nre as i32, "api.*", 200),
711                (MatcherType::Re as i32, "(?x) spark", 200),
712                (MatcherType::Nre as i32, "(?x) spark", 200),
713                (MatcherType::Re as i32, "(?x) spark # comment", 400),
714                (MatcherType::Nre as i32, "(?x) spark # comment", 400),
715                (MatcherType::Re as i32, "a)|(?:b", 400),
716                (MatcherType::Nre as i32, "a)|(?:b", 400),
717            ] {
718                let mut query = remote_query(metric);
719                query.matchers.push(LabelMatcher {
720                    name: "job".to_string(),
721                    value: value.to_string(),
722                    r#type: matcher_type,
723                });
724                let response = post_remote_read(&client, vec![query]).await;
725                assert_eq!(
726                    response.status().as_u16(),
727                    expected_status,
728                    "metric={metric}, type={matcher_type}, value={value}"
729                );
730            }
731        }
732
733        // Missing metrics must retain their result slots in multi-query responses.
734        for metrics in [
735            vec!["missing_metric"],
736            vec!["missing_metric", "another_missing_metric"],
737            vec![
738                "missing_metric",
739                "metric1",
740                "missing_metric",
741                "metric3",
742                "missing_metric",
743            ],
744            vec!["missing_metric", "invalid_metric"],
745        ] {
746            let queries = metrics.iter().map(|metric| remote_query(metric)).collect();
747            let response = post_remote_read(&client, queries).await;
748            if metrics.contains(&"invalid_metric") {
749                assert!(!response.status().is_success());
750                assert!(response.text().await.contains("Ambiguous value column"));
751                continue;
752            }
753            assert_eq!(response.status(), 200, "{}", response.text().await);
754            assert_eq!(response.headers()["content-type"], "application/x-protobuf");
755            assert_eq!(response.headers()["content-encoding"], "snappy");
756            let body = prom_store::snappy_decompress(&response.bytes().await).unwrap();
757            let response = ReadResponse::decode(&body[..]).unwrap();
758            assert_eq!(response.results.len(), metrics.len());
759            for (result, metric) in response.results.iter().zip(metrics) {
760                if metric.ends_with("missing_metric") {
761                    assert!(result.timeseries.is_empty());
762                    continue;
763                }
764                assert!(!result.timeseries.is_empty());
765                for series in &result.timeseries {
766                    assert!(series.labels.iter().any(|label| {
767                        label.name == prom_store::METRIC_NAME_LABEL && label.value == metric
768                    }));
769                    assert_eq!(series.samples.len(), 2);
770                }
771            }
772        }
773
774        // check physical table if provided
775        if let Some(physical_table) = physical_table {
776            let sql = format!("DESC TABLE {physical_table};");
777            instance.do_query(&sql, ctx).await[0].as_ref().unwrap();
778        }
779    }
780
781    #[tokio::test(flavor = "multi_thread")]
782    async fn test_standalone_native_histogram_remote_read() {
783        let standalone =
784            GreptimeDbStandaloneBuilder::new("standalone_native_histogram_remote_read")
785                .build()
786                .await;
787        check_native_histogram_remote_read(standalone.fe_instance()).await;
788    }
789
790    #[tokio::test(flavor = "multi_thread")]
791    async fn test_distributed_native_histogram_remote_read() {
792        let cluster =
793            tests::create_distributed_instance("distributed_native_histogram_remote_read").await;
794        check_native_histogram_remote_read(&cluster.frontend()).await;
795    }
796
797    async fn check_native_histogram_remote_read(instance: &Arc<Instance>) {
798        use api::greptime_proto::io::prometheus::write::v2::{
799            Histogram,
800            histogram::{Count, ZeroCount},
801        };
802        use common_query::native_histogram::{
803            encode_native_histogram, native_histogram_column_schema,
804        };
805        use servers::prom_remote_write::v2::test_util as remote_write_v2;
806
807        let ctx = Arc::new(QueryContext::with(DEFAULT_CATALOG_NAME, "prometheus"));
808        instance
809            .do_query("CREATE DATABASE prometheus", ctx.clone())
810            .await
811            .remove(0)
812            .unwrap();
813        let server = HttpServerBuilder::new(HttpOptions::default())
814            .with_prom_handler(instance.clone(), None, true, Default::default(), None)
815            .build();
816        let client = TestClient::new(server.build(server.make_app()).unwrap()).await;
817        let large = (1u64 << 53) + 1;
818        let histograms = vec![
819            Histogram {
820                count: Some(Count::CountInt(6)),
821                timestamp: 1000,
822                positive_spans: vec![api::greptime_proto::io::prometheus::write::v2::BucketSpan {
823                    offset: 0,
824                    length: 3,
825                }],
826                positive_deltas: vec![1, 2, -1],
827                sum: 10.0,
828                ..Default::default()
829            },
830            Histogram {
831                count: Some(Count::CountInt(large)),
832                zero_count: Some(ZeroCount::ZeroCountInt(large)),
833                timestamp: 2000,
834                ..Default::default()
835            },
836            Histogram {
837                count: Some(Count::CountFloat(6.0)),
838                zero_count: Some(ZeroCount::ZeroCountFloat(0.5)),
839                positive_spans: vec![api::greptime_proto::io::prometheus::write::v2::BucketSpan {
840                    offset: 3,
841                    length: 2,
842                }],
843                positive_counts: vec![2.0, 3.5],
844                timestamp: 3000,
845                ..Default::default()
846            },
847            Histogram {
848                count: Some(Count::CountInt(0)),
849                sum: f64::from_bits(0x7ff0_0000_0000_0002),
850                timestamp: 4000,
851                ..Default::default()
852            },
853        ];
854        let request = remote_write_v2::request_with_labels_and_histograms(
855            vec![(prom_store::METRIC_NAME_LABEL, "latency"), ("job", "api")],
856            histograms,
857        );
858        let response = client
859            .post("/v1/prometheus/write?db=prometheus")
860            .header(
861                "content-type",
862                "application/x-protobuf;proto=io.prometheus.write.v2.Request",
863            )
864            .header("content-encoding", "snappy")
865            .body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap())
866            .send()
867            .await;
868        assert_eq!(response.status(), 204, "{}", response.text().await);
869        let (scalar, _) = to_grpc_row_insert_requests(&WriteRequest {
870            timeseries: vec![prom_store::mock_timeseries()[0].clone()],
871            ..Default::default()
872        })
873        .unwrap();
874        instance.write(scalar, ctx.clone(), true).await.unwrap();
875
876        for flushed in [false, true] {
877            if flushed {
878                instance
879                    .do_query("ADMIN FLUSH_TABLE('latency')", ctx.clone())
880                    .await
881                    .remove(0)
882                    .unwrap();
883            }
884            for accepted_response_types in [vec![], vec![0], vec![1, 0]] {
885                let mut query = remote_query("latency");
886                query.end_timestamp_ms = 4000;
887                query.matchers.push(LabelMatcher {
888                    name: "job".to_string(),
889                    value: "api".to_string(),
890                    r#type: MatcherType::Eq as i32,
891                });
892                let mut empty = query.clone();
893                empty.start_timestamp_ms = 5000;
894                empty.end_timestamp_ms = 6000;
895                let request = ReadRequest {
896                    queries: vec![
897                        remote_query("metric1"),
898                        query,
899                        remote_query("missing_metric"),
900                        empty,
901                    ],
902                    accepted_response_types,
903                };
904                let response = client
905                    .post("/v1/prometheus/read?db=prometheus")
906                    .body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap())
907                    .send()
908                    .await;
909                assert_eq!(response.status(), 200, "{}", response.text().await);
910                assert_eq!(response.headers()["content-type"], "application/x-protobuf");
911                assert_eq!(response.headers()["content-encoding"], "snappy");
912                let body = prom_store::snappy_decompress(&response.bytes().await).unwrap();
913                let response = ReadResponse::decode(body.as_slice()).unwrap();
914                assert_eq!(response.results.len(), 4);
915                assert!(!response.results[0].timeseries[0].samples.is_empty());
916                assert!(response.results[2].timeseries.is_empty());
917                assert!(response.results[3].timeseries.is_empty());
918                let series = &response.results[1].timeseries[0];
919                assert_eq!(
920                    series.labels,
921                    vec![
922                        Label {
923                            name: "__name__".to_string(),
924                            value: "latency".to_string()
925                        },
926                        Label {
927                            name: "job".to_string(),
928                            value: "api".to_string()
929                        }
930                    ]
931                );
932                assert_eq!(series.histograms.len(), 4);
933                assert_eq!(series.histograms[0].positive_deltas, vec![1, 2, -1]);
934                assert_eq!(series.histograms[2].positive_counts, vec![2.0, 3.5]);
935                assert_eq!(
936                    series.histograms[1].count,
937                    Some(api::prom_store::remote::histogram::Count::CountInt(large))
938                );
939                assert_eq!(
940                    series.histograms[1].zero_count,
941                    Some(api::prom_store::remote::histogram::ZeroCount::ZeroCountInt(
942                        large
943                    ))
944                );
945                assert_eq!(
946                    series.histograms[2].count,
947                    Some(api::prom_store::remote::histogram::Count::CountFloat(6.0))
948                );
949                assert_eq!(series.histograms[3].sum.to_bits(), 0x7ff0_0000_0000_0002);
950            }
951        }
952        let response = client
953            .post("/v1/prometheus/read?db=prometheus")
954            .body(
955                prom_store::snappy_compress(
956                    &ReadRequest {
957                        queries: vec![remote_query("latency")],
958                        accepted_response_types: vec![1],
959                    }
960                    .encode_to_vec(),
961                )
962                .unwrap(),
963            )
964            .send()
965            .await;
966        assert!(!response.status().is_success());
967        assert!(
968            response
969                .text()
970                .await
971                .contains("does not support any of the requested response types")
972        );
973
974        // Typed inserts supply sub-millisecond pre-epoch times that neither OTLP
975        // nor millisecond Remote Write can express, and use a legacy field name.
976        for (table, datatype, timestamp) in [
977            (
978                "legacy_us",
979                ColumnDataType::TimestampMicrosecond,
980                ValueData::TimestampMicrosecondValue(-1001),
981            ),
982            (
983                "legacy_ns",
984                ColumnDataType::TimestampNanosecond,
985                ValueData::TimestampNanosecondValue(-1_001_000),
986            ),
987        ] {
988            let mut histogram_schema = native_histogram_column_schema().unwrap();
989            histogram_schema.column_name = "payload".to_string();
990            let histogram = encode_native_histogram(&Histogram {
991                count: Some(Count::CountInt(large)),
992                zero_count: Some(ZeroCount::ZeroCountInt(large)),
993                ..Default::default()
994            })
995            .unwrap();
996            instance
997                .write(
998                    RowInsertRequests {
999                        inserts: vec![RowInsertRequest {
1000                            table_name: table.to_string(),
1001                            rows: Some(Rows {
1002                                schema: vec![
1003                                    api::v1::ColumnSchema {
1004                                        column_name: "event_time".to_string(),
1005                                        datatype: datatype as i32,
1006                                        semantic_type: SemanticType::Timestamp as i32,
1007                                        ..Default::default()
1008                                    },
1009                                    histogram_schema,
1010                                ],
1011                                rows: vec![Row {
1012                                    values: vec![
1013                                        Value {
1014                                            value_data: Some(timestamp),
1015                                        },
1016                                        Value {
1017                                            value_data: Some(histogram),
1018                                        },
1019                                    ],
1020                                }],
1021                            }),
1022                        }],
1023                    },
1024                    ctx.clone(),
1025                    false,
1026                )
1027                .await
1028                .unwrap();
1029            for flushed in [false, true] {
1030                if flushed {
1031                    instance
1032                        .do_query(&format!("ADMIN FLUSH_TABLE('{table}')"), ctx.clone())
1033                        .await
1034                        .remove(0)
1035                        .unwrap();
1036                }
1037                let mut query = remote_query(table);
1038                query.start_timestamp_ms = -2;
1039                query.end_timestamp_ms = 0;
1040                let response = post_remote_read(&client, vec![query]).await;
1041                assert_eq!(response.status(), 200, "{}", response.text().await);
1042                let body = prom_store::snappy_decompress(&response.bytes().await).unwrap();
1043                let response = ReadResponse::decode(body.as_slice()).unwrap();
1044                let series = &response.results[0].timeseries[0];
1045                assert_eq!(series.labels.len(), 1);
1046                assert_eq!(series.histograms[0].timestamp, -2);
1047                assert_eq!(
1048                    series.histograms[0].count,
1049                    Some(api::prom_store::remote::histogram::Count::CountInt(large))
1050                );
1051            }
1052        }
1053        // Bypass ingestion's histogram codec to exercise whole-request failure.
1054        let ValueData::StructValue(mut malformed) =
1055            encode_native_histogram(&Histogram::default()).unwrap()
1056        else {
1057            unreachable!()
1058        };
1059        malformed.items[14].value_data = Some(ValueData::F64Value(0.0));
1060        instance
1061            .write(
1062                RowInsertRequests {
1063                    inserts: vec![RowInsertRequest {
1064                        table_name: "malformed_histogram".to_string(),
1065                        rows: Some(Rows {
1066                            schema: vec![
1067                                api::v1::ColumnSchema {
1068                                    column_name: "ts".to_string(),
1069                                    datatype: ColumnDataType::TimestampMillisecond as i32,
1070                                    semantic_type: SemanticType::Timestamp as i32,
1071                                    ..Default::default()
1072                                },
1073                                native_histogram_column_schema().unwrap(),
1074                            ],
1075                            rows: vec![Row {
1076                                values: vec![
1077                                    Value {
1078                                        value_data: Some(ValueData::TimestampMillisecondValue(
1079                                            1000,
1080                                        )),
1081                                    },
1082                                    Value {
1083                                        value_data: Some(ValueData::StructValue(malformed)),
1084                                    },
1085                                ],
1086                            }],
1087                        }),
1088                    }],
1089                },
1090                ctx.clone(),
1091                false,
1092            )
1093            .await
1094            .unwrap();
1095        for queries in [
1096            vec![remote_query("metric1"), remote_query("malformed_histogram")],
1097            vec![remote_query("malformed_histogram"), remote_query("metric1")],
1098        ] {
1099            let response = post_remote_read(&client, queries).await;
1100            assert_eq!(response.status(), 500);
1101        }
1102    }
1103
1104    fn remote_query(metric: &str) -> Query {
1105        Query {
1106            start_timestamp_ms: 1000,
1107            end_timestamp_ms: 2000,
1108            matchers: vec![LabelMatcher {
1109                name: prom_store::METRIC_NAME_LABEL.to_string(),
1110                value: metric.to_string(),
1111                r#type: MatcherType::Eq as i32,
1112            }],
1113            ..Default::default()
1114        }
1115    }
1116
1117    async fn post_remote_read(client: &TestClient, queries: Vec<Query>) -> TestResponse {
1118        let request = ReadRequest {
1119            queries,
1120            ..Default::default()
1121        };
1122        client
1123            .post("/v1/prometheus/read?db=prometheus")
1124            .body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap())
1125            .send()
1126            .await
1127    }
1128}