diff --git a/src/service/promql/functions/delta.rs b/src/service/promql/functions/delta.rs index 62d9b236c1..dbea303ac3 100644 --- a/src/service/promql/functions/delta.rs +++ b/src/service/promql/functions/delta.rs @@ -49,3 +49,77 @@ impl RangeFunc for DeltaFunc { ) } } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use config::meta::promql::value::{Labels, RangeValue, TimeWindow}; + + use super::*; + + // Test helper + fn delta_test_helper(data: Value, eval_ts: i64) -> Result { + let eval_ctx = EvalContext::new(eval_ts, eval_ts, 0, "test".to_string()); + delta(data, &eval_ctx) + } + + #[test] + fn test_delta_function_single_sample() { + // Single sample should return empty or None + let samples = vec![Sample::new(1000, 10.0)]; + + let range_value = RangeValue { + labels: Labels::default(), + samples, + exemplars: None, + time_window: Some(TimeWindow { + range: Duration::from_secs(2), + offset: Duration::ZERO, + }), + }; + + let matrix = Value::Matrix(vec![range_value]); + let result = delta_test_helper(matrix, 3000).unwrap(); + + match result { + Value::Matrix(m) => { + // With single sample, delta should return empty result + assert_eq!(m.len(), 0); + } + _ => panic!("Expected Matrix result"), + } + } + + #[test] + fn test_delta_function_no_samples() { + // Empty samples should return empty result + let samples = vec![]; + + let range_value = RangeValue { + labels: Labels::default(), + samples, + exemplars: None, + time_window: Some(TimeWindow { + range: Duration::from_secs(2), + offset: Duration::ZERO, + }), + }; + + let matrix = Value::Matrix(vec![range_value]); + let result = delta_test_helper(matrix, 3000).unwrap(); + + match result { + Value::Matrix(m) => { + assert_eq!(m.len(), 0); + } + _ => panic!("Expected Matrix result"), + } + } + + #[test] + fn test_delta_function_name() { + let func = DeltaFunc::new(); + assert_eq!(func.name(), "delta"); + } +} diff --git a/src/service/search/datafusion/distributed_plan/codec/physical_plan_node.rs b/src/service/search/datafusion/distributed_plan/codec/physical_plan_node.rs index 94911dff87..3855d6afc2 100644 --- a/src/service/search/datafusion/distributed_plan/codec/physical_plan_node.rs +++ b/src/service/search/datafusion/distributed_plan/codec/physical_plan_node.rs @@ -28,7 +28,8 @@ use proto::cluster_rpc; use { crate::service::search::datafusion::distributed_plan::enrichment_exec::EnrichmentExec, o2_enterprise::enterprise::search::datafusion::distributed_plan::{ - agg_topk_exec::AggregateTopkExec, streaming_aggs_exec::StreamingAggsExec, tmp_exec::TmpExec, + agg_topk_exec::AggregateTopkExec, streaming_aggs_exec::exec::StreamingAggsExec, + tmp_exec::TmpExec, }, }; diff --git a/src/service/search/datafusion/distributed_plan/codec/streaming_aggs_exec.rs b/src/service/search/datafusion/distributed_plan/codec/streaming_aggs_exec.rs index c420872be1..8da2cc66ee 100644 --- a/src/service/search/datafusion/distributed_plan/codec/streaming_aggs_exec.rs +++ b/src/service/search/datafusion/distributed_plan/codec/streaming_aggs_exec.rs @@ -22,7 +22,7 @@ use datafusion::{ physical_plan::{ExecutionPlan, aggregates::AggregateExec}, }; use datafusion_proto::{physical_plan::AsExecutionPlan, protobuf::PhysicalPlanNode}; -use o2_enterprise::enterprise::search::datafusion::distributed_plan::streaming_aggs_exec::StreamingAggsExec; +use o2_enterprise::enterprise::search::datafusion::distributed_plan::streaming_aggs_exec::exec::StreamingAggsExec; use prost::Message; use proto::cluster_rpc; @@ -116,7 +116,7 @@ mod tests { use datafusion_proto::bytes::{ physical_plan_from_bytes_with_extension_codec, physical_plan_to_bytes_with_extension_codec, }; - use o2_enterprise::enterprise::search::datafusion::distributed_plan::streaming_aggs_exec::StreamingAggsExec; + use o2_enterprise::enterprise::search::datafusion::distributed_plan::streaming_aggs_exec::exec::StreamingAggsExec; use super::*; use crate::service::search::datafusion::udf::str_match_udf::STR_MATCH_UDF;