mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-09-12 16:32:16 +00:00
test: add OTLP nanosecond PromQL performance case
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
@@ -80,6 +80,8 @@ pub(super) struct OtlpTraceLoadThresholds {
|
||||
|
||||
#[derive(Debug, Deserialize, Serialize)]
|
||||
pub(super) struct PromRemoteWritePlan {
|
||||
#[serde(default)]
|
||||
pub(super) input_protocol: InputProtocol,
|
||||
#[serde(default = "default_database")]
|
||||
pub(super) database: String,
|
||||
#[serde(alias = "metric_name")]
|
||||
@@ -114,6 +116,14 @@ pub(super) struct PromRemoteWritePlan {
|
||||
pub(super) read_bench: Option<ReadBenchConfig>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Deserialize, Serialize, Eq, PartialEq)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub(super) enum InputProtocol {
|
||||
#[default]
|
||||
RemoteWrite,
|
||||
OtlpMetrics,
|
||||
}
|
||||
|
||||
pub(super) fn default_database() -> String {
|
||||
"public".to_string()
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
mod case;
|
||||
mod direct_sst;
|
||||
mod inspect_footer;
|
||||
mod otlp_metrics;
|
||||
mod prom_remote_write;
|
||||
mod util;
|
||||
|
||||
@@ -25,6 +26,7 @@ use case::*;
|
||||
use clap::{Args as ClapArgs, Parser, Subcommand};
|
||||
use direct_sst::run_direct_sst;
|
||||
use inspect_footer::{InspectFooterArgs, run_inspect_footer};
|
||||
use otlp_metrics::{OtlpMetricsArgs, run_otlp_metrics};
|
||||
use prom_remote_write::{PromRemoteWriteArgs, run_prom_remote_write};
|
||||
use serde_json::json;
|
||||
|
||||
@@ -43,6 +45,7 @@ struct Args {
|
||||
enum Command {
|
||||
DirectSst(DirectArgs),
|
||||
PromRemoteWrite(PromRemoteWriteArgs),
|
||||
OtlpMetrics(OtlpMetricsArgs),
|
||||
InspectFooter(InspectFooterArgs),
|
||||
Plan(PlanArgs),
|
||||
}
|
||||
@@ -95,6 +98,9 @@ pub async fn run() {
|
||||
Some(Command::PromRemoteWrite(rw)) => run_prom_remote_write(rw)
|
||||
.await
|
||||
.expect("prom remote write failed"),
|
||||
Some(Command::OtlpMetrics(otlp)) => run_otlp_metrics(otlp)
|
||||
.await
|
||||
.expect("OTLP metrics export failed"),
|
||||
Some(Command::InspectFooter(inspect)) => run_inspect_footer(inspect)
|
||||
.await
|
||||
.expect("inspect footer failed"),
|
||||
|
||||
@@ -0,0 +1,294 @@
|
||||
// Copyright 2023 Greptime Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use clap::Args as ClapArgs;
|
||||
use prost::Message;
|
||||
use serde_json::json;
|
||||
|
||||
use crate::query_perf_fixture::case::ValuePattern;
|
||||
use crate::query_perf_fixture::prom_remote_write::deterministic_value;
|
||||
|
||||
#[derive(Debug, ClapArgs)]
|
||||
pub(super) struct OtlpMetricsArgs {
|
||||
#[arg(long, default_value = "http://127.0.0.1:4000/v1/otlp/v1/metrics")]
|
||||
endpoint: String,
|
||||
#[arg(long, default_value = "public")]
|
||||
database: String,
|
||||
#[arg(long)]
|
||||
metric: String,
|
||||
#[arg(long, default_value_t = 8)]
|
||||
series_count: u64,
|
||||
#[arg(long, default_value_t = 30)]
|
||||
samples_per_series: u64,
|
||||
#[arg(long)]
|
||||
start_unix_nanos: u64,
|
||||
#[arg(long)]
|
||||
step_nanos: u64,
|
||||
#[arg(long, alias = "batch-size", default_value_t = 8)]
|
||||
chunk_series_count: u64,
|
||||
#[arg(long, default_value_t = 60)]
|
||||
timeout_seconds: u64,
|
||||
#[arg(long, default_value_t = ValuePattern::Linear)]
|
||||
value_pattern: ValuePattern,
|
||||
#[arg(long, default_value_t = 0.0)]
|
||||
value_base: f64,
|
||||
#[arg(long, default_value_t = 0.125)]
|
||||
value_step: f64,
|
||||
#[arg(long, default_value_t = 97)]
|
||||
value_cardinality: u64,
|
||||
#[arg(long, default_value_t = 0)]
|
||||
value_seed: u64,
|
||||
#[arg(long, default_value_t = 8)]
|
||||
value_run_length: u64,
|
||||
#[arg(long, default_value_t = 100)]
|
||||
value_stall_every: u64,
|
||||
#[arg(long, default_value_t = 16)]
|
||||
value_stall_length: u64,
|
||||
#[arg(long, default_value_t = 5)]
|
||||
value_mixed_every: u64,
|
||||
}
|
||||
|
||||
pub(super) async fn run_otlp_metrics(
|
||||
args: OtlpMetricsArgs,
|
||||
) -> Result<(), Box<dyn std::error::Error>> {
|
||||
let started = std::time::Instant::now();
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(std::time::Duration::from_secs(args.timeout_seconds))
|
||||
.build()?;
|
||||
let mut batches = 0_u64;
|
||||
let mut rows = 0_u64;
|
||||
let mut http_statuses = Vec::new();
|
||||
let mut rejected_data_points = 0_i64;
|
||||
for first in (0..args.series_count).step_by(args.chunk_series_count.max(1) as usize) {
|
||||
let last = (first + args.chunk_series_count.max(1)).min(args.series_count);
|
||||
let request = ExportMetricsServiceRequest {
|
||||
resource_metrics: vec![ResourceMetrics {
|
||||
scope_metrics: vec![ScopeMetrics {
|
||||
metrics: vec![Metric {
|
||||
name: args.metric.clone(),
|
||||
data: Some(metric::Data::Gauge(Gauge {
|
||||
data_points: (first..last)
|
||||
.flat_map(|series_idx| {
|
||||
(0..args.samples_per_series).map(move |sample_idx| {
|
||||
NumberDataPoint {
|
||||
attributes: vec![
|
||||
string_attribute(
|
||||
"host",
|
||||
format!("host{:04}", series_idx % 1024),
|
||||
),
|
||||
string_attribute(
|
||||
"instance",
|
||||
format!("instance{:06}", series_idx),
|
||||
),
|
||||
],
|
||||
time_unix_nano: args.start_unix_nanos
|
||||
+ sample_idx * args.step_nanos,
|
||||
value: Some(number_data_point::Value::AsDouble(
|
||||
deterministic_value(
|
||||
args.value_pattern,
|
||||
args.value_base,
|
||||
args.value_step,
|
||||
args.value_cardinality,
|
||||
args.value_seed,
|
||||
args.value_run_length,
|
||||
args.value_stall_every,
|
||||
args.value_stall_length,
|
||||
args.value_mixed_every,
|
||||
series_idx,
|
||||
sample_idx,
|
||||
args.samples_per_series,
|
||||
),
|
||||
)),
|
||||
}
|
||||
})
|
||||
})
|
||||
.collect(),
|
||||
})),
|
||||
}],
|
||||
}],
|
||||
}],
|
||||
};
|
||||
let batch_rows = (last - first) * args.samples_per_series;
|
||||
let response = client
|
||||
.post(&args.endpoint)
|
||||
.header("content-type", "application/x-protobuf")
|
||||
.header("x-greptime-db-name", &args.database)
|
||||
.body(request.encode_to_vec())
|
||||
.send()
|
||||
.await?;
|
||||
let status = response.status();
|
||||
let bytes = response.bytes().await?;
|
||||
http_statuses.push(status.as_u16());
|
||||
if !status.is_success() {
|
||||
return Err(format!(
|
||||
"OTLP metrics export failed with status {status}: {}",
|
||||
String::from_utf8_lossy(&bytes)
|
||||
)
|
||||
.into());
|
||||
}
|
||||
let response = ExportMetricsServiceResponse::decode(bytes.as_ref())?;
|
||||
let rejected = response
|
||||
.partial_success
|
||||
.map(|partial| partial.rejected_data_points)
|
||||
.unwrap_or_default();
|
||||
if rejected != 0 {
|
||||
return Err(format!("OTLP metrics export rejected {rejected} data points").into());
|
||||
}
|
||||
rejected_data_points += rejected;
|
||||
rows += batch_rows;
|
||||
batches += 1;
|
||||
}
|
||||
println!(
|
||||
"{}",
|
||||
json!({"status":"ok","endpoint":args.endpoint,"database":args.database,"metric":args.metric,"input_protocol":"otlp_metrics","timestamp_precision":"ns","series_count":args.series_count,"samples_per_series":args.samples_per_series,"rows":rows,"samples_written":rows,"batches":batches,"elapsed_seconds":started.elapsed().as_secs_f64(),"http_statuses":http_statuses,"rejected_data_points":rejected_data_points})
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn string_attribute(key: &str, value: String) -> KeyValue {
|
||||
KeyValue {
|
||||
key: key.to_string(),
|
||||
value: Some(AnyValue {
|
||||
value: Some(any_value::Value::StringValue(value)),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
// The metrics endpoint uses the otel-arrow generated metric messages. Keep this
|
||||
// small wire encoder local to the fixture binary so production protocol code is
|
||||
// untouched.
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct ExportMetricsServiceRequest {
|
||||
#[prost(message, repeated, tag = "1")]
|
||||
resource_metrics: Vec<ResourceMetrics>,
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct ResourceMetrics {
|
||||
#[prost(message, repeated, tag = "2")]
|
||||
scope_metrics: Vec<ScopeMetrics>,
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct ScopeMetrics {
|
||||
#[prost(message, repeated, tag = "2")]
|
||||
metrics: Vec<Metric>,
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct Metric {
|
||||
#[prost(string, tag = "1")]
|
||||
name: String,
|
||||
#[prost(oneof = "metric::Data", tags = "5")]
|
||||
data: Option<metric::Data>,
|
||||
}
|
||||
|
||||
mod metric {
|
||||
#[derive(Clone, PartialEq, prost::Oneof)]
|
||||
pub(super) enum Data {
|
||||
#[prost(message, tag = "5")]
|
||||
Gauge(super::Gauge),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct Gauge {
|
||||
#[prost(message, repeated, tag = "1")]
|
||||
data_points: Vec<NumberDataPoint>,
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct NumberDataPoint {
|
||||
#[prost(message, repeated, tag = "7")]
|
||||
attributes: Vec<KeyValue>,
|
||||
#[prost(fixed64, tag = "3")]
|
||||
time_unix_nano: u64,
|
||||
#[prost(oneof = "number_data_point::Value", tags = "4")]
|
||||
value: Option<number_data_point::Value>,
|
||||
}
|
||||
|
||||
mod number_data_point {
|
||||
#[derive(Clone, Copy, PartialEq, prost::Oneof)]
|
||||
pub(super) enum Value {
|
||||
#[prost(double, tag = "4")]
|
||||
AsDouble(f64),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct KeyValue {
|
||||
#[prost(string, tag = "1")]
|
||||
key: String,
|
||||
#[prost(message, optional, tag = "2")]
|
||||
value: Option<AnyValue>,
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct AnyValue {
|
||||
#[prost(oneof = "any_value::Value", tags = "1")]
|
||||
value: Option<any_value::Value>,
|
||||
}
|
||||
|
||||
mod any_value {
|
||||
#[derive(Clone, PartialEq, prost::Oneof)]
|
||||
pub(super) enum Value {
|
||||
#[prost(string, tag = "1")]
|
||||
StringValue(String),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct ExportMetricsServiceResponse {
|
||||
#[prost(message, optional, tag = "1")]
|
||||
partial_success: Option<ExportMetricsPartialSuccess>,
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Message)]
|
||||
struct ExportMetricsPartialSuccess {
|
||||
#[prost(int64, tag = "1")]
|
||||
rejected_data_points: i64,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn encodes_gauge_timestamp_with_nanosecond_remainder() {
|
||||
let point = NumberDataPoint {
|
||||
attributes: vec![string_attribute("host", "host0000".to_string())],
|
||||
time_unix_nano: 1_704_067_200_000_000_123,
|
||||
value: Some(number_data_point::Value::AsDouble(1.0)),
|
||||
};
|
||||
let decoded = NumberDataPoint::decode(point.encode_to_vec().as_slice()).unwrap();
|
||||
assert_eq!(decoded.time_unix_nano % 1_000_000_000, 123);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_partial_success_rejections() {
|
||||
let response = ExportMetricsServiceResponse {
|
||||
partial_success: Some(ExportMetricsPartialSuccess {
|
||||
rejected_data_points: 2,
|
||||
}),
|
||||
};
|
||||
assert_eq!(
|
||||
ExportMetricsServiceResponse::decode(response.encode_to_vec().as_slice())
|
||||
.unwrap()
|
||||
.partial_success
|
||||
.unwrap()
|
||||
.rejected_data_points,
|
||||
2
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -147,58 +147,74 @@ fn prom_label(name: &str, value: &str) -> api::prom_store::remote::Label {
|
||||
}
|
||||
}
|
||||
fn deterministic_prom_value(args: &PromRemoteWriteArgs, series_idx: u64, sample_idx: u64) -> f64 {
|
||||
let ordinal = series_idx
|
||||
* args
|
||||
.value_total_samples_per_series
|
||||
.unwrap_or(args.samples_per_series)
|
||||
+ args.value_sample_offset
|
||||
+ sample_idx;
|
||||
let local = args.value_sample_offset + sample_idx;
|
||||
match args.value_pattern {
|
||||
ValuePattern::Linear => {
|
||||
args.value_base + (series_idx % 97) as f64 + local as f64 * args.value_step
|
||||
}
|
||||
ValuePattern::Constant => args.value_base,
|
||||
ValuePattern::Modulo => {
|
||||
args.value_base + (ordinal % args.value_cardinality.max(1)) as f64 * args.value_step
|
||||
}
|
||||
ValuePattern::Unique => args.value_base + ordinal as f64 * args.value_step,
|
||||
deterministic_value(
|
||||
args.value_pattern,
|
||||
args.value_base,
|
||||
args.value_step,
|
||||
args.value_cardinality,
|
||||
args.value_seed,
|
||||
args.value_run_length,
|
||||
args.value_stall_every,
|
||||
args.value_stall_length,
|
||||
args.value_mixed_every,
|
||||
series_idx,
|
||||
args.value_sample_offset + sample_idx,
|
||||
args.value_total_samples_per_series
|
||||
.unwrap_or(args.samples_per_series),
|
||||
)
|
||||
}
|
||||
|
||||
pub(super) fn deterministic_value(
|
||||
pattern: ValuePattern,
|
||||
base: f64,
|
||||
step: f64,
|
||||
cardinality: u64,
|
||||
seed: u64,
|
||||
run_length: u64,
|
||||
stall_every: u64,
|
||||
stall_length: u64,
|
||||
mixed_every: u64,
|
||||
series_idx: u64,
|
||||
sample_idx: u64,
|
||||
total_samples_per_series: u64,
|
||||
) -> f64 {
|
||||
let ordinal = series_idx * total_samples_per_series + sample_idx;
|
||||
match pattern {
|
||||
ValuePattern::Linear => base + (series_idx % 97) as f64 + sample_idx as f64 * step,
|
||||
ValuePattern::Constant => base,
|
||||
ValuePattern::Modulo => base + (ordinal % cardinality.max(1)) as f64 * step,
|
||||
ValuePattern::Unique => base + ordinal as f64 * step,
|
||||
ValuePattern::SeededRandom => {
|
||||
args.value_base
|
||||
+ (splitmix64(ordinal ^ args.value_seed) % args.value_cardinality.max(1)) as f64
|
||||
* args.value_step
|
||||
base + (splitmix64(ordinal ^ seed) % cardinality.max(1)) as f64 * step
|
||||
}
|
||||
ValuePattern::RunLength => {
|
||||
args.value_base
|
||||
+ ((ordinal / args.value_run_length.max(1)) % args.value_cardinality.max(1)) as f64
|
||||
* args.value_step
|
||||
base + ((ordinal / run_length.max(1)) % cardinality.max(1)) as f64 * step
|
||||
}
|
||||
ValuePattern::QuantizedSignal => {
|
||||
args.value_base
|
||||
+ (((series_idx % args.value_cardinality.max(1))
|
||||
+ (local / args.value_run_length.max(1)))
|
||||
% args.value_cardinality.max(1)) as f64
|
||||
* args.value_step
|
||||
base + (((series_idx % cardinality.max(1)) + (sample_idx / run_length.max(1)))
|
||||
% cardinality.max(1)) as f64
|
||||
* step
|
||||
}
|
||||
ValuePattern::SignalWithSporadicStalls => {
|
||||
let every = args.value_stall_every.max(1);
|
||||
let phase = local % every;
|
||||
let effective = if phase < args.value_stall_length.min(every) {
|
||||
local - phase
|
||||
let every = stall_every.max(1);
|
||||
let phase = sample_idx % every;
|
||||
let effective = if phase < stall_length.min(every) {
|
||||
sample_idx - phase
|
||||
} else {
|
||||
local
|
||||
sample_idx
|
||||
};
|
||||
args.value_base + (series_idx % 97) as f64 + effective as f64 * args.value_step
|
||||
base + (series_idx % 97) as f64 + effective as f64 * step
|
||||
}
|
||||
ValuePattern::MixedSignalRepeated => {
|
||||
if local.is_multiple_of(args.value_mixed_every.max(1)) {
|
||||
args.value_base
|
||||
if sample_idx.is_multiple_of(mixed_every.max(1)) {
|
||||
base
|
||||
} else {
|
||||
args.value_base + (series_idx % 97) as f64 + local as f64 * args.value_step
|
||||
base + (series_idx % 97) as f64 + sample_idx as f64 * step
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn splitmix64(mut value: u64) -> u64 {
|
||||
value = value.wrapping_add(0x9e3779b97f4a7c15);
|
||||
let mut mixed = value;
|
||||
|
||||
@@ -489,6 +489,7 @@ mod tests {
|
||||
append_mode: Some(true),
|
||||
sst_format: Some("flat".to_string()),
|
||||
validate_show_create_engine: true,
|
||||
validate_timestamp_nanos: None,
|
||||
};
|
||||
assert_eq!(
|
||||
create_table_sql(&table).unwrap(),
|
||||
@@ -536,6 +537,7 @@ mod tests {
|
||||
append_mode: Some(true),
|
||||
sst_format: Some("flat".to_string()),
|
||||
validate_show_create_engine: true,
|
||||
validate_timestamp_nanos: None,
|
||||
};
|
||||
assert_eq!(
|
||||
create_table_sql(&table).unwrap(),
|
||||
@@ -568,6 +570,7 @@ mod tests {
|
||||
append_mode: None,
|
||||
sst_format: None,
|
||||
validate_show_create_engine: true,
|
||||
validate_timestamp_nanos: None,
|
||||
};
|
||||
assert_eq!(
|
||||
create_table_sql(&table).unwrap(),
|
||||
|
||||
@@ -49,14 +49,23 @@ pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> {
|
||||
let client = Client::builder()
|
||||
.timeout(Duration::from_secs_f64(args.http_timeout))
|
||||
.build()?;
|
||||
let base = run_target(args.base_http_port, &tables, &configured_queries, &client).await;
|
||||
let candidate = run_target(
|
||||
let mut base = run_target(
|
||||
args.base_http_port,
|
||||
&tables,
|
||||
&configured_queries,
|
||||
&client,
|
||||
false,
|
||||
)
|
||||
.await;
|
||||
let mut candidate = run_target(
|
||||
args.candidate_http_port,
|
||||
&tables,
|
||||
&configured_queries,
|
||||
&client,
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
compare_measured_results(&configured_queries, &mut base, &mut candidate);
|
||||
let thresholds = enforce_thresholds(&configured_queries, &base, &candidate)?;
|
||||
let status = if base.status == "failed"
|
||||
|| candidate.status == "failed"
|
||||
@@ -74,6 +83,7 @@ pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> {
|
||||
"scenario": scenario_value,
|
||||
"queries": configured_queries,
|
||||
"query_mode": "endpoint",
|
||||
"input_protocol": scenario_value.pointer("/remote_write/input_protocol").cloned().unwrap_or(Value::String("direct_sst".to_string())),
|
||||
"http_timeout": args.http_timeout,
|
||||
"targets": [target_report("base", args.base_http_port, base), target_report("candidate", args.candidate_http_port, candidate)],
|
||||
"thresholds": thresholds,
|
||||
@@ -131,6 +141,7 @@ async fn run_target(
|
||||
tables: &[Table],
|
||||
configured_queries: &[Query],
|
||||
client: &Client,
|
||||
validate_candidate_plan: bool,
|
||||
) -> QueryResult {
|
||||
let mut queries = configured_queries.to_vec();
|
||||
if queries.is_empty() {
|
||||
@@ -144,6 +155,11 @@ async fn run_target(
|
||||
warmup: 0,
|
||||
iterations: 1,
|
||||
thresholds: Map::new(),
|
||||
expected_cardinality: None,
|
||||
compare_results: false,
|
||||
measure: true,
|
||||
plan_contains: None,
|
||||
plan_absent: None,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -170,18 +186,30 @@ async fn run_target(
|
||||
}
|
||||
validation.push(sample);
|
||||
}
|
||||
let first = post_query(client, port, &queries[0], db).await;
|
||||
if !first["ok"].as_bool().unwrap_or(false) {
|
||||
validation_errors.push(json!({
|
||||
"sql": queries[0].query,
|
||||
"error": first.get("error"),
|
||||
"response": first.get("response"),
|
||||
}));
|
||||
}
|
||||
validation.push(first);
|
||||
|
||||
let mut measurements = Vec::with_capacity(queries.len());
|
||||
for query in &queries {
|
||||
if !query.measure {
|
||||
let sample = post_query(client, port, query, db).await;
|
||||
if !sample["ok"].as_bool().unwrap_or(false) {
|
||||
validation_errors.push(json!({"sql": query.query, "error": sample.get("error"), "response": sample.get("response")}));
|
||||
}
|
||||
if validate_candidate_plan
|
||||
&& let Some(plan_contains) = &query.plan_contains
|
||||
&& !response_text(sample.get("response").unwrap_or(&Value::Null))
|
||||
.contains(plan_contains)
|
||||
{
|
||||
validation_errors.push(json!({"sql": query.query, "error": format!("plan evidence does not contain {plan_contains}"), "response": sample.get("response")}));
|
||||
}
|
||||
if validate_candidate_plan
|
||||
&& let Some(plan_absent) = &query.plan_absent
|
||||
&& response_text(sample.get("response").unwrap_or(&Value::Null))
|
||||
.contains(plan_absent)
|
||||
{
|
||||
validation_errors.push(json!({"sql": query.query, "error": format!("plan evidence unexpectedly contains {plan_absent}"), "response": sample.get("response")}));
|
||||
}
|
||||
validation.push(sample);
|
||||
continue;
|
||||
}
|
||||
for _ in 0..query.warmup {
|
||||
let warmup = post_query(client, port, query, db).await;
|
||||
if !warmup["ok"].as_bool().unwrap_or(false) {
|
||||
@@ -239,6 +267,92 @@ async fn run_target(
|
||||
}
|
||||
}
|
||||
|
||||
fn compare_measured_results(
|
||||
queries: &[Query],
|
||||
base: &mut QueryResult,
|
||||
candidate: &mut QueryResult,
|
||||
) {
|
||||
for query in queries {
|
||||
if !query.measure || (!query.compare_results && query.expected_cardinality.is_none()) {
|
||||
continue;
|
||||
}
|
||||
let Some(base_measurement) = base
|
||||
.measurements
|
||||
.iter()
|
||||
.find(|measurement| measurement.name == query.name)
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let Some(candidate_measurement) = candidate
|
||||
.measurements
|
||||
.iter()
|
||||
.find(|measurement| measurement.name == query.name)
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
for (iteration, (base_sample, candidate_sample)) in base_measurement
|
||||
.samples
|
||||
.iter()
|
||||
.zip(&candidate_measurement.samples)
|
||||
.enumerate()
|
||||
{
|
||||
let base_rows = crate::query_regression_runner::sql::extract_rows(
|
||||
base_sample.get("response").unwrap_or(&Value::Null),
|
||||
);
|
||||
let candidate_rows = crate::query_regression_runner::sql::extract_rows(
|
||||
candidate_sample.get("response").unwrap_or(&Value::Null),
|
||||
);
|
||||
if let Some(expected) = query.expected_cardinality
|
||||
&& (base_rows.len() as u64 != expected || candidate_rows.len() as u64 != expected)
|
||||
{
|
||||
candidate.validation_errors.push(json!({"query": query.name, "iteration": iteration, "error": format!("expected cardinality {expected}, got base={} candidate={}", base_rows.len(), candidate_rows.len())}));
|
||||
}
|
||||
if query.compare_results
|
||||
&& normalized_rows(base_rows) != normalized_rows(candidate_rows)
|
||||
{
|
||||
candidate.validation_errors.push(json!({"query": query.name, "iteration": iteration, "error": "base/candidate response rows differ"}));
|
||||
}
|
||||
}
|
||||
}
|
||||
if !candidate.validation_errors.is_empty() {
|
||||
candidate.status = "failed".to_string();
|
||||
}
|
||||
}
|
||||
|
||||
fn normalized_rows(rows: Vec<Value>) -> Vec<String> {
|
||||
let mut rows = rows
|
||||
.into_iter()
|
||||
.map(|mut row| {
|
||||
strip_timing_metadata(&mut row);
|
||||
serde_json::to_string(&row).unwrap_or_default()
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
rows.sort_unstable();
|
||||
rows
|
||||
}
|
||||
|
||||
fn strip_timing_metadata(value: &mut Value) {
|
||||
match value {
|
||||
Value::Object(values) => {
|
||||
values.retain(|key, _| {
|
||||
!matches!(
|
||||
key.as_str(),
|
||||
"execution_time" | "execution_time_ms" | "elapsed" | "elapsed_ms" | "cost"
|
||||
)
|
||||
});
|
||||
for value in values.values_mut() {
|
||||
strip_timing_metadata(value);
|
||||
}
|
||||
}
|
||||
Value::Array(values) => {
|
||||
for value in values {
|
||||
strip_timing_metadata(value);
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_show_create(result: &Value, table: &Table) -> Vec<&'static str> {
|
||||
let text = result
|
||||
.get("response")
|
||||
@@ -258,6 +372,15 @@ fn validate_show_create(result: &Value, table: &Table) -> Vec<&'static str> {
|
||||
if table.sst_format.is_some() && !text.contains("sst_format") {
|
||||
errors.push("SHOW CREATE output does not mention sst_format");
|
||||
}
|
||||
if table.validate_timestamp_nanos.is_some()
|
||||
&& (!text.contains("greptime_timestamp")
|
||||
|| !text.contains("timestamp(9)")
|
||||
|| !text.contains("primary key")
|
||||
|| !text.contains("host")
|
||||
|| !text.contains("instance"))
|
||||
{
|
||||
errors.push("SHOW CREATE output does not contain the expected native OTLP Mito schema");
|
||||
}
|
||||
errors
|
||||
}
|
||||
|
||||
@@ -441,6 +564,11 @@ mod tests {
|
||||
step: None,
|
||||
warmup: 0,
|
||||
iterations: 1,
|
||||
expected_cardinality: None,
|
||||
compare_results: false,
|
||||
measure: true,
|
||||
plan_contains: None,
|
||||
plan_absent: None,
|
||||
thresholds: Map::from_iter([
|
||||
("max_candidate_latency_regression_pct".to_string(), json!(0)),
|
||||
("other".to_string(), json!(1)),
|
||||
|
||||
@@ -83,6 +83,8 @@ pub(super) struct Table {
|
||||
pub(super) sst_format: Option<String>,
|
||||
#[serde(default = "default_show_create_engine")]
|
||||
pub(super) validate_show_create_engine: bool,
|
||||
#[serde(default)]
|
||||
pub(super) validate_timestamp_nanos: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
@@ -104,6 +106,8 @@ const fn default_show_create_engine() -> bool {
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
pub(super) struct RemoteWrite {
|
||||
#[serde(default)]
|
||||
pub(super) input_protocol: InputProtocol,
|
||||
pub(super) database: String,
|
||||
pub(super) metric: String,
|
||||
pub(super) physical_table: String,
|
||||
@@ -122,6 +126,14 @@ pub(super) struct RemoteWrite {
|
||||
pub(super) read_bench: Option<ReadBenchConfig>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Serialize, Eq, PartialEq, Default)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub(super) enum InputProtocol {
|
||||
#[default]
|
||||
RemoteWrite,
|
||||
OtlpMetrics,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
pub(super) struct PromStore {
|
||||
pub(super) pending_rows_flush_interval: String,
|
||||
@@ -222,12 +234,26 @@ pub(super) struct Query {
|
||||
pub(super) iterations: usize,
|
||||
#[serde(default)]
|
||||
pub(super) thresholds: Map<String, Value>,
|
||||
#[serde(default)]
|
||||
pub(super) expected_cardinality: Option<u64>,
|
||||
#[serde(default)]
|
||||
pub(super) compare_results: bool,
|
||||
#[serde(default = "default_measure")]
|
||||
pub(super) measure: bool,
|
||||
#[serde(default)]
|
||||
pub(super) plan_contains: Option<String>,
|
||||
#[serde(default)]
|
||||
pub(super) plan_absent: Option<String>,
|
||||
}
|
||||
|
||||
const fn one() -> usize {
|
||||
1
|
||||
}
|
||||
|
||||
const fn default_measure() -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
pub(super) struct QueryResult {
|
||||
pub(super) validation: Vec<Value>,
|
||||
|
||||
@@ -20,7 +20,7 @@ use serde_json::Value;
|
||||
|
||||
use crate::query_regression_runner::Result;
|
||||
use crate::query_regression_runner::model::{
|
||||
Layout, OtlpLoad, Query, RemoteWrite, Scenario, Table,
|
||||
InputProtocol, Layout, OtlpLoad, Query, RemoteWrite, Scenario, Table,
|
||||
};
|
||||
|
||||
pub(super) fn load_plan(generator: &PathBuf, case_path: &PathBuf) -> Result<Value> {
|
||||
@@ -46,17 +46,35 @@ pub(super) fn normalize_scenario(scenario: Scenario) -> Result<(Vec<Table>, Vec<
|
||||
remote_write,
|
||||
queries,
|
||||
} => Ok((
|
||||
vec![Table {
|
||||
database: remote_write.database,
|
||||
name: remote_write.metric,
|
||||
engine: "metric".to_string(),
|
||||
columns: vec![],
|
||||
primary_key: vec![],
|
||||
time_index: None,
|
||||
append_mode: None,
|
||||
sst_format: None,
|
||||
validate_show_create_engine: false,
|
||||
}],
|
||||
vec![
|
||||
if remote_write.input_protocol == InputProtocol::OtlpMetrics {
|
||||
Table {
|
||||
database: remote_write.database,
|
||||
name: remote_write.metric,
|
||||
engine: "mito".to_string(),
|
||||
columns: vec![],
|
||||
primary_key: vec!["host".to_string(), "instance".to_string()],
|
||||
time_index: Some("greptime_timestamp".to_string()),
|
||||
append_mode: None,
|
||||
sst_format: None,
|
||||
validate_show_create_engine: true,
|
||||
validate_timestamp_nanos: Some(123),
|
||||
}
|
||||
} else {
|
||||
Table {
|
||||
database: remote_write.database,
|
||||
name: remote_write.metric,
|
||||
engine: "metric".to_string(),
|
||||
columns: vec![],
|
||||
primary_key: vec![],
|
||||
time_index: None,
|
||||
append_mode: None,
|
||||
sst_format: None,
|
||||
validate_show_create_engine: false,
|
||||
validate_timestamp_nanos: None,
|
||||
}
|
||||
},
|
||||
],
|
||||
queries,
|
||||
)),
|
||||
Scenario::OtlpTraceLoad { .. } => {
|
||||
|
||||
@@ -20,7 +20,7 @@ use std::time::{Duration, Instant};
|
||||
use reqwest::Client;
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use crate::query_regression_runner::model::{PromStore, RemoteWrite};
|
||||
use crate::query_regression_runner::model::{InputProtocol, RemoteWrite};
|
||||
use crate::query_regression_runner::plan::normalized_remote_write;
|
||||
use crate::query_regression_runner::sql::{
|
||||
extract_count_value, http_post_sql, sql_ident, sql_string, value_f64, value_u64,
|
||||
@@ -29,13 +29,15 @@ use crate::query_regression_runner::{PrepareRemoteArgs, RenderRemoteConfigArgs,
|
||||
|
||||
pub(super) async fn run_render_remote_config(args: RenderRemoteConfigArgs) -> Result<()> {
|
||||
let (_, remote) = normalized_remote_write(&args.fixture_generator, &args.case)?;
|
||||
fs::write(args.output, frontend_prom_config(&remote.prom_store)?)?;
|
||||
fs::write(args.output, frontend_remote_config(&remote)?)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn frontend_prom_config(prom: &PromStore) -> Result<String> {
|
||||
fn frontend_remote_config(remote: &RemoteWrite) -> Result<String> {
|
||||
let prom = &remote.prom_store;
|
||||
let with_metric_engine = remote.input_protocol == InputProtocol::RemoteWrite;
|
||||
Ok(format!(
|
||||
"[prom_store]\nenable = true\nwith_metric_engine = true\npending_rows_flush_interval = {}\nmax_batch_rows = {}\nmax_concurrent_flushes = {}\nworker_channel_capacity = {}\nmax_inflight_requests = {}\n",
|
||||
"[prom_store]\nenable = true\nwith_metric_engine = {with_metric_engine}\npending_rows_flush_interval = {}\nmax_batch_rows = {}\nmax_concurrent_flushes = {}\nworker_channel_capacity = {}\nmax_inflight_requests = {}\n",
|
||||
serde_json::to_string(&prom.pending_rows_flush_interval)?,
|
||||
prom.max_batch_rows,
|
||||
prom.max_concurrent_flushes,
|
||||
@@ -71,6 +73,7 @@ pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> {
|
||||
let report = json!({
|
||||
"case_path": case_path,
|
||||
"scenario": "prom_remote_write_then_query",
|
||||
"input_protocol": remote.input_protocol,
|
||||
"base": base,
|
||||
"candidate": candidate,
|
||||
"status": "ok",
|
||||
@@ -107,6 +110,14 @@ async fn prepare_remote_target(
|
||||
)
|
||||
.into());
|
||||
}
|
||||
if remote.input_protocol == InputProtocol::OtlpMetrics {
|
||||
let create = create_otlp_metric_table(client, port, remote).await;
|
||||
if !create["ok"].as_bool().unwrap_or(false) {
|
||||
return Err(
|
||||
format!("CREATE TABLE {} failed for {name}: {create}", remote.metric).into(),
|
||||
);
|
||||
}
|
||||
}
|
||||
let (remote_write, flushes) = ingest_remote_write(generator, port, remote, client).await?;
|
||||
let expected_rows = remote
|
||||
.series_count
|
||||
@@ -121,16 +132,66 @@ async fn prepare_remote_target(
|
||||
remote.visibility_timeout_seconds,
|
||||
)
|
||||
.await?;
|
||||
let timestamp_precision = if remote.input_protocol == InputProtocol::OtlpMetrics {
|
||||
verify_otlp_timestamp_precision(client, port, remote).await?
|
||||
} else {
|
||||
json!({"status": "skipped", "reason": "input_protocol is remote_write"})
|
||||
};
|
||||
Ok(json!({
|
||||
"name": name,
|
||||
"create_database": create_database,
|
||||
"input_protocol": remote.input_protocol,
|
||||
"otlp_metric_compat": if remote.input_protocol == InputProtocol::OtlpMetrics { Value::String("absent (frontend legacy-data compatibility)".to_string()) } else { Value::Null },
|
||||
"remote_write": remote_write,
|
||||
"flushes": flushes,
|
||||
"visibility": visibility,
|
||||
"timestamp_precision": timestamp_precision,
|
||||
"status": "ok",
|
||||
}))
|
||||
}
|
||||
|
||||
async fn create_otlp_metric_table(client: &Client, port: u16, remote: &RemoteWrite) -> Value {
|
||||
http_post_sql(
|
||||
client,
|
||||
port,
|
||||
&format!(
|
||||
"CREATE TABLE {} (host STRING, instance STRING, greptime_timestamp TIMESTAMP(9) TIME INDEX, greptime_value DOUBLE, PRIMARY KEY(host, instance)) ENGINE=mito",
|
||||
sql_ident(&remote.metric),
|
||||
),
|
||||
&remote.database,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn verify_otlp_timestamp_precision(
|
||||
client: &Client,
|
||||
port: u16,
|
||||
remote: &RemoteWrite,
|
||||
) -> Result<Value> {
|
||||
let result = http_post_sql(
|
||||
client,
|
||||
port,
|
||||
&format!(
|
||||
"SELECT greptime_timestamp FROM {} ORDER BY greptime_timestamp LIMIT 1",
|
||||
sql_ident(&remote.metric),
|
||||
),
|
||||
&remote.database,
|
||||
)
|
||||
.await;
|
||||
let stored = crate::query_regression_runner::sql::extract_rows(
|
||||
result.get("response").unwrap_or(&Value::Null),
|
||||
)
|
||||
.first()
|
||||
.and_then(|row| crate::query_regression_runner::sql::row_value(row, 0, "greptime_timestamp"))
|
||||
.map(crate::query_regression_runner::sql::value_text)
|
||||
.unwrap_or_default();
|
||||
let exact_remainder = stored.contains("000000123");
|
||||
if !result["ok"].as_bool().unwrap_or(false) || !exact_remainder {
|
||||
return Err(format!("OTLP timestamp precision check failed: {result}").into());
|
||||
}
|
||||
Ok(json!({"status": "ok", "stored_timestamp": stored, "expected_nanosecond_remainder": 123}))
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
struct SampleChunk {
|
||||
index: u64,
|
||||
@@ -302,23 +363,42 @@ fn remote_write_command(
|
||||
));
|
||||
let mut command = vec![
|
||||
generator.to_string_lossy().to_string(),
|
||||
"prom-remote-write".to_string(),
|
||||
match remote.input_protocol {
|
||||
InputProtocol::RemoteWrite => "prom-remote-write",
|
||||
InputProtocol::OtlpMetrics => "otlp-metrics",
|
||||
}
|
||||
.to_string(),
|
||||
"--endpoint".to_string(),
|
||||
format!("http://127.0.0.1:{port}/v1/prometheus/write"),
|
||||
match remote.input_protocol {
|
||||
InputProtocol::RemoteWrite => format!("http://127.0.0.1:{port}/v1/prometheus/write"),
|
||||
InputProtocol::OtlpMetrics => format!("http://127.0.0.1:{port}/v1/otlp/v1/metrics"),
|
||||
},
|
||||
"--database".to_string(),
|
||||
remote.database.clone(),
|
||||
"--metric".to_string(),
|
||||
remote.metric.clone(),
|
||||
"--physical-table".to_string(),
|
||||
remote.physical_table.clone(),
|
||||
"--series-count".to_string(),
|
||||
remote.series_count.to_string(),
|
||||
"--samples-per-series".to_string(),
|
||||
samples_per_series.to_string(),
|
||||
"--start-unix-millis".to_string(),
|
||||
start_unix_millis.to_string(),
|
||||
"--step-millis".to_string(),
|
||||
remote.step_millis.to_string(),
|
||||
match remote.input_protocol {
|
||||
InputProtocol::RemoteWrite => "--start-unix-millis",
|
||||
InputProtocol::OtlpMetrics => "--start-unix-nanos",
|
||||
}
|
||||
.to_string(),
|
||||
match remote.input_protocol {
|
||||
InputProtocol::RemoteWrite => start_unix_millis.to_string(),
|
||||
InputProtocol::OtlpMetrics => (start_unix_millis as u64 * 1_000_000 + 123).to_string(),
|
||||
},
|
||||
match remote.input_protocol {
|
||||
InputProtocol::RemoteWrite => "--step-millis",
|
||||
InputProtocol::OtlpMetrics => "--step-nanos",
|
||||
}
|
||||
.to_string(),
|
||||
match remote.input_protocol {
|
||||
InputProtocol::RemoteWrite => remote.step_millis.to_string(),
|
||||
InputProtocol::OtlpMetrics => (remote.step_millis as u64 * 1_000_000).to_string(),
|
||||
},
|
||||
"--chunk-series-count".to_string(),
|
||||
remote.chunk_series_count.to_string(),
|
||||
"--timeout-seconds".to_string(),
|
||||
@@ -342,13 +422,23 @@ fn remote_write_command(
|
||||
"--value-mixed-every".to_string(),
|
||||
remote.value.mixed_every.to_string(),
|
||||
];
|
||||
if let Some(sample_offset) = sample_offset {
|
||||
if remote.input_protocol == InputProtocol::RemoteWrite {
|
||||
command.extend([
|
||||
"--physical-table".to_string(),
|
||||
remote.physical_table.clone(),
|
||||
]);
|
||||
}
|
||||
if remote.input_protocol == InputProtocol::RemoteWrite
|
||||
&& let Some(sample_offset) = sample_offset
|
||||
{
|
||||
command.extend([
|
||||
"--value-sample-offset".to_string(),
|
||||
sample_offset.to_string(),
|
||||
]);
|
||||
}
|
||||
if let Some(total_samples_per_series) = total_samples_per_series {
|
||||
if remote.input_protocol == InputProtocol::RemoteWrite
|
||||
&& let Some(total_samples_per_series) = total_samples_per_series
|
||||
{
|
||||
command.extend([
|
||||
"--value-total-samples-per-series".to_string(),
|
||||
total_samples_per_series.to_string(),
|
||||
@@ -364,25 +454,26 @@ async fn flush_remote_table(
|
||||
reason: &str,
|
||||
chunk_index: Option<u64>,
|
||||
) -> Result<Value> {
|
||||
let table = match remote.input_protocol {
|
||||
InputProtocol::RemoteWrite => &remote.physical_table,
|
||||
InputProtocol::OtlpMetrics => &remote.metric,
|
||||
};
|
||||
let mut result = http_post_sql(
|
||||
client,
|
||||
port,
|
||||
&format!("ADMIN FLUSH_TABLE({})", sql_string(&remote.physical_table)),
|
||||
&format!("ADMIN FLUSH_TABLE({})", sql_string(table)),
|
||||
&remote.database,
|
||||
)
|
||||
.await;
|
||||
if !result["ok"].as_bool().unwrap_or(false) {
|
||||
return Err(format!(
|
||||
"ADMIN FLUSH_TABLE {} failed: {result}",
|
||||
remote.physical_table
|
||||
)
|
||||
.into());
|
||||
return Err(format!("ADMIN FLUSH_TABLE {} failed: {result}", table).into());
|
||||
}
|
||||
result
|
||||
.as_object_mut()
|
||||
.ok_or("flush result must be an object")?
|
||||
.extend([
|
||||
("physical_table".to_string(), json!(remote.physical_table)),
|
||||
("flush_table".to_string(), json!(table)),
|
||||
("reason".to_string(), json!(reason)),
|
||||
("chunk_index".to_string(), json!(chunk_index)),
|
||||
]);
|
||||
@@ -440,6 +531,7 @@ mod tests {
|
||||
#[test]
|
||||
fn schedules_remote_sample_chunks_and_flushes() {
|
||||
let remote = RemoteWrite {
|
||||
input_protocol: InputProtocol::RemoteWrite,
|
||||
database: "public".to_string(),
|
||||
metric: "metric".to_string(),
|
||||
physical_table: "physical".to_string(),
|
||||
|
||||
+29
-3
@@ -163,9 +163,9 @@ frontend path. Treat all performance conclusions as release-only; debug builds
|
||||
are suitable only for command wiring and correctness checks.
|
||||
|
||||
`prepare-remote` creates the configured database if needed. The outer driver writes a per-target
|
||||
frontend config enabling `[prom_store]` with metric engine storage and a non-zero
|
||||
`pending_rows_flush_interval`, and validates that the logical metric table reaches
|
||||
`series_count * samples_per_series` rows before trusting the query measurements.
|
||||
frontend config enabling `[prom_store]` with a non-zero `pending_rows_flush_interval`; the default
|
||||
`input_protocol = "remote_write"` also enables metric-engine storage. It validates that the logical
|
||||
metric table reaches `series_count * samples_per_series` rows before trusting the query measurements.
|
||||
Use `--fixture-generator /path/to/query_perf_fixture` to provide the Rust helper
|
||||
to the outer driver.
|
||||
|
||||
@@ -448,3 +448,29 @@ Additional SQL optimizer cases:
|
||||
`LIMIT`.
|
||||
- `sql_join_filter_order`: two direct-SST tables joined on a shared tag with
|
||||
time filters, aggregate ordering, and `LIMIT`.
|
||||
|
||||
### Native OTLP nanosecond LastRow case
|
||||
|
||||
`otlp_ns_last_row_9070` reuses the remote-write-then-query lifecycle with
|
||||
`input_protocol = "otlp_metrics"`. Before each fresh target is loaded,
|
||||
`prepare-remote` creates its lowercase `ENGINE=mito` metric table with
|
||||
`greptime_timestamp TIMESTAMP(9)`, then sends binary OTLP gauge protobufs to
|
||||
`/v1/otlp/v1/metrics`. The load report records `input_protocol`, HTTP statuses,
|
||||
and decoded OTLP `partial_success.rejected_data_points`; a nonzero rejection
|
||||
fails preparation. It flushes the logical metric table, checks all 262144 rows,
|
||||
and verifies the stored timestamp retains the `+123ns` remainder.
|
||||
|
||||
The measured requests are ordinary `TQL EVAL` instant and range selectors (3
|
||||
warmups, 9 iterations). Untimed `TQL EXPLAIN VERBOSE` evidence is retained in
|
||||
the report, and every measured response is row-normalized and compared with
|
||||
the base target, ignoring only timing metadata. This is explicitly
|
||||
legacy-data query coverage; it does not assert the new default OTLP
|
||||
millisecond table behavior. Use the usual remote-case dispatch:
|
||||
|
||||
```bash
|
||||
uv run --no-project python .github/scripts/query-regression-run.py \
|
||||
--cases tests/perf/query_cases/otlp_ns_last_row_9070/case.toml \
|
||||
--base-bin /path/to/base/greptime --candidate-bin /path/to/candidate/greptime \
|
||||
--fixture-generator /path/to/query_perf_fixture \
|
||||
--runner /path/to/query_regression_runner --work-dir /path/to/work
|
||||
```
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
# Native OTLP metrics regression coverage for #9070. The table is explicitly
|
||||
# pre-created so the native OTLP path preserves nanosecond timestamps rather
|
||||
# than selecting a legacy compatibility schema.
|
||||
|
||||
[case]
|
||||
name = "otlp_ns_last_row_9070"
|
||||
description = "Native OTLP gauge ingestion with nanosecond timestamps and ordinary PromQL LastRow queries"
|
||||
issue = "https://github.com/GreptimeTeam/greptimedb/issues/9070"
|
||||
|
||||
[scenario]
|
||||
kind = "prom_remote_write_then_query"
|
||||
|
||||
[scenario.remote_write]
|
||||
input_protocol = "otlp_metrics"
|
||||
database = "public"
|
||||
metric = "otlp_ns_last_row_9070"
|
||||
physical_table = "greptime_physical_table"
|
||||
series_count = 256
|
||||
samples_per_series = 1024
|
||||
start_unix_millis = 1_704_067_200_000
|
||||
step_millis = 100
|
||||
chunk_series_count = 256
|
||||
timeout_seconds = 300
|
||||
visibility_timeout_seconds = 120
|
||||
|
||||
# Legacy-data query coverage only: this case intentionally does not test the
|
||||
# new default OTLP millisecond table behavior.
|
||||
|
||||
[[scenario.queries]]
|
||||
name = "instant_selector_last_row_candidate"
|
||||
kind = "tql"
|
||||
query = "TQL EVAL (1704067303, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}"
|
||||
warmup = 3
|
||||
iterations = 9
|
||||
compare_results = true
|
||||
expected_cardinality = 256
|
||||
|
||||
[scenario.queries.thresholds]
|
||||
max_candidate_latency_regression_pct = 25
|
||||
|
||||
[[scenario.queries]]
|
||||
name = "range_selector_last_row_control"
|
||||
kind = "tql"
|
||||
query = "TQL EVAL (1704067293, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}"
|
||||
warmup = 3
|
||||
iterations = 9
|
||||
compare_results = true
|
||||
expected_cardinality = 2816
|
||||
|
||||
[scenario.queries.thresholds]
|
||||
max_candidate_latency_regression_pct = 25
|
||||
|
||||
# Keep plan evidence untimed. It is reported with validation and never enters
|
||||
# the latency sample set.
|
||||
[[scenario.queries]]
|
||||
name = "instant_last_row_plan"
|
||||
kind = "tql"
|
||||
query = "TQL EXPLAIN VERBOSE (1704067303, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}"
|
||||
measure = false
|
||||
plan_contains = "LastRow"
|
||||
|
||||
[[scenario.queries]]
|
||||
name = "range_no_last_row_plan"
|
||||
kind = "tql"
|
||||
query = "TQL EXPLAIN VERBOSE (1704067293, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}"
|
||||
measure = false
|
||||
plan_absent = "LastRow"
|
||||
Reference in New Issue
Block a user