mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-13 07:22:32 +00:00
fix: preserve OTLP metric nanosecond timestamps
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
This commit is contained in:
Lei, HUANG
@@ -409,8 +409,8 @@ mod tests {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use api::v1::RowInsertRequests;
|
||||
use api::v1::meta::Role;
|
||||
use api::v1::{RowInsertRequest, RowInsertRequests, Rows};
|
||||
use async_trait::async_trait;
|
||||
use auth::{
|
||||
PermissionChecker, PermissionCheckerRef, PermissionReq, PermissionResp,
|
||||
@@ -507,6 +507,46 @@ mod tests {
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_all_empty_metric_insert_is_noop_without_physical_table() {
|
||||
let instance = test_instance().await;
|
||||
let ctx = metric_query_ctx(true);
|
||||
let physical_table = "empty_metric_physical";
|
||||
let requests = RowInsertRequests {
|
||||
inserts: vec![
|
||||
RowInsertRequest {
|
||||
table_name: "missing_rows".to_string(),
|
||||
rows: None,
|
||||
},
|
||||
RowInsertRequest {
|
||||
table_name: "empty_rows".to_string(),
|
||||
rows: Some(Rows::default()),
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
let output = instance
|
||||
.handle_metric_row_inserts(requests, ctx.clone(), physical_table.to_string())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(matches!(output.data, client::OutputData::AffectedRows(0)));
|
||||
assert!(output.meta.plan.is_none());
|
||||
assert!(
|
||||
instance
|
||||
.catalog_manager()
|
||||
.table(
|
||||
DEFAULT_CATALOG_NAME,
|
||||
DEFAULT_SCHEMA_NAME,
|
||||
physical_table,
|
||||
Some(ctx.as_ref()),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct CountingPermissionChecker {
|
||||
checks: AtomicUsize,
|
||||
|
||||
@@ -37,7 +37,8 @@ use common_recordbatch::RecordBatches;
|
||||
use common_telemetry::{debug, tracing};
|
||||
use operator::insert::{
|
||||
AutoCreateTableType, InserterRef, PerTableSemanticIndex, apply_per_table_semantic_options,
|
||||
build_create_table_expr, fill_table_options_for_create, parse_per_table_semantic_index,
|
||||
build_create_table_expr, fill_table_options_for_create, metric_timestamp_datatype,
|
||||
parse_per_table_semantic_index,
|
||||
};
|
||||
use operator::statement::StatementExecutor;
|
||||
use prost::Message;
|
||||
@@ -300,10 +301,12 @@ impl PendingRowsSchemaAlterer for Instance {
|
||||
|
||||
let create_type = auto_create_table_type_for_prom_remote_write(&ctx, with_metric_engine);
|
||||
if let Some(physical_table) = required_physical_table_for_create_type(&create_type) {
|
||||
let timestamp_datatype = metric_timestamp_datatype_for_server(tables.iter().copied())?;
|
||||
self.create_metric_physical_table_if_missing(
|
||||
catalog,
|
||||
schema,
|
||||
physical_table,
|
||||
timestamp_datatype,
|
||||
ctx.clone(),
|
||||
)
|
||||
.await?;
|
||||
@@ -664,6 +667,7 @@ impl Instance {
|
||||
catalog: &str,
|
||||
schema: &str,
|
||||
physical_table: &str,
|
||||
timestamp_datatype: ColumnDataType,
|
||||
ctx: QueryContextRef,
|
||||
) -> ServerResult<()> {
|
||||
let table = self
|
||||
@@ -677,22 +681,7 @@ impl Instance {
|
||||
}
|
||||
|
||||
let table_ref = TableReference::full(catalog, schema, physical_table);
|
||||
let default_schema = vec![
|
||||
api::v1::ColumnSchema {
|
||||
column_name: common_query::prelude::greptime_timestamp().to_string(),
|
||||
datatype: api::v1::ColumnDataType::TimestampMillisecond as i32,
|
||||
semantic_type: api::v1::SemanticType::Timestamp as i32,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
},
|
||||
api::v1::ColumnSchema {
|
||||
column_name: common_query::prelude::greptime_value().to_string(),
|
||||
datatype: api::v1::ColumnDataType::Float64 as i32,
|
||||
semantic_type: api::v1::SemanticType::Field as i32,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
},
|
||||
];
|
||||
let default_schema = metric_physical_table_schema(timestamp_datatype);
|
||||
let mut create_table_expr = build_create_table_expr(
|
||||
&table_ref,
|
||||
&default_schema,
|
||||
@@ -713,6 +702,33 @@ impl Instance {
|
||||
}
|
||||
}
|
||||
|
||||
fn metric_timestamp_datatype_for_server<'a>(
|
||||
tables: impl IntoIterator<Item = (&'a str, &'a [api::v1::ColumnSchema])>,
|
||||
) -> ServerResult<ColumnDataType> {
|
||||
metric_timestamp_datatype(tables)
|
||||
.map_err(BoxedError::new)
|
||||
.context(error::ExecuteGrpcQuerySnafu)
|
||||
}
|
||||
|
||||
fn metric_physical_table_schema(timestamp_datatype: ColumnDataType) -> Vec<api::v1::ColumnSchema> {
|
||||
vec![
|
||||
api::v1::ColumnSchema {
|
||||
column_name: common_query::prelude::greptime_timestamp().to_string(),
|
||||
datatype: timestamp_datatype as i32,
|
||||
semantic_type: SemanticType::Timestamp as i32,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
},
|
||||
api::v1::ColumnSchema {
|
||||
column_name: common_query::prelude::greptime_value().to_string(),
|
||||
datatype: ColumnDataType::Float64 as i32,
|
||||
semantic_type: SemanticType::Field as i32,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
/// This handler is mainly used for `frontend` or `standalone` to directly import
|
||||
/// the metrics collected by itself, thereby avoiding importing metrics through the network,
|
||||
/// thus reducing compression and network transmission overhead,
|
||||
@@ -802,10 +818,51 @@ mod tests {
|
||||
use std::sync::Arc;
|
||||
|
||||
use api::prom_store::remote::LabelMatcher;
|
||||
use common_error::ext::ErrorExt;
|
||||
use common_error::status_code::StatusCode;
|
||||
use session::context::QueryContext;
|
||||
|
||||
use super::*;
|
||||
|
||||
fn metric_schema(timestamp_datatype: ColumnDataType) -> Vec<api::v1::ColumnSchema> {
|
||||
vec![api::v1::ColumnSchema {
|
||||
column_name: "ts".to_string(),
|
||||
datatype: timestamp_datatype as i32,
|
||||
semantic_type: SemanticType::Timestamp as i32,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
}]
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_metric_physical_schema_uses_derived_nanosecond_timestamp() {
|
||||
let first = metric_schema(ColumnDataType::TimestampNanosecond);
|
||||
let second = metric_schema(ColumnDataType::TimestampNanosecond);
|
||||
let tables = [("first", first.as_slice()), ("second", second.as_slice())];
|
||||
|
||||
let datatype = metric_timestamp_datatype_for_server(tables).unwrap();
|
||||
let physical_schema = metric_physical_table_schema(datatype);
|
||||
|
||||
assert_eq!(ColumnDataType::TimestampNanosecond, datatype);
|
||||
assert_eq!(
|
||||
ColumnDataType::TimestampNanosecond as i32,
|
||||
physical_schema[0].datatype
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_invalid_metric_timestamp_schema_preserves_client_error() {
|
||||
let error =
|
||||
metric_timestamp_datatype_for_server([("missing_timestamp", &[][..])]).unwrap_err();
|
||||
|
||||
assert_eq!(StatusCode::InvalidArguments, error.status_code());
|
||||
assert!(
|
||||
error.output_msg().contains(
|
||||
"Metric table `missing_timestamp` must have exactly one timestamp column"
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_auto_create_table_type_for_prom_remote_write_metric_engine() {
|
||||
let mut query_ctx = QueryContext::with(
|
||||
|
||||
+140
-3
@@ -319,10 +319,23 @@ impl Inserter {
|
||||
.unwrap_or_default()
|
||||
});
|
||||
validate_column_count_match(&requests)?;
|
||||
if requests.inserts.is_empty() {
|
||||
return Ok(Output::new_with_affected_rows(0));
|
||||
}
|
||||
let timestamp_datatype =
|
||||
metric_timestamp_datatype(requests.inserts.iter().map(|request| {
|
||||
let rows = request.rows.as_ref().unwrap();
|
||||
(request.table_name.as_str(), rows.schema.as_slice())
|
||||
}))?;
|
||||
|
||||
// check and create physical table
|
||||
self.create_physical_table_on_demand(&ctx, physical_table.clone(), statement_executor)
|
||||
.await?;
|
||||
self.create_physical_table_on_demand(
|
||||
&ctx,
|
||||
physical_table.clone(),
|
||||
timestamp_datatype,
|
||||
statement_executor,
|
||||
)
|
||||
.await?;
|
||||
|
||||
// check and create logical tables
|
||||
let CreateAlterTableResult {
|
||||
@@ -900,6 +913,7 @@ impl Inserter {
|
||||
&self,
|
||||
ctx: &QueryContextRef,
|
||||
physical_table: String,
|
||||
timestamp_datatype: ColumnDataType,
|
||||
statement_executor: &StatementExecutor,
|
||||
) -> Result<()> {
|
||||
let catalog_name = ctx.current_catalog();
|
||||
@@ -931,7 +945,7 @@ impl Inserter {
|
||||
let default_schema = vec![
|
||||
ColumnSchema {
|
||||
column_name: greptime_timestamp().to_string(),
|
||||
datatype: ColumnDataType::TimestampMillisecond as _,
|
||||
datatype: timestamp_datatype as _,
|
||||
semantic_type: SemanticType::Timestamp as _,
|
||||
datatype_extension: None,
|
||||
options: None,
|
||||
@@ -1285,6 +1299,70 @@ fn validate_column_count_match(requests: &RowInsertRequests) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn metric_table_timestamp_datatype(
|
||||
table_name: &str,
|
||||
schema: &[ColumnSchema],
|
||||
) -> Result<ColumnDataType> {
|
||||
let mut timestamp_columns = schema
|
||||
.iter()
|
||||
.filter(|column| column.semantic_type == SemanticType::Timestamp as i32);
|
||||
let timestamp_column = timestamp_columns
|
||||
.next()
|
||||
.with_context(|| InvalidInsertRequestSnafu {
|
||||
reason: format!("Metric table `{table_name}` must have exactly one timestamp column"),
|
||||
})?;
|
||||
ensure!(
|
||||
timestamp_columns.next().is_none(),
|
||||
InvalidInsertRequestSnafu {
|
||||
reason: format!("Metric table `{table_name}` must have exactly one timestamp column"),
|
||||
}
|
||||
);
|
||||
|
||||
let datatype = ColumnDataType::try_from(timestamp_column.datatype)
|
||||
.ok()
|
||||
.with_context(|| InvalidInsertRequestSnafu {
|
||||
reason: format!(
|
||||
"Metric table `{table_name}` has unsupported timestamp datatype {}",
|
||||
timestamp_column.datatype
|
||||
),
|
||||
})?;
|
||||
ensure!(
|
||||
matches!(
|
||||
datatype,
|
||||
ColumnDataType::TimestampMillisecond | ColumnDataType::TimestampNanosecond
|
||||
),
|
||||
InvalidInsertRequestSnafu {
|
||||
reason: format!(
|
||||
"Metric table `{table_name}` has unsupported timestamp datatype {}",
|
||||
timestamp_column.datatype
|
||||
),
|
||||
}
|
||||
);
|
||||
Ok(datatype)
|
||||
}
|
||||
|
||||
/// Validates metric table timestamp schemas and returns their common datatype.
|
||||
pub fn metric_timestamp_datatype<'a>(
|
||||
tables: impl IntoIterator<Item = (&'a str, &'a [ColumnSchema])>,
|
||||
) -> Result<ColumnDataType> {
|
||||
let mut operation_datatype = None;
|
||||
for (table_name, schema) in tables {
|
||||
let datatype = metric_table_timestamp_datatype(table_name, schema)?;
|
||||
ensure!(
|
||||
operation_datatype.is_none_or(|expected| expected == datatype),
|
||||
InvalidInsertRequestSnafu {
|
||||
reason: "Metric tables in one batch must use the same timestamp datatype"
|
||||
.to_string(),
|
||||
}
|
||||
);
|
||||
operation_datatype = Some(datatype);
|
||||
}
|
||||
|
||||
operation_datatype.with_context(|| InvalidInsertRequestSnafu {
|
||||
reason: "Metric table batch has no schemas".to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Fill table options for a new table by create type.
|
||||
pub fn fill_table_options_for_create(
|
||||
table_options: &mut std::collections::HashMap<String, String>,
|
||||
@@ -1568,6 +1646,65 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::tests::{create_partition_rule_manager, prepare_mocked_backend};
|
||||
|
||||
fn metric_schema(timestamp_datatype: ColumnDataType) -> Vec<api::v1::ColumnSchema> {
|
||||
vec![time_index_column_schema("ts", timestamp_datatype)]
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_metric_timestamp_datatype_derivation() {
|
||||
for datatype in [
|
||||
ColumnDataType::TimestampNanosecond,
|
||||
ColumnDataType::TimestampMillisecond,
|
||||
] {
|
||||
let first = metric_schema(datatype);
|
||||
let second = metric_schema(datatype);
|
||||
assert_eq!(
|
||||
datatype,
|
||||
metric_timestamp_datatype([
|
||||
("metric_0", first.as_slice()),
|
||||
("metric_1", second.as_slice()),
|
||||
])
|
||||
.unwrap()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_metric_timestamp_datatype_rejects_mixed_units() {
|
||||
let milliseconds = metric_schema(ColumnDataType::TimestampMillisecond);
|
||||
let nanoseconds = metric_schema(ColumnDataType::TimestampNanosecond);
|
||||
|
||||
assert!(matches!(
|
||||
metric_timestamp_datatype([
|
||||
("milliseconds", milliseconds.as_slice()),
|
||||
("nanoseconds", nanoseconds.as_slice()),
|
||||
]),
|
||||
Err(crate::error::Error::InvalidInsertRequest { .. })
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_metric_timestamp_datatype_rejects_malformed_schemas() {
|
||||
let missing = vec![];
|
||||
let mut multiple = metric_schema(ColumnDataType::TimestampMillisecond);
|
||||
multiple.push(time_index_column_schema(
|
||||
"another_ts",
|
||||
ColumnDataType::TimestampMillisecond,
|
||||
));
|
||||
let unsupported = metric_schema(ColumnDataType::TimestampMicrosecond);
|
||||
|
||||
for (table_name, schema) in [
|
||||
("missing", missing.as_slice()),
|
||||
("multiple", multiple.as_slice()),
|
||||
("unsupported", unsupported.as_slice()),
|
||||
] {
|
||||
assert!(matches!(
|
||||
metric_timestamp_datatype([(table_name, schema)]),
|
||||
Err(crate::error::Error::InvalidInsertRequest { .. })
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
fn make_table_ref_with_schema(
|
||||
ts_name: &str,
|
||||
field_name: &str,
|
||||
|
||||
+126
-44
@@ -645,7 +645,7 @@ fn encode_exponential_histogram(
|
||||
})?;
|
||||
let mut emitted = false;
|
||||
for (index, data_point) in histogram.data_points.iter().enumerate() {
|
||||
let (value, timestamp_nanos) = match exponential_histogram_value(data_point) {
|
||||
let (value, _) = match exponential_histogram_value(data_point) {
|
||||
Ok(value) => value,
|
||||
Err(reason) => {
|
||||
reject_data_points(outcome, 1, || {
|
||||
@@ -667,7 +667,7 @@ fn encode_exponential_histogram(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
timestamp_nanos,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
row_writer::write_by_schema(
|
||||
@@ -967,29 +967,23 @@ fn write_attributes(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn write_timestamp(
|
||||
table: &mut TableData,
|
||||
row: &mut Vec<Value>,
|
||||
time_nano: i64,
|
||||
legacy_mode: bool,
|
||||
) -> Result<()> {
|
||||
if legacy_mode {
|
||||
row_writer::write_ts_to_nanos(
|
||||
table,
|
||||
greptime_timestamp(),
|
||||
Some(time_nano),
|
||||
Precision::Nanosecond,
|
||||
row,
|
||||
)
|
||||
} else {
|
||||
row_writer::write_ts_to_millis(
|
||||
table,
|
||||
greptime_timestamp(),
|
||||
Some(time_nano / 1000000),
|
||||
Precision::Millisecond,
|
||||
row,
|
||||
)
|
||||
}
|
||||
fn write_timestamp(table: &mut TableData, row: &mut Vec<Value>, time_nano: u64) -> Result<()> {
|
||||
let time_nano = i64::try_from(time_nano).map_err(|_| {
|
||||
error::TimestampOverflowSnafu {
|
||||
error: format!(
|
||||
"timestamp {time_nano} overflow with precision {}",
|
||||
Precision::Nanosecond
|
||||
),
|
||||
}
|
||||
.build()
|
||||
})?;
|
||||
row_writer::write_ts_to_nanos(
|
||||
table,
|
||||
greptime_timestamp(),
|
||||
Some(time_nano),
|
||||
Precision::Nanosecond,
|
||||
row,
|
||||
)
|
||||
}
|
||||
|
||||
fn write_data_point_value(
|
||||
@@ -1037,7 +1031,7 @@ fn write_tags_and_timestamp(
|
||||
resource_attrs: Option<&Vec<KeyValue>>,
|
||||
scope_attrs: Option<&Vec<KeyValue>>,
|
||||
data_point_attrs: Option<&Vec<KeyValue>>,
|
||||
timestamp_nanos: i64,
|
||||
timestamp_nanos: u64,
|
||||
metric_ctx: &OtlpMetricCtx,
|
||||
) -> Result<()> {
|
||||
if metric_ctx.is_legacy {
|
||||
@@ -1075,7 +1069,7 @@ fn write_tags_and_timestamp(
|
||||
)?;
|
||||
}
|
||||
|
||||
write_timestamp(table, row, timestamp_nanos, metric_ctx.is_legacy)?;
|
||||
write_timestamp(table, row, timestamp_nanos)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -1106,7 +1100,7 @@ fn encode_gauge(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
data_point.time_unix_nano as i64,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
|
||||
@@ -1145,7 +1139,7 @@ fn encode_sum(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
data_point.time_unix_nano as i64,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
write_temporality_tag(table, &mut row, is_delta)?;
|
||||
@@ -1257,7 +1251,7 @@ fn encode_histogram(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
data_point.time_unix_nano as i64,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
write_temporality_tag(bucket_table, &mut bucket_row, is_delta)?;
|
||||
@@ -1280,7 +1274,7 @@ fn encode_histogram(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
data_point.time_unix_nano as i64,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
write_temporality_tag(sum_table, &mut sum_row, is_delta)?;
|
||||
@@ -1305,7 +1299,7 @@ fn encode_histogram(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
data_point.time_unix_nano as i64,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
write_temporality_tag(count_table, &mut count_row, is_delta)?;
|
||||
@@ -1402,7 +1396,7 @@ fn encode_summary(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
data_point.time_unix_nano as i64,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
|
||||
@@ -1443,7 +1437,7 @@ fn encode_summary(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
data_point.time_unix_nano as i64,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
row_writer::write_tag(quantile_table, "quantile", quantile.quantile, &mut row)?;
|
||||
@@ -1469,7 +1463,7 @@ fn encode_summary(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
data_point.time_unix_nano as i64,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
|
||||
@@ -1496,7 +1490,7 @@ fn encode_summary(
|
||||
resource_attrs,
|
||||
scope_attrs,
|
||||
Some(data_point.attributes.as_ref()),
|
||||
data_point.time_unix_nano as i64,
|
||||
data_point.time_unix_nano,
|
||||
metric_ctx,
|
||||
)?;
|
||||
|
||||
@@ -1745,6 +1739,88 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
fn boundary_gauge_metric(time_unix_nano: u64) -> Metric {
|
||||
Metric {
|
||||
name: "boundary_gauge".to_string(),
|
||||
data: Some(metric::Data::Gauge(Gauge {
|
||||
data_points: vec![NumberDataPoint {
|
||||
time_unix_nano,
|
||||
value: Some(Value::AsInt(1)),
|
||||
..Default::default()
|
||||
}],
|
||||
})),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_gauge_accepts_and_preserves_i64_max_timestamp() {
|
||||
let ctx = OtlpMetricCtx {
|
||||
metric_type: MetricType::Gauge,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let mut tables = MultiTableData::default();
|
||||
let mut semantic_index = SemanticIndex::default();
|
||||
let mut outcome = MetricsIngestOutcome::default();
|
||||
encode_metrics(
|
||||
&mut tables,
|
||||
&boundary_gauge_metric(i64::MAX as u64),
|
||||
None,
|
||||
None,
|
||||
&ctx,
|
||||
&mut semantic_index,
|
||||
&mut outcome,
|
||||
)
|
||||
.unwrap();
|
||||
let (requests, rows) = tables.into_row_insert_requests();
|
||||
assert_eq!(outcome.accepted_data_points, 1);
|
||||
assert_eq!(rows, 1);
|
||||
let rows = requests.inserts[0].rows.as_ref().unwrap();
|
||||
let timestamp_index = rows
|
||||
.schema
|
||||
.iter()
|
||||
.position(|column| column.column_name == greptime_timestamp())
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
rows.rows[0].values[timestamp_index].value_data,
|
||||
Some(ValueData::TimestampNanosecondValue(i64::MAX))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_gauge_rejects_timestamp_above_i64_max() {
|
||||
let ctx = OtlpMetricCtx {
|
||||
metric_type: MetricType::Gauge,
|
||||
..Default::default()
|
||||
};
|
||||
let mut tables = MultiTableData::default();
|
||||
let mut semantic_index = SemanticIndex::default();
|
||||
let mut outcome = MetricsIngestOutcome::default();
|
||||
let error = encode_metrics(
|
||||
&mut tables,
|
||||
&boundary_gauge_metric(i64::MAX as u64 + 1),
|
||||
None,
|
||||
None,
|
||||
&ctx,
|
||||
&mut semantic_index,
|
||||
&mut outcome,
|
||||
)
|
||||
.unwrap_err();
|
||||
let message = error.to_string();
|
||||
assert_eq!(outcome.accepted_data_points, 0);
|
||||
assert_eq!(
|
||||
tables
|
||||
.get_or_default_table_data("boundary_gauge", 0, 0)
|
||||
.num_rows(),
|
||||
0
|
||||
);
|
||||
assert!(message.len() <= MAX_REJECTION_MESSAGE_BYTES, "{message}");
|
||||
assert!(message.contains("Timestamp overflow"), "{message}");
|
||||
assert!(message.contains("9223372036854775808"), "{message}");
|
||||
assert!(message.contains("Precision::Nanosecond"), "{message}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_encode_sum() {
|
||||
let mut tables = MultiTableData::default();
|
||||
@@ -2679,10 +2755,12 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_exponential_histogram_legacy_and_new_modes_share_struct() {
|
||||
use api::v1::ColumnDataType;
|
||||
use common_query::prelude::greptime_native_histogram;
|
||||
|
||||
let mut point = exponential_point();
|
||||
point.sum = Some(42.0);
|
||||
point.time_unix_nano = 2_000_123;
|
||||
let request = metrics_request(vec![exponential_metric(
|
||||
"request.duration",
|
||||
vec![point],
|
||||
@@ -2724,14 +2802,18 @@ mod tests {
|
||||
.unwrap()]
|
||||
.clone();
|
||||
assert_eq!(new_histogram, legacy_histogram);
|
||||
let new_timestamp_index = new_rows
|
||||
.schema
|
||||
.iter()
|
||||
.position(|column| column.column_name == greptime_timestamp())
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
new_rows.schema[new_timestamp_index].datatype,
|
||||
ColumnDataType::TimestampNanosecond as i32
|
||||
);
|
||||
assert!(matches!(
|
||||
new_rows.rows[0].values[new_rows
|
||||
.schema
|
||||
.iter()
|
||||
.position(|column| column.column_name == greptime_timestamp())
|
||||
.unwrap()]
|
||||
.value_data,
|
||||
Some(ValueData::TimestampMillisecondValue(2))
|
||||
new_rows.rows[0].values[new_timestamp_index].value_data,
|
||||
Some(ValueData::TimestampNanosecondValue(2_000_123))
|
||||
));
|
||||
assert!(matches!(
|
||||
legacy_rows.rows[0].values[legacy_rows
|
||||
@@ -2740,7 +2822,7 @@ mod tests {
|
||||
.position(|column| column.column_name == greptime_timestamp())
|
||||
.unwrap()]
|
||||
.value_data,
|
||||
Some(ValueData::TimestampNanosecondValue(2_000_000))
|
||||
Some(ValueData::TimestampNanosecondValue(2_000_123))
|
||||
));
|
||||
}
|
||||
|
||||
|
||||
@@ -538,8 +538,10 @@ fn is_scalar_metric_schema(schema: &[ColumnSchema]) -> bool {
|
||||
let timestamp_count = schema
|
||||
.iter()
|
||||
.filter(|column| {
|
||||
column.datatype == ColumnDataType::TimestampMillisecond as i32
|
||||
&& column.semantic_type == SemanticType::Timestamp as i32
|
||||
matches!(
|
||||
ColumnDataType::try_from(column.datatype),
|
||||
Ok(ColumnDataType::TimestampMillisecond | ColumnDataType::TimestampNanosecond)
|
||||
) && column.semantic_type == SemanticType::Timestamp as i32
|
||||
})
|
||||
.count();
|
||||
let field_count = schema
|
||||
@@ -553,8 +555,10 @@ fn is_scalar_metric_schema(schema: &[ColumnSchema]) -> bool {
|
||||
timestamp_count == 1
|
||||
&& field_count == 1
|
||||
&& schema.iter().all(|column| {
|
||||
(column.datatype == ColumnDataType::TimestampMillisecond as i32
|
||||
&& column.semantic_type == SemanticType::Timestamp as i32)
|
||||
(matches!(
|
||||
ColumnDataType::try_from(column.datatype),
|
||||
Ok(ColumnDataType::TimestampMillisecond | ColumnDataType::TimestampNanosecond)
|
||||
) && column.semantic_type == SemanticType::Timestamp as i32)
|
||||
|| (column.datatype == ColumnDataType::Float64 as i32
|
||||
&& column.semantic_type == SemanticType::Field as i32)
|
||||
|| (column.datatype == ColumnDataType::String as i32
|
||||
@@ -1448,7 +1452,7 @@ fn columns_taxonomy(
|
||||
essential_column_indices.push(index);
|
||||
}
|
||||
}
|
||||
ArrowDataType::Timestamp(TimeUnit::Millisecond, _) => {
|
||||
ArrowDataType::Timestamp(TimeUnit::Millisecond | TimeUnit::Nanosecond, _) => {
|
||||
ensure!(
|
||||
timestamp_index.replace(index).is_none(),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
@@ -2103,7 +2107,9 @@ mod tests {
|
||||
ColumnDataType, ColumnSchema, Row, RowInsertRequest, RowInsertRequests, Rows, SemanticType,
|
||||
StructValue, Value,
|
||||
};
|
||||
use arrow::array::{BinaryArray, BooleanArray, StringArray, TimestampMillisecondArray};
|
||||
use arrow::array::{
|
||||
BinaryArray, BooleanArray, StringArray, TimestampMillisecondArray, TimestampNanosecondArray,
|
||||
};
|
||||
use arrow::datatypes::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
|
||||
use arrow::record_batch::RecordBatch;
|
||||
use async_trait::async_trait;
|
||||
@@ -2314,13 +2320,13 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_scalar_metric_batchability_rejects_non_millisecond_timestamp() {
|
||||
fn test_scalar_metric_batchability_accepts_nanosecond_timestamp() {
|
||||
let mut requests = scalar_request();
|
||||
let rows = requests.inserts[0].rows.as_mut().unwrap();
|
||||
rows.schema[0].datatype = ColumnDataType::TimestampNanosecond as i32;
|
||||
rows.rows[0].values[0].value_data = Some(ValueData::TimestampNanosecondValue(1_000_000));
|
||||
rows.rows[0].values[0].value_data = Some(ValueData::TimestampNanosecondValue(2_000_123));
|
||||
|
||||
assert!(!is_scalar_metric_batchable(&requests).unwrap());
|
||||
assert!(is_scalar_metric_batchable(&requests).unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -4243,6 +4249,51 @@ mod tests {
|
||||
assert_eq!("greptime_value", modified[0].schema().field(2).name());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_transform_logical_batches_to_physical_preserves_nanosecond_timestamp() {
|
||||
let batch = RecordBatch::try_new(
|
||||
Arc::new(ArrowSchema::new(vec![
|
||||
Field::new(
|
||||
"greptime_timestamp",
|
||||
ArrowDataType::Timestamp(arrow::datatypes::TimeUnit::Nanosecond, None),
|
||||
false,
|
||||
),
|
||||
Field::new("greptime_value", ArrowDataType::Float64, true),
|
||||
Field::new("tag1", ArrowDataType::Utf8, true),
|
||||
])),
|
||||
vec![
|
||||
Arc::new(TimestampNanosecondArray::from(vec![2_000_123])),
|
||||
Arc::new(arrow::array::Float64Array::from(vec![1.0])),
|
||||
Arc::new(StringArray::from(vec!["v1"])),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let table_batches = vec![TableBatch {
|
||||
table_name: "t1".to_string(),
|
||||
table_id: 1,
|
||||
batches: vec![RecordBatchWithTsIdx::try_new(batch, 0).unwrap()],
|
||||
row_count: 1,
|
||||
}];
|
||||
|
||||
let modified = transform_logical_batches_to_physical(
|
||||
&table_batches,
|
||||
&HashMap::from([("tag1".to_string(), 1)]),
|
||||
&HashSet::new(),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
&ArrowDataType::Timestamp(arrow::datatypes::TimeUnit::Nanosecond, None),
|
||||
modified[0].schema().field(1).data_type()
|
||||
);
|
||||
let timestamps = modified[0]
|
||||
.column(1)
|
||||
.as_any()
|
||||
.downcast_ref::<TimestampNanosecondArray>()
|
||||
.unwrap();
|
||||
assert_eq!(2_000_123, timestamps.value(0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_transform_logical_batches_to_physical_taxonomy_failure() {
|
||||
let batch = mock_aligned_tag_batch("tag1", "v1", 1000, 1.0);
|
||||
|
||||
@@ -115,6 +115,18 @@ pub(crate) fn rows_to_aligned_record_batch(
|
||||
target_ts_name
|
||||
),
|
||||
})?;
|
||||
let target_timestamp_unit = match target_schema.field(timestamp_index).data_type() {
|
||||
ArrowDataType::Timestamp(unit, _) => unit,
|
||||
actual => {
|
||||
return error::UnexpectedResultSnafu {
|
||||
reason: format!(
|
||||
"Resolved timestamp column '{}' has non-timestamp type {}",
|
||||
target_ts_name, actual
|
||||
),
|
||||
}
|
||||
.fail();
|
||||
}
|
||||
};
|
||||
|
||||
// Map effective target column name → (source column index, source arrow type).
|
||||
// Handles prom renames: Timestamp → target ts name, Float64 → target field name.
|
||||
@@ -131,11 +143,20 @@ pub(crate) fn rows_to_aligned_record_batch(
|
||||
}
|
||||
ArrowDataType::Timestamp(unit, _) => {
|
||||
ensure!(
|
||||
unit == &TimeUnit::Millisecond,
|
||||
matches!(unit, TimeUnit::Millisecond | TimeUnit::Nanosecond),
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: format!(
|
||||
"Unexpected remote write batch timestamp unit, expect millisecond, got: {}",
|
||||
unit
|
||||
"Unsupported metric row timestamp unit, expected Millisecond or Nanosecond, actual {:?}",
|
||||
unit,
|
||||
)
|
||||
}
|
||||
);
|
||||
ensure!(
|
||||
unit == target_timestamp_unit,
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: format!(
|
||||
"Metric row timestamp unit mismatch, expected {:?}, actual {:?}",
|
||||
target_timestamp_unit, unit,
|
||||
)
|
||||
}
|
||||
);
|
||||
@@ -207,7 +228,13 @@ pub fn build_metric_create_table_schema_from_proto(
|
||||
rows_schema
|
||||
.iter()
|
||||
.map(|col| {
|
||||
let semantic_type = if col.datatype == api::v1::ColumnDataType::TimestampMillisecond as i32 {
|
||||
let semantic_type = if matches!(
|
||||
api::v1::ColumnDataType::try_from(col.datatype),
|
||||
Ok(
|
||||
api::v1::ColumnDataType::TimestampMillisecond
|
||||
| api::v1::ColumnDataType::TimestampNanosecond
|
||||
)
|
||||
) {
|
||||
SemanticType::Timestamp
|
||||
} else if col.datatype == api::v1::ColumnDataType::Float64 as i32 {
|
||||
SemanticType::Field
|
||||
@@ -305,7 +332,9 @@ fn build_arrow_array(
|
||||
mod tests {
|
||||
use api::v1::value::ValueData;
|
||||
use api::v1::{ColumnDataType, ColumnSchema, Row, Rows, SemanticType, Value};
|
||||
use arrow::array::{Array, Float64Array, StringArray, TimestampMillisecondArray};
|
||||
use arrow::array::{
|
||||
Array, Float64Array, StringArray, TimestampMillisecondArray, TimestampNanosecondArray,
|
||||
};
|
||||
use arrow::datatypes::{DataType, Field, Schema as ArrowSchema, TimeUnit};
|
||||
|
||||
use super::{
|
||||
@@ -409,6 +438,100 @@ mod tests {
|
||||
assert_eq!(values.value(1), 99.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rows_to_aligned_record_batch_nanosecond_preserves_value() {
|
||||
let rows = Rows {
|
||||
schema: vec![
|
||||
ColumnSchema {
|
||||
column_name: "greptime_timestamp".to_string(),
|
||||
datatype: ColumnDataType::TimestampNanosecond as i32,
|
||||
semantic_type: SemanticType::Timestamp as i32,
|
||||
..Default::default()
|
||||
},
|
||||
ColumnSchema {
|
||||
column_name: "greptime_value".to_string(),
|
||||
datatype: ColumnDataType::Float64 as i32,
|
||||
semantic_type: SemanticType::Field as i32,
|
||||
..Default::default()
|
||||
},
|
||||
],
|
||||
rows: vec![Row {
|
||||
values: vec![
|
||||
Value {
|
||||
value_data: Some(ValueData::TimestampNanosecondValue(2_000_123)),
|
||||
},
|
||||
Value {
|
||||
value_data: Some(ValueData::F64Value(1.0)),
|
||||
},
|
||||
],
|
||||
}],
|
||||
};
|
||||
let target = ArrowSchema::new(vec![
|
||||
Field::new(
|
||||
"my_ts",
|
||||
DataType::Timestamp(TimeUnit::Nanosecond, None),
|
||||
false,
|
||||
),
|
||||
Field::new("my_value", DataType::Float64, true),
|
||||
]);
|
||||
|
||||
let aligned_batch = rows_to_aligned_record_batch(&rows, &target).unwrap();
|
||||
let (batch, timestamp_index) = aligned_batch.into_parts();
|
||||
let timestamps = batch
|
||||
.column(timestamp_index)
|
||||
.as_any()
|
||||
.downcast_ref::<TimestampNanosecondArray>()
|
||||
.unwrap();
|
||||
assert_eq!(2_000_123, timestamps.value(0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rows_to_aligned_record_batch_nanosecond_rejects_millisecond_target() {
|
||||
let rows = Rows {
|
||||
schema: vec![
|
||||
ColumnSchema {
|
||||
column_name: "greptime_timestamp".to_string(),
|
||||
datatype: ColumnDataType::TimestampNanosecond as i32,
|
||||
semantic_type: SemanticType::Timestamp as i32,
|
||||
..Default::default()
|
||||
},
|
||||
ColumnSchema {
|
||||
column_name: "greptime_value".to_string(),
|
||||
datatype: ColumnDataType::Float64 as i32,
|
||||
semantic_type: SemanticType::Field as i32,
|
||||
..Default::default()
|
||||
},
|
||||
],
|
||||
rows: vec![Row {
|
||||
values: vec![
|
||||
Value {
|
||||
value_data: Some(ValueData::TimestampNanosecondValue(2_000_123)),
|
||||
},
|
||||
Value {
|
||||
value_data: Some(ValueData::F64Value(1.0)),
|
||||
},
|
||||
],
|
||||
}],
|
||||
};
|
||||
let target = ArrowSchema::new(vec![
|
||||
Field::new(
|
||||
"my_ts",
|
||||
DataType::Timestamp(TimeUnit::Millisecond, None),
|
||||
false,
|
||||
),
|
||||
Field::new("my_value", DataType::Float64, true),
|
||||
]);
|
||||
|
||||
let error = rows_to_aligned_record_batch(&rows, &target).unwrap_err();
|
||||
let message = error.to_string();
|
||||
assert!(
|
||||
matches!(error, crate::error::Error::InvalidPromRemoteRequest { .. }),
|
||||
"{message}"
|
||||
);
|
||||
assert!(message.contains("expected Millisecond"), "{message}");
|
||||
assert!(message.contains("actual Nanosecond"), "{message}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rows_to_aligned_record_batch_fills_nulls() {
|
||||
let rows = Rows {
|
||||
@@ -568,4 +691,29 @@ mod tests {
|
||||
assert_eq!(SemanticType::Field as i32, schema[2].semantic_type);
|
||||
assert_eq!(ColumnDataType::Float64 as i32, schema[2].datatype);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_build_metric_create_table_schema_preserves_nanosecond_timestamp_semantics() {
|
||||
let rows_schema = vec![
|
||||
ColumnSchema {
|
||||
column_name: "greptime_timestamp".to_string(),
|
||||
datatype: ColumnDataType::TimestampNanosecond as i32,
|
||||
semantic_type: SemanticType::Timestamp as i32,
|
||||
..Default::default()
|
||||
},
|
||||
ColumnSchema {
|
||||
column_name: "greptime_value".to_string(),
|
||||
datatype: ColumnDataType::Float64 as i32,
|
||||
semantic_type: SemanticType::Field as i32,
|
||||
..Default::default()
|
||||
},
|
||||
];
|
||||
|
||||
let schema = build_metric_create_table_schema_from_proto(&rows_schema).unwrap();
|
||||
assert_eq!(
|
||||
ColumnDataType::TimestampNanosecond as i32,
|
||||
schema[0].datatype
|
||||
);
|
||||
assert_eq!(SemanticType::Timestamp as i32, schema[0].semantic_type);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6596,7 +6596,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
|
||||
.await;
|
||||
|
||||
// CREATE TABLE IF NOT EXISTS "claude_code_cost_usage_USD_total" (
|
||||
// "greptime_timestamp" TIMESTAMP(3) NOT NULL,
|
||||
// "greptime_timestamp" TIMESTAMP(9) NOT NULL,
|
||||
// "greptime_value" DOUBLE NULL,
|
||||
// "host_arch" STRING NULL,
|
||||
// "job" STRING NULL,
|
||||
@@ -6620,7 +6620,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
|
||||
// on_physical_table = 'greptime_physical_table',
|
||||
// otlp_metric_compat = 'prom'
|
||||
// )
|
||||
let expected = "[[\"claude_code_cost_usage_USD_total\",\"CREATE TABLE IF NOT EXISTS \\\"claude_code_cost_usage_USD_total\\\" (\\n \\\"greptime_timestamp\\\" TIMESTAMP(3) NOT NULL,\\n \\\"greptime_value\\\" DOUBLE NULL,\\n \\\"host_arch\\\" STRING NULL,\\n \\\"job\\\" STRING NULL,\\n \\\"model\\\" STRING NULL,\\n \\\"os_version\\\" STRING NULL,\\n \\\"otel_scope_name\\\" STRING NULL,\\n \\\"otel_scope_schema_url\\\" STRING NULL,\\n \\\"otel_scope_version\\\" STRING NULL,\\n \\\"otlp_aggregation_temporality\\\" STRING NULL,\\n \\\"service_name\\\" STRING NULL,\\n \\\"service_version\\\" STRING NULL,\\n \\\"session_id\\\" STRING NULL,\\n \\\"terminal_type\\\" STRING NULL,\\n \\\"user_id\\\" STRING NULL,\\n TIME INDEX (\\\"greptime_timestamp\\\"),\\n PRIMARY KEY (\\\"host_arch\\\", \\\"job\\\", \\\"model\\\", \\\"os_version\\\", \\\"otel_scope_name\\\", \\\"otel_scope_schema_url\\\", \\\"otel_scope_version\\\", \\\"otlp_aggregation_temporality\\\", \\\"service_name\\\", \\\"service_version\\\", \\\"session_id\\\", \\\"terminal_type\\\", \\\"user_id\\\")\\n)\\n\\nENGINE=metric\\nWITH(\\n 'comment' = 'Created on insertion',\\n 'greptime.semantic.metric.metadata_quality' = 'declared',\\n 'greptime.semantic.metric.original_name' = 'claude_code.cost.usage',\\n 'greptime.semantic.metric.temporality' = 'delta',\\n 'greptime.semantic.metric.type' = 'counter',\\n 'greptime.semantic.metric.unit' = 'USD',\\n 'greptime.semantic.signal_type' = 'metric',\\n 'greptime.semantic.source' = 'opentelemetry',\\n on_physical_table = 'greptime_physical_table',\\n otlp_metric_compat = 'prom'\\n)\"]]";
|
||||
let expected = "[[\"claude_code_cost_usage_USD_total\",\"CREATE TABLE IF NOT EXISTS \\\"claude_code_cost_usage_USD_total\\\" (\\n \\\"greptime_timestamp\\\" TIMESTAMP(9) NOT NULL,\\n \\\"greptime_value\\\" DOUBLE NULL,\\n \\\"host_arch\\\" STRING NULL,\\n \\\"job\\\" STRING NULL,\\n \\\"model\\\" STRING NULL,\\n \\\"os_version\\\" STRING NULL,\\n \\\"otel_scope_name\\\" STRING NULL,\\n \\\"otel_scope_schema_url\\\" STRING NULL,\\n \\\"otel_scope_version\\\" STRING NULL,\\n \\\"otlp_aggregation_temporality\\\" STRING NULL,\\n \\\"service_name\\\" STRING NULL,\\n \\\"service_version\\\" STRING NULL,\\n \\\"session_id\\\" STRING NULL,\\n \\\"terminal_type\\\" STRING NULL,\\n \\\"user_id\\\" STRING NULL,\\n TIME INDEX (\\\"greptime_timestamp\\\"),\\n PRIMARY KEY (\\\"host_arch\\\", \\\"job\\\", \\\"model\\\", \\\"os_version\\\", \\\"otel_scope_name\\\", \\\"otel_scope_schema_url\\\", \\\"otel_scope_version\\\", \\\"otlp_aggregation_temporality\\\", \\\"service_name\\\", \\\"service_version\\\", \\\"session_id\\\", \\\"terminal_type\\\", \\\"user_id\\\")\\n)\\n\\nENGINE=metric\\nWITH(\\n 'comment' = 'Created on insertion',\\n 'greptime.semantic.metric.metadata_quality' = 'declared',\\n 'greptime.semantic.metric.original_name' = 'claude_code.cost.usage',\\n 'greptime.semantic.metric.temporality' = 'delta',\\n 'greptime.semantic.metric.type' = 'counter',\\n 'greptime.semantic.metric.unit' = 'USD',\\n 'greptime.semantic.signal_type' = 'metric',\\n 'greptime.semantic.source' = 'opentelemetry',\\n on_physical_table = 'greptime_physical_table',\\n otlp_metric_compat = 'prom'\\n)\"]]";
|
||||
validate_data(
|
||||
"otlp_metrics_all_show_create_table",
|
||||
&client,
|
||||
@@ -6630,7 +6630,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
|
||||
.await;
|
||||
|
||||
// select metrics data
|
||||
let expected = "[[1753780559836,2.244618,\"arm64\",\"claude-code\",\"claude-sonnet-4-20250514\",\"25.0.0\",\"com.anthropic.claude_code\",\"\",\"1.0.62\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"],[1753780559836,0.0052544,\"arm64\",\"claude-code\",\"claude-3-5-haiku-20241022\",\"25.0.0\",\"com.anthropic.claude_code\",\"\",\"1.0.62\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"]]";
|
||||
let expected = "[[1753780559836000000,2.244618,\"arm64\",\"claude-code\",\"claude-sonnet-4-20250514\",\"25.0.0\",\"com.anthropic.claude_code\",\"\",\"1.0.62\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"],[1753780559836000000,0.0052544,\"arm64\",\"claude-code\",\"claude-3-5-haiku-20241022\",\"25.0.0\",\"com.anthropic.claude_code\",\"\",\"1.0.62\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"]]";
|
||||
validate_data(
|
||||
"otlp_metrics_all_select",
|
||||
&client,
|
||||
@@ -6705,7 +6705,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
|
||||
assert_eq!(StatusCode::OK, res.status());
|
||||
|
||||
// CREATE TABLE IF NOT EXISTS "claude_code_cost_usage_USD_total" (
|
||||
// "greptime_timestamp" TIMESTAMP(3) NOT NULL,
|
||||
// "greptime_timestamp" TIMESTAMP(9) NOT NULL,
|
||||
// "greptime_value" DOUBLE NULL,
|
||||
// "job" STRING NULL,
|
||||
// "model" STRING NULL,
|
||||
@@ -6726,7 +6726,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
|
||||
// on_physical_table = 'greptime_physical_table',
|
||||
// otlp_metric_compat = 'prom'
|
||||
// )
|
||||
let expected = "[[\"claude_code_cost_usage_USD_total\",\"CREATE TABLE IF NOT EXISTS \\\"claude_code_cost_usage_USD_total\\\" (\\n \\\"greptime_timestamp\\\" TIMESTAMP(3) NOT NULL,\\n \\\"greptime_value\\\" DOUBLE NULL,\\n \\\"job\\\" STRING NULL,\\n \\\"model\\\" STRING NULL,\\n \\\"os_type\\\" STRING NULL,\\n \\\"os_version\\\" STRING NULL,\\n \\\"otlp_aggregation_temporality\\\" STRING NULL,\\n \\\"service_name\\\" STRING NULL,\\n \\\"service_version\\\" STRING NULL,\\n \\\"session_id\\\" STRING NULL,\\n \\\"terminal_type\\\" STRING NULL,\\n \\\"user_id\\\" STRING NULL,\\n TIME INDEX (\\\"greptime_timestamp\\\"),\\n PRIMARY KEY (\\\"job\\\", \\\"model\\\", \\\"os_type\\\", \\\"os_version\\\", \\\"otlp_aggregation_temporality\\\", \\\"service_name\\\", \\\"service_version\\\", \\\"session_id\\\", \\\"terminal_type\\\", \\\"user_id\\\")\\n)\\n\\nENGINE=metric\\nWITH(\\n 'comment' = 'Created on insertion',\\n 'greptime.semantic.metric.metadata_quality' = 'declared',\\n 'greptime.semantic.metric.original_name' = 'claude_code.cost.usage',\\n 'greptime.semantic.metric.temporality' = 'delta',\\n 'greptime.semantic.metric.type' = 'counter',\\n 'greptime.semantic.metric.unit' = 'USD',\\n 'greptime.semantic.signal_type' = 'metric',\\n 'greptime.semantic.source' = 'opentelemetry',\\n on_physical_table = 'greptime_physical_table',\\n otlp_metric_compat = 'prom'\\n)\"]]";
|
||||
let expected = "[[\"claude_code_cost_usage_USD_total\",\"CREATE TABLE IF NOT EXISTS \\\"claude_code_cost_usage_USD_total\\\" (\\n \\\"greptime_timestamp\\\" TIMESTAMP(9) NOT NULL,\\n \\\"greptime_value\\\" DOUBLE NULL,\\n \\\"job\\\" STRING NULL,\\n \\\"model\\\" STRING NULL,\\n \\\"os_type\\\" STRING NULL,\\n \\\"os_version\\\" STRING NULL,\\n \\\"otlp_aggregation_temporality\\\" STRING NULL,\\n \\\"service_name\\\" STRING NULL,\\n \\\"service_version\\\" STRING NULL,\\n \\\"session_id\\\" STRING NULL,\\n \\\"terminal_type\\\" STRING NULL,\\n \\\"user_id\\\" STRING NULL,\\n TIME INDEX (\\\"greptime_timestamp\\\"),\\n PRIMARY KEY (\\\"job\\\", \\\"model\\\", \\\"os_type\\\", \\\"os_version\\\", \\\"otlp_aggregation_temporality\\\", \\\"service_name\\\", \\\"service_version\\\", \\\"session_id\\\", \\\"terminal_type\\\", \\\"user_id\\\")\\n)\\n\\nENGINE=metric\\nWITH(\\n 'comment' = 'Created on insertion',\\n 'greptime.semantic.metric.metadata_quality' = 'declared',\\n 'greptime.semantic.metric.original_name' = 'claude_code.cost.usage',\\n 'greptime.semantic.metric.temporality' = 'delta',\\n 'greptime.semantic.metric.type' = 'counter',\\n 'greptime.semantic.metric.unit' = 'USD',\\n 'greptime.semantic.signal_type' = 'metric',\\n 'greptime.semantic.source' = 'opentelemetry',\\n on_physical_table = 'greptime_physical_table',\\n otlp_metric_compat = 'prom'\\n)\"]]";
|
||||
validate_data(
|
||||
"otlp_metrics_show_create_table",
|
||||
&client,
|
||||
@@ -6736,7 +6736,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
|
||||
.await;
|
||||
|
||||
// select metrics data
|
||||
let expected = "[[1753780559836,2.244618,\"claude-code\",\"claude-sonnet-4-20250514\",\"darwin\",\"25.0.0\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"],[1753780559836,0.0052544,\"claude-code\",\"claude-3-5-haiku-20241022\",\"darwin\",\"25.0.0\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"]]";
|
||||
let expected = "[[1753780559836000000,2.244618,\"claude-code\",\"claude-sonnet-4-20250514\",\"darwin\",\"25.0.0\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"],[1753780559836000000,0.0052544,\"claude-code\",\"claude-3-5-haiku-20241022\",\"darwin\",\"25.0.0\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"]]";
|
||||
validate_data(
|
||||
"otlp_metrics_select",
|
||||
&client,
|
||||
@@ -6772,7 +6772,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
|
||||
assert_eq!(StatusCode::OK, res.status());
|
||||
|
||||
// CREATE TABLE IF NOT EXISTS "claude_code_cost_usage_USD_total" (
|
||||
// "greptime_timestamp" TIMESTAMP(3) NOT NULL,
|
||||
// "greptime_timestamp" TIMESTAMP(9) NOT NULL,
|
||||
// "greptime_value" DOUBLE NULL,
|
||||
// "job" STRING NULL,
|
||||
// "model" STRING NULL,
|
||||
@@ -6791,7 +6791,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
|
||||
// on_physical_table = 'greptime_physical_table',
|
||||
// otlp_metric_compat = 'prom'
|
||||
// )
|
||||
let expected = "[[\"claude_code_cost_usage_USD_total\",\"CREATE TABLE IF NOT EXISTS \\\"claude_code_cost_usage_USD_total\\\" (\\n \\\"greptime_timestamp\\\" TIMESTAMP(3) NOT NULL,\\n \\\"greptime_value\\\" DOUBLE NULL,\\n \\\"job\\\" STRING NULL,\\n \\\"model\\\" STRING NULL,\\n \\\"otlp_aggregation_temporality\\\" STRING NULL,\\n \\\"service_name\\\" STRING NULL,\\n \\\"service_version\\\" STRING NULL,\\n \\\"session_id\\\" STRING NULL,\\n \\\"terminal_type\\\" STRING NULL,\\n \\\"user_id\\\" STRING NULL,\\n TIME INDEX (\\\"greptime_timestamp\\\"),\\n PRIMARY KEY (\\\"job\\\", \\\"model\\\", \\\"otlp_aggregation_temporality\\\", \\\"service_name\\\", \\\"service_version\\\", \\\"session_id\\\", \\\"terminal_type\\\", \\\"user_id\\\")\\n)\\n\\nENGINE=metric\\nWITH(\\n 'comment' = 'Created on insertion',\\n 'greptime.semantic.metric.metadata_quality' = 'declared',\\n 'greptime.semantic.metric.original_name' = 'claude_code.cost.usage',\\n 'greptime.semantic.metric.temporality' = 'delta',\\n 'greptime.semantic.metric.type' = 'counter',\\n 'greptime.semantic.metric.unit' = 'USD',\\n 'greptime.semantic.signal_type' = 'metric',\\n 'greptime.semantic.source' = 'opentelemetry',\\n on_physical_table = 'greptime_physical_table',\\n otlp_metric_compat = 'prom'\\n)\"]]";
|
||||
let expected = "[[\"claude_code_cost_usage_USD_total\",\"CREATE TABLE IF NOT EXISTS \\\"claude_code_cost_usage_USD_total\\\" (\\n \\\"greptime_timestamp\\\" TIMESTAMP(9) NOT NULL,\\n \\\"greptime_value\\\" DOUBLE NULL,\\n \\\"job\\\" STRING NULL,\\n \\\"model\\\" STRING NULL,\\n \\\"otlp_aggregation_temporality\\\" STRING NULL,\\n \\\"service_name\\\" STRING NULL,\\n \\\"service_version\\\" STRING NULL,\\n \\\"session_id\\\" STRING NULL,\\n \\\"terminal_type\\\" STRING NULL,\\n \\\"user_id\\\" STRING NULL,\\n TIME INDEX (\\\"greptime_timestamp\\\"),\\n PRIMARY KEY (\\\"job\\\", \\\"model\\\", \\\"otlp_aggregation_temporality\\\", \\\"service_name\\\", \\\"service_version\\\", \\\"session_id\\\", \\\"terminal_type\\\", \\\"user_id\\\")\\n)\\n\\nENGINE=metric\\nWITH(\\n 'comment' = 'Created on insertion',\\n 'greptime.semantic.metric.metadata_quality' = 'declared',\\n 'greptime.semantic.metric.original_name' = 'claude_code.cost.usage',\\n 'greptime.semantic.metric.temporality' = 'delta',\\n 'greptime.semantic.metric.type' = 'counter',\\n 'greptime.semantic.metric.unit' = 'USD',\\n 'greptime.semantic.signal_type' = 'metric',\\n 'greptime.semantic.source' = 'opentelemetry',\\n on_physical_table = 'greptime_physical_table',\\n otlp_metric_compat = 'prom'\\n)\"]]";
|
||||
validate_data(
|
||||
"otlp_metrics_show_create_table_none",
|
||||
&client,
|
||||
@@ -6801,7 +6801,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
|
||||
.await;
|
||||
|
||||
// select metrics data
|
||||
let expected = "[[1753780559836,2.244618,\"claude-code\",\"claude-sonnet-4-20250514\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"],[1753780559836,0.0052544,\"claude-code\",\"claude-3-5-haiku-20241022\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"]]";
|
||||
let expected = "[[1753780559836000000,2.244618,\"claude-code\",\"claude-sonnet-4-20250514\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"],[1753780559836000000,0.0052544,\"claude-code\",\"claude-3-5-haiku-20241022\",\"delta\",\"claude-code\",\"1.0.62\",\"736525A3-F5D4-496B-933E-827AF23A5B97\",\"ghostty\",\"6DA02FD9-B5C5-4E61-9355-9FE8EC9A0CF4\"]]";
|
||||
validate_data(
|
||||
"otlp_metrics_select_none",
|
||||
&client,
|
||||
@@ -7005,7 +7005,7 @@ async fn test_otlp_metrics_batching_flushes_multiple_small_requests() {
|
||||
let responses = tokio::join!(
|
||||
send_otlp_metrics_json(
|
||||
&client,
|
||||
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"task6_gauge","unit":"1","gauge":{"dataPoints":[{"timeUnixNano":"1000000","asDouble":1.25}]}}]}]}]}"#,
|
||||
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"task6_gauge","unit":"1","gauge":{"dataPoints":[{"timeUnixNano":"1000000123","asDouble":1.25}]}}]}]}]}"#,
|
||||
),
|
||||
send_otlp_metrics_json(
|
||||
&client,
|
||||
@@ -7030,14 +7030,21 @@ async fn test_otlp_metrics_batching_flushes_multiple_small_requests() {
|
||||
"otlp_batched_gauge_rows",
|
||||
&client,
|
||||
"select greptime_timestamp, greptime_value from task6_gauge order by greptime_timestamp;",
|
||||
"[[1,1.25],[3,3.75]]",
|
||||
"[[3000000,3.75],[1000000123,1.25]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
"otlp_batched_sum_rows",
|
||||
&client,
|
||||
"select greptime_timestamp, greptime_value from task6_sum order by greptime_timestamp;",
|
||||
"[[2,2.5],[4,4.5]]",
|
||||
"[[2000000,2.5],[4000000,4.5]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
"otlp_batched_nanosecond_timestamp_schema",
|
||||
&client,
|
||||
"SELECT table_name, lower(data_type) FROM information_schema.columns WHERE table_schema = 'public' AND table_name IN ('greptime_physical_table', 'task6_gauge') AND column_name = 'greptime_timestamp' ORDER BY table_name;",
|
||||
r#"[["greptime_physical_table","timestamp(9)"],["task6_gauge","timestamp(9)"]]"#,
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -7062,8 +7069,8 @@ async fn test_otlp_metrics_batching_respects_existing_logical_table_placement()
|
||||
assert_eq!(StatusCode::OK, response.status());
|
||||
|
||||
for sql in [
|
||||
r#"CREATE TABLE otlp_custom_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("physical_metric_table" = "")"#,
|
||||
r#"CREATE TABLE placement_metric (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("on_physical_table" = "otlp_custom_physical", "otlp_metric_compat" = "prom")"#,
|
||||
r#"CREATE TABLE otlp_custom_physical (greptime_timestamp TIMESTAMP(9) TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("physical_metric_table" = "")"#,
|
||||
r#"CREATE TABLE placement_metric (greptime_timestamp TIMESTAMP(9) TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("on_physical_table" = "otlp_custom_physical", "otlp_metric_compat" = "prom")"#,
|
||||
] {
|
||||
let response = execute_sql(&client, sql).await;
|
||||
assert_eq!(
|
||||
@@ -7076,7 +7083,7 @@ async fn test_otlp_metrics_batching_respects_existing_logical_table_placement()
|
||||
|
||||
let response = send_otlp_metrics_json(
|
||||
&client,
|
||||
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"placement_metric","gauge":{"dataPoints":[{"timeUnixNano":"2000000","asDouble":2.0}]}}]}]}]}"#,
|
||||
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"placement_metric","gauge":{"dataPoints":[{"timeUnixNano":"2000000456","asDouble":2.0}]}}]}]}]}"#,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(StatusCode::OK, response.status());
|
||||
@@ -7084,8 +7091,8 @@ async fn test_otlp_metrics_batching_respects_existing_logical_table_placement()
|
||||
validate_data(
|
||||
"otlp_batched_custom_physical_table",
|
||||
&client,
|
||||
"SELECT greptime_value FROM placement_metric",
|
||||
"[[2.0]]",
|
||||
"SELECT greptime_timestamp, greptime_value FROM placement_metric",
|
||||
"[[2000000456,2.0]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
@@ -7095,6 +7102,72 @@ async fn test_otlp_metrics_batching_respects_existing_logical_table_placement()
|
||||
"[[1.0]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
"otlp_fallback_nanosecond_timestamp_schema",
|
||||
&client,
|
||||
"SELECT table_name, lower(data_type) FROM information_schema.columns WHERE table_schema = 'public' AND table_name IN ('otlp_custom_physical', 'placement_metric') AND column_name = 'greptime_timestamp' ORDER BY table_name;",
|
||||
r#"[["otlp_custom_physical","timestamp(9)"],["placement_metric","timestamp(9)"]]"#,
|
||||
)
|
||||
.await;
|
||||
|
||||
guard.remove_all().await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn test_otlp_metrics_batching_rejects_millisecond_metric_tables() {
|
||||
common_telemetry::init_default_ut_logging();
|
||||
let (app, mut guard) = setup_test_otlp_metrics_app_with_frontend_batched(
|
||||
StorageType::File,
|
||||
"test_otlp_metrics_batching_rejects_millisecond_metric_tables",
|
||||
)
|
||||
.await;
|
||||
let client = TestClient::new(app).await;
|
||||
|
||||
for sql in [
|
||||
r#"CREATE TABLE greptime_physical_table (greptime_timestamp TIMESTAMP(3) TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("physical_metric_table" = "")"#,
|
||||
r#"CREATE TABLE millisecond_metric (greptime_timestamp TIMESTAMP(3) TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("on_physical_table" = "greptime_physical_table", "otlp_metric_compat" = "prom")"#,
|
||||
] {
|
||||
let response = execute_sql(&client, sql).await;
|
||||
assert_eq!(
|
||||
StatusCode::OK,
|
||||
response.status(),
|
||||
"{}",
|
||||
response.text().await
|
||||
);
|
||||
}
|
||||
|
||||
let response = send_otlp_metrics_json(
|
||||
&client,
|
||||
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"millisecond_metric","gauge":{"dataPoints":[{"timeUnixNano":"1000000123","asDouble":1.0}]}}]}]}]}"#,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(StatusCode::BAD_REQUEST, response.status());
|
||||
let body: Value = response.json().await;
|
||||
let error = body["error"].as_str().unwrap();
|
||||
for expected in [
|
||||
"timestamp unit mismatch",
|
||||
"expected Millisecond",
|
||||
"actual Nanosecond",
|
||||
] {
|
||||
assert!(
|
||||
error.contains(expected),
|
||||
"expected {expected:?} in error body: {body:?}"
|
||||
);
|
||||
}
|
||||
validate_data(
|
||||
"otlp_millisecond_metric_table_remains_empty",
|
||||
&client,
|
||||
"SELECT count(*) FROM millisecond_metric;",
|
||||
"[[0]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
"otlp_millisecond_physical_table_remains_empty",
|
||||
&client,
|
||||
"SELECT count(*) FROM greptime_physical_table;",
|
||||
"[[0]]",
|
||||
)
|
||||
.await;
|
||||
|
||||
guard.remove_all().await;
|
||||
}
|
||||
@@ -7125,8 +7198,8 @@ async fn test_otlp_metrics_batching_preserves_disabled_auto_create_policy() {
|
||||
.await;
|
||||
|
||||
for sql in [
|
||||
r#"CREATE TABLE greptime_physical_table (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("physical_metric_table" = "")"#,
|
||||
r#"CREATE TABLE existing_metric (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("on_physical_table" = "greptime_physical_table", "otlp_metric_compat" = "prom")"#,
|
||||
r#"CREATE TABLE greptime_physical_table (greptime_timestamp TIMESTAMP(9) TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("physical_metric_table" = "")"#,
|
||||
r#"CREATE TABLE existing_metric (greptime_timestamp TIMESTAMP(9) TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("on_physical_table" = "greptime_physical_table", "otlp_metric_compat" = "prom")"#,
|
||||
] {
|
||||
let response = execute_sql(&client, sql).await;
|
||||
assert_eq!(
|
||||
@@ -7193,42 +7266,42 @@ async fn test_otlp_metrics_batching_preserves_fanout_semantic_options() {
|
||||
"otlp_batched_histogram_rows",
|
||||
&client,
|
||||
"select greptime_timestamp, route, le, greptime_value from task6_histogram_bucket order by le;",
|
||||
"[[1,\"read\",\"1\",1.0],[1,\"read\",\"inf\",3.0]]",
|
||||
"[[1000000,\"read\",\"1\",1.0],[1000000,\"read\",\"inf\",3.0]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
"otlp_batched_histogram_count",
|
||||
&client,
|
||||
"select greptime_timestamp, route, greptime_value from task6_histogram_count;",
|
||||
"[[1,\"read\",3.0]]",
|
||||
"[[1000000,\"read\",3.0]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
"otlp_batched_histogram_sum",
|
||||
&client,
|
||||
"select greptime_timestamp, route, greptime_value from task6_histogram_sum;",
|
||||
"[[1,\"read\",6.0]]",
|
||||
"[[1000000,\"read\",6.0]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
"otlp_batched_summary_rows",
|
||||
&client,
|
||||
"select greptime_timestamp, route, quantile, greptime_value from task6_summary order by quantile;",
|
||||
"[[2,\"write\",\"0.5\",2.0],[2,\"write\",\"0.9\",4.0]]",
|
||||
"[[2000000,\"write\",\"0.5\",2.0],[2000000,\"write\",\"0.9\",4.0]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
"otlp_batched_summary_count",
|
||||
&client,
|
||||
"select greptime_timestamp, route, greptime_value from task6_summary_count;",
|
||||
"[[2,\"write\",4.0]]",
|
||||
"[[2000000,\"write\",4.0]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
"otlp_batched_summary_sum",
|
||||
&client,
|
||||
"select greptime_timestamp, route, greptime_value from task6_summary_sum;",
|
||||
"[[2,\"write\",10.0]]",
|
||||
"[[2000000,\"write\",10.0]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
@@ -7386,7 +7459,7 @@ pub async fn test_otlp_exponential_histogram(store_type: StorageType) {
|
||||
"otlp_exponential_histogram_row",
|
||||
&client,
|
||||
"select greptime_timestamp, greptime_native_histogram from otlp_exponential_latency;",
|
||||
"[[3000,{\"count_f64\":null,\"count_i64\":4,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[],\"negative_span_lengths\":[],\"negative_span_offsets\":[],\"positive_buckets_f64\":[],\"positive_buckets_i64\":[1,2],\"positive_span_lengths\":[2],\"positive_span_offsets\":[0],\"reset_hint\":0,\"schema\":0,\"start_timestamp\":1000,\"sum\":8.0,\"zero_count_f64\":null,\"zero_count_i64\":1,\"zero_threshold\":0.0}]]",
|
||||
"[[3000000000,{\"count_f64\":null,\"count_i64\":4,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[],\"negative_span_lengths\":[],\"negative_span_offsets\":[],\"positive_buckets_f64\":[],\"positive_buckets_i64\":[1,2],\"positive_span_lengths\":[2],\"positive_span_offsets\":[0],\"reset_hint\":0,\"schema\":0,\"start_timestamp\":1000,\"sum\":8.0,\"zero_count_f64\":null,\"zero_count_i64\":1,\"zero_threshold\":0.0}]]",
|
||||
)
|
||||
.await;
|
||||
validate_data(
|
||||
|
||||
Reference in New Issue
Block a user