diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index 74ecb6b3cd..2bfcc617cb 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -1496,7 +1496,7 @@ mod tests { } #[tokio::test] - async fn test_accommodate_existing_schema_logic() { + async fn test_accommodate_existing_schema_and_reject_kind_changes() { let ts_name = "my_ts"; let field_name = "my_field"; let table = @@ -1548,6 +1548,66 @@ mod tests { let req_schema = req.rows.as_ref().unwrap().schema.clone(); assert_eq!(req_schema[0].column_name, ts_name); assert_eq!(req_schema[1].column_name, field_name); + + let (datatype, datatype_extension) = + ColumnDataTypeWrapper::try_from(native_histogram_value_type().clone()) + .unwrap() + .into_parts(); + let mut histogram_req = RowInsertRequest { + table_name: "test_table".to_string(), + rows: Some(Rows { + schema: vec![ + time_index_column_schema("ts", ColumnDataType::TimestampMillisecond), + api::v1::ColumnSchema { + column_name: greptime_native_histogram().to_string(), + datatype: datatype as i32, + semantic_type: SemanticType::Field as i32, + datatype_extension, + options: None, + }, + ], + rows: vec![], + }), + }; + let error = inserter + .get_alter_table_expr_on_demand(&mut histogram_req, &table, &ctx, true, true, true) + .unwrap_err(); + assert!( + error + .to_string() + .contains("cannot mix native histogram and float sample fields") + ); + + let histogram_table = make_table_ref_with_schema( + "ts", + greptime_native_histogram(), + native_histogram_value_type().clone(), + ); + let mut sample_req = RowInsertRequest { + table_name: "test_table".to_string(), + rows: Some(Rows { + schema: vec![ + time_index_column_schema("ts", ColumnDataType::TimestampMillisecond), + field_column_schema(greptime_value(), ColumnDataType::Float64), + ], + rows: vec![], + }), + }; + let error = inserter + .get_alter_table_expr_on_demand( + &mut sample_req, + &histogram_table, + &ctx, + true, + true, + true, + ) + .unwrap_err(); + assert!( + error + .to_string() + .contains("cannot mix native histogram and float sample fields") + ); } #[test] diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index 30a10aebed..56ab053033 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/src/servers/src/otlp/metrics.rs @@ -374,7 +374,8 @@ fn encode_metrics( metric_ctx, )?; } - // TODO(sunng87) leave ExponentialHistogram for next release + // TODO: Convert OTLP exponential histograms into the canonical native-histogram + // Struct. See docs/rfcs/2026-08-04-native-histograms.md. metric::Data::ExponentialHistogram(_hist) => {} } } @@ -714,12 +715,6 @@ fn encode_histogram( Ok(()) } -#[allow(dead_code)] -fn encode_exponential_histogram(_name: &str, _hist: &ExponentialHistogram) -> Result<()> { - // TODO(sunng87): implement this using a prometheus compatible way - Ok(()) -} - fn encode_summary( table_writer: &mut MultiTableData, name: &str, diff --git a/src/servers/src/prom_remote_write/decode.rs b/src/servers/src/prom_remote_write/decode.rs index 3f108cdc2f..6f635c7caf 100644 --- a/src/servers/src/prom_remote_write/decode.rs +++ b/src/servers/src/prom_remote_write/decode.rs @@ -56,6 +56,12 @@ pub(crate) struct PromTimeSeries { impl Clear for PromTimeSeries { fn clear(&mut self) { self.table_name.clear(); + self.schema.clear(); + self.physical_table.clear(); + // Labels borrow from the request buffer, which may be replaced after this reset. + for label in self.labels.iter_mut() { + label.clear(); + } self.labels.clear(); self.samples.clear(); } @@ -93,27 +99,31 @@ impl PromTimeSeries { } #[allow(deprecated)] - match label.name { + let is_special_label = match label.name { METRIC_NAME_LABEL_BYTES => { self.table_name = prom_validation_mode.decode_string(label.value)?; - self.labels.truncate(self.labels.len() - 1); + true } SCHEMA_LABEL_BYTES => { self.schema = Some(prom_validation_mode.decode_string(label.value)?); - self.labels.truncate(self.labels.len() - 1); + true } DATABASE_LABEL_BYTES | DATABASE_LABEL_ALT_BYTES => { if self.schema.is_none() { self.schema = Some(prom_validation_mode.decode_string(label.value)?); } - self.labels.truncate(self.labels.len() - 1); + true } PHYSICAL_TABLE_LABEL_BYTES | PHYSICAL_TABLE_LABEL_ALT_BYTES => { self.physical_table = Some(prom_validation_mode.decode_string(label.value)?); - self.labels.truncate(self.labels.len() - 1); + true } - _ => {} + _ => false, + }; + if is_special_label { + label.clear(); + self.labels.truncate(self.labels.len() - 1); } Ok(()) @@ -129,6 +139,9 @@ impl PromTimeSeries { Ok(()) } 3u32 => prost::encoding::skip_field(wire_type, tag, buf, Default::default()), + 4u32 => Err(DecodeError::new( + "remote write v1 native histogram ingestion is unsupported; use remote write v2", + )), _ => prost::encoding::skip_field(wire_type, tag, buf, Default::default()), } } @@ -170,6 +183,7 @@ pub struct PromWriteRequest<'a> { impl<'a> Clear for PromWriteRequest<'a> { fn clear(&mut self) { + self.series.clear(); self.table_data.clear(); } } @@ -186,6 +200,7 @@ impl<'a> PromWriteRequest<'a> { processor: &mut PromSeriesProcessor, ) -> Result<(), DecodeError> { const STRUCT_NAME: &str = "PromWriteRequest"; + self.clear(); self.table_data.set_raw_data(buf); let mut offset = 0; while offset < self.table_data.raw_data.len() { @@ -247,8 +262,7 @@ impl<'a> PromWriteRequest<'a> { } if decoded_timeseries { - self.series.labels.clear(); - self.series.samples.clear(); + self.series.clear(); } } @@ -371,7 +385,7 @@ impl PromSeriesProcessor { mod tests { use std::collections::HashMap; - use api::prom_store::remote::WriteRequest; + use api::prom_store::remote::{Histogram, Label, Sample, TimeSeries, WriteRequest}; use api::v1::{Row, RowInsertRequests, Rows}; use bytes::Bytes; use prost::Message; @@ -453,6 +467,130 @@ mod tests { } } + #[test] + fn test_decode_rejects_remote_write_v1_native_histograms() { + for samples in [ + Vec::new(), + vec![Sample { + value: 1.0, + timestamp: 1000, + }], + ] { + let request = WriteRequest { + timeseries: vec![TimeSeries { + samples, + histograms: vec![Histogram::default()], + ..Default::default() + }], + ..Default::default() + }; + let mut processor = PromSeriesProcessor::default_processor(); + let mut write_request = PromWriteRequest::default(); + let error = write_request + .decode( + request.encode_to_vec(), + PromValidationMode::Strict, + &mut processor, + ) + .unwrap_err(); + + assert!(error.to_string().contains( + "remote write v1 native histogram ingestion is unsupported; use remote write v2" + )); + assert_eq!( + write_request.as_row_insert_requests().ref_all_req().count(), + 0 + ); + } + } + + #[test] + fn test_decode_clears_state_after_error() { + let label = |name: &str, value: &str| Label { + name: name.to_string(), + value: value.to_string(), + }; + let sample = Sample { + value: 1.0, + timestamp: 1000, + }; + let failed_request = WriteRequest { + timeseries: vec![ + TimeSeries { + labels: vec![ + label("__name__", "stale_metric"), + label("stale_label", "stale_value"), + ], + samples: vec![sample], + ..Default::default() + }, + TimeSeries { + labels: vec![ + label("__name__", "rejected_metric"), + label("rejected_label", "rejected_value"), + label("__schema__", "rejected_schema"), + label("x_greptime_physical_table", "rejected_physical_table"), + ], + samples: vec![sample], + histograms: vec![Histogram::default()], + ..Default::default() + }, + ], + ..Default::default() + }; + let successful_request = WriteRequest { + timeseries: vec![TimeSeries { + labels: vec![ + label("__name__", "fresh_metric"), + label("fresh_label", "fresh_value"), + ], + samples: vec![sample], + ..Default::default() + }], + ..Default::default() + }; + let mut processor = PromSeriesProcessor::default_processor(); + let mut write_request = PromWriteRequest::default(); + + write_request + .decode( + failed_request.encode_to_vec(), + PromValidationMode::Strict, + &mut processor, + ) + .unwrap_err(); + write_request + .decode( + successful_request.encode_to_vec(), + PromValidationMode::Strict, + &mut processor, + ) + .unwrap(); + + assert_eq!(write_request.table_data.tables.len(), 1); + let (prom_ctx, tables) = write_request.table_data.tables.iter().next().unwrap(); + assert_eq!(prom_ctx.schema, None); + assert_eq!(prom_ctx.physical_table, None); + assert_eq!(tables.len(), 1); + assert!(tables.contains_key("fresh_metric")); + + let requests = write_request + .as_row_insert_requests() + .all_req() + .collect::>(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].table_name, "fresh_metric"); + let rows = requests[0].rows.as_ref().unwrap(); + assert_eq!(rows.rows.len(), 1); + assert_eq!( + rows.schema + .iter() + .map(|column| column.column_name.as_str()) + .collect::>(), + vec![greptime_timestamp(), greptime_value(), "fresh_label"] + ); + } + #[test] fn test_decode_string_strict_mode_valid_utf8() { let valid_utf8 = Bytes::from("hello world"); diff --git a/src/servers/src/prom_remote_write/v2.rs b/src/servers/src/prom_remote_write/v2.rs index 229fac0166..cfa772f691 100644 --- a/src/servers/src/prom_remote_write/v2.rs +++ b/src/servers/src/prom_remote_write/v2.rs @@ -45,7 +45,6 @@ use crate::row_writer::{self, TableData}; type PromTags = Vec<(String, String)>; type ResolvedSeriesLabels = (PromCtx, String, PromTags); -const MIN_REMOTE_WRITE_V2_SCHEMA: i32 = -4; const MAX_REMOTE_WRITE_V2_SCHEMA: i32 = 8; const MAX_REDUCIBLE_REMOTE_WRITE_V2_SCHEMA: i32 = 52; @@ -346,7 +345,7 @@ fn native_histogram_struct_value(histogram: &Histogram) -> Result { } fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> Result<()> { - validate_native_histogram_schema(histogram.schema)?; + let exponential_overflow_index = validate_native_histogram_schema(histogram.schema)?; validate_native_histogram_custom_values(histogram)?; if histogram.schema == CUSTOM_BUCKETS_SCHEMA { @@ -379,8 +378,11 @@ fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> histogram.negative_deltas.len(), ) }; - let custom_max_index = if histogram.schema == CUSTOM_BUCKETS_SCHEMA { - Some( + let bucket_index_range = if let Some(overflow_index) = exponential_overflow_index { + (i32::MIN, overflow_index) + } else { + ( + 0, i32::try_from(histogram.custom_values.len()).ok().context( error::InvalidPromRemoteRequestSnafu { msg: "remote write v2 custom native histogram has too many custom_values" @@ -388,30 +390,30 @@ fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> }, )?, ) - } else { - None }; validate_native_histogram_spans( "positive", &histogram.positive_spans, positive_buckets, - custom_max_index, + bucket_index_range, )?; validate_native_histogram_spans( "negative", &histogram.negative_spans, negative_buckets, - None, + bucket_index_range, )?; Ok(()) } -fn validate_native_histogram_schema(schema: i32) -> Result<()> { - if schema == CUSTOM_BUCKETS_SCHEMA - || (MIN_REMOTE_WRITE_V2_SCHEMA..=MAX_REMOTE_WRITE_V2_SCHEMA).contains(&schema) - { - return Ok(()); +fn validate_native_histogram_schema(schema: i32) -> Result> { + if schema == CUSTOM_BUCKETS_SCHEMA { + return Ok(None); + } + + if let Some(overflow_index) = exponential_overflow_bucket_index(schema) { + return Ok(Some(overflow_index)); } if (MAX_REMOTE_WRITE_V2_SCHEMA + 1..=MAX_REDUCIBLE_REMOTE_WRITE_V2_SCHEMA).contains(&schema) { @@ -467,7 +469,7 @@ fn validate_native_histogram_spans( name: &str, spans: &[BucketSpan], bucket_count: usize, - custom_max_index: Option, + bucket_index_range: (i32, i32), ) -> Result<()> { let span_len = spans .iter() @@ -487,7 +489,7 @@ fn validate_native_histogram_spans( let mut current_index = 0i32; for (span_index, span) in spans.iter().enumerate() { ensure!( - span.offset >= 0 || (span_index == 0 && custom_max_index.is_none()), + span.offset >= 0 || (span_index == 0 && bucket_index_range.0 == i32::MIN), error::InvalidPromRemoteRequestSnafu { msg: format!( "remote write v2 native histogram {name} span {} has negative offset {}", @@ -509,16 +511,14 @@ fn validate_native_histogram_spans( }; for _ in 0..span.length { - if let Some(max_index) = custom_max_index { - ensure!( - (0..=max_index).contains(¤t_index), - error::InvalidPromRemoteRequestSnafu { - msg: format!( - "remote write v2 custom native histogram {name} bucket index {current_index} is out of range" - ), - } - ); - } + ensure!( + (bucket_index_range.0..=bucket_index_range.1).contains(¤t_index), + error::InvalidPromRemoteRequestSnafu { + msg: format!( + "remote write v2 native histogram {name} bucket index {current_index} is out of range" + ), + } + ); current_index = current_index .checked_add(1) @@ -1473,6 +1473,82 @@ mod tests { ); } + #[test] + fn test_into_context_req_rejects_metric_kind_conflict_across_label_sets() { + let request = Request { + symbols: vec![ + "".to_string(), + METRIC_NAME_LABEL.to_string(), + "metric".to_string(), + "job".to_string(), + "api".to_string(), + "worker".to_string(), + ], + timeseries: vec![ + TimeSeries { + labels_refs: vec![1, 2, 3, 4], + samples: vec![Sample { + value: 1.0, + timestamp: 1000, + start_timestamp: 0, + }], + ..Default::default() + }, + TimeSeries { + labels_refs: vec![1, 2, 3, 5], + histograms: vec![Histogram::default()], + ..Default::default() + }, + ], + }; + + assert_invalid( + "same metric kind conflict across label sets", + request, + "contains both samples and native histograms", + ); + } + + #[test] + fn test_into_context_req_validates_exponential_overflow_bucket_index() { + for schema in [-4, 0, 8] { + let max_index = exponential_overflow_bucket_index(schema).unwrap(); + for positive in [true, false] { + let mut histogram = Histogram { + schema, + count: Some(Count::CountInt(1)), + ..Default::default() + }; + if positive { + histogram.positive_spans = vec![BucketSpan { + offset: max_index, + length: 1, + }]; + histogram.positive_deltas = vec![1]; + } else { + histogram.negative_spans = vec![BucketSpan { + offset: max_index, + length: 1, + }]; + histogram.negative_deltas = vec![1]; + } + into_write_requests(request_with_histogram(histogram.clone())).unwrap(); + + let beyond = max_index + 1; + if positive { + histogram.positive_spans[0].offset = beyond; + } else { + histogram.negative_spans[0].offset = beyond; + } + assert_invalid( + "exponential overflow bucket index", + request_with_histogram(histogram), + &format!("bucket index {beyond} is out of range"), + ); + } + } + } + #[test] fn test_into_context_req_converts_histograms_and_ignores_exemplars() { let request = Request { diff --git a/src/servers/src/prom_store.rs b/src/servers/src/prom_store.rs index 9988f7051f..772689f0e0 100644 --- a/src/servers/src/prom_store.rs +++ b/src/servers/src/prom_store.rs @@ -381,6 +381,8 @@ fn recordbatch_to_timeseries(table: &str, recordbatch: RecordBatch) -> Result