diff --git a/src/metric-engine/src/engine/create.rs b/src/metric-engine/src/engine/create.rs index 952643cd7a4..9fa6521abbd 100644 --- a/src/metric-engine/src/engine/create.rs +++ b/src/metric-engine/src/engine/create.rs @@ -234,8 +234,9 @@ impl MetricEngineInner { request_unit == unit, UnexpectedRequestSnafu { reason: format!( - "Metric has differenttime unit ({:?}) than the physical region ({:?})", - request_unit, unit + "Metric has different time unit ({request_unit:?}) than the physical \ + table ({unit:?}); the time index unit of a logical table must match \ + the physical table's", ), } ); @@ -1109,6 +1110,37 @@ mod test { assert!(is_tsid_col(tsid_metadata)); } + #[tokio::test] + async fn test_create_logical_regions_time_index_unit_mismatch() { + let env = TestEnv::new().await; + let engine = env.metric(); + let physical_region_id = env.default_physical_region_id(); + let logical_region_id = env.default_logical_region_id(); + env.create_physical_region_with_ts_type( + physical_region_id, + &TestEnv::default_table_dir(), + vec![], + ConcreteDataType::timestamp_microsecond_datatype(), + ) + .await; + + // A logical region request with a millisecond time index is rejected + // with the mismatch error naming both units. + let request = create_logical_region_request( + &["job"], + physical_region_id, + &store_api::path_utils::table_dir("test", logical_region_id.table_id()), + ); + let err = engine + .handle_request(logical_region_id, RegionRequest::Create(request)) + .await + .unwrap_err(); + let message = err.to_string(); + assert!(message.contains("Millisecond"), "{message}"); + assert!(message.contains("Microsecond"), "{message}"); + assert!(message.contains("different time unit"), "{message}"); + } + #[tokio::test] async fn test_create_logical_regions() { let env = TestEnv::new().await; diff --git a/src/metric-engine/src/engine/put.rs b/src/metric-engine/src/engine/put.rs index 56d4af2284f..fdf9b02a5ff 100644 --- a/src/metric-engine/src/engine/put.rs +++ b/src/metric-engine/src/engine/put.rs @@ -791,7 +791,9 @@ mod tests { use common_function::utils::partition_expr_version; use common_query::prelude::{greptime_native_histogram, greptime_timestamp, greptime_value}; use common_recordbatch::RecordBatches; - use datatypes::arrow::array::{Float64Array, TimestampMillisecondArray}; + use datatypes::arrow::array::{ + Float64Array, TimestampMicrosecondArray, TimestampMillisecondArray, + }; use datatypes::prelude::ConcreteDataType; use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema}; use datatypes::value::Value as PartitionValue; @@ -1116,6 +1118,91 @@ mod tests { expr.as_json_str().unwrap() } + #[tokio::test] + async fn test_put_and_scan_microsecond_physical_region() { + let env = TestEnv::new().await; + let engine = env.metric(); + let physical_region_id = env.default_physical_region_id(); + let logical_region_id = env.default_logical_region_id(); + env.create_physical_region_with_ts_type( + physical_region_id, + &TestEnv::default_table_dir(), + vec![], + ConcreteDataType::timestamp_microsecond_datatype(), + ) + .await; + + // A logical region with a matching microsecond time index is accepted. + let region_create_request = test_util::create_logical_region_request_with_ts_type( + &["job"], + physical_region_id, + &table_dir("test", logical_region_id.table_id()), + ConcreteDataType::timestamp_microsecond_datatype(), + ); + engine + .handle_request( + logical_region_id, + RegionRequest::Create(region_create_request), + ) + .await + .unwrap(); + + // Writing microsecond rows works. + let affected_rows = engine + .handle_request( + logical_region_id, + RegionRequest::Put(RegionPutRequest { + rows: Rows { + schema: test_util::row_schema_with_tags_and_ts_datatype( + &["job"], + ColumnDataType::TimestampMicrosecond, + ), + rows: test_util::build_rows_with_ts_datatype( + 1, + 2, + ColumnDataType::TimestampMicrosecond, + ), + }, + hint: None, + partition_expr_version: None, + skip_wal: false, + }), + ) + .await + .unwrap(); + assert_eq!(affected_rows.affected_rows, 2); + + // The scan returns timestamps in the physical region's unit. + let stream = engine + .scan_to_stream(logical_region_id, ScanRequest::default()) + .await + .unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + let mut rows = Vec::new(); + for batch in batches.iter() { + let batch = batch.df_record_batch(); + let timestamps = batch + .column(batch.schema().index_of(greptime_timestamp()).unwrap()) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(batch.schema().index_of(greptime_value()).unwrap()) + .as_any() + .downcast_ref::() + .unwrap(); + rows.extend( + timestamps + .values() + .iter() + .copied() + .zip(values.values().iter().copied()), + ); + } + rows.sort_unstable_by_key(|(timestamp, _)| *timestamp); + assert_eq!(rows, vec![(0, 0.0), (1, 1.0)]); + } + async fn create_logical_region_with_tags( env: &TestEnv, physical_region_id: RegionId, diff --git a/src/metric-engine/src/test_util.rs b/src/metric-engine/src/test_util.rs index 1f177370988..0671351ddb3 100644 --- a/src/metric-engine/src/test_util.rs +++ b/src/metric-engine/src/test_util.rs @@ -158,6 +158,24 @@ impl TestEnv { physical_region_id: RegionId, table_dir: &str, options: Vec<(String, String)>, + ) { + self.create_physical_region_with_ts_type( + physical_region_id, + table_dir, + options, + ConcreteDataType::timestamp_millisecond_datatype(), + ) + .await + } + + /// Create regions in [MetricEngine] with specific `physical_region_id` + /// and time index column type. + pub async fn create_physical_region_with_ts_type( + &self, + physical_region_id: RegionId, + table_dir: &str, + options: Vec<(String, String)>, + ts_type: ConcreteDataType, ) { let region_create_request = RegionCreateRequest { engine: METRIC_ENGINE_NAME.to_string(), @@ -165,11 +183,7 @@ impl TestEnv { ColumnMetadata { column_id: 0, semantic_type: SemanticType::Timestamp, - column_schema: ColumnSchema::new( - greptime_timestamp(), - ConcreteDataType::timestamp_millisecond_datatype(), - false, - ), + column_schema: ColumnSchema::new(greptime_timestamp(), ts_type, false), }, ColumnMetadata { column_id: 1, @@ -330,16 +344,28 @@ pub fn create_logical_region_request( tags: &[&str], physical_region_id: RegionId, table_dir: &str, +) -> RegionCreateRequest { + create_logical_region_request_with_ts_type( + tags, + physical_region_id, + table_dir, + ConcreteDataType::timestamp_millisecond_datatype(), + ) +} + +/// Generate a [RegionCreateRequest] for logical region with the given time +/// index column type. +pub fn create_logical_region_request_with_ts_type( + tags: &[&str], + physical_region_id: RegionId, + table_dir: &str, + ts_type: ConcreteDataType, ) -> RegionCreateRequest { let mut column_metadatas = vec![ ColumnMetadata { column_id: 0, semantic_type: SemanticType::Timestamp, - column_schema: ColumnSchema::new( - greptime_timestamp(), - ConcreteDataType::timestamp_millisecond_datatype(), - false, - ), + column_schema: ColumnSchema::new(greptime_timestamp(), ts_type, false), }, ColumnMetadata { column_id: 1, @@ -407,10 +433,20 @@ pub fn alter_logical_region_request(tags: &[&str]) -> RegionAlterRequest { /// /// The result will also contains default timestamp and value column at beginning. pub fn row_schema_with_tags(tags: &[&str]) -> Vec { + row_schema_with_tags_and_ts_datatype(tags, ColumnDataType::TimestampMillisecond) +} + +/// Generate a row schema with given tag columns and time index datatype. +/// +/// The result will also contains default timestamp and value column at beginning. +pub fn row_schema_with_tags_and_ts_datatype( + tags: &[&str], + ts_datatype: ColumnDataType, +) -> Vec { let mut schema = vec![ PbColumnSchema { column_name: greptime_timestamp().to_string(), - datatype: ColumnDataType::TimestampMillisecond as i32, + datatype: ts_datatype as i32, semantic_type: SemanticType::Timestamp as _, datatype_extension: None, options: None, @@ -440,11 +476,33 @@ pub fn row_schema_with_tags(tags: &[&str]) -> Vec { /// The schema is generated by [row_schema_with_tags]. `num_tags` doesn't need to be precise, /// it's used to determine the column id for new columns. pub fn build_rows(num_tags: usize, num_rows: usize) -> Vec { + build_rows_with_ts_datatype(num_tags, num_rows, ColumnDataType::TimestampMillisecond) +} + +/// Build [Row]s whose time index values use the given timestamp datatype. +/// +/// The schema is generated by [row_schema_with_tags_and_ts_datatype]. +/// `num_tags` doesn't need to be precise, it's used to determine the column id +/// for new columns. +pub fn build_rows_with_ts_datatype( + num_tags: usize, + num_rows: usize, + ts_datatype: ColumnDataType, +) -> Vec { + let ts_value = |i: usize| -> ValueData { + match ts_datatype { + ColumnDataType::TimestampSecond => ValueData::TimestampSecondValue(i as _), + ColumnDataType::TimestampMillisecond => ValueData::TimestampMillisecondValue(i as _), + ColumnDataType::TimestampMicrosecond => ValueData::TimestampMicrosecondValue(i as _), + ColumnDataType::TimestampNanosecond => ValueData::TimestampNanosecondValue(i as _), + _ => unreachable!("not a timestamp datatype"), + } + }; let mut rows = vec![]; for i in 0..num_rows { let mut values = vec![ Value { - value_data: Some(ValueData::TimestampMillisecondValue(i as _)), + value_data: Some(ts_value(i)), }, Value { value_data: Some(ValueData::F64Value(i as f64)), diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index 35727ae9b35..9be1084f46e 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -22,6 +22,7 @@ use api::v1::region::{ InsertRequest as RegionInsertRequest, InsertRequests as RegionInsertRequests, RegionRequestHeader, }; +use api::v1::value::ValueData; use api::v1::{ AlterTableExpr, ColumnDataType, ColumnSchema, CreateTableExpr, InsertRequests, RowInsertRequest, RowInsertRequests, Rows, SemanticType, @@ -46,6 +47,8 @@ use common_query::native_histogram::{is_native_histogram_value_type, native_hist use common_query::prelude::{greptime_timestamp, greptime_value}; use common_telemetry::tracing_context::TracingContext; use common_telemetry::{debug, error, warn}; +use common_time::Timestamp; +use common_time::timestamp::TimeUnit; use datatypes::schema::SkippingIndexOptions; use futures_util::future; use meter_macros::write_meter; @@ -436,9 +439,25 @@ impl Inserter { validate_column_count_match(&requests)?; // check and create physical table - self.create_physical_table_on_demand(&ctx, physical_table.clone(), statement_executor) + let physical_table_ref = self + .create_physical_table_on_demand(&ctx, physical_table.clone(), statement_executor) .await?; + // Align the requests' time index unit with the physical table's before + // creating logical tables or writing: the metric engine requires the + // logical tables (hence the requests) to use the same time index unit + // as the physical table. Ingestion endpoints encode timestamps in a + // fixed unit (Prometheus remote write always uses millisecond, OTLP + // uses nanosecond on the metric engine path), so rewrite the schema + // and values to the physical table's unit here. Converting to a + // coarser unit truncates the sub-unit part, following + // `Timestamp::convert_to`'s floor semantics. + // + // This reuses the table lookup from `create_physical_table_on_demand` + // (no extra catalog access) and is a no-op apart from a schema scan + // when the units already match. + align_row_insert_requests_time_unit(&physical_table_ref, &mut requests)?; + // check and create logical tables let CreateAlterTableResult { instant_table_ids, @@ -1052,17 +1071,16 @@ impl Inserter { ctx: &QueryContextRef, physical_table: String, statement_executor: &StatementExecutor, - ) -> Result<()> { + ) -> Result { let catalog_name = ctx.current_catalog(); let schema_name = ctx.current_schema(); // check if exist - if self + if let Some(table) = self .get_table(catalog_name, &schema_name, &physical_table) .await? - .is_some() { - return Ok(()); + return Ok(table); } // Gate here too, otherwise a disabled switch would still leak the physical table. @@ -1114,7 +1132,7 @@ impl Inserter { .await; match res { - Ok(_) => Ok(()), + Ok(table) => Ok(table), Err(err) => { error!(err; "Failed to create table {table_reference}"); Err(err) @@ -1388,6 +1406,124 @@ fn request_is_native_histogram(request_schema: &[ColumnSchema]) -> bool { ) } +/// Rewrites the time index column of every request in `requests` to the given +/// physical metric table's time index unit. +/// +/// This only rewrites requests whose time index unit differs from the +/// physical table's; converting to a coarser unit truncates the sub-unit part +/// (floor, see `Timestamp::convert_to`). A metric table without a timestamp +/// column is left alone: the metric engine rejects it anyway. +fn align_row_insert_requests_time_unit( + physical_table: &TableRef, + requests: &mut RowInsertRequests, +) -> Result<()> { + let Some(target_unit) = physical_table + .table_info() + .meta + .schema + .timestamp_column() + .and_then(|col| col.data_type.as_timestamp().map(|ts| ts.unit())) + else { + return Ok(()); + }; + convert_row_insert_requests_time_unit(requests, target_unit) +} + +/// Converts the time index column of each request to `target_unit`. +fn convert_row_insert_requests_time_unit( + requests: &mut RowInsertRequests, + target_unit: TimeUnit, +) -> Result<()> { + for request in &mut requests.inserts { + let Some(rows) = request.rows.as_mut() else { + continue; + }; + convert_rows_time_unit(rows, target_unit)?; + } + Ok(()) +} + +fn convert_rows_time_unit(rows: &mut Rows, target_unit: TimeUnit) -> Result<()> { + let Some(ts_index) = rows + .schema + .iter() + .position(|col| col.semantic_type == SemanticType::Timestamp as i32) + else { + return Ok(()); + }; + let Some(source_unit) = timestamp_unit_from_datatype(rows.schema[ts_index].datatype) else { + return Ok(()); + }; + if source_unit == target_unit { + return Ok(()); + } + + rows.schema[ts_index].datatype = column_datatype_from_time_unit(target_unit) as i32; + // Timestamp columns never carry a datatype extension. + rows.schema[ts_index].datatype_extension = None; + + for row in &mut rows.rows { + let Some(value) = row.values.get_mut(ts_index) else { + continue; + }; + let Some(value_data) = value.value_data.take() else { + continue; + }; + value.value_data = + convert_timestamp_value_data(value_data, source_unit, target_unit, ts_index)?; + } + Ok(()) +} + +fn timestamp_unit_from_datatype(datatype: i32) -> Option { + ColumnDataType::try_from(datatype) + .ok() + .and_then(|datatype| match datatype { + ColumnDataType::TimestampSecond => Some(TimeUnit::Second), + ColumnDataType::TimestampMillisecond => Some(TimeUnit::Millisecond), + ColumnDataType::TimestampMicrosecond => Some(TimeUnit::Microsecond), + ColumnDataType::TimestampNanosecond => Some(TimeUnit::Nanosecond), + _ => None, + }) +} + +fn column_datatype_from_time_unit(unit: TimeUnit) -> ColumnDataType { + match unit { + TimeUnit::Second => ColumnDataType::TimestampSecond, + TimeUnit::Millisecond => ColumnDataType::TimestampMillisecond, + TimeUnit::Microsecond => ColumnDataType::TimestampMicrosecond, + TimeUnit::Nanosecond => ColumnDataType::TimestampNanosecond, + } +} + +fn convert_timestamp_value_data( + value_data: ValueData, + source_unit: TimeUnit, + target_unit: TimeUnit, + column_index: usize, +) -> Result> { + let timestamp = match value_data { + ValueData::TimestampSecondValue(v) => Timestamp::new_second(v), + ValueData::TimestampMillisecondValue(v) => Timestamp::new_millisecond(v), + ValueData::TimestampMicrosecondValue(v) => Timestamp::new_microsecond(v), + ValueData::TimestampNanosecondValue(v) => Timestamp::new_nanosecond(v), + // Null or non-timestamp value; nothing to convert. + other => return Ok(Some(other)), + }; + let converted = timestamp.convert_to(target_unit).context(InvalidInsertRequestSnafu { + reason: format!( + "timestamp column {column_index} value {} in unit {source_unit:?} overflows when converting to unit {target_unit:?}", + timestamp.value() + ), + })?; + Ok(Some(match target_unit { + TimeUnit::Second => ValueData::TimestampSecondValue(converted.value()), + TimeUnit::Millisecond => ValueData::TimestampMillisecondValue(converted.value()), + TimeUnit::Microsecond => ValueData::TimestampMicrosecondValue(converted.value()), + TimeUnit::Nanosecond => ValueData::TimestampNanosecondValue(converted.value()), + })) +} + fn table_is_native_histogram(table: &TableRef) -> bool { let mut fields = table.field_columns(); let Some(col) = fields.next() else { @@ -1705,7 +1841,7 @@ mod tests { use api::helper::ColumnDataTypeWrapper; use api::v1::helper::{field_column_schema, time_index_column_schema}; - use api::v1::{RowInsertRequest, Rows, Value}; + use api::v1::{Row, RowInsertRequest, Rows, Value}; use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME}; use common_meta::cache::new_table_flownode_set_cache; use common_meta::ddl::test_util::datanode_handler::NaiveDatanodeHandler; @@ -1770,6 +1906,204 @@ mod tests { )) } + fn make_metric_physical_table_ref_with_time_unit(unit: TimeUnit) -> TableRef { + let schema = datatypes::schema::SchemaBuilder::try_from_columns(vec![ + ColumnSchema::new( + greptime_timestamp(), + ConcreteDataType::timestamp_datatype(unit), + false, + ) + .with_time_index(true), + ColumnSchema::new(greptime_value(), ConcreteDataType::float64_datatype(), true), + ]) + .unwrap() + .build() + .unwrap(); + let meta = TableMetaBuilder::empty() + .schema(Arc::new(schema)) + .primary_key_indices(vec![]) + .value_indices(vec![1]) + .engine("metric") + .next_column_id(0) + .options(Default::default()) + .created_on(Default::default()) + .build() + .unwrap(); + let info = Arc::new( + TableInfoBuilder::default() + .table_id(1) + .table_version(0) + .name("greptime_physical_table") + .schema_name(DEFAULT_SCHEMA_NAME) + .catalog_name(DEFAULT_CATALOG_NAME) + .desc(None) + .table_type(TableType::Base) + .meta(meta) + .build() + .unwrap(), + ); + Arc::new(table::Table::new( + info, + table::metadata::FilterPushDownType::Unsupported, + Arc::new(DummyDataSource), + )) + } + + fn ms_row_insert_request(timestamp_ms: i64) -> RowInsertRequest { + RowInsertRequest { + table_name: "my_metric".to_string(), + rows: Some(Rows { + schema: vec![ + time_index_column_schema( + greptime_timestamp(), + ColumnDataType::TimestampMillisecond, + ), + field_column_schema(greptime_value(), ColumnDataType::Float64), + ], + rows: vec![Row { + values: vec![ + Value { + value_data: Some(ValueData::TimestampMillisecondValue(timestamp_ms)), + }, + Value { + value_data: Some(ValueData::F64Value(1.0)), + }, + ], + }], + }), + } + } + + #[test] + fn test_convert_row_insert_requests_time_unit_noop_when_matching() { + let mut requests = RowInsertRequests { + inserts: vec![ms_row_insert_request(123)], + }; + convert_row_insert_requests_time_unit(&mut requests, TimeUnit::Millisecond).unwrap(); + let rows = requests.inserts[0].rows.as_ref().unwrap(); + assert_eq!( + rows.schema[0].datatype, + ColumnDataType::TimestampMillisecond as i32 + ); + assert!(matches!( + rows.rows[0].values[0].value_data, + Some(ValueData::TimestampMillisecondValue(123)) + )); + } + + #[test] + fn test_convert_row_insert_requests_time_unit_widens_losslessly() { + let mut requests = RowInsertRequests { + inserts: vec![ms_row_insert_request(123)], + }; + convert_row_insert_requests_time_unit(&mut requests, TimeUnit::Microsecond).unwrap(); + let rows = requests.inserts[0].rows.as_ref().unwrap(); + assert_eq!( + rows.schema[0].datatype, + ColumnDataType::TimestampMicrosecond as i32 + ); + assert!(matches!( + rows.rows[0].values[0].value_data, + Some(ValueData::TimestampMicrosecondValue(123_000)) + )); + } + + #[test] + fn test_convert_row_insert_requests_time_unit_truncates_on_narrowing() { + // 123_456_789 ns floors to 123_456 us and 123 ms; negative values + // floor towards negative infinity, matching `Timestamp::convert_to`. + let requests = |value_ns: i64| RowInsertRequests { + inserts: vec![RowInsertRequest { + table_name: "my_metric".to_string(), + rows: Some(Rows { + schema: vec![ + time_index_column_schema( + greptime_timestamp(), + ColumnDataType::TimestampNanosecond, + ), + field_column_schema(greptime_value(), ColumnDataType::Float64), + ], + rows: vec![Row { + values: vec![ + Value { + value_data: Some(ValueData::TimestampNanosecondValue(value_ns)), + }, + Value { + value_data: Some(ValueData::F64Value(1.0)), + }, + ], + }], + }), + }], + }; + + let mut reqs = requests(123_456_789); + convert_row_insert_requests_time_unit(&mut reqs, TimeUnit::Microsecond).unwrap(); + let rows = reqs.inserts[0].rows.as_ref().unwrap(); + assert!(matches!( + rows.rows[0].values[0].value_data, + Some(ValueData::TimestampMicrosecondValue(123_456)) + )); + + let mut reqs = requests(123_456_789); + convert_row_insert_requests_time_unit(&mut reqs, TimeUnit::Millisecond).unwrap(); + let rows = reqs.inserts[0].rows.as_ref().unwrap(); + assert!(matches!( + rows.rows[0].values[0].value_data, + Some(ValueData::TimestampMillisecondValue(123)) + )); + + let mut reqs = requests(-123_456_789); + convert_row_insert_requests_time_unit(&mut reqs, TimeUnit::Microsecond).unwrap(); + let rows = reqs.inserts[0].rows.as_ref().unwrap(); + assert!(matches!( + rows.rows[0].values[0].value_data, + Some(ValueData::TimestampMicrosecondValue(-123_457)) + )); + } + + #[test] + fn test_convert_row_insert_requests_time_unit_overflow() { + let mut requests = RowInsertRequests { + inserts: vec![ms_row_insert_request(i64::MAX)], + }; + let err = + convert_row_insert_requests_time_unit(&mut requests, TimeUnit::Nanosecond).unwrap_err(); + assert!(err.to_string().contains("overflows"), "{err}"); + } + + #[test] + fn test_align_row_insert_requests_time_unit() { + // Microsecond physical table, millisecond request (prometheus remote + // write shape): the request is rewritten to the physical unit. + let table = make_metric_physical_table_ref_with_time_unit(TimeUnit::Microsecond); + let mut requests = RowInsertRequests { + inserts: vec![ms_row_insert_request(123)], + }; + align_row_insert_requests_time_unit(&table, &mut requests).unwrap(); + let rows = requests.inserts[0].rows.as_ref().unwrap(); + assert_eq!( + rows.schema[0].datatype, + ColumnDataType::TimestampMicrosecond as i32 + ); + assert!(matches!( + rows.rows[0].values[0].value_data, + Some(ValueData::TimestampMicrosecondValue(123_000)) + )); + + // Millisecond physical table (the auto-created default): no-op. + let table = make_metric_physical_table_ref_with_time_unit(TimeUnit::Millisecond); + let mut requests = RowInsertRequests { + inserts: vec![ms_row_insert_request(123)], + }; + align_row_insert_requests_time_unit(&table, &mut requests).unwrap(); + let rows = requests.inserts[0].rows.as_ref().unwrap(); + assert_eq!( + rows.schema[0].datatype, + ColumnDataType::TimestampMillisecond as i32 + ); + } + #[tokio::test] async fn test_accommodate_existing_schema_and_reject_kind_changes() { let ts_name = "my_ts"; diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index 6b7bbd3fab1..07bde7d2905 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/src/servers/src/otlp/metrics.rs @@ -971,9 +971,15 @@ fn write_timestamp( table: &mut TableData, row: &mut Vec, time_nano: i64, - legacy_mode: bool, + metric_ctx: &OtlpMetricCtx, ) -> Result<()> { - if legacy_mode { + // Keep the full nanosecond precision whenever the request is headed for + // the metric engine: `Inserter::handle_metric_row_inserts` converts the + // timestamps to the physical table's time index unit (which may be + // micro/nanosecond). Only the non-metric prometheus-compatible path is + // fixed to milliseconds, to keep auto-created mito tables on the + // millisecond time index. + if metric_ctx.is_legacy || metric_ctx.with_metric_engine { row_writer::write_ts_to_nanos( table, greptime_timestamp(), @@ -1075,7 +1081,7 @@ fn write_tags_and_timestamp( )?; } - write_timestamp(table, row, timestamp_nanos, metric_ctx.is_legacy)?; + write_timestamp(table, row, timestamp_nanos, metric_ctx)?; Ok(()) } @@ -1512,6 +1518,7 @@ fn encode_summary( #[cfg(test)] mod tests { + use api::v1::ColumnDataType; use common_query::prelude::set_default_prefix; use otel_arrow_rust::proto::opentelemetry::common::v1::AnyValue; use otel_arrow_rust::proto::opentelemetry::common::v1::any_value::Value as Val; @@ -2497,6 +2504,81 @@ mod tests { } } + #[test] + fn test_metric_engine_path_keeps_nanosecond_precision() { + let time_unix_nano = 1_704_067_200_123_456_789u64; + let request = metrics_request(vec![Metric { + name: "my_gauge".to_string(), + data: Some(metric::Data::Gauge(Gauge { + data_points: vec![NumberDataPoint { + time_unix_nano, + value: Some(Value::AsDouble(1.0)), + ..Default::default() + }], + })), + ..Default::default() + }]); + + // The metric engine path keeps the full nanosecond precision here; + // `Inserter::handle_metric_row_inserts` converts the timestamps to + // the physical table's time index unit afterwards. + let mut metric_ctx = OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + }; + let MetricsConversion { requests, .. } = + to_grpc_insert_requests(request, &mut metric_ctx).unwrap(); + + let rows = requests.inserts[0].rows.as_ref().unwrap(); + let ts_index = rows + .schema + .iter() + .position(|column| column.column_name == greptime_timestamp()) + .unwrap(); + assert_eq!( + rows.schema[ts_index].datatype, + ColumnDataType::TimestampNanosecond as i32 + ); + assert!(matches!( + rows.rows[0].values[ts_index].value_data, + Some(ValueData::TimestampNanosecondValue( + 1_704_067_200_123_456_789 + )) + )); + + // The non-metric prometheus-compatible path stays millisecond so + // auto-created mito tables keep the millisecond time index. + let mut compat_ctx = OtlpMetricCtx::default(); + let request = metrics_request(vec![Metric { + name: "my_gauge".to_string(), + data: Some(metric::Data::Gauge(Gauge { + data_points: vec![NumberDataPoint { + time_unix_nano, + value: Some(Value::AsDouble(1.0)), + ..Default::default() + }], + })), + ..Default::default() + }]); + let MetricsConversion { requests, .. } = + to_grpc_insert_requests(request, &mut compat_ctx).unwrap(); + + let rows = requests.inserts[0].rows.as_ref().unwrap(); + let ts_index = rows + .schema + .iter() + .position(|column| column.column_name == greptime_timestamp()) + .unwrap(); + assert_eq!( + rows.schema[ts_index].datatype, + ColumnDataType::TimestampMillisecond as i32 + ); + assert!(matches!( + rows.rows[0].values[ts_index].value_data, + Some(ValueData::TimestampMillisecondValue(1_704_067_200_123)) + )); + } + #[test] fn test_exponential_histogram_gate_and_partial_outcome() { let request = metrics_request(vec![ diff --git a/src/servers/src/prom_store.rs b/src/servers/src/prom_store.rs index 980504515a8..73f36d1424d 100644 --- a/src/servers/src/prom_store.rs +++ b/src/servers/src/prom_store.rs @@ -23,9 +23,10 @@ use api::prom_store::remote::label_matcher::Type as MatcherType; use api::prom_store::remote::{Label, Query, Sample, TimeSeries, WriteRequest}; use api::v1::RowInsertRequests; use arrow::array::{ - Array, AsArray, DictionaryArray, LargeStringArray, StringArray, StringViewArray, + Array, ArrayRef, AsArray, DictionaryArray, LargeStringArray, StringArray, StringViewArray, }; -use arrow::datatypes::{Float64Type, TimestampMillisecondType, UInt32Type}; +use arrow::compute::kernels::cast as casts; +use arrow::datatypes::{DataType, Float64Type, TimeUnit, TimestampMillisecondType, UInt32Type}; use common_grpc::precision::Precision; use common_query::prelude::{greptime_timestamp, greptime_value}; use common_recordbatch::{RecordBatch, RecordBatches}; @@ -383,6 +384,35 @@ fn recordbatch_to_timeseries( .with_context(|| error::InvalidPromRemoteReadQueryResultSnafu { msg: format!("missing timestamp column '{timestamp_column_name}' in query result"), })?; + // The Prometheus remote read wire format carries millisecond timestamps, + // while the table's time index can use any time unit (e.g. a metric + // physical table created with TIMESTAMP(6)). Narrowing casts truncate the + // sub-millisecond part. + let ts_column: ArrayRef = match ts_column.data_type() { + DataType::Timestamp(TimeUnit::Millisecond, _) => ts_column.clone(), + DataType::Timestamp(_, _) => casts::cast( + ts_column, + &DataType::Timestamp(TimeUnit::Millisecond, None), + ) + .map_err(|e| { + error::InvalidPromRemoteReadQueryResultSnafu { + msg: format!( + "failed to cast timestamp column '{timestamp_column_name}' of datatype {:?} to millisecond: {e}", + ts_column.data_type() + ), + } + .build() + })?, + _ => { + return error::InvalidPromRemoteReadQueryResultSnafu { + msg: format!( + "Expect timestamp column of datatype Timestamp(Millisecond), actual {:?}", + ts_column.data_type() + ), + } + .fail(); + } + }; let ts_column = ts_column .as_primitive_opt::() .with_context(|| error::InvalidPromRemoteReadQueryResultSnafu { diff --git a/tests-integration/src/otlp.rs b/tests-integration/src/otlp.rs index fe4dfb8a9e5..c6e4d07034a 100644 --- a/tests-integration/src/otlp.rs +++ b/tests-integration/src/otlp.rs @@ -19,6 +19,7 @@ mod test { use client::{DEFAULT_CATALOG_NAME, OutputData}; use common_recordbatch::RecordBatches; use datatypes::arrow::array::AsArray; + use datatypes::arrow::datatypes::TimestampMicrosecondType; use frontend::instance::Instance; use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest; @@ -37,6 +38,7 @@ mod test { use servers::query_handler::OpenTelemetryProtocolHandler; use servers::query_handler::sql::SqlQueryHandler; use session::context::QueryContext; + use session::protocol_ctx::{OtlpMetricCtx, ProtocolCtx}; use crate::standalone::GreptimeDbStandaloneBuilder; use crate::tests; @@ -888,6 +890,99 @@ WITH( ); } + #[tokio::test(flavor = "multi_thread")] + pub async fn test_otlp_metrics_into_microsecond_physical_table_on_standalone() { + let standalone = GreptimeDbStandaloneBuilder::new("test_otlp_us_physical") + .build() + .await; + + test_otlp_metrics_into_microsecond_physical_table(standalone.fe_instance()).await; + } + + #[tokio::test(flavor = "multi_thread")] + pub async fn test_otlp_metrics_into_microsecond_physical_table_on_distributed() { + let instance = tests::create_distributed_instance("test_otlp_us_physical_dist").await; + + test_otlp_metrics_into_microsecond_physical_table(&instance.frontend()).await; + } + + /// Regression test for : + /// a physical metric table pre-created with a TIMESTAMP(6) (microsecond) + /// time index accepts OTLP ingestion; nanosecond samples are converted to + /// the physical table's unit, keeping microsecond precision. + async fn test_otlp_metrics_into_microsecond_physical_table(instance: &Arc) { + let db = "otlp_us_physical"; + let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, db); + // Route the request to the metric engine, like the OTLP HTTP handler + // does with the default `prom_store.with_metric_engine = true`. + ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx { + with_metric_engine: true, + ..Default::default() + })); + let ctx = Arc::new(ctx); + assert!( + SqlQueryHandler::do_query( + instance.as_ref(), + &format!("CREATE DATABASE IF NOT EXISTS {db}"), + ctx.clone(), + ) + .await + .first() + .unwrap() + .is_ok() + ); + + let mut output = instance + .do_query( + "CREATE TABLE greptime_physical_table (\ + greptime_timestamp TIMESTAMP(6) NOT NULL, \ + greptime_value DOUBLE NULL, \ + TIME INDEX (greptime_timestamp)) \ + ENGINE = metric WITH ('physical_metric_table' = 'true')", + ctx.clone(), + ) + .await; + assert!(output.remove(0).is_ok()); + + let request = ExportMetricsServiceRequest { + resource_metrics: vec![ResourceMetrics { + scope_metrics: vec![ScopeMetrics { + metrics: vec![Metric { + name: "my_gauge".to_string(), + data: Some(metric::Data::Gauge(Gauge { + data_points: vec![NumberDataPoint { + attributes: vec![keyvalue("host", "h1")], + time_unix_nano: 1_704_067_200_123_456_789, + value: Some(Value::AsDouble(1.0)), + ..Default::default() + }], + })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }], + }; + instance.metrics(request, ctx.clone()).await.unwrap(); + + let mut output = instance + .do_query("SELECT greptime_timestamp FROM my_gauge", ctx.clone()) + .await; + let OutputData::Stream(stream) = output.remove(0).unwrap().data else { + unreachable!() + }; + let batches = RecordBatches::try_collect(stream).await.unwrap().take(); + assert_eq!(batches[0].num_rows(), 1); + // The logical table's time index keeps the physical table's + // microsecond unit and precision: 1704067200123456789ns -> + // 1704067200123456us. + let timestamps = batches[0] + .column(0) + .as_primitive::(); + assert_eq!(timestamps.value(0), 1_704_067_200_123_456); + } + fn build_request() -> ExportMetricsServiceRequest { let data_points = vec![ NumberDataPoint { diff --git a/tests-integration/src/prom_store.rs b/tests-integration/src/prom_store.rs index 0649b2dd778..fe58a451311 100644 --- a/tests-integration/src/prom_store.rs +++ b/tests-integration/src/prom_store.rs @@ -20,7 +20,10 @@ mod tests { use api::prom_store::remote::{ Label, LabelMatcher, Query, ReadRequest, ReadResponse, Sample, WriteRequest, }; + use client::OutputData; use common_catalog::consts::DEFAULT_CATALOG_NAME; + use datatypes::arrow::array::AsArray; + use datatypes::arrow::datatypes::TimestampMicrosecondType; use frontend::instance::Instance; use prost::Message; use servers::http::prom_store::PHYSICAL_TABLE_PARAM; @@ -83,6 +86,121 @@ mod tests { .await; } + #[tokio::test(flavor = "multi_thread")] + async fn test_standalone_prom_store_remote_rw_microsecond_physical_table() { + common_telemetry::init_default_ut_logging(); + let standalone = + GreptimeDbStandaloneBuilder::new("test_prom_store_remote_rw_us_physical_table") + .build() + .await; + let instance = standalone.fe_instance(); + + test_prom_store_remote_rw_microsecond_physical_table(instance).await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_distributed_prom_store_remote_rw_microsecond_physical_table() { + common_telemetry::init_default_ut_logging(); + let distributed = + tests::create_distributed_instance("test_prom_store_remote_rw_us_physical_table").await; + test_prom_store_remote_rw_microsecond_physical_table(&distributed.frontend()).await; + } + + /// Regression test for : + /// prometheus remote write/read against a physical metric table + /// pre-created with a TIMESTAMP(6) (microsecond) time index. Millisecond + /// samples are widened to the physical table's unit on write and narrowed + /// back to milliseconds on remote read. + async fn test_prom_store_remote_rw_microsecond_physical_table(instance: &Arc) { + let db = "prometheus_us"; + let ctx = Arc::new(QueryContext::with(DEFAULT_CATALOG_NAME, db)); + assert!( + SqlQueryHandler::do_query( + instance.as_ref(), + &format!("CREATE DATABASE IF NOT EXISTS {db}"), + ctx.clone(), + ) + .await + .first() + .unwrap() + .is_ok() + ); + + let mut output = instance + .do_query( + "CREATE TABLE greptime_physical_table (\ + greptime_timestamp TIMESTAMP(6) NOT NULL, \ + greptime_value DOUBLE NULL, \ + TIME INDEX (greptime_timestamp)) \ + ENGINE = metric WITH ('physical_metric_table' = 'true')", + ctx.clone(), + ) + .await; + assert!(output.remove(0).is_ok()); + + let write_request = WriteRequest { + timeseries: vec![prom_store::mock_timeseries()[0].clone()], + ..Default::default() + }; + let (row_inserts, _) = to_grpc_row_insert_requests(&write_request).unwrap(); + instance + .write(row_inserts, ctx.clone(), true) + .await + .unwrap(); + + let read_request = ReadRequest { + queries: vec![Query { + start_timestamp_ms: 1000, + end_timestamp_ms: 2000, + matchers: vec![LabelMatcher { + name: prom_store::METRIC_NAME_LABEL.to_string(), + value: "metric1".to_string(), + r#type: 0, + }], + ..Default::default() + }], + ..Default::default() + }; + let resp = instance.read(read_request, ctx.clone()).await.unwrap(); + let body = prom_store::snappy_decompress(&resp.body).unwrap(); + let read_response = ReadResponse::decode(&body[..]).unwrap(); + assert_eq!(1, read_response.results.len()); + assert_eq!(1, read_response.results[0].timeseries.len()); + let timeseries = &read_response.results[0].timeseries[0]; + assert_eq!( + timeseries.samples, + vec![ + Sample { + value: 1.0, + timestamp: 1000, + }, + Sample { + value: 2.0, + timestamp: 2000, + }, + ] + ); + + // The stored timestamps keep the physical table's microsecond unit. + let mut output = instance + .do_query("SELECT greptime_timestamp FROM metric1", ctx.clone()) + .await; + let OutputData::Stream(stream) = output.remove(0).unwrap().data else { + unreachable!() + }; + let batches = common_recordbatch::RecordBatches::try_collect(stream) + .await + .unwrap() + .take(); + assert_eq!(batches[0].num_rows(), 2); + let timestamps = batches[0] + .column(0) + .as_primitive::(); + // Millisecond samples are widened to the microsecond time index. + assert_eq!(timestamps.value(0), 1_000_000); + assert_eq!(timestamps.value(1), 2_000_000); + } + async fn test_prom_store_remote_rw(instance: &Arc, physical_table: Option) { let write_request = WriteRequest { timeseries: prom_store::mock_timeseries(),