rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-08-18 12:08:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(promql): preserve ordinary NaN samples (#8494)

Browse Source 
* fix(promql): distinguish stale markers from NaN

Signed-off-by: discord9 <discord9@163.com>

* fix(promql): preserve ordinary NaN samples

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-07-15 09:54:49 +08:00
committed by GitHub
parent 1a476f76e2
commit 2123108db0
14 changed files with 507 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -11020,6 +11020,7 @@ dependencies = [
"bytemuck",
"common-error",
"common-macro",
"common-query",
"common-recordbatch",
"common-telemetry",
"criterion 0.7.0",
src/common/query/src/lib.rs
+1
View File
@@ -17,6 +17,7 @@ pub mod error;
pub mod logical_plan;
pub mod native_histogram;
pub mod prelude;
pub mod prometheus;
pub mod request;
pub mod stream;
#[cfg(any(test, feature = "testing"))]
src/common/query/src/prometheus.rs
+37
View File
@@ -0,0 +1,37 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/// Canonical Prometheus stale-marker NaN bit pattern.
pub const PROMETHEUS_STALE_NAN_BITS: u64 = 0x7ff0_0000_0000_0002;
/// Returns whether `value` is the canonical Prometheus stale-marker NaN.
#[inline]
pub fn is_prometheus_stale_nan(value: f64) -> bool {
value.to_bits() == PROMETHEUS_STALE_NAN_BITS
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn recognizes_only_the_canonical_stale_marker() {
assert!(is_prometheus_stale_nan(f64::from_bits(
0x7ff0_0000_0000_0002
)));
assert!(!is_prometheus_stale_nan(f64::from_bits(
0x7ff8_0000_0000_0000
)));
}
}
src/promql/Cargo.toml
+1
View File
@@ -13,6 +13,7 @@ async-trait.workspace = true
bytemuck.workspace = true
common-error.workspace = true
common-macro.workspace = true
common-query.workspace = true
common-telemetry.workspace = true
datafusion.workspace = true
datafusion-common.workspace = true
src/promql/src/extension_plan/instant_manipulate.rs
+203 -12
View File
@@ -18,6 +18,7 @@ use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use common_query::prometheus::is_prometheus_stale_nan;
use datafusion::arrow::array::{Array, Float64Array, TimestampMillisecondArray, UInt64Array};
use datafusion::arrow::datatypes::{DataType, SchemaRef};
use datafusion::arrow::record_batch::RecordBatch;
@@ -579,9 +580,10 @@ impl InstantManipulateStream {
match curr.cmp(&expected_ts) {
Ordering::Equal => {
if let Some(field_column) = &field_column
&& field_column.value(cursor).is_nan()
&& field_column.is_valid(cursor)
&& is_prometheus_stale_nan(field_column.value(cursor))
{
// ignore the NaN value
// Ignore the stale marker.
} else {
take_indices.push(cursor as u64);
aligned_ts.push(expected_ts);
@@ -613,9 +615,10 @@ impl InstantManipulateStream {
if prev_ts + self.lookback_delta > expected_ts {
// only use the point in the time range
if let Some(field_column) = &field_column
&& field_column.value(prev_cursor).is_nan()
&& field_column.is_valid(prev_cursor)
&& is_prometheus_stale_nan(field_column.value(prev_cursor))
{
// if the newest value is NaN, it means the value is stale, so we should not use it
// Do not use a stale marker as the newest value.
continue;
}
// use this point
@@ -624,9 +627,10 @@ impl InstantManipulateStream {
}
}
} else if let Some(field_column) = &field_column
&& field_column.value(cursor).is_nan()
&& field_column.is_valid(cursor)
&& is_prometheus_stale_nan(field_column.value(cursor))
{
// if the newest value is NaN, it means the value is stale, so we should not use it
// Do not use a stale marker as the newest value.
} else {
// use this point
take_indices.push(cursor as u64);
@@ -688,6 +692,7 @@ fn reuse_constant_column(array: &Arc<dyn Array>, len: usize) -> DataFusionResult
#[cfg(test)]
mod test {
use datafusion::arrow::buffer::NullBuffer;
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::common::ToDFSchema;
use datafusion::datasource::memory::MemorySourceConfig;
@@ -697,7 +702,7 @@ mod test {
use super::*;
use crate::extension_plan::test_util::{
TIME_INDEX_COLUMN, prepare_test_data, prepare_test_data_with_nan,
TIME_INDEX_COLUMN, prepare_test_data, prepare_test_data_with_stale_marker,
};
async fn do_normalize_test(
@@ -706,10 +711,10 @@ mod test {
lookback_delta: Millisecond,
interval: Millisecond,
expected: String,
contains_nan: bool,
contains_stale_marker: bool,
) {
let memory_exec = if contains_nan {
Arc::new(prepare_test_data_with_nan())
let memory_exec = if contains_stale_marker {
Arc::new(prepare_test_data_with_stale_marker())
} else {
Arc::new(prepare_test_data())
};
@@ -1255,7 +1260,7 @@ mod test {
}
#[tokio::test]
async fn lookback_10s_interval_10s_with_nan() {
async fn lookback_10s_interval_10s_with_stale_marker() {
let expected = String::from(
"+---------------------+-------+\
\n| timestamp | value |\
@@ -1269,7 +1274,7 @@ mod test {
}
#[tokio::test]
async fn lookback_10s_interval_10s_with_nan_unaligned() {
async fn lookback_10s_interval_10s_with_stale_marker_unaligned() {
let expected = String::from(
"+-------------------------+-------+\
\n| timestamp | value |\
@@ -1303,4 +1308,190 @@ mod test {
)
.await;
}
#[tokio::test]
async fn ordinary_nan_is_selected_for_exact_and_lookback() {
let schema = Arc::new(Schema::new(vec![
Field::new(
TIME_INDEX_COLUMN,
DataType::Timestamp(datafusion::arrow::datatypes::TimeUnit::Millisecond, None),
false,
),
Field::new("value", DataType::Float64, true),
]));
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(TimestampMillisecondArray::from(vec![1_000])),
Arc::new(Float64Array::from(vec![f64::NAN])),
],
)
.unwrap();
let input = Arc::new(DataSourceExec::new(Arc::new(
MemorySourceConfig::try_new(&[vec![batch]], schema, None).unwrap(),
)));
let exec = Arc::new(InstantManipulateExec {
start: 1_000,
end: 1_500,
lookback_delta: 1_000,
interval: 500,
time_index_column: TIME_INDEX_COLUMN.to_string(),
field_column: Some("value".to_string()),
reuse_tsid_column: false,
input,
metric: ExecutionPlanMetricsSet::new(),
});
let context = SessionContext::default();
let batches = datafusion::physical_plan::collect(exec, context.task_ctx())
.await
.unwrap();
let values = batches
.iter()
.flat_map(|batch| {
batch
.column(1)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap()
.values()
.iter()
.copied()
})
.collect::<Vec<_>>();
let timestamps = batches
.iter()
.flat_map(|batch| {
batch
.column(0)
.as_any()
.downcast_ref::<TimestampMillisecondArray>()
.unwrap()
.values()
.iter()
.copied()
})
.collect::<Vec<_>>();
assert_eq!(values.len(), 2);
assert_eq!(timestamps, vec![1_000, 1_500]);
assert!(values.iter().all(|value| value.is_nan()));
assert_eq!(
values
.iter()
.map(|value| value.to_bits())
.collect::<Vec<_>>(),
vec![f64::NAN.to_bits(); 2]
);
}
#[tokio::test]
async fn prometheus_stale_nan_suppresses_exact_and_lookback() {
let schema = Arc::new(Schema::new(vec![
Field::new(
TIME_INDEX_COLUMN,
DataType::Timestamp(datafusion::arrow::datatypes::TimeUnit::Millisecond, None),
false,
),
Field::new("value", DataType::Float64, true),
]));
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(TimestampMillisecondArray::from(vec![500, 1_000])),
Arc::new(Float64Array::from(vec![
42.0,
f64::from_bits(0x7ff0_0000_0000_0002),
])),
],
)
.unwrap();
let input = Arc::new(DataSourceExec::new(Arc::new(
MemorySourceConfig::try_new(&[vec![batch]], schema, None).unwrap(),
)));
let exec = Arc::new(InstantManipulateExec {
start: 1_000,
end: 1_500,
lookback_delta: 1_001,
interval: 500,
time_index_column: TIME_INDEX_COLUMN.to_string(),
field_column: Some("value".to_string()),
reuse_tsid_column: false,
input,
metric: ExecutionPlanMetricsSet::new(),
});
let context = SessionContext::default();
let batches = datafusion::physical_plan::collect(exec, context.task_ctx())
.await
.unwrap();
assert_eq!(
batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
0,
"the stale marker must suppress both the exact and lookback selections rather than \
falling back to 42.0"
);
}
#[tokio::test]
async fn null_value_backed_by_stale_bits_is_selected_for_exact_and_lookback() {
let schema = Arc::new(Schema::new(vec![
Field::new(
TIME_INDEX_COLUMN,
DataType::Timestamp(datafusion::arrow::datatypes::TimeUnit::Millisecond, None),
false,
),
Field::new("value", DataType::Float64, true),
]));
let field_column = Float64Array::new(
vec![f64::from_bits(0x7ff0_0000_0000_0002)].into(),
Some(NullBuffer::from(vec![false])),
);
assert!(!field_column.is_valid(0));
assert_eq!(field_column.value(0).to_bits(), 0x7ff0_0000_0000_0002);
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(TimestampMillisecondArray::from(vec![1_000])),
Arc::new(field_column),
],
)
.unwrap();
let input = Arc::new(DataSourceExec::new(Arc::new(
MemorySourceConfig::try_new(&[vec![batch]], schema, None).unwrap(),
)));
let exec = Arc::new(InstantManipulateExec {
start: 1_000,
end: 1_500,
lookback_delta: 1_000,
interval: 500,
time_index_column: TIME_INDEX_COLUMN.to_string(),
field_column: Some("value".to_string()),
reuse_tsid_column: false,
input,
metric: ExecutionPlanMetricsSet::new(),
});
let context = SessionContext::default();
let batches = datafusion::physical_plan::collect(exec, context.task_ctx())
.await
.unwrap();
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
let batch = batches.iter().find(|batch| batch.num_rows() == 2).unwrap();
let timestamps = batch
.column(0)
.as_any()
.downcast_ref::<TimestampMillisecondArray>()
.unwrap();
let values = batch
.column(1)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
assert_eq!(timestamps.values(), &[1_000, 1_500]);
assert!(!values.is_valid(0));
assert!(!values.is_valid(1));
}
}
src/promql/src/extension_plan/normalize.rs
+101 -26
View File
@@ -17,7 +17,8 @@ use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use datafusion::arrow::array::{BooleanArray, Float64Array};
use common_query::prometheus::is_prometheus_stale_nan;
use datafusion::arrow::array::{Array, BooleanArray, Float64Array};
use datafusion::arrow::compute;
use datafusion::common::{DFSchema, DFSchemaRef, Result as DataFusionResult, Statistics};
use datafusion::error::DataFusionError;
@@ -52,12 +53,12 @@ use crate::metrics::PROMQL_SERIES_COUNT;
/// Roughly speaking, this method does these things:
/// - bias sample's timestamp by offset
/// - sort the record batch based on timestamp column
/// - remove NaN values (optional)
/// - remove Prometheus stale markers (optional)
#[derive(Debug, PartialEq, Eq, Hash, PartialOrd)]
pub struct SeriesNormalize {
offset: Millisecond,
time_index_column_name: String,
need_filter_out_nan: bool,
filter_stale_markers: bool,
tag_columns: Vec<String>,
input: LogicalPlan,
@@ -122,7 +123,7 @@ impl UserDefinedLogicalNodeCore for SeriesNormalize {
write!(
f,
"PromSeriesNormalize: offset=[{}], time index=[{}], filter NaN: [{}]",
self.offset, self.time_index_column_name, self.need_filter_out_nan
self.offset, self.time_index_column_name, self.filter_stale_markers
)
}
@@ -158,7 +159,7 @@ impl UserDefinedLogicalNodeCore for SeriesNormalize {
Ok(Self {
offset: self.offset,
time_index_column_name,
need_filter_out_nan: self.need_filter_out_nan,
filter_stale_markers: self.filter_stale_markers,
tag_columns,
input,
unfix: None,
@@ -167,7 +168,7 @@ impl UserDefinedLogicalNodeCore for SeriesNormalize {
Ok(Self {
offset: self.offset,
time_index_column_name: self.time_index_column_name.clone(),
need_filter_out_nan: self.need_filter_out_nan,
filter_stale_markers: self.filter_stale_markers,
tag_columns: self.tag_columns.clone(),
input,
unfix: None,
@@ -180,14 +181,14 @@ impl SeriesNormalize {
pub fn new<N: AsRef<str>>(
offset: Millisecond,
time_index_column_name: N,
need_filter_out_nan: bool,
filter_stale_markers: bool,
tag_columns: Vec<String>,
input: LogicalPlan,
) -> Self {
Self {
offset,
time_index_column_name: time_index_column_name.as_ref().to_string(),
need_filter_out_nan,
filter_stale_markers,
tag_columns,
input,
unfix: None,
@@ -202,7 +203,7 @@ impl SeriesNormalize {
Arc::new(SeriesNormalizeExec {
offset: self.offset,
time_index_column_name: self.time_index_column_name.clone(),
need_filter_out_nan: self.need_filter_out_nan,
filter_stale_markers: self.filter_stale_markers,
input: exec_input,
tag_columns: self.tag_columns.clone(),
metric: ExecutionPlanMetricsSet::new(),
@@ -222,7 +223,7 @@ impl SeriesNormalize {
pb::SeriesNormalize {
offset: self.offset,
time_index_idx,
filter_nan: self.need_filter_out_nan,
filter_nan: self.filter_stale_markers,
tag_column_indices,
..Default::default()
}
@@ -244,7 +245,7 @@ impl SeriesNormalize {
Ok(Self {
offset: pb_normalize.offset,
time_index_column_name: String::new(),
need_filter_out_nan: pb_normalize.filter_nan,
filter_stale_markers: pb_normalize.filter_nan,
tag_columns: Vec::new(),
input: placeholder_plan,
unfix: Some(unfix),
@@ -256,7 +257,7 @@ impl SeriesNormalize {
pub struct SeriesNormalizeExec {
offset: Millisecond,
time_index_column_name: String,
need_filter_out_nan: bool,
filter_stale_markers: bool,
tag_columns: Vec<String>,
input: Arc<dyn ExecutionPlan>,
@@ -303,7 +304,7 @@ impl ExecutionPlan for SeriesNormalizeExec {
Ok(Arc::new(Self {
offset: self.offset,
time_index_column_name: self.time_index_column_name.clone(),
need_filter_out_nan: self.need_filter_out_nan,
filter_stale_markers: self.filter_stale_markers,
input: children[0].clone(),
tag_columns: self.tag_columns.clone(),
metric: self.metric.clone(),
@@ -334,7 +335,7 @@ impl ExecutionPlan for SeriesNormalizeExec {
Ok(Box::pin(SeriesNormalizeStream {
offset: self.offset,
time_index,
need_filter_out_nan: self.need_filter_out_nan,
filter_stale_markers: self.filter_stale_markers,
schema,
input,
metric: baseline_metric,
@@ -364,7 +365,7 @@ impl DisplayAs for SeriesNormalizeExec {
write!(
f,
"PromSeriesNormalizeExec: offset=[{}], time index=[{}], filter NaN: [{}]",
self.offset, self.time_index_column_name, self.need_filter_out_nan
self.offset, self.time_index_column_name, self.filter_stale_markers
)
}
}
@@ -375,7 +376,7 @@ pub struct SeriesNormalizeStream {
offset: Millisecond,
// Column index of TIME INDEX column's position in schema
time_index: usize,
need_filter_out_nan: bool,
filter_stale_markers: bool,
schema: SchemaRef,
input: SendableRecordBatchStream,
@@ -408,25 +409,25 @@ impl SeriesNormalizeStream {
columns[self.time_index] = ts_column_biased;
let result_batch = RecordBatch::try_new(input.schema(), columns)?;
if !self.need_filter_out_nan {
if !self.filter_stale_markers {
return Ok(result_batch);
}
// TODO(ruihang): consider the "special NaN"
// filter out NaN
let mut filter = vec![true; input.num_rows()];
// Filter out Prometheus stale markers.
let mut stale_marker_filter = vec![true; input.num_rows()];
for column in result_batch.columns() {
if let Some(float_column) = column.as_any().downcast_ref::<Float64Array>() {
for (i, flag) in filter.iter_mut().enumerate() {
if float_column.value(i).is_nan() {
for (i, flag) in stale_marker_filter.iter_mut().enumerate() {
if float_column.is_valid(i) && is_prometheus_stale_nan(float_column.value(i)) {
*flag = false;
}
}
}
}
let result = compute::filter_record_batch(&result_batch, &BooleanArray::from(filter))
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?;
let result =
compute::filter_record_batch(&result_batch, &BooleanArray::from(stale_marker_filter))
.map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?;
Ok(result)
}
}
@@ -462,6 +463,7 @@ impl Stream for SeriesNormalizeStream {
#[cfg(test)]
mod test {
use datafusion::arrow::array::Float64Array;
use datafusion::arrow::buffer::NullBuffer;
use datafusion::arrow::datatypes::{
ArrowPrimitiveType, DataType, Field, Schema, TimestampMillisecondType,
};
@@ -522,7 +524,7 @@ mod test {
let normalize_exec = Arc::new(SeriesNormalizeExec {
offset: 0,
time_index_column_name: TIME_INDEX_COLUMN.to_string(),
need_filter_out_nan: true,
filter_stale_markers: true,
input: memory_exec,
tag_columns: vec!["path".to_string()],
metric: ExecutionPlanMetricsSet::new(),
@@ -556,7 +558,7 @@ mod test {
let normalize_exec = Arc::new(SeriesNormalizeExec {
offset: 1_000,
time_index_column_name: TIME_INDEX_COLUMN.to_string(),
need_filter_out_nan: true,
filter_stale_markers: true,
input: memory_exec,
metric: ExecutionPlanMetricsSet::new(),
tag_columns: vec!["path".to_string()],
@@ -583,4 +585,77 @@ mod test {
assert_eq!(result_literal, expected);
}
#[tokio::test]
async fn filters_stale_markers_and_preserves_ordinary_nan() {
let schema = Arc::new(Schema::new(vec![
Field::new(
TIME_INDEX_COLUMN,
TimestampMillisecondType::DATA_TYPE,
false,
),
Field::new("value", DataType::Float64, true),
Field::new("auxiliary", DataType::Float64, true),
]));
let value_column = Float64Array::new(
vec![
42.0,
f64::from_bits(0x7ff0_0000_0000_0002),
24.0,
f64::from_bits(0x7ff0_0000_0000_0002),
]
.into(),
Some(NullBuffer::from(vec![true, true, true, false])),
);
assert!(!value_column.is_valid(3));
assert_eq!(value_column.value(3).to_bits(), 0x7ff0_0000_0000_0002);
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(TimestampMillisecondArray::from(vec![
1_000, 2_000, 3_000, 4_000,
])),
Arc::new(value_column),
Arc::new(Float64Array::from(vec![
f64::from_bits(0x7ff8_0000_0000_0000),
1.0,
f64::from_bits(0x7ff0_0000_0000_0002),
2.0,
])),
],
)
.unwrap();
let input = Arc::new(DataSourceExec::new(Arc::new(
MemorySourceConfig::try_new(&[vec![batch]], schema, None).unwrap(),
)));
let exec = Arc::new(SeriesNormalizeExec {
offset: 0,
time_index_column_name: TIME_INDEX_COLUMN.to_string(),
filter_stale_markers: true,
tag_columns: Vec::new(),
input,
metric: ExecutionPlanMetricsSet::new(),
});
let context = SessionContext::default();
let batches = datafusion::physical_plan::collect(exec, context.task_ctx())
.await
.unwrap();
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
let batch = batches.iter().find(|batch| batch.num_rows() == 2).unwrap();
let value = batch
.column(1)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
let auxiliary = batch
.column(2)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
assert_eq!(value.value(0), 42.0);
assert_eq!(auxiliary.value(0).to_bits(), 0x7ff8_0000_0000_0000);
assert!(!value.is_valid(1));
}
}
src/promql/src/extension_plan/test_util.rs
+9 -2
View File
@@ -16,6 +16,7 @@
use std::sync::Arc;
use common_query::prometheus::PROMETHEUS_STALE_NAN_BITS;
use common_recordbatch::DfRecordBatch as RecordBatch;
use datafusion::arrow::array::Float64Array;
use datafusion::arrow::datatypes::{
@@ -52,7 +53,7 @@ pub(crate) fn prepare_test_data() -> DataSourceExec {
))
}
pub(crate) fn prepare_test_data_with_nan() -> DataSourceExec {
pub(crate) fn prepare_test_data_with_stale_marker() -> DataSourceExec {
let schema = Arc::new(Schema::new(vec![
Field::new(TIME_INDEX_COLUMN, TimestampMillisecondType::DATA_TYPE, true),
Field::new("value", DataType::Float64, true),
@@ -60,7 +61,13 @@ pub(crate) fn prepare_test_data_with_nan() -> DataSourceExec {
let timestamp_column = Arc::new(TimestampMillisecondArray::from(vec![
0, 30_000, 60_000, 90_000, 120_000, // every 30s
])) as _;
let field_column = Arc::new(Float64Array::from(vec![0.0, f64::NAN, 6.0, f64::NAN, 12.0])) as _;
let field_column = Arc::new(Float64Array::from(vec![
0.0,
f64::from_bits(PROMETHEUS_STALE_NAN_BITS),
6.0,
f64::from_bits(PROMETHEUS_STALE_NAN_BITS),
12.0,
])) as _;
let data = RecordBatch::try_new(schema.clone(), vec![timestamp_column, field_column]).unwrap();
DataSourceExec::new(Arc::new(
src/promql/src/functions/aggr_over_time.rs
+70 -2
View File
@@ -43,7 +43,14 @@ pub fn avg_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Op
display_name = prom_min_over_time
)]
pub fn min_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option<f64> {
compute::min(values)
let mut valid_values = values.iter().flatten();
let mut min = valid_values.next()?;
for value in valid_values {
if value < min || min.is_nan() {
min = value;
}
}
Some(min)
}
/// The maximum value of all points in the specified interval.
@@ -53,7 +60,14 @@ pub fn min_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Op
display_name = prom_max_over_time
)]
pub fn max_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option<f64> {
compute::max(values)
let mut valid_values = values.iter().flatten();
let mut max = valid_values.next()?;
for value in valid_values {
if value > max || max.is_nan() {
max = value;
}
}
Some(max)
}
/// The sum of all values in the specified interval.
@@ -183,6 +197,60 @@ mod test {
use super::*;
use crate::functions::test_util::simple_range_udf_runner;
fn assert_over_time_value(actual: Option<f64>, expected: Option<f64>) {
match (actual, expected) {
(Some(actual), Some(expected)) if expected.is_nan() => assert!(actual.is_nan()),
(Some(actual), Some(expected)) => assert_eq!(actual, expected),
(None, None) => {}
(actual, expected) => panic!("expected {expected:?}, got {actual:?}"),
}
}
fn assert_min_max(
values: Vec<Option<f64>>,
expected_min: Option<f64>,
expected_max: Option<f64>,
) {
let timestamps = TimestampMillisecondArray::from(vec![0; values.len()]);
let values = Float64Array::from(values);
assert_over_time_value(min_over_time(&timestamps, &values), expected_min);
assert_over_time_value(max_over_time(&timestamps, &values), expected_max);
}
#[test]
fn min_max_over_time_ignore_ordinary_nan_when_finite_values_exist() {
let ordinary_nan = f64::from_bits(0x7ff8_0000_0000_0000);
assert_min_max(
vec![Some(ordinary_nan), Some(3.0), Some(-2.0)],
Some(-2.0),
Some(3.0),
);
assert_min_max(
vec![Some(3.0), Some(ordinary_nan), Some(-2.0)],
Some(-2.0),
Some(3.0),
);
assert_min_max(
vec![Some(-2.0), Some(3.0), Some(ordinary_nan)],
Some(-2.0),
Some(3.0),
);
assert_min_max(
vec![Some(ordinary_nan), Some(ordinary_nan)],
Some(ordinary_nan),
Some(ordinary_nan),
);
assert_min_max(
vec![Some(3.0), Some(-2.0), Some(1.0)],
Some(-2.0),
Some(3.0),
);
assert_min_max(vec![], None, None);
assert_min_max(vec![None, None], None, None);
}
// build timestamp range and value range arrays for test
fn build_test_range_arrays() -> (RangeArray, RangeArray) {
let ts_array = Arc::new(TimestampMillisecondArray::from_iter(
src/servers/src/http/result/prometheus_resp.rs
+59 -2
View File
@@ -24,6 +24,7 @@ use axum::http::HeaderValue;
use axum::response::{IntoResponse, Response};
use common_error::ext::ErrorExt;
use common_error::status_code::StatusCode;
use common_query::prometheus::is_prometheus_stale_nan;
use common_query::{Output, OutputData};
use common_recordbatch::RecordBatches;
use datatypes::prelude::ConcreteDataType;
@@ -268,8 +269,8 @@ impl PrometheusJsonResponse {
// retrieve value
if field_column.is_valid(row_index) {
let v = field_column.value(row_index);
// ignore all NaN values to reduce the amount of data to be sent.
if v.is_nan() {
// Ignore Prometheus stale markers.
if is_prometheus_stale_nan(v) {
continue;
}
@@ -350,3 +351,59 @@ impl PrometheusJsonResponse {
Ok(data)
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use common_recordbatch::{RecordBatch, RecordBatches};
use datatypes::data_type::ConcreteDataType;
use datatypes::schema::{ColumnSchema, Schema};
use datatypes::vectors::{Float64Vector, TimestampMillisecondVector};
use super::*;
#[test]
fn matrix_response_preserves_ordinary_nan_and_filters_stale_markers() {
let schema = Arc::new(Schema::new(vec![
ColumnSchema::new(
"timestamp",
ConcreteDataType::timestamp_millisecond_datatype(),
false,
),
ColumnSchema::new("value", ConcreteDataType::float64_datatype(), true),
]));
let batch = RecordBatch::new(
schema.clone(),
vec![
Arc::new(TimestampMillisecondVector::from_vec(vec![
1_000, 2_000, 3_000, 4_000,
])) as _,
Arc::new(Float64Vector::from(vec![
Some(1.0),
Some(f64::from_bits(0x7ff8_0000_0000_0000)),
Some(f64::from_bits(0x7ff0_0000_0000_0002)),
None,
])) as _,
],
)
.unwrap();
let batches = RecordBatches::try_new(schema, vec![batch]).unwrap();
let response =
PrometheusJsonResponse::record_batches_to_data(batches, None, ValueType::Matrix)
.unwrap();
let PrometheusResponse::PromData(data) = response else {
panic!("expected Prometheus data response");
};
let PromQueryResult::Matrix(series) = data.result else {
panic!("expected matrix result");
};
assert_eq!(series.len(), 1);
assert_eq!(
series[0].values,
vec![(1.0, "1".to_string()), (2.0, "NaN".to_string())]
);
}
}
tests/cases/standalone/common/promql/scalar.result
+10 -7
View File
@@ -586,8 +586,8 @@ INSERT INTO TABLE presence_metric VALUES
Affected Rows: 15
-- NaN drops `cpu0` from the grouped count, while the NULL sample on `cpu2`
-- still leaves a zero-valued row in `count(...) by (cpu)`.
-- Ordinary NaN keeps `cpu0` present in the grouped count, while the NULL sample
-- on `cpu2` still leaves a zero-valued row in `count(...) by (cpu)`.
-- SQLNESS SORT_RESULT 3 1
TQL EVAL (0, 600, '200s') count(presence_metric{instance="i1"}) by (cpu);
@@ -595,6 +595,8 @@ TQL EVAL (0, 600, '200s') count(presence_metric{instance="i1"}) by (cpu);
| cpu | ts | count(presence_metric.val) |
+------+---------------------+----------------------------+
| cpu0 | 1970-01-01T00:00:00 | 2 |
| cpu0 | 1970-01-01T00:03:20 | 2 |
| cpu0 | 1970-01-01T00:06:40 | 2 |
| cpu0 | 1970-01-01T00:10:00 | 2 |
| cpu1 | 1970-01-01T00:00:00 | 1 |
| cpu1 | 1970-01-01T00:03:20 | 1 |
@@ -605,7 +607,8 @@ TQL EVAL (0, 600, '200s') count(presence_metric{instance="i1"}) by (cpu);
| cpu2 | 1970-01-01T00:06:40 | 0 |
+------+---------------------+----------------------------+
-- Nested-count rewrite should preserve grouped presence after stale-NaN filtering and null-value pruning.
-- Nested-count rewrite should preserve grouped presence after stale-marker filtering
-- and null-value pruning.
-- SQLNESS SORT_RESULT 3 1
TQL EVAL (0, 600, '200s') scalar(count(count(presence_metric{instance="i1"}) by (cpu)));
@@ -613,8 +616,8 @@ TQL EVAL (0, 600, '200s') scalar(count(count(presence_metric{instance="i1"}) by
| ts | scalar(count(count(presence_metric.val))) |
+---------------------+-------------------------------------------+
| 1970-01-01T00:00:00 | 3.0 |
| 1970-01-01T00:03:20 | 2.0 |
| 1970-01-01T00:06:40 | 2.0 |
| 1970-01-01T00:03:20 | 3.0 |
| 1970-01-01T00:06:40 | 3.0 |
| 1970-01-01T00:10:00 | 2.0 |
+---------------------+-------------------------------------------+
@@ -626,8 +629,8 @@ TQL EVAL (0, 600, '200s') scalar(count(sum(presence_metric{instance="i1"}) by (c
| ts | scalar(count(sum(presence_metric.val))) |
+---------------------+-----------------------------------------+
| 1970-01-01T00:00:00 | 3.0 |
| 1970-01-01T00:03:20 | 1.0 |
| 1970-01-01T00:06:40 | 1.0 |
| 1970-01-01T00:03:20 | 2.0 |
| 1970-01-01T00:06:40 | 2.0 |
| 1970-01-01T00:10:00 | 2.0 |
+---------------------+-----------------------------------------+
tests/cases/standalone/common/promql/scalar.sql
+4 -3
View File
@@ -186,12 +186,13 @@ INSERT INTO TABLE presence_metric VALUES
(600000, 'i1', 'cpu0', 'b', 8.0),
(600000, 'i2', 'cpu9', 'a', 103.0);
-- NaN drops `cpu0` from the grouped count, while the NULL sample on `cpu2`
-- still leaves a zero-valued row in `count(...) by (cpu)`.
-- Ordinary NaN keeps `cpu0` present in the grouped count, while the NULL sample
-- on `cpu2` still leaves a zero-valued row in `count(...) by (cpu)`.
-- SQLNESS SORT_RESULT 3 1
TQL EVAL (0, 600, '200s') count(presence_metric{instance="i1"}) by (cpu);
-- Nested-count rewrite should preserve grouped presence after stale-NaN filtering and null-value pruning.
-- Nested-count rewrite should preserve grouped presence after stale-marker filtering
-- and null-value pruning.
-- SQLNESS SORT_RESULT 3 1
TQL EVAL (0, 600, '200s') scalar(count(count(presence_metric{instance="i1"}) by (cpu)));
tests/cases/standalone/common/tql/aggr_over_time.result
+4 -1
View File
@@ -295,6 +295,7 @@ tql eval (60, 60, '1s') min_over_time(data[2m]);
| 1970-01-01T00:01:00 | 0.0 | some_nan |
| 1970-01-01T00:01:00 | 0.0 | some_nan3 |
| 1970-01-01T00:01:00 | 1.0 | some_nan2 |
| 1970-01-01T00:01:00 | NaN | only_nan |
+---------------------+----------------------------------+-----------+
-- eval instant at 1m max_over_time(data[2m])
@@ -313,6 +314,7 @@ tql eval (60, 60, '1s') max_over_time(data[2m]);
| 1970-01-01T00:01:00 | 2.0 | some_nan |
| 1970-01-01T00:01:00 | 2.0 | some_nan2 |
| 1970-01-01T00:01:00 | 3.0 | numbers |
| 1970-01-01T00:01:00 | NaN | only_nan |
+---------------------+----------------------------------+-----------+
-- eval instant at 1m last_over_time(data[2m])
@@ -327,10 +329,11 @@ tql eval (60, 60, '1s') last_over_time(data[2m]);
+---------------------+-----------------------------------+-----------+
| ts | prom_last_over_time(ts_range,val) | ty |
+---------------------+-----------------------------------+-----------+
| 1970-01-01T00:01:00 | 0.0 | some_nan |
| 1970-01-01T00:01:00 | 1.0 | some_nan2 |
| 1970-01-01T00:01:00 | 1.0 | some_nan3 |
| 1970-01-01T00:01:00 | 3.0 | numbers |
| 1970-01-01T00:01:00 | NaN | only_nan |
| 1970-01-01T00:01:00 | NaN | some_nan |
+---------------------+-----------------------------------+-----------+
drop table data;
tests/cases/standalone/common/tql/operator.result
+6 -3
View File
@@ -40,11 +40,14 @@ tql eval (60, 60, '1s') trigy atan2 trigx;
-- eval instant at 1m trigy atan2 trigNaN
-- trigy{} NaN
-- This query doesn't have result because `trignan` is NaN and will be filtered out.
-- Ordinary NaN is a valid sample and propagates through `atan2`.
tql eval (60, 60, '1s') trigy atan2 trignan;
++
++
+---------------------+------------------------------+
| ts | atan2(trigy.val,trignan.val) |
+---------------------+------------------------------+
| 1970-01-01T00:01:00 | NaN |
+---------------------+------------------------------+
-- eval instant at 1m 10 atan2 20
-- 0.4636476090008061
tests/cases/standalone/common/tql/operator.sql
+1 -2
View File
@@ -23,7 +23,7 @@ tql eval (60, 60, '1s') trigy atan2 trigx;
-- eval instant at 1m trigy atan2 trigNaN
-- trigy{} NaN
-- This query doesn't have result because `trignan` is NaN and will be filtered out.
-- Ordinary NaN is a valid sample and propagates through `atan2`.
tql eval (60, 60, '1s') trigy atan2 trignan;
-- eval instant at 1m 10 atan2 20
@@ -73,4 +73,3 @@ VALUES
tql eval (1743465600.5, 1743465610, '1s') irate(t[2s]);
drop table t;