chore: support streaming agg read cache when disable/enable utf8view (#9803)

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<Value> {
+        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");
+    }
+}

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,
     },
 };
 

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;