From 2123108db08711fac45aa8030658353aa2d7a67c Mon Sep 17 00:00:00 2001 From: discord9 Date: Wed, 15 Jul 2026 09:54:49 +0800 Subject: [PATCH] fix(promql): preserve ordinary NaN samples (#8494) * fix(promql): distinguish stale markers from NaN Signed-off-by: discord9 * fix(promql): preserve ordinary NaN samples Signed-off-by: discord9 --------- Signed-off-by: discord9 --- Cargo.lock | 1 + src/common/query/src/lib.rs | 1 + src/common/query/src/prometheus.rs | 37 +++ src/promql/Cargo.toml | 1 + .../src/extension_plan/instant_manipulate.rs | 215 +++++++++++++++++- src/promql/src/extension_plan/normalize.rs | 127 ++++++++--- src/promql/src/extension_plan/test_util.rs | 11 +- src/promql/src/functions/aggr_over_time.rs | 72 +++++- .../src/http/result/prometheus_resp.rs | 61 ++++- .../standalone/common/promql/scalar.result | 17 +- .../cases/standalone/common/promql/scalar.sql | 7 +- .../common/tql/aggr_over_time.result | 5 +- .../standalone/common/tql/operator.result | 9 +- .../cases/standalone/common/tql/operator.sql | 3 +- 14 files changed, 507 insertions(+), 60 deletions(-) create mode 100644 src/common/query/src/prometheus.rs diff --git a/Cargo.lock b/Cargo.lock index cb6cb42c2e..a813b9468a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11020,6 +11020,7 @@ dependencies = [ "bytemuck", "common-error", "common-macro", + "common-query", "common-recordbatch", "common-telemetry", "criterion 0.7.0", diff --git a/src/common/query/src/lib.rs b/src/common/query/src/lib.rs index bdba49eaf7..e70c4d97e7 100644 --- a/src/common/query/src/lib.rs +++ b/src/common/query/src/lib.rs @@ -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"))] diff --git a/src/common/query/src/prometheus.rs b/src/common/query/src/prometheus.rs new file mode 100644 index 0000000000..709aa5d769 --- /dev/null +++ b/src/common/query/src/prometheus.rs @@ -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 + ))); + } +} diff --git a/src/promql/Cargo.toml b/src/promql/Cargo.toml index 460be8ddd9..36ac213fef 100644 --- a/src/promql/Cargo.toml +++ b/src/promql/Cargo.toml @@ -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 diff --git a/src/promql/src/extension_plan/instant_manipulate.rs b/src/promql/src/extension_plan/instant_manipulate.rs index 7318b9cba5..1897b453f3 100644 --- a/src/promql/src/extension_plan/instant_manipulate.rs +++ b/src/promql/src/extension_plan/instant_manipulate.rs @@ -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, 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::() + .unwrap() + .values() + .iter() + .copied() + }) + .collect::>(); + let timestamps = batches + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values() + .iter() + .copied() + }) + .collect::>(); + + 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![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::(), + 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::(), 2); + let batch = batches.iter().find(|batch| batch.num_rows() == 2).unwrap(); + let timestamps = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + + assert_eq!(timestamps.values(), &[1_000, 1_500]); + assert!(!values.is_valid(0)); + assert!(!values.is_valid(1)); + } } diff --git a/src/promql/src/extension_plan/normalize.rs b/src/promql/src/extension_plan/normalize.rs index 6e24f3869b..a84d5fc5c0 100644 --- a/src/promql/src/extension_plan/normalize.rs +++ b/src/promql/src/extension_plan/normalize.rs @@ -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, 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>( offset: Millisecond, time_index_column_name: N, - need_filter_out_nan: bool, + filter_stale_markers: bool, tag_columns: Vec, 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, input: Arc, @@ -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::() { - 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::(), 2); + let batch = batches.iter().find(|batch| batch.num_rows() == 2).unwrap(); + let value = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + let auxiliary = batch + .column(2) + .as_any() + .downcast_ref::() + .unwrap(); + + assert_eq!(value.value(0), 42.0); + assert_eq!(auxiliary.value(0).to_bits(), 0x7ff8_0000_0000_0000); + assert!(!value.is_valid(1)); + } } diff --git a/src/promql/src/extension_plan/test_util.rs b/src/promql/src/extension_plan/test_util.rs index 5521d151c2..43d0d68603 100644 --- a/src/promql/src/extension_plan/test_util.rs +++ b/src/promql/src/extension_plan/test_util.rs @@ -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( diff --git a/src/promql/src/functions/aggr_over_time.rs b/src/promql/src/functions/aggr_over_time.rs index 518c8689ef..05b2b1b2e2 100644 --- a/src/promql/src/functions/aggr_over_time.rs +++ b/src/promql/src/functions/aggr_over_time.rs @@ -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 { - 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 { - 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, expected: Option) { + 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>, + expected_min: Option, + expected_max: Option, + ) { + let timestamps = TimestampMillisecondArray::from(vec![0; values.len()]); + let values = Float64Array::from(values); + + assert_over_time_value(min_over_time(×tamps, &values), expected_min); + assert_over_time_value(max_over_time(×tamps, &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( diff --git a/src/servers/src/http/result/prometheus_resp.rs b/src/servers/src/http/result/prometheus_resp.rs index 6f2b115686..f0d1689436 100644 --- a/src/servers/src/http/result/prometheus_resp.rs +++ b/src/servers/src/http/result/prometheus_resp.rs @@ -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())] + ); + } +} diff --git a/tests/cases/standalone/common/promql/scalar.result b/tests/cases/standalone/common/promql/scalar.result index c3292b4f5c..ec1c768f69 100644 --- a/tests/cases/standalone/common/promql/scalar.result +++ b/tests/cases/standalone/common/promql/scalar.result @@ -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 | +---------------------+-----------------------------------------+ diff --git a/tests/cases/standalone/common/promql/scalar.sql b/tests/cases/standalone/common/promql/scalar.sql index 662f9665fe..b0c85d89e3 100644 --- a/tests/cases/standalone/common/promql/scalar.sql +++ b/tests/cases/standalone/common/promql/scalar.sql @@ -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))); diff --git a/tests/cases/standalone/common/tql/aggr_over_time.result b/tests/cases/standalone/common/tql/aggr_over_time.result index c846d73c05..e39c9d5238 100644 --- a/tests/cases/standalone/common/tql/aggr_over_time.result +++ b/tests/cases/standalone/common/tql/aggr_over_time.result @@ -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; diff --git a/tests/cases/standalone/common/tql/operator.result b/tests/cases/standalone/common/tql/operator.result index bc105b1454..bc0971c3f3 100644 --- a/tests/cases/standalone/common/tql/operator.result +++ b/tests/cases/standalone/common/tql/operator.result @@ -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 diff --git a/tests/cases/standalone/common/tql/operator.sql b/tests/cases/standalone/common/tql/operator.sql index 2ab77dceb9..8d10f62bcc 100644 --- a/tests/cases/standalone/common/tql/operator.sql +++ b/tests/cases/standalone/common/tql/operator.sql @@ -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; -