From 9cb5f992a97a64ea56cbe3ebe2c2df50b26514a0 Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" Date: Tue, 8 Sep 2026 14:40:16 +0800 Subject: [PATCH] fix: preserve OTLP metric nanosecond timestamps Signed-off-by: Lei, HUANG --- src/frontend/src/instance/otlp.rs | 42 +++++- src/frontend/src/instance/prom_store.rs | 91 ++++++++++--- src/operator/src/insert.rs | 143 +++++++++++++++++++- src/servers/src/otlp/metrics.rs | 170 ++++++++++++++++++------ src/servers/src/pending_rows_batcher.rs | 69 ++++++++-- src/servers/src/prom_row_builder.rs | 158 +++++++++++++++++++++- tests-integration/tests/http.rs | 125 +++++++++++++---- 7 files changed, 693 insertions(+), 105 deletions(-) diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 2f21cc88259..99449cbc5df 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/src/frontend/src/instance/otlp.rs @@ -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, diff --git a/src/frontend/src/instance/prom_store.rs b/src/frontend/src/instance/prom_store.rs index ee07d9e8fd5..331ed158249 100644 --- a/src/frontend/src/instance/prom_store.rs +++ b/src/frontend/src/instance/prom_store.rs @@ -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, +) -> ServerResult { + metric_timestamp_datatype(tables) + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu) +} + +fn metric_physical_table_schema(timestamp_datatype: ColumnDataType) -> Vec { + 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 { + 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( diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index fabdf21ef7e..daa47ff359a 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -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 { + 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, +) -> Result { + 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, @@ -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 { + 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, diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index 6b7bbd3fab1..0f2dea314c8 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/src/servers/src/otlp/metrics.rs @@ -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, - 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, 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>, scope_attrs: Option<&Vec>, data_point_attrs: Option<&Vec>, - 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)) )); } diff --git a/src/servers/src/pending_rows_batcher.rs b/src/servers/src/pending_rows_batcher.rs index e08db004f95..d0eb1616090 100644 --- a/src/servers/src/pending_rows_batcher.rs +++ b/src/servers/src/pending_rows_batcher.rs @@ -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::() + .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); diff --git a/src/servers/src/prom_row_builder.rs b/src/servers/src/prom_row_builder.rs index eb620010da7..c7596f94b21 100644 --- a/src/servers/src/prom_row_builder.rs +++ b/src/servers/src/prom_row_builder.rs @@ -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::() + .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); + } } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 0745a3537c3..f8dddfdec05 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -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(