mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-08-18 12:08:22 +00:00
feat!: update native histogram unsigned int types (#8824)
* feat(native-histogram): store counts and span lengths as signed integers Native histograms are unreleased, so the on-disk integer payload columns are switched from unsigned to signed types without backward-compat: - count_u64 / zero_count_u64: uint64 -> int64 - positive_span_lengths / negative_span_lengths: list(uint32) -> list(int32) - Span.length (query-time model): u32 -> i32 The Prometheus remote-write v2 source carries these as uint64/uint32, so the unsigned->signed conversion at the ingestion boundary is overflow checked: an integer count >= 2^63 or a span length >= 2^31 is rejected with an explicit error rather than silently wrapping to a negative value. read_spans additionally rejects negative stored lengths to keep the non-negative invariant sound for downstream `as usize` casts. The UDAF accumulator's own observation counter (transient aggregation state, not part of the persisted histogram value) is intentionally left as uint64. Signed-off-by: Ning Sun <sunning@greptime.com> * refactor(native-histogram): rename count/zero_count fields to _i64 Now that the integer payload columns are stored as int64, rename the field constants and persisted names to match: COUNT_U64_FIELD ("count_u64") -> COUNT_I64_FIELD ("count_i64") ZERO_COUNT_U64_FIELD ("zero_count_u64") -> ZERO_COUNT_I64_FIELD ("zero_count_i64") The local builder variables and the docs/JSON snapshot are updated to match. No backward-compat (unreleased feature). Signed-off-by: Ning Sun <sunning@greptime.com> * test(native-histogram): refresh planner plan snapshot for signed types The mixed native-histogram range test embeds the full histogram Struct type in its expected plan string, which still carried the pre-rename unsigned fields. Update the snapshot to match the signed schema: positive/negative_span_lengths: List(UInt32) -> List(Int32) count_u64/zero_count_u64: UInt64 -> count_i64/zero_count_i64: Int64 Signed-off-by: Ning Sun <sunning@greptime.com> --------- Signed-off-by: Ning Sun <sunning@greptime.com>
This commit is contained in:
Ning Sun
GitHub
@@ -23,13 +23,13 @@ use std::collections::BTreeMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use datafusion::arrow::array::{
|
||||
Array, ArrayRef, Float64Array, Int32Array, ListArray, PrimitiveArray, StructArray,
|
||||
TimestampMillisecondArray, UInt64Array,
|
||||
Array, ArrayRef, Float64Array, Int32Array, Int64Array, ListArray, PrimitiveArray, StructArray,
|
||||
TimestampMillisecondArray,
|
||||
};
|
||||
use datafusion::arrow::buffer::NullBuffer;
|
||||
use datafusion::arrow::datatypes::{
|
||||
ArrowPrimitiveType, DataType as ArrowDataType, Field, Float64Type, Int32Type, Int64Type,
|
||||
TimestampMillisecondType, UInt32Type, UInt64Type,
|
||||
TimestampMillisecondType,
|
||||
};
|
||||
use datafusion_common::{DataFusionError, Result as DfResult};
|
||||
use datatypes::data_type::{ConcreteDataType, DataType};
|
||||
@@ -50,8 +50,8 @@ pub const POSITIVE_SPAN_OFFSETS_FIELD: &str = "positive_span_offsets";
|
||||
pub const POSITIVE_SPAN_LENGTHS_FIELD: &str = "positive_span_lengths";
|
||||
pub const NEGATIVE_SPAN_OFFSETS_FIELD: &str = "negative_span_offsets";
|
||||
pub const NEGATIVE_SPAN_LENGTHS_FIELD: &str = "negative_span_lengths";
|
||||
pub const COUNT_U64_FIELD: &str = "count_u64";
|
||||
pub const ZERO_COUNT_U64_FIELD: &str = "zero_count_u64";
|
||||
pub const COUNT_I64_FIELD: &str = "count_i64";
|
||||
pub const ZERO_COUNT_I64_FIELD: &str = "zero_count_i64";
|
||||
pub const POSITIVE_BUCKETS_I64_FIELD: &str = "positive_buckets_i64";
|
||||
pub const NEGATIVE_BUCKETS_I64_FIELD: &str = "negative_buckets_i64";
|
||||
pub const COUNT_F64_FIELD: &str = "count_f64";
|
||||
@@ -72,8 +72,8 @@ pub const NATIVE_HISTOGRAM_FIELD_NAMES: &[&str] = &[
|
||||
POSITIVE_SPAN_LENGTHS_FIELD,
|
||||
NEGATIVE_SPAN_OFFSETS_FIELD,
|
||||
NEGATIVE_SPAN_LENGTHS_FIELD,
|
||||
COUNT_U64_FIELD,
|
||||
ZERO_COUNT_U64_FIELD,
|
||||
COUNT_I64_FIELD,
|
||||
ZERO_COUNT_I64_FIELD,
|
||||
POSITIVE_BUCKETS_I64_FIELD,
|
||||
NEGATIVE_BUCKETS_I64_FIELD,
|
||||
COUNT_F64_FIELD,
|
||||
@@ -108,9 +108,9 @@ pub fn native_histogram_field_type(name: &str) -> Option<ConcreteDataType> {
|
||||
ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::int32_datatype())),
|
||||
),
|
||||
POSITIVE_SPAN_LENGTHS_FIELD | NEGATIVE_SPAN_LENGTHS_FIELD => Some(
|
||||
ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::uint32_datatype())),
|
||||
ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::int32_datatype())),
|
||||
),
|
||||
COUNT_U64_FIELD | ZERO_COUNT_U64_FIELD => Some(ConcreteDataType::uint64_datatype()),
|
||||
COUNT_I64_FIELD | ZERO_COUNT_I64_FIELD => Some(ConcreteDataType::int64_datatype()),
|
||||
POSITIVE_BUCKETS_I64_FIELD | NEGATIVE_BUCKETS_I64_FIELD => Some(
|
||||
ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::int64_datatype())),
|
||||
),
|
||||
@@ -195,7 +195,7 @@ pub struct Span {
|
||||
/// The first bucket index, or the gap after the preceding span.
|
||||
pub offset: i32,
|
||||
/// Number of consecutive buckets in the span.
|
||||
pub length: u32,
|
||||
pub length: i32,
|
||||
}
|
||||
|
||||
/// Inclusion rules for a materialized bucket's lower and upper bounds.
|
||||
@@ -1313,7 +1313,7 @@ where
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn read_spans(offsets: Vec<i32>, lengths: Vec<u32>, name: &str) -> DfResult<Vec<Span>> {
|
||||
fn read_spans(offsets: Vec<i32>, lengths: Vec<i32>, name: &str) -> DfResult<Vec<Span>> {
|
||||
if offsets.len() != lengths.len() {
|
||||
return Err(DataFusionError::Execution(format!(
|
||||
"native histogram {name} span offsets and lengths mismatch: {} vs {}",
|
||||
@@ -1321,11 +1321,18 @@ fn read_spans(offsets: Vec<i32>, lengths: Vec<u32>, name: &str) -> DfResult<Vec<
|
||||
lengths.len()
|
||||
)));
|
||||
}
|
||||
Ok(offsets
|
||||
offsets
|
||||
.into_iter()
|
||||
.zip(lengths)
|
||||
.map(|(offset, length)| Span { offset, length })
|
||||
.collect())
|
||||
.map(|(offset, length)| {
|
||||
if length < 0 {
|
||||
return Err(DataFusionError::Execution(format!(
|
||||
"native histogram {name} span has negative length {length}"
|
||||
)));
|
||||
}
|
||||
Ok(Span { offset, length })
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn check_span_bucket_count(spans: &[Span], buckets: usize, name: &str) -> DfResult<()> {
|
||||
@@ -1352,12 +1359,12 @@ pub fn read_histogram(array: &StructArray, row: usize) -> DfResult<Option<Native
|
||||
let schema = required_primitive::<Int32Type>(array, SCHEMA_FIELD, row)?;
|
||||
let positive_spans = read_spans(
|
||||
list_values::<Int32Type>(array, POSITIVE_SPAN_OFFSETS_FIELD, row)?,
|
||||
list_values::<UInt32Type>(array, POSITIVE_SPAN_LENGTHS_FIELD, row)?,
|
||||
list_values::<Int32Type>(array, POSITIVE_SPAN_LENGTHS_FIELD, row)?,
|
||||
"positive",
|
||||
)?;
|
||||
let negative_spans = read_spans(
|
||||
list_values::<Int32Type>(array, NEGATIVE_SPAN_OFFSETS_FIELD, row)?,
|
||||
list_values::<UInt32Type>(array, NEGATIVE_SPAN_LENGTHS_FIELD, row)?,
|
||||
list_values::<Int32Type>(array, NEGATIVE_SPAN_LENGTHS_FIELD, row)?,
|
||||
"negative",
|
||||
)?;
|
||||
|
||||
@@ -1372,8 +1379,8 @@ pub fn read_histogram(array: &StructArray, row: usize) -> DfResult<Option<Native
|
||||
)
|
||||
} else {
|
||||
(
|
||||
required_primitive::<UInt64Type>(array, COUNT_U64_FIELD, row)? as f64,
|
||||
optional_primitive::<UInt64Type>(array, ZERO_COUNT_U64_FIELD, row)?
|
||||
required_primitive::<Int64Type>(array, COUNT_I64_FIELD, row)? as f64,
|
||||
optional_primitive::<Int64Type>(array, ZERO_COUNT_I64_FIELD, row)?
|
||||
.unwrap_or_default() as f64,
|
||||
list_values::<Int64Type>(array, POSITIVE_BUCKETS_I64_FIELD, row)?
|
||||
.into_iter()
|
||||
@@ -1429,8 +1436,8 @@ pub fn build_histogram_array(values: &[Option<NativeHistogram>]) -> ArrayRef {
|
||||
let mut positive_span_lengths = Vec::with_capacity(values.len());
|
||||
let mut negative_span_offsets = Vec::with_capacity(values.len());
|
||||
let mut negative_span_lengths = Vec::with_capacity(values.len());
|
||||
let mut count_u64 = Vec::<Option<u64>>::with_capacity(values.len());
|
||||
let mut zero_count_u64 = Vec::<Option<u64>>::with_capacity(values.len());
|
||||
let mut count_i64 = Vec::<Option<i64>>::with_capacity(values.len());
|
||||
let mut zero_count_i64 = Vec::<Option<i64>>::with_capacity(values.len());
|
||||
let mut positive_buckets_i64 = Vec::with_capacity(values.len());
|
||||
let mut negative_buckets_i64 = Vec::with_capacity(values.len());
|
||||
let mut count_f64 = Vec::with_capacity(values.len());
|
||||
@@ -1476,8 +1483,8 @@ pub fn build_histogram_array(values: &[Option<NativeHistogram>]) -> ArrayRef {
|
||||
.map(|span| span.length)
|
||||
.collect(),
|
||||
));
|
||||
count_u64.push(None);
|
||||
zero_count_u64.push(None);
|
||||
count_i64.push(None);
|
||||
zero_count_i64.push(None);
|
||||
positive_buckets_i64.push(list_opt(Vec::<i64>::new()));
|
||||
negative_buckets_i64.push(list_opt(Vec::<i64>::new()));
|
||||
count_f64.push(Some(histogram.count));
|
||||
@@ -1495,8 +1502,8 @@ pub fn build_histogram_array(values: &[Option<NativeHistogram>]) -> ArrayRef {
|
||||
positive_span_lengths.push(None);
|
||||
negative_span_offsets.push(None);
|
||||
negative_span_lengths.push(None);
|
||||
count_u64.push(None);
|
||||
zero_count_u64.push(None);
|
||||
count_i64.push(None);
|
||||
zero_count_i64.push(None);
|
||||
positive_buckets_i64.push(None);
|
||||
negative_buckets_i64.push(None);
|
||||
count_f64.push(None);
|
||||
@@ -1532,7 +1539,7 @@ pub fn build_histogram_array(values: &[Option<NativeHistogram>]) -> ArrayRef {
|
||||
),
|
||||
(
|
||||
POSITIVE_SPAN_LENGTHS_FIELD,
|
||||
Arc::new(ListArray::from_iter_primitive::<UInt32Type, _, _>(
|
||||
Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(
|
||||
positive_span_lengths,
|
||||
)),
|
||||
),
|
||||
@@ -1544,14 +1551,14 @@ pub fn build_histogram_array(values: &[Option<NativeHistogram>]) -> ArrayRef {
|
||||
),
|
||||
(
|
||||
NEGATIVE_SPAN_LENGTHS_FIELD,
|
||||
Arc::new(ListArray::from_iter_primitive::<UInt32Type, _, _>(
|
||||
Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(
|
||||
negative_span_lengths,
|
||||
)),
|
||||
),
|
||||
(COUNT_U64_FIELD, Arc::new(UInt64Array::from(count_u64))),
|
||||
(COUNT_I64_FIELD, Arc::new(Int64Array::from(count_i64))),
|
||||
(
|
||||
ZERO_COUNT_U64_FIELD,
|
||||
Arc::new(UInt64Array::from(zero_count_u64)),
|
||||
ZERO_COUNT_I64_FIELD,
|
||||
Arc::new(Int64Array::from(zero_count_i64)),
|
||||
),
|
||||
(
|
||||
POSITIVE_BUCKETS_I64_FIELD,
|
||||
@@ -1662,7 +1669,7 @@ mod tests {
|
||||
|
||||
assert_eq!(read_histogram(array, 0).unwrap(), Some(expected));
|
||||
assert!(
|
||||
primitive_child::<UInt64Type>(array, COUNT_U64_FIELD)
|
||||
primitive_child::<Int64Type>(array, COUNT_I64_FIELD)
|
||||
.unwrap()
|
||||
.is_null(0)
|
||||
);
|
||||
|
||||
@@ -2751,7 +2751,7 @@ mod tests {
|
||||
custom_values: Vec::new(),
|
||||
positive_spans: vec![Span {
|
||||
offset: 0,
|
||||
length: positive_buckets.len() as u32,
|
||||
length: positive_buckets.len() as i32,
|
||||
}],
|
||||
negative_spans: Vec::new(),
|
||||
count,
|
||||
|
||||
@@ -10026,19 +10026,19 @@ mod test {
|
||||
.unwrap()
|
||||
.display_indent_schema()
|
||||
.to_string();
|
||||
let expected = r#"Filter: greptime_value IS NOT NULL OR greptime_native_histogram IS NOT NULL [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
Projection: some_metric.timestamp, prom_mixed_range_float(Utf8("sum_over_time"), timestamp_range, greptime_value, greptime_native_histogram) AS greptime_value, prom_mixed_range_histogram(Utf8("sum_over_time"), timestamp_range, greptime_value, greptime_native_histogram) AS greptime_native_histogram, some_metric.tag_0 [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
PromRangeManipulate: req range=[0..100000000], interval=[5000], eval range=[600000], time index=[timestamp], values=["greptime_value", "greptime_native_histogram"] [timestamp:Timestamp(ms), greptime_value:Dictionary(Int64, Float64);N, greptime_native_histogram:Dictionary(Int64, Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64)));N, tag_0:Utf8, timestamp_range:Dictionary(Int64, Timestamp(ms))]
|
||||
PromSeriesDivide: tags=["tag_0"] [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
Filter: greptime_value IS NOT NULL OR greptime_native_histogram IS NOT NULL [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
Projection: some_metric.timestamp, prom_mixed_range_float(Utf8("rate"), timestamp_range, greptime_value, greptime_native_histogram, some_metric.timestamp, Int64(300000)) AS greptime_value, prom_mixed_range_histogram(Utf8("rate"), timestamp_range, greptime_value, greptime_native_histogram, some_metric.timestamp, Int64(300000)) AS greptime_native_histogram, some_metric.tag_0 [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
PromRangeManipulate: req range=[-540000..100000000], interval=[60000], eval range=[300000], time index=[timestamp], values=["greptime_native_histogram", "greptime_value"] [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Dictionary(Int64, Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64)));N, greptime_value:Dictionary(Int64, Float64);N, timestamp_range:Dictionary(Int64, Timestamp(ms))]
|
||||
PromSeriesNormalize: offset=[0], time index=[timestamp], filter NaN: [true] [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]
|
||||
PromSeriesDivide: tags=["tag_0"] [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]
|
||||
Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]
|
||||
Filter: some_metric.timestamp >= TimestampMillisecond(-839999, None) AND some_metric.timestamp <= TimestampMillisecond(100000000, None) [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]
|
||||
TableScan: some_metric [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(UInt32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(UInt32), "count_u64": UInt64, "zero_count_u64": UInt64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]"#;
|
||||
let expected = r#"Filter: greptime_value IS NOT NULL OR greptime_native_histogram IS NOT NULL [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
Projection: some_metric.timestamp, prom_mixed_range_float(Utf8("sum_over_time"), timestamp_range, greptime_value, greptime_native_histogram) AS greptime_value, prom_mixed_range_histogram(Utf8("sum_over_time"), timestamp_range, greptime_value, greptime_native_histogram) AS greptime_native_histogram, some_metric.tag_0 [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
PromRangeManipulate: req range=[0..100000000], interval=[5000], eval range=[600000], time index=[timestamp], values=["greptime_value", "greptime_native_histogram"] [timestamp:Timestamp(ms), greptime_value:Dictionary(Int64, Float64);N, greptime_native_histogram:Dictionary(Int64, Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64)));N, tag_0:Utf8, timestamp_range:Dictionary(Int64, Timestamp(ms))]
|
||||
PromSeriesDivide: tags=["tag_0"] [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
Filter: greptime_value IS NOT NULL OR greptime_native_histogram IS NOT NULL [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
Projection: some_metric.timestamp, prom_mixed_range_float(Utf8("rate"), timestamp_range, greptime_value, greptime_native_histogram, some_metric.timestamp, Int64(300000)) AS greptime_value, prom_mixed_range_histogram(Utf8("rate"), timestamp_range, greptime_value, greptime_native_histogram, some_metric.timestamp, Int64(300000)) AS greptime_native_histogram, some_metric.tag_0 [timestamp:Timestamp(ms), greptime_value:Float64;N, greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, tag_0:Utf8]
|
||||
PromRangeManipulate: req range=[-540000..100000000], interval=[60000], eval range=[300000], time index=[timestamp], values=["greptime_native_histogram", "greptime_value"] [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Dictionary(Int64, Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64)));N, greptime_value:Dictionary(Int64, Float64);N, timestamp_range:Dictionary(Int64, Timestamp(ms))]
|
||||
PromSeriesNormalize: offset=[0], time index=[timestamp], filter NaN: [true] [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]
|
||||
PromSeriesDivide: tags=["tag_0"] [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]
|
||||
Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]
|
||||
Filter: some_metric.timestamp >= TimestampMillisecond(-839999, None) AND some_metric.timestamp <= TimestampMillisecond(100000000, None) [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]
|
||||
TableScan: some_metric [tag_0:Utf8, timestamp:Timestamp(ms), greptime_native_histogram:Struct("schema": Int32, "zero_threshold": Float64, "sum": Float64, "reset_hint": Int32, "start_timestamp": Timestamp(ms), "custom_values": List(Float64), "positive_span_offsets": List(Int32), "positive_span_lengths": List(Int32), "negative_span_offsets": List(Int32), "negative_span_lengths": List(Int32), "count_i64": Int64, "zero_count_i64": Int64, "positive_buckets_i64": List(Int64), "negative_buckets_i64": List(Int64), "count_f64": Float64, "zero_count_f64": Float64, "positive_buckets_f64": List(Float64), "negative_buckets_f64": List(Float64));N, greptime_value:Float64;N]"#;
|
||||
assert_eq!(plan, expected);
|
||||
}
|
||||
|
||||
|
||||
@@ -47,7 +47,7 @@ Each histogram row stores one Struct field named
|
||||
|
||||
- common scalar children: `schema`, `zero_threshold`, `sum`, `reset_hint`,
|
||||
`start_timestamp`;
|
||||
- count children: `count_u64` / `zero_count_u64` or `count_f64` / `zero_count_f64`;
|
||||
- count children: `count_i64` / `zero_count_i64` or `count_f64` / `zero_count_f64`;
|
||||
- list children for custom values, spans, and positive/negative buckets;
|
||||
- original Prometheus labels as Greptime tags.
|
||||
|
||||
|
||||
@@ -371,6 +371,8 @@ fn native_histogram_struct_value(histogram: &Histogram) -> Result<ValueData> {
|
||||
validate_native_histogram(histogram, uses_float_counts)?;
|
||||
|
||||
let mut items = Vec::with_capacity(NATIVE_HISTOGRAM_FIELD_NAMES.len());
|
||||
let positive_span_lengths = i32_span_lengths("positive", &histogram.positive_spans)?;
|
||||
let negative_span_lengths = i32_span_lengths("negative", &histogram.negative_spans)?;
|
||||
items.extend([
|
||||
pb_value(ValueData::I32Value(histogram.schema)),
|
||||
pb_value(ValueData::F64Value(histogram.zero_threshold)),
|
||||
@@ -381,9 +383,9 @@ fn native_histogram_struct_value(histogram: &Histogram) -> Result<ValueData> {
|
||||
)),
|
||||
f64_list_value(histogram.custom_values.iter().copied()),
|
||||
i32_list_value(histogram.positive_spans.iter().map(|span| span.offset)),
|
||||
u32_list_value(histogram.positive_spans.iter().map(|span| span.length)),
|
||||
i32_list_value(positive_span_lengths.iter().copied()),
|
||||
i32_list_value(histogram.negative_spans.iter().map(|span| span.offset)),
|
||||
u32_list_value(histogram.negative_spans.iter().map(|span| span.length)),
|
||||
i32_list_value(negative_span_lengths.iter().copied()),
|
||||
]);
|
||||
|
||||
if uses_float_counts {
|
||||
@@ -418,9 +420,23 @@ fn native_histogram_struct_value(histogram: &Histogram) -> Result<ValueData> {
|
||||
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)?;
|
||||
let count = i64::try_from(count)
|
||||
.ok()
|
||||
.context(error::InvalidPromRemoteRequestSnafu {
|
||||
msg: format!(
|
||||
"remote write v2 native histogram integer count {count} overflows i64"
|
||||
),
|
||||
})?;
|
||||
let zero_count = i64::try_from(zero_count).ok().context(
|
||||
error::InvalidPromRemoteRequestSnafu {
|
||||
msg: format!(
|
||||
"remote write v2 native histogram integer zero_count {zero_count} overflows i64"
|
||||
),
|
||||
},
|
||||
)?;
|
||||
items.extend([
|
||||
pb_value(ValueData::U64Value(count)),
|
||||
pb_value(ValueData::U64Value(zero_count)),
|
||||
pb_value(ValueData::I64Value(count)),
|
||||
pb_value(ValueData::I64Value(zero_count)),
|
||||
i64_list_value(positive_buckets.iter().copied()),
|
||||
i64_list_value(negative_buckets.iter().copied()),
|
||||
null_pb_value(),
|
||||
@@ -779,8 +795,20 @@ fn i32_list_value(values: impl IntoIterator<Item = i32>) -> Value {
|
||||
list_value(values.into_iter().map(ValueData::I32Value))
|
||||
}
|
||||
|
||||
fn u32_list_value(values: impl IntoIterator<Item = u32>) -> Value {
|
||||
list_value(values.into_iter().map(ValueData::U32Value))
|
||||
fn i32_span_lengths(name: &str, spans: &[BucketSpan]) -> Result<Vec<i32>> {
|
||||
spans
|
||||
.iter()
|
||||
.map(|span| {
|
||||
i32::try_from(span.length)
|
||||
.ok()
|
||||
.context(error::InvalidPromRemoteRequestSnafu {
|
||||
msg: format!(
|
||||
"remote write v2 native histogram {name} span length {} overflows i32",
|
||||
span.length
|
||||
),
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn i64_list_value(values: impl IntoIterator<Item = i64>) -> Value {
|
||||
@@ -1708,8 +1736,8 @@ mod tests {
|
||||
Some(ValueData::I32Value(0))
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_field_value(&rows, 0, COUNT_U64_FIELD),
|
||||
Some(ValueData::U64Value(0))
|
||||
histogram_field_value(&rows, 0, COUNT_I64_FIELD),
|
||||
Some(ValueData::I64Value(0))
|
||||
);
|
||||
assert_eq!(histogram_field_value(&rows, 0, COUNT_F64_FIELD), None);
|
||||
}
|
||||
@@ -1825,8 +1853,8 @@ mod tests {
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
histogram_field_value(&rows, 0, COUNT_U64_FIELD),
|
||||
Some(ValueData::U64Value(0))
|
||||
histogram_field_value(&rows, 0, COUNT_I64_FIELD),
|
||||
Some(ValueData::I64Value(0))
|
||||
);
|
||||
assert_eq!(histogram_field_value(&rows, 0, COUNT_F64_FIELD), None);
|
||||
assert!(matches!(
|
||||
@@ -1839,7 +1867,7 @@ mod tests {
|
||||
POSITIVE_BUCKETS_F64_FIELD
|
||||
)));
|
||||
|
||||
assert_eq!(histogram_field_value(&rows, 1, COUNT_U64_FIELD), None);
|
||||
assert_eq!(histogram_field_value(&rows, 1, COUNT_I64_FIELD), None);
|
||||
assert_eq!(
|
||||
histogram_field_value(&rows, 1, COUNT_F64_FIELD),
|
||||
Some(ValueData::F64Value(6.0))
|
||||
|
||||
@@ -18,7 +18,7 @@ use api::v1::value::ValueData;
|
||||
use api::v1::{ColumnSchema, Rows};
|
||||
use bytes::Bytes;
|
||||
use common_query::native_histogram::{
|
||||
COUNT_U64_FIELD, NATIVE_HISTOGRAM_FIELD_NAMES, POSITIVE_BUCKETS_F64_FIELD,
|
||||
COUNT_I64_FIELD, NATIVE_HISTOGRAM_FIELD_NAMES, POSITIVE_BUCKETS_F64_FIELD,
|
||||
POSITIVE_BUCKETS_I64_FIELD, POSITIVE_SPAN_OFFSETS_FIELD, SCHEMA_FIELD,
|
||||
};
|
||||
use common_query::prelude::greptime_native_histogram;
|
||||
@@ -100,8 +100,8 @@ fn test_decode_remote_write_v2_native_histogram_dump() {
|
||||
Some(ValueData::I32Value(3))
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_field_value(rows, 0, COUNT_U64_FIELD),
|
||||
Some(ValueData::U64Value(24))
|
||||
histogram_field_value(rows, 0, COUNT_I64_FIELD),
|
||||
Some(ValueData::I64Value(24))
|
||||
);
|
||||
assert_eq!(
|
||||
list_i32_values(histogram_field_value(rows, 0, POSITIVE_SPAN_OFFSETS_FIELD)),
|
||||
|
||||
@@ -2878,7 +2878,7 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag
|
||||
"prometheus_remote_write_v2_native_histogram_rows",
|
||||
&client,
|
||||
"select greptime_timestamp, greptime_native_histogram, job, instance from remote_write_v2_latency_seconds order by greptime_timestamp;",
|
||||
"[[3000,{\"count_f64\":null,\"count_u64\":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_u64\":1,\"zero_threshold\":0.001},\"api\",\"localhost:9090\"],[4000,{\"count_f64\":6.0,\"count_u64\":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_u64\":null,\"zero_threshold\":0.002},\"api\",\"localhost:9090\"]]",
|
||||
"[[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\"]]",
|
||||
)
|
||||
.await;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user