From 5e9bb3c689234b3c0a88fd3ece43987548345df8 Mon Sep 17 00:00:00 2001 From: shuiyisong <113876041+shuiyisong@users.noreply.github.com> Date: Tue, 27 Oct 2026 03:09:13 +0000 Subject: [PATCH] feat: support native histograms in sampled remote read (#9971) * feat: support native histograms in sampled remote read Signed-off-by: shuiyisong * fix: align native histogram ingestion with remote read validation Signed-off-by: shuiyisong --------- Signed-off-by: shuiyisong --- docs/rfcs/2026-08-04-native-histograms.md | 76 +- src/common/query/src/native_histogram.rs | 36 +- .../query/src/native_histogram/encoding.rs | 1101 +++++++++++++++-- src/frontend/src/instance/prom_store.rs | 22 + src/servers/Cargo.toml | 1 + src/servers/src/prom_store.rs | 342 ++++- src/servers/tests/http/prom_store_test.rs | 58 +- tests-integration/src/otlp.rs | 80 ++ tests-integration/src/prom_store.rs | 323 +++++ tests-integration/tests/http.rs | 145 ++- tests/compatibility/README.md | 8 + .../native_histogram_remote_read/README.md | 184 +++ .../native_histogram_remote_read/response.pb | Bin 0 -> 420 bytes 13 files changed, 2270 insertions(+), 106 deletions(-) create mode 100644 tests/compatibility/wire/native_histogram_remote_read/README.md create mode 100644 tests/compatibility/wire/native_histogram_remote_read/response.pb diff --git a/docs/rfcs/2026-08-04-native-histograms.md b/docs/rfcs/2026-08-04-native-histograms.md index 6b0a9ebb757..cc31a8df5e4 100644 --- a/docs/rfcs/2026-08-04-native-histograms.md +++ b/docs/rfcs/2026-08-04-native-histograms.md @@ -16,8 +16,8 @@ Struct before persistence and are queried only as native histograms. Prometheus Remote Write 2.0 native histograms and cumulative OTLP/HTTP exponential histograms are enabled by default. Remote Write 1.0 histogram payloads are rejected instead of being acknowledged and dropped. -Native-histogram Remote Read is deferred; the existing Remote Read path -continues to return scalar samples only. +Remote Read returns scalar samples and native histograms through `SAMPLES`, +including exact stored integer counts. Streamed responses remain deferred. # Goals @@ -25,13 +25,15 @@ continues to return scalar samples only. 2. Keep one stable Prometheus-compatible persisted representation across Prometheus and OTLP ingestion. 3. Match Prometheus query behavior where it is observable and practical. -4. State intentional limitations explicitly so incomplete behavior is not +4. Return stored native histograms through sampled Remote Read without rounding + integer counts or losing supported histogram fields. +5. State intentional limitations explicitly so incomplete behavior is not mistaken for support. # Non-Goals - Supporting native histograms in Remote Write 1.0. -- Returning native histograms through Prometheus Remote Read. +- Streaming Remote Read responses (`STREAMED_XOR_CHUNKS`). - Persisting Remote Write metric metadata. - Persisting exemplars. - Converting OTLP delta values to cumulative values during ingestion. Supported @@ -107,19 +109,57 @@ beyond it. The overflow index is `(1024 << schema) + 1` for non-negative schemas and `(1024 >> -schema) + 1` for negative schemas. Inputs beyond that index are rejected rather than clamped or merged into infinity. +Ingestion and sampled Remote Read share histogram layout validation. An initial +exponential span offset of `i32::MIN` is rejected on either side, including for +empty spans; other negative initial offsets remain supported. For custom +histograms with `N` bounds, bucket index `N` is the overflow bucket, and empty +spans may end at `N + 1`. Span extents beyond that boundary are rejected. + +This tightens ingestion acceptance: these layouts were previously accepted on +write but rejected by Remote Read. A validation failure rejects the entire +Remote Write 2.0 request with HTTP 400 and all written-count headers set to zero, +before any samples or histograms are persisted. Healthy samples in the same +request are also rejected; senders that do not retry HTTP 400 can drop them. + Exemplars and declared metric metadata are accepted on the wire but are not persisted by this feature. ## Remote Read -Native-histogram Remote Read is deferred. GreptimeDB's existing Remote Read path -continues to negotiate `SAMPLES` responses and serialize scalar samples only. +`/v1/prometheus/read` returns scalar samples and integer or float native +histograms through the existing Snappy-compressed `SAMPLES` protobuf response. +Omitting accepted types, requesting `SAMPLES`, or requesting +`[STREAMED_XOR_CHUNKS, SAMPLES]` selects this format. Streaming-only requests +remain unsupported; the response is still buffered as a whole. -A follow-up can map the stored Struct into `TimeSeries.histograms`. It must -preserve integer counts without passing through `f64`, retain the stored -histogram fields, and decide sampled versus streamed response coverage. The -sampled protobuf has no start-timestamp field, so that omission must remain -explicit if sampled responses are implemented. +One request can query both scalar and histogram metrics. A histogram series +can contain integer and float histogram samples, but the one-sample-kind-per- +metric-table rule remains unchanged. The selected field must be `Float64` or +the canonical native-histogram Struct, regardless of its stored column name. + +Integer counts remain exact within the existing nonnegative `i64` storage +range, including above `2^53`. Stored absolute integer buckets become signed +wire deltas, retaining their sequence across span boundaries. Float bucket +counts remain absolute. Schema, spans, custom bounds, zero threshold, sum +(including stale-NaN bits), reset hint, and millisecond timestamps are retained. +The sampled protobuf has no `start_timestamp` field, so that value remains in +storage but is omitted from the response. Raw-delta OTLP histograms retain +their temporality labels and gauge hints; Remote Read does not accumulate them +or reconstruct the original OTLP point. + +Malformed or wire-incompatible non-null stored histograms fail the entire +`ReadRequest` with HTTP 500, including healthy queries in that request. Read +validation remains necessary for older rows and typed inserts that bypass the +histogram ingestion encoder. Incompatible stored rows are not repaired by the +stricter ingestion checks. Null histograms remain absent samples. + +Custom-bucket interoperability is exercised against Prometheus revision +`e471c7708c6b0b13e2050222bbd5a907b36f4113`, including `AtHistogram` and +`AtFloatHistogram` on integer custom buckets. This is a tested revision, not +a minimum-release claim; older readers can lose custom bounds during integer- +to-float conversion. Integer payloads are not downgraded for those readers. +The [wire fixture and iterator check](../../tests/compatibility/wire/native_histogram_remote_read/README.md) +record the independent expectations and reproduction commands. ## OTLP @@ -340,16 +380,20 @@ partial-success behavior, kind exclusivity, stale-marker selector semantics, synthetic-zero rates, incompatible empty layouts, overflow-safe averages, layout-sensitive equality, infinite custom midpoints, and exponential overflow indices for schemas `-4`, `0`, and `8`. -Behavioral coverage also verifies frontend PromQL warning and info responses. +Behavioral coverage also verifies frontend PromQL warning and info responses, +sampled Remote Read before and after flush in standalone and distributed +setups, exact large integers, stale-marker bits, timestamp units, and OTLP +normalization. Frozen upstream protobuf fixtures guard the wire encoder. This work does not change GreptimeDB's persisted Struct or protobuf -dependencies. It adds the disabled-by-default public OTLP gate; existing -native-histogram data remains readable. +dependencies. Remote Read uses the existing endpoint and negotiation without +a new configuration option; existing native-histogram data remains readable. # Future Work -- Native-histogram Remote Read, including exact integer round-trips and streamed - chunks with start timestamps. +- TODO: Add native histogram and float histogram chunks to + `STREAMED_XOR_CHUNKS`, with checksummed framing and bounded buffering. Decide + separately whether to include start-timestamp-bearing chunk encodings. - Raw-delta OTLP exponential-histogram storage and temporality-aware native histogram queries, as described above; no delta-to-cumulative conversion. - Persistent Remote Write metadata and accurate help/unit updates. diff --git a/src/common/query/src/native_histogram.rs b/src/common/query/src/native_histogram.rs index d8a559a2dc5..38a7895c27a 100644 --- a/src/common/query/src/native_histogram.rs +++ b/src/common/query/src/native_histogram.rs @@ -36,7 +36,10 @@ use datafusion::arrow::datatypes::{ use datafusion_common::{DataFusionError, Result as DfResult}; use datatypes::data_type::{ConcreteDataType, DataType}; use datatypes::types::{StructField, StructType}; -pub use encoding::{NativeHistogramError, encode_native_histogram, native_histogram_column_schema}; +pub use encoding::{ + NativeHistogramError, decode_native_histogram_for_remote_read, encode_native_histogram, + native_histogram_column_schema, +}; use once_cell::sync::Lazy; use crate::prelude::greptime_native_histogram; @@ -136,6 +139,37 @@ pub fn is_native_histogram_value_schema(name: &str, data_type: &ConcreteDataType name == greptime_native_histogram() && is_native_histogram_value_type(data_type) } +#[cfg(any(test, feature = "testing"))] +pub mod testing { + use api::helper::pb_value_to_value_ref; + use api::v1::{Value, value::ValueData}; + use datafusion::arrow::array::{AsArray, StructArray}; + use datatypes::vectors::Helper; + + use crate::native_histogram::{native_histogram_column_schema, native_histogram_value_type}; + + /// Builds stored arrays through the same protobuf-to-vector path as ingestion. + pub fn histogram_array(values: &[Option]) -> StructArray { + let schema = native_histogram_column_schema().unwrap(); + let rows = values + .iter() + .map(|value_data| { + datatypes::value::Value::from(pb_value_to_value_ref( + &Value { + value_data: value_data.clone(), + }, + schema.datatype_extension.as_ref(), + )) + }) + .collect::>(); + Helper::try_from_row_into_vector(&rows, native_histogram_value_type()) + .unwrap() + .to_arrow_array() + .as_struct() + .clone() + } +} + /// Schema identifier for native histograms with explicit custom bucket bounds. pub const CUSTOM_BUCKETS_SCHEMA: i32 = -53; const MIN_EXPONENTIAL_SCHEMA: i32 = -4; diff --git a/src/common/query/src/native_histogram/encoding.rs b/src/common/query/src/native_histogram/encoding.rs index be7b35ec454..d5e1e545f84 100644 --- a/src/common/query/src/native_histogram/encoding.rs +++ b/src/common/query/src/native_histogram/encoding.rs @@ -15,12 +15,15 @@ use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount}; use api::greptime_proto::io::prometheus::write::v2::{BucketSpan, Histogram}; use api::helper::ColumnDataTypeWrapper; +use api::prom_store::remote; use api::v1::value::ValueData; use api::v1::{ColumnSchema, ListValue, SemanticType, Value}; +use datafusion::arrow::array::{Array, StructArray}; +use datafusion::arrow::datatypes::{Float64Type, Int32Type, Int64Type, TimestampMillisecondType}; use snafu::{Snafu, ensure}; use crate::native_histogram::{ - CUSTOM_BUCKETS_SCHEMA, MAX_EXPONENTIAL_SCHEMA, NATIVE_HISTOGRAM_FIELD_NAMES, + self, CUSTOM_BUCKETS_SCHEMA, MAX_EXPONENTIAL_SCHEMA, NATIVE_HISTOGRAM_FIELD_NAMES, exponential_overflow_bucket_index, native_histogram_value_type, }; use crate::prelude::greptime_native_histogram; @@ -78,7 +81,6 @@ pub fn encode_native_histogram(histogram: &Histogram) -> Result { ]); if uses_float_counts { - validate_float_native_histogram_counts(histogram)?; let count = match histogram.count.as_ref() { Some(Count::CountFloat(count)) => *count, _ => 0.0, @@ -87,6 +89,12 @@ pub fn encode_native_histogram(histogram: &Histogram) -> Result { Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count, _ => 0.0, }; + validate_float_native_histogram_counts( + count, + zero_count, + &histogram.positive_counts, + &histogram.negative_counts, + )?; items.extend([ null_pb_value(), null_pb_value(), @@ -108,7 +116,13 @@ pub fn encode_native_histogram(histogram: &Histogram) -> Result { }; let positive_buckets = bucket_counts_from_deltas(&histogram.positive_deltas)?; let negative_buckets = bucket_counts_from_deltas(&histogram.negative_deltas)?; - validate_integer_native_histogram_counts(histogram, &positive_buckets, &negative_buckets)?; + validate_integer_native_histogram_counts( + count, + zero_count, + histogram.sum, + &positive_buckets, + &negative_buckets, + )?; let count = i64::try_from(count).map_err(|_| NativeHistogramError { message: format!("native histogram integer count {count} overflows i64"), })?; @@ -130,27 +144,215 @@ pub fn encode_native_histogram(histogram: &Histogram) -> Result { Ok(ValueData::StructValue(api::v1::StructValue { items })) } -fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> Result<()> { - let exponential_overflow_index = validate_native_histogram_schema(histogram.schema)?; - validate_native_histogram_custom_values(histogram)?; +/// Decodes a stored histogram without rounding integer counts through `f64`. +/// +/// Null structs are absent samples. Malformed or sampled-wire-incompatible +/// values are errors. The sampled format has no field for `start_timestamp`. +pub fn decode_native_histogram_for_remote_read( + array: &StructArray, + row: usize, + timestamp: i64, +) -> Result> { + use crate::native_histogram::{ + COUNT_F64_FIELD, COUNT_I64_FIELD, CUSTOM_VALUES_FIELD, NEGATIVE_BUCKETS_F64_FIELD, + NEGATIVE_BUCKETS_I64_FIELD, NEGATIVE_SPAN_LENGTHS_FIELD, NEGATIVE_SPAN_OFFSETS_FIELD, + POSITIVE_BUCKETS_F64_FIELD, POSITIVE_BUCKETS_I64_FIELD, POSITIVE_SPAN_LENGTHS_FIELD, + POSITIVE_SPAN_OFFSETS_FIELD, RESET_HINT_FIELD, SCHEMA_FIELD, START_TIMESTAMP_FIELD, + SUM_FIELD, ZERO_COUNT_F64_FIELD, ZERO_COUNT_I64_FIELD, ZERO_THRESHOLD_FIELD, + }; - if histogram.schema == CUSTOM_BUCKETS_SCHEMA { - ensure!( - histogram.zero_threshold == 0.0 && native_histogram_zero_count_is_zero(histogram), - NativeHistogramSnafu { - message: "custom native histogram must not use a zero bucket" - } - ); - ensure!( - histogram.negative_spans.is_empty() - && histogram.negative_deltas.is_empty() - && histogram.negative_counts.is_empty(), - NativeHistogramSnafu { - message: "custom native histogram must not use negative buckets" - } - ); + ensure!( + row < array.len(), + NativeHistogramSnafu { + message: "native histogram row is out of bounds" + } + ); + if array.is_null(row) { + return Ok(None); } + let invalid_field = |error| NativeHistogramError { + message: format!("invalid stored native histogram: {error}"), + }; + let read_spans = |name, offsets, lengths| -> Result> { + let offsets = native_histogram::list_values::(array, offsets, row) + .map_err(invalid_field)?; + let lengths = native_histogram::list_values::(array, lengths, row) + .map_err(invalid_field)?; + native_histogram::read_spans(offsets, lengths, name) + .map_err(invalid_field)? + .into_iter() + .map(|span| { + Ok(remote::BucketSpan { + offset: span.offset, + length: u32::try_from(span.length).map_err(|_| NativeHistogramError { + message: format!("native histogram {name} span length is negative"), + })?, + }) + }) + .collect() + }; + let mut histogram = remote::Histogram { + schema: native_histogram::required_primitive::(array, SCHEMA_FIELD, row) + .map_err(invalid_field)?, + zero_threshold: native_histogram::required_primitive::( + array, + ZERO_THRESHOLD_FIELD, + row, + ) + .map_err(invalid_field)?, + sum: native_histogram::required_primitive::(array, SUM_FIELD, row) + .map_err(invalid_field)?, + reset_hint: native_histogram::required_primitive::(array, RESET_HINT_FIELD, row) + .map_err(invalid_field)?, + custom_values: native_histogram::list_values::( + array, + CUSTOM_VALUES_FIELD, + row, + ) + .map_err(invalid_field)?, + positive_spans: read_spans( + "positive", + POSITIVE_SPAN_OFFSETS_FIELD, + POSITIVE_SPAN_LENGTHS_FIELD, + )?, + negative_spans: read_spans( + "negative", + NEGATIVE_SPAN_OFFSETS_FIELD, + NEGATIVE_SPAN_LENGTHS_FIELD, + )?, + timestamp, + ..Default::default() + }; + // Validate the stored child even though the sampled wire format cannot carry it. + native_histogram::optional_primitive::( + array, + START_TIMESTAMP_FIELD, + row, + ) + .map_err(invalid_field)?; + let count_i64 = native_histogram::optional_primitive::(array, COUNT_I64_FIELD, row) + .map_err(invalid_field)?; + let count_f64 = + native_histogram::optional_primitive::(array, COUNT_F64_FIELD, row) + .map_err(invalid_field)?; + let zero_i64 = + native_histogram::optional_primitive::(array, ZERO_COUNT_I64_FIELD, row) + .map_err(invalid_field)?; + let zero_f64 = + native_histogram::optional_primitive::(array, ZERO_COUNT_F64_FIELD, row) + .map_err(invalid_field)?; + let positive_i64 = + native_histogram::list_values::(array, POSITIVE_BUCKETS_I64_FIELD, row) + .map_err(invalid_field)?; + let negative_i64 = + native_histogram::list_values::(array, NEGATIVE_BUCKETS_I64_FIELD, row) + .map_err(invalid_field)?; + let positive_f64 = + native_histogram::list_values::(array, POSITIVE_BUCKETS_F64_FIELD, row) + .map_err(invalid_field)?; + let negative_f64 = + native_histogram::list_values::(array, NEGATIVE_BUCKETS_F64_FIELD, row) + .map_err(invalid_field)?; + let (zero_count_is_zero, positive_len, negative_len) = match (count_i64, count_f64) { + (Some(count), None) => { + ensure!( + zero_f64.is_none() && positive_f64.is_empty() && negative_f64.is_empty(), + NativeHistogramSnafu { + message: "integer native histogram has populated float fields" + } + ); + let count = u64::try_from(count).map_err(|_| NativeHistogramError { + message: "native histogram integer count is negative".to_string(), + })?; + let zero_count = + u64::try_from(zero_i64.unwrap_or_default()).map_err(|_| NativeHistogramError { + message: "native histogram integer zero_count is negative".to_string(), + })?; + validate_integer_native_histogram_counts( + count, + zero_count, + histogram.sum, + &positive_i64, + &negative_i64, + )?; + histogram.count = Some(remote::histogram::Count::CountInt(count)); + histogram.zero_count = Some(remote::histogram::ZeroCount::ZeroCountInt(zero_count)); + histogram.positive_deltas = bucket_deltas_from_counts(&positive_i64)?; + histogram.negative_deltas = bucket_deltas_from_counts(&negative_i64)?; + (zero_count == 0, positive_i64.len(), negative_i64.len()) + } + (None, Some(count)) => { + ensure!( + zero_i64.is_none() && positive_i64.is_empty() && negative_i64.is_empty(), + NativeHistogramSnafu { + message: "float native histogram has populated integer fields" + } + ); + let zero_count = zero_f64.unwrap_or_default(); + validate_float_native_histogram_counts( + count, + zero_count, + &positive_f64, + &negative_f64, + )?; + histogram.count = Some(remote::histogram::Count::CountFloat(count)); + histogram.zero_count = Some(remote::histogram::ZeroCount::ZeroCountFloat(zero_count)); + histogram.positive_counts = positive_f64; + histogram.negative_counts = negative_f64; + ( + zero_count == 0.0, + histogram.positive_counts.len(), + histogram.negative_counts.len(), + ) + } + _ => { + return NativeHistogramSnafu { + message: "stored native histogram must have exactly one populated count family", + } + .fail(); + } + }; + validate_native_histogram_layout( + histogram.schema, + histogram.zero_threshold, + zero_count_is_zero, + &histogram.custom_values, + ( + histogram + .positive_spans + .iter() + .map(|s| (s.offset, s.length)), + positive_len, + ), + ( + histogram + .negative_spans + .iter() + .map(|s| (s.offset, s.length)), + negative_len, + ), + )?; + Ok(Some(histogram)) +} + +fn bucket_deltas_from_counts(counts: &[i64]) -> Result> { + let mut previous = 0i64; + counts + .iter() + .map(|&count| { + let delta = count + .checked_sub(previous) + .ok_or_else(|| NativeHistogramError { + message: "native histogram bucket delta overflows i64".to_string(), + })?; + previous = count; + Ok(delta) + }) + .collect() +} + +fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> Result<()> { let (positive_buckets, negative_buckets) = if uses_float_counts { ( histogram.positive_counts.len(), @@ -162,28 +364,82 @@ fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> histogram.negative_deltas.len(), ) }; + validate_native_histogram_layout( + histogram.schema, + histogram.zero_threshold, + native_histogram_zero_count_is_zero(histogram), + &histogram.custom_values, + ( + histogram + .positive_spans + .iter() + .map(|s| (s.offset, s.length)), + positive_buckets, + ), + ( + histogram + .negative_spans + .iter() + .map(|s| (s.offset, s.length)), + negative_buckets, + ), + ) +} + +fn validate_native_histogram_layout( + schema: i32, + zero_threshold: f64, + zero_count_is_zero: bool, + custom_values: &[f64], + positive: (impl ExactSizeIterator + Clone, usize), + negative: (impl ExactSizeIterator + Clone, usize), +) -> Result<()> { + let exponential_overflow_index = validate_native_histogram_schema(schema)?; + validate_native_histogram_custom_values(schema, custom_values)?; + if schema == CUSTOM_BUCKETS_SCHEMA { + ensure!( + zero_threshold == 0.0 && zero_count_is_zero, + NativeHistogramSnafu { + message: "custom native histogram must not use a zero bucket" + } + ); + ensure!( + negative.0.len() == 0 && negative.1 == 0, + NativeHistogramSnafu { + message: "custom native histogram must not use negative buckets" + } + ); + } + 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()).map_err(|_| NativeHistogramError { + i32::try_from(custom_values.len()).map_err(|_| NativeHistogramError { message: "custom native histogram has too many custom_values".to_string(), })?, ) }; - validate_native_histogram_spans( - "positive", - &histogram.positive_spans, - positive_buckets, - bucket_index_range, - )?; - validate_native_histogram_spans( - "negative", - &histogram.negative_spans, - negative_buckets, - bucket_index_range, - )?; + let positive_extent = + validate_native_histogram_spans("positive", positive.0, positive.1, bucket_index_range)?; + validate_native_histogram_spans("negative", negative.0, negative.1, bucket_index_range)?; + + if schema == CUSTOM_BUCKETS_SCHEMA { + let max_extent = + bucket_index_range + .1 + .checked_add(1) + .ok_or_else(|| NativeHistogramError { + message: "custom native histogram extent overflows i32".to_string(), + })?; + ensure!( + positive_extent <= max_extent, + NativeHistogramSnafu { + message: "custom native histogram span extent exceeds custom_values plus the overflow bucket" + } + ); + } Ok(()) } @@ -208,10 +464,10 @@ fn validate_native_histogram_schema(schema: i32) -> Result> { } } -fn validate_native_histogram_custom_values(histogram: &Histogram) -> Result<()> { - if histogram.schema != CUSTOM_BUCKETS_SCHEMA { +fn validate_native_histogram_custom_values(schema: i32, custom_values: &[f64]) -> Result<()> { + if schema != CUSTOM_BUCKETS_SCHEMA { ensure!( - histogram.custom_values.is_empty(), + custom_values.is_empty(), NativeHistogramSnafu { message: "standard native histogram must not use custom_values" } @@ -219,7 +475,7 @@ fn validate_native_histogram_custom_values(histogram: &Histogram) -> Result<()> return Ok(()); } - for value in &histogram.custom_values { + for value in custom_values { ensure!( !value.is_nan() && *value != f64::INFINITY, NativeHistogramSnafu { @@ -227,7 +483,7 @@ fn validate_native_histogram_custom_values(histogram: &Histogram) -> Result<()> } ); } - for values in histogram.custom_values.windows(2) { + for values in custom_values.windows(2) { ensure!( values[0] < values[1], NativeHistogramSnafu { @@ -241,12 +497,12 @@ fn validate_native_histogram_custom_values(histogram: &Histogram) -> Result<()> fn validate_native_histogram_spans( name: &str, - spans: &[BucketSpan], + spans: impl Iterator + Clone, bucket_count: usize, bucket_index_range: (i32, i32), -) -> Result<()> { - let span_len = spans.iter().try_fold(0usize, |sum, span| { - let length = usize::try_from(span.length).map_err(|_| NativeHistogramError { +) -> Result { + let span_len = spans.clone().try_fold(0usize, |sum, (_, length)| { + let length = usize::try_from(length).map_err(|_| NativeHistogramError { message: format!("native histogram {name} span length exceeds usize"), })?; sum.checked_add(length).ok_or_else(|| NativeHistogramError { @@ -263,28 +519,36 @@ fn validate_native_histogram_spans( ); let mut current_index = 0i32; - for (span_index, span) in spans.iter().enumerate() { + for (span_index, (offset, length)) in spans.enumerate() { ensure!( - span.offset >= 0 || (span_index == 0 && bucket_index_range.0 == i32::MIN), + offset >= 0 || (span_index == 0 && bucket_index_range.0 == i32::MIN), NativeHistogramSnafu { message: format!( "native histogram {name} span {} has negative offset {}", span_index + 1, - span.offset + offset + ) + } + ); + ensure!( + span_index != 0 || offset != i32::MIN, + NativeHistogramSnafu { + message: format!( + "native histogram {name} initial span offset i32::MIN is incompatible with Remote Read" ) } ); current_index = if span_index == 0 { - span.offset + offset } else { current_index - .checked_add(span.offset) + .checked_add(offset) .ok_or_else(|| NativeHistogramError { message: format!("native histogram {name} span index overflows i32"), })? }; - for _ in 0..span.length { + for _ in 0..length { ensure!( (bucket_index_range.0..=bucket_index_range.1).contains(¤t_index), NativeHistogramSnafu { @@ -301,14 +565,15 @@ fn validate_native_histogram_spans( } } - Ok(()) + Ok(current_index) } -fn validate_float_native_histogram_counts(histogram: &Histogram) -> Result<()> { - let count = match histogram.count.as_ref() { - Some(Count::CountFloat(count)) => *count, - _ => 0.0, - }; +fn validate_float_native_histogram_counts( + count: f64, + zero_count: f64, + positive_buckets: &[f64], + negative_buckets: &[f64], +) -> Result<()> { ensure!( count >= 0.0 || count.is_nan(), NativeHistogramSnafu { @@ -316,10 +581,6 @@ fn validate_float_native_histogram_counts(histogram: &Histogram) -> Result<()> { } ); - let zero_count = match histogram.zero_count.as_ref() { - Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count, - _ => 0.0, - }; ensure!( zero_count >= 0.0 || zero_count.is_nan(), NativeHistogramSnafu { @@ -328,8 +589,8 @@ fn validate_float_native_histogram_counts(histogram: &Histogram) -> Result<()> { ); for (name, counts) in [ - ("positive", &histogram.positive_counts), - ("negative", &histogram.negative_counts), + ("positive", positive_buckets), + ("negative", negative_buckets), ] { for (index, count) in counts.iter().enumerate() { ensure!( @@ -348,18 +609,12 @@ fn validate_float_native_histogram_counts(histogram: &Histogram) -> Result<()> { } fn validate_integer_native_histogram_counts( - histogram: &Histogram, + count: u64, + zero_count: u64, + sum: f64, positive_buckets: &[i64], negative_buckets: &[i64], ) -> Result<()> { - let count = match histogram.count.as_ref() { - Some(Count::CountInt(count)) => *count, - _ => 0, - }; - let zero_count = match histogram.zero_count.as_ref() { - Some(ZeroCount::ZeroCountInt(zero_count)) => *zero_count, - _ => 0, - }; let bucket_count = positive_buckets .iter() @@ -375,7 +630,7 @@ fn validate_integer_native_histogram_counts( }) })?; ensure!( - if histogram.sum.is_nan() { + if sum.is_nan() { bucket_count <= count } else { bucket_count == count @@ -510,6 +765,710 @@ fn bucket_counts_from_deltas(deltas: &[i64]) -> Result> { mod tests { use super::*; + use std::sync::Arc; + + use datafusion::arrow::array::{ArrayRef, Float64Array, Int32Array}; + use datafusion::arrow::datatypes::Field; + + use crate::native_histogram::testing::histogram_array; + use crate::native_histogram::*; + + fn empty_histogram(float: bool) -> Histogram { + Histogram { + count: Some(if float { + Count::CountFloat(0.0) + } else { + Count::CountInt(0) + }), + zero_count: Some(if float { + ZeroCount::ZeroCountFloat(0.0) + } else { + ZeroCount::ZeroCountInt(0) + }), + ..Default::default() + } + } + + fn stored(histogram: &Histogram) -> StructArray { + histogram_array(&[Some(encode_native_histogram(histogram).unwrap())]) + } + + fn modified(histogram: &Histogram, changes: &[(&str, Value)]) -> StructArray { + let ValueData::StructValue(mut value) = encode_native_histogram(histogram).unwrap() else { + unreachable!() + }; + for (name, replacement) in changes { + let index = NATIVE_HISTOGRAM_FIELD_NAMES + .iter() + .position(|field| field == name) + .unwrap(); + value.items[index] = replacement.clone(); + } + histogram_array(&[Some(ValueData::StructValue(value))]) + } + + fn assert_invalid(array: &StructArray, message: &str) { + let error = decode_native_histogram_for_remote_read(array, 0, 123).unwrap_err(); + assert!( + error.to_string().contains(message), + "{error}: expected {message}" + ); + } + + #[test] + fn native_histogram_remote_read_preserves_integer_counts_and_deltas() { + for count in [(1u64 << 53) + 1, i64::MAX as u64] { + let histogram = Histogram { + count: Some(Count::CountInt(count)), + zero_count: Some(ZeroCount::ZeroCountInt(count - 12)), + positive_spans: vec![ + BucketSpan { + offset: -2, + length: 2, + }, + BucketSpan { + offset: 3, + length: 1, + }, + ], + negative_spans: vec![ + BucketSpan { + offset: -3, + length: 1, + }, + BucketSpan { + offset: 2, + length: 2, + }, + ], + positive_deltas: vec![1, 2, -1], + negative_deltas: vec![1, 2, -1], + sum: 10.0, + reset_hint: 42, + start_timestamp: 123, + ..Default::default() + }; + let array = stored(&histogram); + assert_eq!( + native_histogram::list_values::(&array, POSITIVE_BUCKETS_I64_FIELD, 0) + .unwrap(), + vec![1, 3, 2] + ); + assert_eq!( + native_histogram::list_values::(&array, NEGATIVE_BUCKETS_I64_FIELD, 0) + .unwrap(), + vec![1, 3, 2] + ); + let decoded = decode_native_histogram_for_remote_read(&array, 0, -2) + .unwrap() + .unwrap(); + assert_eq!( + decoded.count, + Some(remote::histogram::Count::CountInt(count)) + ); + assert_eq!( + decoded.zero_count, + Some(remote::histogram::ZeroCount::ZeroCountInt(count - 12)) + ); + assert_eq!(decoded.positive_deltas, vec![1, 2, -1]); + assert_eq!(decoded.negative_deltas, vec![1, 2, -1]); + assert_eq!( + decoded + .positive_spans + .iter() + .map(|s| (s.offset, s.length)) + .collect::>(), + vec![(-2, 2), (3, 1)] + ); + assert_eq!(decoded.reset_hint, 42); + assert_eq!(decoded.timestamp, -2); + } + // A NaN sum permits unobserved NaNs; an incorrect [3, 2] delta vector + // would also pass a total check, so assert the deltas independently. + let histogram = Histogram { + count: Some(Count::CountInt(8)), + sum: f64::NAN, + positive_spans: vec![BucketSpan { + offset: 0, + length: 2, + }], + positive_deltas: vec![3, -1], + ..Default::default() + }; + let decoded = decode_native_histogram_for_remote_read(&stored(&histogram), 0, 0) + .unwrap() + .unwrap(); + assert_eq!(decoded.positive_deltas, vec![3, -1]); + } + + #[test] + fn native_histogram_remote_read_zero_families_and_float_bits() { + for float in [false, true] { + let histogram = empty_histogram(float); + let array = modified( + &histogram, + &[ + ( + if float { + ZERO_COUNT_F64_FIELD + } else { + ZERO_COUNT_I64_FIELD + }, + null_pb_value(), + ), + (POSITIVE_BUCKETS_I64_FIELD, null_pb_value()), + (POSITIVE_BUCKETS_F64_FIELD, null_pb_value()), + ], + ); + let decoded = decode_native_histogram_for_remote_read(&array, 0, 1) + .unwrap() + .unwrap(); + assert_eq!( + matches!( + decoded.count, + Some(remote::histogram::Count::CountFloat(0.0)) + ), + float + ); + assert_eq!( + matches!(decoded.count, Some(remote::histogram::Count::CountInt(0))), + !float + ); + assert_eq!( + matches!( + decoded.zero_count, + Some(remote::histogram::ZeroCount::ZeroCountFloat(0.0)) + ), + float + ); + assert_eq!( + matches!( + decoded.zero_count, + Some(remote::histogram::ZeroCount::ZeroCountInt(0)) + ), + !float + ); + for bits in [0x7ff8_0000_0000_0123, 0x7ff0_0000_0000_0002] { + let histogram = Histogram { + sum: f64::from_bits(bits), + ..histogram.clone() + }; + let decoded = decode_native_histogram_for_remote_read(&stored(&histogram), 0, 1) + .unwrap() + .unwrap(); + assert_eq!(decoded.sum.to_bits(), bits); + } + } + let histogram = Histogram { + count: Some(Count::CountFloat(0.3)), + positive_spans: vec![BucketSpan { + offset: 0, + length: 2, + }], + positive_counts: vec![0.1, 0.2], + ..empty_histogram(true) + }; + let decoded = decode_native_histogram_for_remote_read(&stored(&histogram), 0, 1) + .unwrap() + .unwrap(); + assert_eq!(decoded.positive_counts, vec![0.1, 0.2]); + for value in [f64::INFINITY, f64::from_bits(0x7ff8_0000_0000_0123)] { + let histogram = Histogram { + count: Some(Count::CountFloat(value)), + zero_count: Some(ZeroCount::ZeroCountFloat(value)), + positive_counts: vec![value, value], + ..histogram.clone() + }; + let decoded = decode_native_histogram_for_remote_read(&stored(&histogram), 0, 1) + .unwrap() + .unwrap(); + let Some(remote::histogram::Count::CountFloat(count)) = decoded.count else { + panic!() + }; + assert_eq!(count.to_bits(), value.to_bits()); + assert_eq!(decoded.positive_counts[0].to_bits(), value.to_bits()); + } + assert!( + decode_native_histogram_for_remote_read(&histogram_array(&[None]), 0, 0) + .unwrap() + .is_none() + ); + } + + #[test] + fn native_histogram_remote_read_rejects_invalid_stored_children() { + let integer = empty_histogram(false); + let float = empty_histogram(true); + for field in [ + SCHEMA_FIELD, + ZERO_THRESHOLD_FIELD, + SUM_FIELD, + RESET_HINT_FIELD, + ] { + assert_invalid(&modified(&integer, &[(field, null_pb_value())]), "is null"); + } + for (histogram, changes, message) in [ + ( + &integer, + vec![(COUNT_I64_FIELD, null_pb_value())], + "exactly one", + ), + ( + &integer, + vec![(COUNT_F64_FIELD, pb_value(ValueData::F64Value(0.0)))], + "exactly one", + ), + ( + &integer, + vec![(ZERO_COUNT_F64_FIELD, pb_value(ValueData::F64Value(0.0)))], + "populated float", + ), + ( + &integer, + vec![(POSITIVE_BUCKETS_F64_FIELD, f64_list_value([0.0]))], + "populated float", + ), + ( + &integer, + vec![(NEGATIVE_BUCKETS_F64_FIELD, f64_list_value([0.0]))], + "populated float", + ), + ( + &float, + vec![(ZERO_COUNT_I64_FIELD, pb_value(ValueData::I64Value(0)))], + "populated integer", + ), + ( + &float, + vec![(POSITIVE_BUCKETS_I64_FIELD, i64_list_value([0]))], + "populated integer", + ), + ( + &float, + vec![(NEGATIVE_BUCKETS_I64_FIELD, i64_list_value([0]))], + "populated integer", + ), + ( + &integer, + vec![(COUNT_I64_FIELD, pb_value(ValueData::I64Value(-1)))], + "count is negative", + ), + ( + &integer, + vec![(ZERO_COUNT_I64_FIELD, pb_value(ValueData::I64Value(-1)))], + "zero_count is negative", + ), + ( + &integer, + vec![(POSITIVE_BUCKETS_I64_FIELD, i64_list_value([-1]))], + "count is negative", + ), + ( + &integer, + vec![(NEGATIVE_BUCKETS_I64_FIELD, i64_list_value([-1]))], + "count is negative", + ), + ( + &float, + vec![(COUNT_F64_FIELD, pb_value(ValueData::F64Value(-1.0)))], + "must not be negative", + ), + ( + &float, + vec![(ZERO_COUNT_F64_FIELD, pb_value(ValueData::F64Value(-1.0)))], + "must not be negative", + ), + ( + &float, + vec![(POSITIVE_BUCKETS_F64_FIELD, f64_list_value([-1.0]))], + "must not be negative", + ), + ( + &float, + vec![(NEGATIVE_BUCKETS_F64_FIELD, f64_list_value([-1.0]))], + "must not be negative", + ), + ( + &integer, + vec![ + (COUNT_I64_FIELD, pb_value(ValueData::I64Value(1))), + ( + POSITIVE_BUCKETS_I64_FIELD, + i64_list_value([i64::MAX, i64::MAX, 3]), + ), + ], + "overflows u64", + ), + ( + &integer, + vec![(COUNT_I64_FIELD, pb_value(ValueData::I64Value(1)))], + "observations", + ), + ( + &integer, + vec![(SCHEMA_FIELD, pb_value(ValueData::I32Value(53)))], + "unsupported", + ), + ( + &integer, + vec![(POSITIVE_SPAN_OFFSETS_FIELD, i32_list_value([0]))], + "offsets and lengths mismatch", + ), + ( + &integer, + vec![ + (POSITIVE_SPAN_OFFSETS_FIELD, i32_list_value([0])), + (POSITIVE_SPAN_LENGTHS_FIELD, i32_list_value([-1])), + ], + "negative length", + ), + ( + &integer, + vec![ + (POSITIVE_SPAN_OFFSETS_FIELD, i32_list_value([0])), + (POSITIVE_SPAN_LENGTHS_FIELD, i32_list_value([1])), + ], + "describe 1 buckets", + ), + ( + &integer, + vec![ + (POSITIVE_SPAN_OFFSETS_FIELD, i32_list_value([i32::MAX, 1])), + (POSITIVE_SPAN_LENGTHS_FIELD, i32_list_value([0, 0])), + ], + "overflows i32", + ), + ( + &integer, + vec![( + CUSTOM_VALUES_FIELD, + pb_value(ValueData::ListValue(ListValue { + items: vec![null_pb_value()], + })), + )], + "contains null", + ), + ] { + assert_invalid(&modified(histogram, &changes), message); + } + let array = stored(&integer); + let mut fields = array.fields().to_vec(); + let mut children = array.columns().to_vec(); + fields.remove(0); + children.remove(0); + assert_invalid( + &StructArray::new(fields.into(), children, None), + "missing field schema", + ); + let mut fields = array.fields().to_vec(); + let mut children = array.columns().to_vec(); + children[0] = Arc::new(Float64Array::from(vec![0.0])) as ArrayRef; + fields[0] = Arc::new(Field::new( + SCHEMA_FIELD, + children[0].data_type().clone(), + true, + )); + assert_invalid( + &StructArray::new(fields.into(), children, None), + "invalid type", + ); + // Even inactive children have to retain their declared types. + let mut fields = array.fields().to_vec(); + let mut children = array.columns().to_vec(); + children[14] = Arc::new(Int32Array::from(vec![None])) as ArrayRef; + fields[14] = Arc::new(Field::new( + COUNT_F64_FIELD, + children[14].data_type().clone(), + true, + )); + assert_invalid( + &StructArray::new(fields.into(), children, None), + "invalid type", + ); + } + + #[test] + fn native_histogram_remote_read_schema_rules_match_ingestion() { + for float in [false, true] { + let base = empty_histogram(float); + for bounds in [ + vec![f64::NAN], + vec![f64::INFINITY], + vec![1.0, 1.0], + vec![2.0, 1.0], + ] { + let invalid = Histogram { + schema: CUSTOM_BUCKETS_SCHEMA, + custom_values: bounds.clone(), + ..base.clone() + }; + assert!(encode_native_histogram(&invalid).is_err()); + assert_invalid( + &modified( + &base, + &[ + ( + SCHEMA_FIELD, + pb_value(ValueData::I32Value(CUSTOM_BUCKETS_SCHEMA)), + ), + (CUSTOM_VALUES_FIELD, f64_list_value(bounds)), + ], + ), + "custom_values", + ); + } + let invalid = Histogram { + custom_values: vec![1.0], + ..base.clone() + }; + assert!(encode_native_histogram(&invalid).is_err()); + assert_invalid( + &modified(&base, &[(CUSTOM_VALUES_FIELD, f64_list_value([1.0]))]), + "must not use custom_values", + ); + let custom = Histogram { + schema: CUSTOM_BUCKETS_SCHEMA, + custom_values: vec![-f64::INFINITY, 1.0], + ..base + }; + let decoded = decode_native_histogram_for_remote_read(&stored(&custom), 0, 0) + .unwrap() + .unwrap(); + assert_eq!(decoded.custom_values, vec![-f64::INFINITY, 1.0]); + let invalid = Histogram { + zero_threshold: 1.0, + ..custom.clone() + }; + assert!(encode_native_histogram(&invalid).is_err()); + assert_invalid( + &modified( + &custom, + &[(ZERO_THRESHOLD_FIELD, pb_value(ValueData::F64Value(1.0)))], + ), + "must not use a zero bucket", + ); + let (count_field, zero_field, bucket_field, one, buckets) = if float { + ( + COUNT_F64_FIELD, + ZERO_COUNT_F64_FIELD, + NEGATIVE_BUCKETS_F64_FIELD, + pb_value(ValueData::F64Value(1.0)), + f64_list_value([1.0]), + ) + } else { + ( + COUNT_I64_FIELD, + ZERO_COUNT_I64_FIELD, + NEGATIVE_BUCKETS_I64_FIELD, + pb_value(ValueData::I64Value(1)), + i64_list_value([1]), + ) + }; + let invalid = Histogram { + count: Some(if float { + Count::CountFloat(1.0) + } else { + Count::CountInt(1) + }), + zero_count: Some(if float { + ZeroCount::ZeroCountFloat(1.0) + } else { + ZeroCount::ZeroCountInt(1) + }), + ..custom.clone() + }; + assert!(encode_native_histogram(&invalid).is_err()); + assert_invalid( + &modified( + &custom, + &[(count_field, one.clone()), (zero_field, one.clone())], + ), + "must not use a zero bucket", + ); + for length in [0, 1] { + let mut invalid = custom.clone(); + invalid.negative_spans = vec![BucketSpan { offset: 1, length }]; + let mut changes = vec![ + (NEGATIVE_SPAN_OFFSETS_FIELD, i32_list_value([1])), + (NEGATIVE_SPAN_LENGTHS_FIELD, i32_list_value([length as i32])), + ]; + if length == 1 { + invalid.count = Some(if float { + Count::CountFloat(1.0) + } else { + Count::CountInt(1) + }); + if float { + invalid.negative_counts = vec![1.0]; + } else { + invalid.negative_deltas = vec![1]; + } + changes.extend([(count_field, one.clone()), (bucket_field, buckets.clone())]); + } + assert!(encode_native_histogram(&invalid).is_err()); + assert_invalid( + &modified(&custom, &changes), + "must not use negative buckets", + ); + } + } + } + + #[test] + fn native_histogram_remote_read_span_rules_match_ingestion() { + for float in [false, true] { + let custom = Histogram { + schema: CUSTOM_BUCKETS_SCHEMA, + custom_values: vec![1.0, 2.0], + ..empty_histogram(float) + }; + for custom_values in [vec![], custom.custom_values.clone()] { + let overflow_index = custom_values.len() as i32; + for (spans, accepted) in [ + (vec![(-1, 0)], false), + (vec![(overflow_index + 2, 0)], false), + (vec![(overflow_index, 0), (2, 0)], false), + (vec![(overflow_index + 1, 0)], true), + (vec![(overflow_index, 0), (1, 0)], true), + (vec![(overflow_index, 1)], true), + (vec![(overflow_index + 1, 1)], false), + ] { + let bucket_count = spans.iter().map(|(_, length)| *length).sum::(); + let base = Histogram { + custom_values: custom_values.clone(), + positive_spans: vec![BucketSpan { + offset: 0, + length: bucket_count, + }], + positive_counts: if float { + vec![0.0; bucket_count as usize] + } else { + vec![] + }, + positive_deltas: if float { + vec![] + } else { + vec![0; bucket_count as usize] + }, + ..custom.clone() + }; + let write = Histogram { + positive_spans: spans + .iter() + .map(|&(offset, length)| BucketSpan { offset, length }) + .collect(), + ..base.clone() + }; + assert_eq!(encode_native_histogram(&write).is_ok(), accepted); + let array = modified( + &base, + &[ + ( + POSITIVE_SPAN_OFFSETS_FIELD, + i32_list_value(spans.iter().map(|s| s.0)), + ), + ( + POSITIVE_SPAN_LENGTHS_FIELD, + i32_list_value(spans.iter().map(|s| s.1 as i32)), + ), + ], + ); + assert_eq!( + decode_native_histogram_for_remote_read(&array, 0, 0).is_ok(), + accepted + ); + } + } + for negative in [false, true] { + for length in [0, 1] { + let mut base = empty_histogram(float); + let spans = vec![BucketSpan { offset: 0, length }]; + if negative { + base.negative_spans = spans; + } else { + base.positive_spans = spans; + } + if length == 1 { + if float { + if negative { + base.negative_counts = vec![0.0]; + } else { + base.positive_counts = vec![0.0]; + } + } else if negative { + base.negative_deltas = vec![0]; + } else { + base.positive_deltas = vec![0]; + } + } + for offset in [i32::MIN, i32::MIN + 1, -2, 0] { + let mut write = base.clone(); + let offset_field = if negative { + write.negative_spans[0].offset = offset; + NEGATIVE_SPAN_OFFSETS_FIELD + } else { + write.positive_spans[0].offset = offset; + POSITIVE_SPAN_OFFSETS_FIELD + }; + if offset == i32::MIN { + assert!(encode_native_histogram(&write).is_err()); + assert_invalid( + &modified(&base, &[(offset_field, i32_list_value([offset]))]), + "initial span offset i32::MIN", + ); + } else { + assert!( + decode_native_histogram_for_remote_read(&stored(&write), 0, 0) + .unwrap() + .is_some() + ); + } + } + } + } + let base = empty_histogram(float); + // Out-of-range buckets are different from valid empty end cursors. + for offset in [-1, 3] { + let custom = Histogram { + positive_spans: vec![BucketSpan { + offset: 0, + length: 1, + }], + positive_counts: if float { vec![0.0] } else { vec![] }, + positive_deltas: if float { vec![] } else { vec![0] }, + ..custom.clone() + }; + assert_invalid( + &modified( + &custom, + &[(POSITIVE_SPAN_OFFSETS_FIELD, i32_list_value([offset]))], + ), + if offset < 0 { + "negative offset" + } else { + "out of range" + }, + ); + } + let exponential = Histogram { + positive_spans: vec![BucketSpan { + offset: 0, + length: 1, + }], + positive_counts: if float { vec![0.0] } else { vec![] }, + positive_deltas: if float { vec![] } else { vec![0] }, + ..base + }; + assert_invalid( + &modified( + &exponential, + &[(POSITIVE_SPAN_OFFSETS_FIELD, i32_list_value([i32::MAX]))], + ), + "out of range", + ); + } + } + #[test] fn shared_errors_are_protocol_neutral() { let error = encode_native_histogram(&Histogram { diff --git a/src/frontend/src/instance/prom_store.rs b/src/frontend/src/instance/prom_store.rs index 7238c38f83d..94687238ed0 100644 --- a/src/frontend/src/instance/prom_store.rs +++ b/src/frontend/src/instance/prom_store.rs @@ -850,6 +850,24 @@ mod tests { use super::*; + #[test] + fn test_native_histogram_remote_read_response_negotiation() { + for accepted in [ + vec![], + vec![ResponseType::Samples as i32], + vec![ + ResponseType::StreamedXorChunks as i32, + ResponseType::Samples as i32, + ], + ] { + assert_eq!( + negotiate_response_type(&accepted).unwrap(), + ResponseType::Samples + ); + } + assert!(negotiate_response_type(&[ResponseType::StreamedXorChunks as i32]).is_err()); + } + #[test] fn test_remote_read_label_types() { use datatypes::schema::ColumnSchema; @@ -864,6 +882,10 @@ mod tests { (ConcreteDataType::vector_datatype(3), false), (ConcreteDataType::int64_datatype(), false), (ConcreteDataType::boolean_datatype(), false), + ( + common_query::native_histogram::native_histogram_value_type().clone(), + false, + ), ] { let schema = Schema::new(vec![ ColumnSchema::new( diff --git a/src/servers/Cargo.toml b/src/servers/Cargo.toml index 1f433318938..4383896e1f8 100644 --- a/src/servers/Cargo.toml +++ b/src/servers/Cargo.toml @@ -155,6 +155,7 @@ catalog = { workspace = true, features = ["testing"] } client = { workspace = true, features = ["testing"] } common-base.workspace = true common-meta = { workspace = true, features = ["testing"] } +common-query = { workspace = true, features = ["testing"] } common-test-util.workspace = true criterion = "0.5" json5 = "0.4" diff --git a/src/servers/src/prom_store.rs b/src/servers/src/prom_store.rs index b991cc9c524..d45cabceaf7 100644 --- a/src/servers/src/prom_store.rs +++ b/src/servers/src/prom_store.rs @@ -33,6 +33,7 @@ use arrow::datatypes::{ }; use common_grpc::precision::Precision; use common_memory_manager::MemoryGuard; +use common_query::native_histogram; use common_query::prelude::{greptime_timestamp, greptime_value}; use common_recordbatch::{RecordBatch, RecordBatches}; use common_telemetry::{tracing, warn}; @@ -458,6 +459,7 @@ pub fn recordbatches_to_timeseries( for ts in &mut timeseries { ts.samples.sort_unstable_by_key(|s| s.timestamp); + ts.histograms.sort_unstable_by_key(|h| h.timestamp); } timeseries @@ -531,21 +533,26 @@ fn append_recordbatch_to_timeseries( ), })?; - // TODO: Add native-histogram encoding when Prometheus Remote Read support is prioritized. - // The current path intentionally returns scalar samples only. let field_column = recordbatch .column_by_name(value_column_name) .with_context(|| error::InvalidPromRemoteReadQueryResultSnafu { msg: format!("missing value column '{value_column_name}' in query result"), })?; - let field_column = field_column - .as_primitive_opt::() - .with_context(|| error::InvalidPromRemoteReadQueryResultSnafu { + let scalar_column = field_column.as_primitive_opt::(); + let histogram_column = recordbatch + .schema + .column_schema_by_name(value_column_name) + .filter(|column| native_histogram::is_native_histogram_value_type(&column.data_type)) + .and_then(|_| field_column.as_struct_opt()); + ensure!( + scalar_column.is_some() || histogram_column.is_some(), + error::InvalidPromRemoteReadQueryResultSnafu { msg: format!( - "Expect value column of datatype Float64, actual {:?}", + "Expect value column of datatype Float64 or canonical native histogram Struct, actual {:?}", field_column.data_type() ), - })?; + } + ); let columns = label_columns(&recordbatch, timestamp_column_name, value_column_name)?; let mut previous_timeseries: Option = None; @@ -577,11 +584,25 @@ fn append_recordbatch_to_timeseries( continue; } - let value = field_column.value(row); let timestamp = ts_column.value(row); - let sample = Sample { value, timestamp }; - - timeseries[timeseries_index].samples.push(sample); + if let Some(column) = scalar_column { + timeseries[timeseries_index].samples.push(Sample { + value: column.value(row), + timestamp, + }); + } else if let Some(column) = histogram_column { + let histogram = + native_histogram::decode_native_histogram_for_remote_read(column, row, timestamp) + .map_err(|error| { + error::InvalidPromRemoteReadQueryResultSnafu { + msg: error.to_string(), + } + .build() + })?; + if let Some(histogram) = histogram { + timeseries[timeseries_index].histograms.push(histogram); + } + } } Ok(()) @@ -1540,6 +1561,305 @@ mod tests { ); } + #[test] + fn test_native_histogram_remote_read_wire_fixture() { + use api::greptime_proto::io::prometheus::write::v2::{ + BucketSpan, Histogram, + histogram::{Count, ZeroCount}, + }; + use api::prom_store::remote::{QueryResult, ReadResponse}; + use common_query::native_histogram::testing::histogram_array; + use prost::Message; + + let histograms = vec![ + Histogram { + count: Some(Count::CountInt((1 << 53) + 13)), + zero_count: Some(ZeroCount::ZeroCountInt((1 << 53) + 1)), + sum: 10.0, + schema: 1, + zero_threshold: 0.001, + reset_hint: 2, + start_timestamp: 123, + positive_spans: vec![ + BucketSpan { + offset: -2, + length: 2, + }, + BucketSpan { + offset: 3, + length: 1, + }, + ], + positive_deltas: vec![1, 2, -1], + negative_spans: vec![ + BucketSpan { + offset: -3, + length: 1, + }, + BucketSpan { + offset: 2, + length: 2, + }, + ], + negative_deltas: vec![1, 2, -1], + ..Default::default() + }, + Histogram { + count: Some(Count::CountFloat(6.0)), + zero_count: Some(ZeroCount::ZeroCountFloat(0.5)), + sum: 20.0, + schema: 2, + zero_threshold: 0.002, + reset_hint: 3, + positive_spans: vec![BucketSpan { + offset: 3, + length: 2, + }], + positive_counts: vec![2.0, 3.5], + ..Default::default() + }, + Histogram { + count: Some(Count::CountInt(1)), + sum: 1.0, + schema: -53, + custom_values: vec![1.0, 2.0], + positive_spans: vec![BucketSpan { + offset: 0, + length: 1, + }], + positive_deltas: vec![1], + ..Default::default() + }, + Histogram { + count: Some(Count::CountFloat(3.5)), + sum: 4.0, + schema: -53, + custom_values: vec![1.0, 2.0], + reset_hint: 3, + positive_spans: vec![BucketSpan { + offset: 0, + length: 2, + }], + positive_counts: vec![1.0, 2.5], + ..Default::default() + }, + Histogram { + count: Some(Count::CountInt(0)), + ..Default::default() + }, + Histogram { + count: Some(Count::CountFloat(0.0)), + ..Default::default() + }, + Histogram { + count: Some(Count::CountInt(8)), + sum: f64::from_bits(0x7ff8_0000_0000_0123), + positive_spans: vec![BucketSpan { + offset: 0, + length: 2, + }], + positive_deltas: vec![3, -1], + ..Default::default() + }, + Histogram { + count: Some(Count::CountInt(0)), + sum: f64::from_bits(0x7ff0_0000_0000_0002), + ..Default::default() + }, + Histogram { + count: Some(Count::CountInt(i64::MAX as u64)), + zero_count: Some(ZeroCount::ZeroCountInt(i64::MAX as u64)), + ..Default::default() + }, + ]; + let values = histograms + .iter() + .map(|h| Some(native_histogram::encode_native_histogram(h).unwrap())) + .collect::>(); + let array = histogram_array(&values); + let buckets = array + .column_by_name("positive_buckets_i64") + .unwrap() + .as_list::() + .value(0); + assert_eq!( + buckets + .as_primitive::() + .values() + .as_ref(), + &[1, 3, 2] + ); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + Field::new( + "ts", + ArrowDataType::Timestamp(TimeUnit::Millisecond, None), + false, + ), + Field::new("histogram", array.data_type().clone(), true), + ])); + let schema = Arc::new(Schema::try_from(arrow_schema.clone()).unwrap()); + let batch = DfRecordBatch::try_new( + arrow_schema, + vec![ + Arc::new(TimestampMillisecondArray::from( + (1..=9).map(|i| i * 1000).collect::>(), + )), + Arc::new(array), + ], + ) + .unwrap(); + let batches = RecordBatches::try_new( + schema.clone(), + vec![RecordBatch::from_df_record_batch(schema, batch)], + ) + .unwrap(); + let response = ReadResponse { + results: vec![QueryResult { + timeseries: recordbatches_to_timeseries( + "native_histogram_remote_read", + "ts", + "histogram", + batches, + ) + .unwrap(), + }], + }; + let body = snappy_compress(&response.encode_to_vec()).unwrap(); + let mut decoded = + ReadResponse::decode(snappy_decompress(&body).unwrap().as_slice()).unwrap(); + let mut expected = ReadResponse::decode( + include_bytes!( + "../../../tests/compatibility/wire/native_histogram_remote_read/response.pb" + ) + .as_slice(), + ) + .unwrap(); + let actual_histograms = &mut decoded.results[0].timeseries[0].histograms; + assert_eq!(actual_histograms[0].positive_deltas, vec![1, 2, -1]); + assert_eq!(actual_histograms[0].negative_deltas, vec![1, 2, -1]); + assert_eq!(actual_histograms[6].positive_deltas, vec![3, -1]); + for (actual, expected) in actual_histograms + .iter_mut() + .zip(&mut expected.results[0].timeseries[0].histograms) + { + assert_eq!(actual.sum.to_bits(), expected.sum.to_bits()); + actual.sum = 0.0; + expected.sum = 0.0; + } + assert_eq!(decoded, expected); + // The optional output is consumed by the pinned Prometheus iterator check. + if let Some(path) = std::env::var_os("GREPTIME_NATIVE_HISTOGRAM_REMOTE_READ_OUTPUT") { + std::fs::write(path, body).unwrap(); + } + } + + #[test] + fn test_native_histogram_remote_read_groups_and_sorts_batches() { + use api::greptime_proto::io::prometheus::write::v2::{Histogram, histogram::Count}; + use common_query::native_histogram::testing::histogram_array; + + let integer = native_histogram::encode_native_histogram(&Histogram { + count: Some(Count::CountInt(0)), + sum: f64::from_bits(0x7ff0_0000_0000_0002), + ..Default::default() + }) + .unwrap(); + let float = native_histogram::encode_native_histogram(&Histogram { + count: Some(Count::CountFloat(0.0)), + ..Default::default() + }) + .unwrap(); + let histograms = histogram_array(&[Some(integer), Some(float.clone()), None, Some(float)]); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + Field::new( + "event_time", + ArrowDataType::Timestamp(TimeUnit::Microsecond, None), + true, + ), + Field::new("legacy_histogram", histograms.data_type().clone(), true), + Field::new_dictionary("instance", ArrowDataType::UInt32, ArrowDataType::Utf8, true), + ])); + let schema = Arc::new(Schema::try_from(arrow_schema.clone()).unwrap()); + let batch = |timestamps: Vec>, labels: Vec>| { + RecordBatch::from_df_record_batch( + schema.clone(), + DfRecordBatch::try_new( + arrow_schema.clone(), + vec![ + Arc::new(arrow::array::TimestampMicrosecondArray::from(timestamps)), + Arc::new(histograms.clone()), + Arc::new(DictionaryArray::::new( + UInt32Array::from(labels), + Arc::new(StringArray::from(vec![Some("b"), Some("a"), None])), + )), + ], + ) + .unwrap(), + ) + }; + let batches = vec![ + batch( + vec![Some(4000), Some(3000), Some(5000), Some(-1001)], + vec![Some(0), None, Some(0), Some(0)], + ), + batch( + vec![Some(2000), Some(1000), Some(6000), None], + vec![Some(0), Some(2), Some(1), Some(0)], + ), + ]; + let series = recordbatches_to_timeseries( + "latency", + "event_time", + "legacy_histogram", + RecordBatches::try_new(schema.clone(), batches).unwrap(), + ) + .unwrap(); + assert_eq!(series.len(), 3); + assert_eq!( + series[0].labels, + vec![new_label( + METRIC_NAME_LABEL.to_string(), + "latency".to_string() + )] + ); + assert_eq!( + series[0] + .histograms + .iter() + .map(|h| h.timestamp) + .collect::>(), + vec![1, 3] + ); + assert_eq!(series[1].labels[1].value, "a"); + assert!(series[1].histograms.is_empty()); + assert_eq!(series[2].labels[1].value, "b"); + assert_eq!( + series[2] + .histograms + .iter() + .map(|h| h.timestamp) + .collect::>(), + vec![-2, 2, 4] + ); + assert!(matches!( + series[2].histograms[0].count, + Some(api::prom_store::remote::histogram::Count::CountFloat(0.0)) + )); + for histogram in &series[2].histograms[1..] { + assert!(matches!( + histogram.count, + Some(api::prom_store::remote::histogram::Count::CountInt(0)) + )); + assert_eq!(histogram.sum.to_bits(), 0x7ff0_0000_0000_0002); + } + assert!( + series + .iter() + .all(|s| s.samples.is_empty() + && s.labels.iter().all(|l| l.name != "legacy_histogram")) + ); + } + #[test] fn test_recordbatches_to_timeseries() { let schema = Arc::new(Schema::new(vec![ diff --git a/src/servers/tests/http/prom_store_test.rs b/src/servers/tests/http/prom_store_test.rs index c7a288ccee0..e4ef6219b30 100644 --- a/src/servers/tests/http/prom_store_test.rs +++ b/src/servers/tests/http/prom_store_test.rs @@ -16,8 +16,8 @@ use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use api::greptime_proto::io::prometheus::write::v2::{ - Request as RemoteWriteV2Request, Sample as RemoteWriteV2Sample, - TimeSeries as RemoteWriteV2TimeSeries, + BucketSpan as RemoteWriteV2BucketSpan, Request as RemoteWriteV2Request, + Sample as RemoteWriteV2Sample, TimeSeries as RemoteWriteV2TimeSeries, }; use api::prom_store::remote::{ LabelMatcher, Query, QueryResult, ReadRequest, ReadResponse, WriteRequest, @@ -28,6 +28,7 @@ use async_trait::async_trait; use axum::Router; use axum::http::HeaderMap; use common_query::Output; +use common_query::native_histogram::CUSTOM_BUCKETS_SCHEMA; use common_query::prelude::{ GREPTIME_PHYSICAL_TABLE, greptime_native_histogram, greptime_timestamp, greptime_value, }; @@ -317,6 +318,59 @@ async fn test_prometheus_remote_write_v2_convert_error_has_written_headers() { assert!(write_rx.try_recv().is_err()); } +#[tokio::test] +async fn test_prometheus_remote_write_v2_rejects_incompatible_histogram_before_writes() { + common_telemetry::init_default_ut_logging(); + let (read_tx, _read_rx) = mpsc::channel(100); + let (write_tx, mut write_rx) = mpsc::channel(100); + let app = make_test_app_with_write_capture(read_tx, write_tx); + let client = TestClient::new(app).await; + + for (schema, offset, message) in [ + (0, i32::MIN, "initial span offset i32::MIN"), + ( + CUSTOM_BUCKETS_SCHEMA, + 2, + "span extent exceeds custom_values", + ), + ] { + let mut histogram = remote_write_v2::histogram(2000); + histogram.schema = schema; + histogram.positive_spans = vec![RemoteWriteV2BucketSpan { offset, length: 0 }]; + let write_request = RemoteWriteV2Request { + symbols: vec![ + String::new(), + prom_store::METRIC_NAME_LABEL.to_string(), + "http_requests_total".to_string(), + "http_request_duration_seconds".to_string(), + ], + timeseries: vec![ + RemoteWriteV2TimeSeries { + labels_refs: vec![1, 2], + samples: vec![RemoteWriteV2Sample { + value: 42.0, + timestamp: 1000, + start_timestamp: 0, + }], + ..Default::default() + }, + RemoteWriteV2TimeSeries { + labels_refs: vec![1, 3], + histograms: vec![remote_write_v2::histogram(1000), histogram], + ..Default::default() + }, + ], + }; + + let result = post_remote_write_v2(&client, &write_request).await; + + assert_eq!(result.status(), 400); + assert_remote_write_v2_written_headers(&result.headers(), "0"); + assert!(result.text().await.contains(message)); + assert!(write_rx.try_recv().is_err()); + } +} + #[tokio::test] async fn test_prometheus_remote_write_v2_write_error_has_partial_written_headers() { common_telemetry::init_default_ut_logging(); diff --git a/tests-integration/src/otlp.rs b/tests-integration/src/otlp.rs index 4998e29e0c8..5ab264d23d0 100644 --- a/tests-integration/src/otlp.rs +++ b/tests-integration/src/otlp.rs @@ -2838,6 +2838,32 @@ WITH( .remove(0) .unwrap(); } + let series = + native_histogram_remote_read(&instance, "delta_distribution", ctx.clone()).await; + assert_eq!(series.len(), 1); + assert!( + series[0] + .labels + .iter() + .any(|l| l.name == "otlp_aggregation_temporality" && l.value == "delta") + ); + assert_eq!(series[0].histograms.len(), 3); + for (histogram, (timestamp, count)) in + series[0] + .histograms + .iter() + .zip([(60_000, 10), (120_000, 20), (180_000, 15)]) + { + assert_eq!(histogram.timestamp, timestamp); + assert_eq!( + histogram.count, + Some(api::prom_store::remote::histogram::Count::CountInt(count)) + ); + assert_eq!(histogram.positive_deltas, vec![count as i64]); + assert_eq!(histogram.positive_spans[0].offset, 1); + assert_eq!(histogram.sum, count as f64 * 1.5); + assert_eq!(histogram.reset_hint, 3); + } for (expression, expected) in [ ( "histogram_count(sum_over_time(delta_distribution[3m]))", @@ -3139,6 +3165,42 @@ WITH( } } + async fn native_histogram_remote_read( + instance: &Arc, + table: &str, + ctx: Arc, + ) -> Vec { + use api::prom_store::remote::{LabelMatcher, Query, ReadRequest, ReadResponse}; + use prost::Message; + use servers::query_handler::PromStoreProtocolHandler; + + let response = instance + .read( + ReadRequest { + queries: vec![Query { + start_timestamp_ms: 0, + end_timestamp_ms: 200_000, + matchers: vec![LabelMatcher { + name: "__name__".to_string(), + value: table.to_string(), + r#type: 0, + }], + ..Default::default() + }], + ..Default::default() + }, + ctx, + ) + .await + .unwrap(); + let body = servers::prom_store::snappy_decompress(&response.body).unwrap(); + ReadResponse::decode(body.as_slice()) + .unwrap() + .results + .remove(0) + .timeseries + } + #[tokio::test(flavor = "multi_thread")] async fn test_otlp_delta_exponential_storage_precision_and_replacement() { use datatypes::arrow::datatypes::Int64Type; @@ -3194,6 +3256,24 @@ WITH( .remove(0) .unwrap(); } + let series = + native_histogram_remote_read(instance, "interval", ctx.clone()).await; + assert_eq!(series.len(), 1); + assert!(series[0].labels.iter().any(|l| l.name == "otlp_aggregation_temporality" && l.value == "delta")); + let histogram = &series[0].histograms[0]; + assert_eq!(histogram.timestamp, 60_000); + assert_eq!( + histogram.count, + Some(api::prom_store::remote::histogram::Count::CountInt(count)) + ); + assert_eq!( + histogram.zero_count, + Some(api::prom_store::remote::histogram::ZeroCount::ZeroCountInt( + count + )) + ); + assert_eq!(histogram.reset_hint, 3); + assert_eq!(histogram.sum, 0.0); let output = instance .do_query( "SELECT greptime_timestamp,greptime_native_histogram FROM interval", diff --git a/tests-integration/src/prom_store.rs b/tests-integration/src/prom_store.rs index 87798c282c1..0608ff88f9c 100644 --- a/tests-integration/src/prom_store.rs +++ b/tests-integration/src/prom_store.rs @@ -778,6 +778,329 @@ mod tests { } } + #[tokio::test(flavor = "multi_thread")] + async fn test_standalone_native_histogram_remote_read() { + let standalone = + GreptimeDbStandaloneBuilder::new("standalone_native_histogram_remote_read") + .build() + .await; + check_native_histogram_remote_read(standalone.fe_instance()).await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_distributed_native_histogram_remote_read() { + let cluster = + tests::create_distributed_instance("distributed_native_histogram_remote_read").await; + check_native_histogram_remote_read(&cluster.frontend()).await; + } + + async fn check_native_histogram_remote_read(instance: &Arc) { + use api::greptime_proto::io::prometheus::write::v2::{ + Histogram, + histogram::{Count, ZeroCount}, + }; + use common_query::native_histogram::{ + encode_native_histogram, native_histogram_column_schema, + }; + use servers::prom_remote_write::v2::test_util as remote_write_v2; + + let ctx = Arc::new(QueryContext::with(DEFAULT_CATALOG_NAME, "prometheus")); + instance + .do_query("CREATE DATABASE prometheus", ctx.clone()) + .await + .remove(0) + .unwrap(); + let server = HttpServerBuilder::new(HttpOptions::default()) + .with_prom_handler(instance.clone(), None, true, Default::default(), None) + .build(); + let client = TestClient::new(server.build(server.make_app()).unwrap()).await; + let large = (1u64 << 53) + 1; + let histograms = vec![ + Histogram { + count: Some(Count::CountInt(6)), + timestamp: 1000, + positive_spans: vec![api::greptime_proto::io::prometheus::write::v2::BucketSpan { + offset: 0, + length: 3, + }], + positive_deltas: vec![1, 2, -1], + sum: 10.0, + ..Default::default() + }, + Histogram { + count: Some(Count::CountInt(large)), + zero_count: Some(ZeroCount::ZeroCountInt(large)), + timestamp: 2000, + ..Default::default() + }, + Histogram { + count: Some(Count::CountFloat(6.0)), + zero_count: Some(ZeroCount::ZeroCountFloat(0.5)), + positive_spans: vec![api::greptime_proto::io::prometheus::write::v2::BucketSpan { + offset: 3, + length: 2, + }], + positive_counts: vec![2.0, 3.5], + timestamp: 3000, + ..Default::default() + }, + Histogram { + count: Some(Count::CountInt(0)), + sum: f64::from_bits(0x7ff0_0000_0000_0002), + timestamp: 4000, + ..Default::default() + }, + ]; + let request = remote_write_v2::request_with_labels_and_histograms( + vec![(prom_store::METRIC_NAME_LABEL, "latency"), ("job", "api")], + histograms, + ); + let response = client + .post("/v1/prometheus/write?db=prometheus") + .header( + "content-type", + "application/x-protobuf;proto=io.prometheus.write.v2.Request", + ) + .header("content-encoding", "snappy") + .body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap()) + .send() + .await; + assert_eq!(response.status(), 204, "{}", response.text().await); + let (scalar, _) = to_grpc_row_insert_requests(&WriteRequest { + timeseries: vec![prom_store::mock_timeseries()[0].clone()], + ..Default::default() + }) + .unwrap(); + instance.write(scalar, ctx.clone(), true).await.unwrap(); + + for flushed in [false, true] { + if flushed { + instance + .do_query("ADMIN FLUSH_TABLE('latency')", ctx.clone()) + .await + .remove(0) + .unwrap(); + } + for accepted_response_types in [vec![], vec![0], vec![1, 0]] { + let mut query = remote_query("latency"); + query.end_timestamp_ms = 4000; + query.matchers.push(LabelMatcher { + name: "job".to_string(), + value: "api".to_string(), + r#type: MatcherType::Eq as i32, + }); + let mut empty = query.clone(); + empty.start_timestamp_ms = 5000; + empty.end_timestamp_ms = 6000; + let request = ReadRequest { + queries: vec![ + remote_query("metric1"), + query, + remote_query("missing_metric"), + empty, + ], + accepted_response_types, + }; + let response = client + .post("/v1/prometheus/read?db=prometheus") + .body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap()) + .send() + .await; + assert_eq!(response.status(), 200, "{}", response.text().await); + assert_eq!(response.headers()["content-type"], "application/x-protobuf"); + assert_eq!(response.headers()["content-encoding"], "snappy"); + let body = prom_store::snappy_decompress(&response.bytes().await).unwrap(); + let response = ReadResponse::decode(body.as_slice()).unwrap(); + assert_eq!(response.results.len(), 4); + assert!(!response.results[0].timeseries[0].samples.is_empty()); + assert!(response.results[2].timeseries.is_empty()); + assert!(response.results[3].timeseries.is_empty()); + let series = &response.results[1].timeseries[0]; + assert_eq!( + series.labels, + vec![ + Label { + name: "__name__".to_string(), + value: "latency".to_string() + }, + Label { + name: "job".to_string(), + value: "api".to_string() + } + ] + ); + assert_eq!(series.histograms.len(), 4); + assert_eq!(series.histograms[0].positive_deltas, vec![1, 2, -1]); + assert_eq!(series.histograms[2].positive_counts, vec![2.0, 3.5]); + assert_eq!( + series.histograms[1].count, + Some(api::prom_store::remote::histogram::Count::CountInt(large)) + ); + assert_eq!( + series.histograms[1].zero_count, + Some(api::prom_store::remote::histogram::ZeroCount::ZeroCountInt( + large + )) + ); + assert_eq!( + series.histograms[2].count, + Some(api::prom_store::remote::histogram::Count::CountFloat(6.0)) + ); + assert_eq!(series.histograms[3].sum.to_bits(), 0x7ff0_0000_0000_0002); + } + } + let response = client + .post("/v1/prometheus/read?db=prometheus") + .body( + prom_store::snappy_compress( + &ReadRequest { + queries: vec![remote_query("latency")], + accepted_response_types: vec![1], + } + .encode_to_vec(), + ) + .unwrap(), + ) + .send() + .await; + assert!(!response.status().is_success()); + assert!( + response + .text() + .await + .contains("does not support any of the requested response types") + ); + + // Typed inserts supply sub-millisecond pre-epoch times that neither OTLP + // nor millisecond Remote Write can express, and use a legacy field name. + for (table, datatype, timestamp) in [ + ( + "legacy_us", + ColumnDataType::TimestampMicrosecond, + ValueData::TimestampMicrosecondValue(-1001), + ), + ( + "legacy_ns", + ColumnDataType::TimestampNanosecond, + ValueData::TimestampNanosecondValue(-1_001_000), + ), + ] { + let mut histogram_schema = native_histogram_column_schema().unwrap(); + histogram_schema.column_name = "payload".to_string(); + let histogram = encode_native_histogram(&Histogram { + count: Some(Count::CountInt(large)), + zero_count: Some(ZeroCount::ZeroCountInt(large)), + ..Default::default() + }) + .unwrap(); + instance + .write( + RowInsertRequests { + inserts: vec![RowInsertRequest { + table_name: table.to_string(), + rows: Some(Rows { + schema: vec![ + api::v1::ColumnSchema { + column_name: "event_time".to_string(), + datatype: datatype as i32, + semantic_type: SemanticType::Timestamp as i32, + ..Default::default() + }, + histogram_schema, + ], + rows: vec![Row { + values: vec![ + Value { + value_data: Some(timestamp), + }, + Value { + value_data: Some(histogram), + }, + ], + }], + }), + }], + }, + ctx.clone(), + false, + ) + .await + .unwrap(); + for flushed in [false, true] { + if flushed { + instance + .do_query(&format!("ADMIN FLUSH_TABLE('{table}')"), ctx.clone()) + .await + .remove(0) + .unwrap(); + } + let mut query = remote_query(table); + query.start_timestamp_ms = -2; + query.end_timestamp_ms = 0; + let response = post_remote_read(&client, vec![query]).await; + assert_eq!(response.status(), 200, "{}", response.text().await); + let body = prom_store::snappy_decompress(&response.bytes().await).unwrap(); + let response = ReadResponse::decode(body.as_slice()).unwrap(); + let series = &response.results[0].timeseries[0]; + assert_eq!(series.labels.len(), 1); + assert_eq!(series.histograms[0].timestamp, -2); + assert_eq!( + series.histograms[0].count, + Some(api::prom_store::remote::histogram::Count::CountInt(large)) + ); + } + } + // Bypass ingestion's histogram codec to exercise whole-request failure. + let ValueData::StructValue(mut malformed) = + encode_native_histogram(&Histogram::default()).unwrap() + else { + unreachable!() + }; + malformed.items[14].value_data = Some(ValueData::F64Value(0.0)); + instance + .write( + RowInsertRequests { + inserts: vec![RowInsertRequest { + table_name: "malformed_histogram".to_string(), + rows: Some(Rows { + schema: vec![ + api::v1::ColumnSchema { + column_name: "ts".to_string(), + datatype: ColumnDataType::TimestampMillisecond as i32, + semantic_type: SemanticType::Timestamp as i32, + ..Default::default() + }, + native_histogram_column_schema().unwrap(), + ], + rows: vec![Row { + values: vec![ + Value { + value_data: Some(ValueData::TimestampMillisecondValue( + 1000, + )), + }, + Value { + value_data: Some(ValueData::StructValue(malformed)), + }, + ], + }], + }), + }], + }, + ctx.clone(), + false, + ) + .await + .unwrap(); + for queries in [ + vec![remote_query("metric1"), remote_query("malformed_histogram")], + vec![remote_query("malformed_histogram"), remote_query("metric1")], + ] { + let response = post_remote_read(&client, queries).await; + assert_eq!(response.status(), 500); + } + } + fn remote_query(metric: &str) -> Query { Query { start_timestamp_ms: 1000, diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 1671f5a2921..11c41431fbc 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -4262,6 +4262,27 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag ], ..Default::default() }); + let large = (1u64 << 53) + 1; + write_request.timeseries[1].histograms.extend([ + Histogram { + count: Some(Count::CountInt(large)), + zero_count: Some(ZeroCount::ZeroCountInt(large)), + timestamp: 5000, + ..Default::default() + }, + Histogram { + count: Some(Count::CountInt(0)), + sum: f64::from_bits(0x7ff0_0000_0000_0002), + timestamp: 6000, + ..Default::default() + }, + Histogram { + count: Some(Count::CountFloat(0.0)), + sum: f64::from_bits(0x7ff8_0000_0000_0123), + timestamp: 7000, + ..Default::default() + }, + ]); let serialized_request = write_request.encode_to_vec(); let compressed_request = prom_store::snappy_compress(&serialized_request).expect("failed to encode snappy"); @@ -4278,7 +4299,7 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag .await; assert_eq!(res.status(), StatusCode::NO_CONTENT); let headers = res.headers(); - assert_remote_write_v2_written_headers_with_histograms(&headers, "1", "2"); + assert_remote_write_v2_written_headers_with_histograms(&headers, "1", "5"); assert!(res.text().await.is_empty()); validate_data( @@ -4289,7 +4310,7 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag ) .await; - let histogram_sql = "select greptime_timestamp, greptime_native_histogram, job, instance from remote_write_v2_latency_seconds order by greptime_timestamp;"; + let histogram_sql = "select greptime_timestamp, greptime_native_histogram, job, instance from remote_write_v2_latency_seconds where greptime_timestamp < 5000::TIMESTAMP order by greptime_timestamp;"; let histogram_rows = "[[3000,{\"count_f64\":null,\"count_i64\":8,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[1],\"negative_span_lengths\":[1],\"negative_span_offsets\":[-2],\"positive_buckets_f64\":[],\"positive_buckets_i64\":[1,3,2],\"positive_span_lengths\":[3],\"positive_span_offsets\":[0],\"reset_hint\":2,\"schema\":1,\"start_timestamp\":1500,\"sum\":10.0,\"zero_count_f64\":null,\"zero_count_i64\":1,\"zero_threshold\":0.001},\"api\",\"localhost:9090\"],[4000,{\"count_f64\":6.0,\"count_i64\":null,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[],\"negative_span_lengths\":[],\"negative_span_offsets\":[],\"positive_buckets_f64\":[2.0,3.5],\"positive_buckets_i64\":[],\"positive_span_lengths\":[2],\"positive_span_offsets\":[3],\"reset_hint\":3,\"schema\":2,\"start_timestamp\":2500,\"sum\":20.0,\"zero_count_f64\":0.5,\"zero_count_i64\":null,\"zero_threshold\":0.002},\"api\",\"localhost:9090\"]]"; validate_data( "prometheus_remote_write_v2_native_histogram_rows", @@ -4299,6 +4320,76 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag ) .await; + for flushed in [false, true] { + if flushed { + flush_table(&client, "remote_write_v2_latency_seconds").await; + } + for accepted_response_types in [vec![], vec![0], vec![1, 0]] { + let query = |name: &str| Query { + start_timestamp_ms: 0, + end_timestamp_ms: 10_000, + matchers: vec![LabelMatcher { + name: prom_store::METRIC_NAME_LABEL.to_string(), + value: name.to_string(), + r#type: 0, + }], + ..Default::default() + }; + let request = ReadRequest { + queries: vec![ + query("remote_write_v2_sample_seconds"), + query("missing_native_histogram"), + query("remote_write_v2_latency_seconds"), + ], + accepted_response_types, + }; + let response = client + .post("/v1/prometheus/read") + .body(prom_store::snappy_compress(&request.encode_to_vec()).unwrap()) + .send() + .await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["content-type"], "application/x-protobuf"); + assert_eq!(response.headers()["content-encoding"], "snappy"); + let body = prom_store::snappy_decompress(&response.bytes().await).unwrap(); + let response = ReadResponse::decode(body.as_slice()).unwrap(); + assert_eq!(response.results.len(), 3); + assert_eq!( + response.results[0].timeseries[0].samples, + vec![Sample { + value: 5.0, + timestamp: 2500 + }] + ); + assert!(response.results[1].timeseries.is_empty()); + let series = &response.results[2].timeseries[0]; + assert!(series.samples.is_empty()); + assert_eq!( + series + .histograms + .iter() + .map(|h| h.timestamp) + .collect::>(), + vec![3000, 4000, 5000, 6000, 7000] + ); + assert_eq!(series.histograms[0].positive_deltas, vec![1, 2, -1]); + assert_eq!(series.histograms[0].negative_deltas, vec![1]); + assert_eq!(series.histograms[1].positive_counts, vec![2.0, 3.5]); + assert_eq!( + series.histograms[2].count, + Some(api::prom_store::remote::histogram::Count::CountInt(large)) + ); + assert_eq!( + series.histograms[2].zero_count, + Some(api::prom_store::remote::histogram::ZeroCount::ZeroCountInt( + large + )) + ); + assert_eq!(series.histograms[3].sum.to_bits(), 0x7ff0_0000_0000_0002); + assert_eq!(series.histograms[4].sum.to_bits(), 0x7ff8_0000_0000_0123); + } + } + // The native histogram column is a struct of nested lists, so its SST // encoding differs from its memtable representation. flush_table(&client, "remote_write_v2_latency_seconds").await; @@ -9133,9 +9224,17 @@ pub async fn test_otlp_exponential_histogram(store_type: StorageType) { )] }; - let (app, mut guard) = - setup_test_http_app_with_frontend(store_type, "test_otlp_exponential_histogram").await; - let client = TestClient::new(app).await; + let standalone = tests_integration::standalone::GreptimeDbStandaloneBuilder::new( + "test_otlp_exponential_histogram", + ) + .with_default_store_type(store_type) + .build() + .await; + let server = build_test_prom_server(standalone.fe_instance().clone(), false, None) + .with_otlp_handler(standalone.fe_instance().clone(), true) + .build(); + let client = TestClient::new(server.build(server.make_app()).unwrap()).await; + let mut guard = standalone.guard; let res = send_req(&client, headers(), "/v1/otlp/v1/metrics", body, false).await; assert_eq!(StatusCode::OK, res.status()); let response = ExportMetricsServiceResponse::decode(res.bytes().await).unwrap(); @@ -9158,6 +9257,42 @@ pub async fn test_otlp_exponential_histogram(store_type: StorageType) { ) .await; + for flushed in [false, true] { + if flushed { + flush_table(&client, "otlp_exponential_latency").await; + } + let response = post_remote_read( + &client, + vec![Query { + start_timestamp_ms: 0, + end_timestamp_ms: 4000, + matchers: vec![LabelMatcher { + name: "__name__".to_string(), + value: "otlp_exponential_latency".to_string(), + r#type: 0, + }], + ..Default::default() + }], + ) + .await; + let histogram = &response.results[0].timeseries[0].histograms[0]; + assert_eq!(histogram.timestamp, 3000); + assert_eq!( + histogram.count, + Some(api::prom_store::remote::histogram::Count::CountInt(4)) + ); + assert_eq!( + histogram.zero_count, + Some(api::prom_store::remote::histogram::ZeroCount::ZeroCountInt( + 1 + )) + ); + assert_eq!(histogram.positive_deltas, vec![1, 1]); + assert_eq!(histogram.positive_spans[0].offset, 0); + assert_eq!(histogram.reset_hint, 0); + assert_eq!(histogram.sum, 8.0); + } + for query in [ "histogram_count(otlp_exponential_latency)", "histogram_sum(otlp_exponential_latency)", diff --git a/tests/compatibility/README.md b/tests/compatibility/README.md index 1daf42cc736..53ecda81048 100644 --- a/tests/compatibility/README.md +++ b/tests/compatibility/README.md @@ -15,6 +15,14 @@ read-preference defaulting and unrelated metadata preservation: cargo nextest run -p session -E 'test(test_legacy_query_context_wire_defaults_to_leader)' ``` +The [native histogram Remote Read fixture](wire/native_histogram_remote_read/README.md) +checks the sampled encoder against independent Prometheus protobuf values, +including exact integer counts, custom bounds, and NaN payloads: + +```shell +cargo nextest run -p servers -E 'test(native_histogram_remote_read_wire_fixture)' +``` + ## Object store WAL fixtures Frozen objects of the object store WAL are checked byte for byte by the format tests of diff --git a/tests/compatibility/wire/native_histogram_remote_read/README.md b/tests/compatibility/wire/native_histogram_remote_read/README.md new file mode 100644 index 00000000000..9db185efcd4 --- /dev/null +++ b/tests/compatibility/wire/native_histogram_remote_read/README.md @@ -0,0 +1,184 @@ +# Native histogram sampled Remote Read wire compatibility + +`response.pb` is a frozen `prompb.ReadResponse` generated from Prometheus +revision `e471c7708c6b0b13e2050222bbd5a907b36f4113`, independently of +`greptime-proto`. It contains one query result, one series named +`native_histogram_remote_read`, and nine samples at 1000ms intervals. + +| Timestamp (ms) | Independent expectation | +| --- | --- | +| 1000 | Integer count `2^53 + 13`, zero count `2^53 + 1`, both bucket sides `[1,3,2]`, wire deltas `[1,2,-1]` across span boundaries, schema 1, reset hint 2 | +| 2000 | Float count 6, zero count 0.5, absolute buckets `[2,3.5]`, schema 2, gauge hint | +| 3000 | Integer custom buckets `[1,2]`, one observation in `(-Inf,1]` | +| 4000 | Float custom buckets `[1,2]`, absolute counts `[1,2.5]`, gauge hint | +| 5000 | Empty integer histogram with explicit count and zero-count oneofs | +| 6000 | Empty float histogram with explicit count and zero-count oneofs | +| 7000 | Integer count 8, buckets `[3,2]`, wire deltas `[3,-1]`, ordinary NaN sum bits `0x7ff8000000000123` | +| 8000 | Empty integer histogram, stale NaN sum bits `0x7ff0000000000002` | +| 9000 | Integer count and zero count `i64::MAX` | + +The 420-byte fixture has SHA-256 `f1414bfa6f0821e920a88e325964abfad0269b05b7c0643a84ca6e908729d1eb`. +Generation used Go `go1.27.3 darwin/arm64`, the pinned Prometheus `go.mod` and +`go.sum` (declared Go version `1.26.10`), and its generated `prompb` types. +Direct wire dependencies were `github.com/gogo/protobuf v1.3.2` and +`github.com/golang/snappy v1.0.0`. No local `protoc` or Greptime protobuf +generator is involved. + +## Automated regression + +From the GreptimeDB checkout: + +```shell +cargo nextest run -p servers -E 'test(native_histogram_remote_read_wire_fixture)' +``` + +This constructs stored arrays from independent Rust inputs, checks fixed +absolute bucket counts, runs the actual record-batch encoder, Snappy-compresses +and protobuf-decodes its output, and compares all fields with the upstream +fixture. NaN payloads are compared by bits. It also checks fixed wire deltas; +converting those deltas back with an upstream helper is not the expected-value +generator. Tests never regenerate the frozen fixture. + +## Reproduce and check with Prometheus + +Use a scratch module to preserve the Prometheus checkout. Set the two checkout +paths, create a scratch directory, and save the Go program below as `main.go` +in that directory: + +```shell +PROMETHEUS_CHECKOUT=/path/to/prometheus +GREPTIMEDB_CHECKOUT=/path/to/greptimedb +NH_SCRATCH=$(mktemp -d) +git -C "$PROMETHEUS_CHECKOUT" rev-parse HEAD +cp "$PROMETHEUS_CHECKOUT/go.mod" "$PROMETHEUS_CHECKOUT/go.sum" "$NH_SCRATCH/" +cd "$NH_SCRATCH" +go mod edit -module=native-histogram-interop +go mod edit -require=github.com/prometheus/prometheus@v0.0.0 +go mod edit -replace="github.com/prometheus/prometheus=$PROMETHEUS_CHECKOUT" +# Save the Go program below as main.go here before running these commands. +go run -mod=mod . generate "$NH_SCRATCH/response.pb" +shasum -a 256 "$NH_SCRATCH/response.pb" +cmp "$NH_SCRATCH/response.pb" "$GREPTIMEDB_CHECKOUT/tests/compatibility/wire/native_histogram_remote_read/response.pb" +cd "$GREPTIMEDB_CHECKOUT" +GREPTIME_NATIVE_HISTOGRAM_REMOTE_READ_OUTPUT="$NH_SCRATCH/greptime-response.snappy" \ + cargo nextest run -p servers -E 'test(native_histogram_remote_read_wire_fixture)' +cd "$NH_SCRATCH" +go run -mod=mod . check "$NH_SCRATCH/greptime-response.snappy" +``` + +The check consumes `storage/remote.FromQueryResult` iterators to exhaustion, +checks both iterator and series-set errors, integer/float types, exact integer +counts, timestamps, spans, bucket placement, custom bounds, reset hints, and +NaN bits. It calls both `AtHistogram` and `AtFloatHistogram` on integer custom +samples, including a single finite-bound bucket. Intentional integer-to-float +conversion is not expected to retain large-integer precision. + +Verified on 2026-10-26 against the response emitted by the Rust test: +`PASS: 9 histogram samples, both iterator accessors, no series or iterator errors`. + +The tested baseline is the exact revision above, not a claimed minimum +Prometheus release. This Go check is a manual implementation check; it is not +part of the existing Go CI. The Rust regression is the automated guard. + +```go +package main + +import ( + "fmt" + "math" + "os" + "reflect" + + "github.com/golang/snappy" + "github.com/prometheus/prometheus/prompb" + "github.com/prometheus/prometheus/storage/remote" + "github.com/prometheus/prometheus/tsdb/chunkenc" +) + +func expected() *prompb.ReadResponse { + hs := []prompb.Histogram{ + {Count: &prompb.Histogram_CountInt{CountInt: (1<<53)+13}, ZeroCount: &prompb.Histogram_ZeroCountInt{ZeroCountInt: (1<<53)+1}, Sum: 10, Schema: 1, ZeroThreshold: 0.001, ResetHint: prompb.Histogram_NO, + PositiveSpans: []prompb.BucketSpan{{Offset: -2, Length: 2}, {Offset: 3, Length: 1}}, PositiveDeltas: []int64{1,2,-1}, NegativeSpans: []prompb.BucketSpan{{Offset: -3, Length: 1}, {Offset: 2, Length: 2}}, NegativeDeltas: []int64{1,2,-1}}, + {Count: &prompb.Histogram_CountFloat{CountFloat: 6}, ZeroCount: &prompb.Histogram_ZeroCountFloat{ZeroCountFloat: 0.5}, Sum: 20, Schema: 2, ZeroThreshold: 0.002, ResetHint: prompb.Histogram_GAUGE, + PositiveSpans: []prompb.BucketSpan{{Offset: 3, Length: 2}}, PositiveCounts: []float64{2,3.5}}, + {Count: &prompb.Histogram_CountInt{CountInt: 1}, ZeroCount: &prompb.Histogram_ZeroCountInt{}, Sum: 1, Schema: -53, CustomValues: []float64{1,2}, PositiveSpans: []prompb.BucketSpan{{Length: 1}}, PositiveDeltas: []int64{1}}, + {Count: &prompb.Histogram_CountFloat{CountFloat: 3.5}, ZeroCount: &prompb.Histogram_ZeroCountFloat{}, Sum: 4, Schema: -53, CustomValues: []float64{1,2}, PositiveSpans: []prompb.BucketSpan{{Length: 2}}, PositiveCounts: []float64{1,2.5}, ResetHint: prompb.Histogram_GAUGE}, + {Count: &prompb.Histogram_CountInt{}, ZeroCount: &prompb.Histogram_ZeroCountInt{}}, + {Count: &prompb.Histogram_CountFloat{}, ZeroCount: &prompb.Histogram_ZeroCountFloat{}}, + {Count: &prompb.Histogram_CountInt{CountInt: 8}, ZeroCount: &prompb.Histogram_ZeroCountInt{}, Sum: math.Float64frombits(0x7ff8000000000123), PositiveSpans: []prompb.BucketSpan{{Length: 2}}, PositiveDeltas: []int64{3,-1}}, + {Count: &prompb.Histogram_CountInt{}, ZeroCount: &prompb.Histogram_ZeroCountInt{}, Sum: math.Float64frombits(0x7ff0000000000002)}, + {Count: &prompb.Histogram_CountInt{CountInt: math.MaxInt64}, ZeroCount: &prompb.Histogram_ZeroCountInt{ZeroCountInt: math.MaxInt64}}, + } + for i := range hs { hs[i].Timestamp = int64(i+1)*1000 } + return &prompb.ReadResponse{Results: []*prompb.QueryResult{{Timeseries: []*prompb.TimeSeries{{Labels: []prompb.Label{{Name:"__name__", Value:"native_histogram_remote_read"}}, Histograms:hs}}}}} +} + +func check(ok bool, args ...any) { if !ok { panic(fmt.Sprint(args...)) } } +func main() { + want := expected() + if os.Args[1] == "generate" { + data, err := want.Marshal(); if err != nil { panic(err) } + if err := os.WriteFile(os.Args[2], data, 0644); err != nil { panic(err) } + fmt.Printf("generated %d bytes\n", len(data)); return + } + data, err := os.ReadFile(os.Args[2]); if err != nil { panic(err) } + data, err = snappy.Decode(nil, data); if err != nil { panic(err) } + var got prompb.ReadResponse + if err = got.Unmarshal(data); err != nil { panic(err) } + check(len(got.Results)==1 && len(got.Results[0].Timeseries)==1, "result/series count") + hs := got.Results[0].Timeseries[0].Histograms + expectedHs := want.Results[0].Timeseries[0].Histograms + check(len(hs)==len(expectedHs), "histogram count") + for i := range hs { + g,w := hs[i], expectedHs[i] + check(math.Float64bits(g.Sum)==math.Float64bits(w.Sum), "sum bits ", i) + g.Sum,w.Sum=0,0 + check(reflect.DeepEqual(g,w), "protobuf fields ", i, ": ", g, " expected ", w) + } + ss := remote.FromQueryResult(true, got.Results[0]) + samples,series := 0,0 + for ss.Next() { + series++ + it := ss.At().Iterator(nil) + for typ := it.Next(); typ != chunkenc.ValNone; typ=it.Next() { + w := expectedHs[samples] + if w.IsFloatHistogram() { + check(typ==chunkenc.ValFloatHistogram, "float type ", samples) + t,h:=it.AtFloatHistogram(nil) + check(t==w.Timestamp && h.Count==w.GetCountFloat() && h.ZeroCount==w.GetZeroCountFloat(), "float count/time ", samples) + check(reflect.DeepEqual(h.PositiveBuckets,w.PositiveCounts), "float buckets ", samples) + check(reflect.DeepEqual(h.CustomValues,w.CustomValues), "float bounds ", samples) + check(math.Float64bits(h.Sum)==math.Float64bits(w.Sum), "float sum ", samples) + check(int32(h.CounterResetHint)==int32(w.ResetHint), "float reset ", samples) + } else { + check(typ==chunkenc.ValHistogram, "integer type ", samples) + t,h:=it.AtHistogram(nil) + check(t==w.Timestamp && h.Count==w.GetCountInt() && h.ZeroCount==w.GetZeroCountInt(), "integer count/time ", samples) + check(reflect.DeepEqual(h.PositiveBuckets,w.PositiveDeltas) && reflect.DeepEqual(h.NegativeBuckets,w.NegativeDeltas), "integer deltas ", samples) + check(reflect.DeepEqual(h.CustomValues,w.CustomValues), "integer bounds ", samples) + check(math.Float64bits(h.Sum)==math.Float64bits(w.Sum), "integer sum ", samples) + check(int32(h.CounterResetHint)==int32(w.ResetHint), "integer reset ", samples) + _,f:=it.AtFloatHistogram(nil) + check(reflect.DeepEqual(f.CustomValues,w.CustomValues), "converted bounds ", samples) + if w.Schema == -53 { + buckets:=f.PositiveBucketIterator() + check(buckets.Next(), "custom bucket missing") + b:=buckets.At() + check(b.Lower==math.Inf(-1) && b.Upper==1 && b.Count==1, "custom bucket placement: ",b) + check(!buckets.Next(), "extra custom bucket") + } + } + _,f:=it.AtFloatHistogram(nil) + check(f.Schema==w.Schema && f.ZeroThreshold==w.ZeroThreshold, "schema/threshold ",samples) + check(len(f.PositiveSpans)==len(w.PositiveSpans) && len(f.NegativeSpans)==len(w.NegativeSpans), "span counts ",samples) + for i,s:=range f.PositiveSpans { check(s.Offset==w.PositiveSpans[i].Offset && s.Length==w.PositiveSpans[i].Length,"positive spans") } + for i,s:=range f.NegativeSpans { check(s.Offset==w.NegativeSpans[i].Offset && s.Length==w.NegativeSpans[i].Length,"negative spans") } + samples++ + } + check(it.Err()==nil,"iterator error: ",it.Err()) + } + check(ss.Err()==nil,"series error: ",ss.Err()) + check(series==1 && samples==9,"series/sample count ",series,"/",samples) + fmt.Println("PASS: 9 histogram samples, both iterator accessors, no series or iterator errors") +} +``` diff --git a/tests/compatibility/wire/native_histogram_remote_read/response.pb b/tests/compatibility/wire/native_histogram_remote_read/response.pb new file mode 100644 index 0000000000000000000000000000000000000000..42d761854204c5403b596993cf7476255e7d13d5 GIT binary patch literal 420 zcmd;D$jmj5nM;FQ^gB*C1*j1DH0A)<4hbBSfIuw*jqv*AJ~KSF=Yur_zVsZn#ZAlx#9x9k`4zW zRQCsag-ip6C?=pCj7$|f#G4qQT0Yprv?_T+t>l4ODd3=x3AKp%=;o{|Iyhomwi$nGEYV23d!F|jdL+%QuT;9!ts0z2vh WP^RL5osu}me<-LofU_!AxB&p8<5}$h literal 0 HcmV?d00001