added a separate error type

Author: milindsrivastava1997

asap-query-engine/src/engines/simple_engine/mod.rs

@@ -16,6 +16,7 @@ use std::sync::{Arc, RwLock};
 use std::time::Instant;
 use tracing::{debug, warn};
 
+use crate::precompute_operators::AccumulatorError;
 use crate::AggregateCore;
 
 use asap_types::enums::WindowType;
@@ -1094,9 +1095,9 @@ impl SimpleEngine {
     fn merge_accumulators(
         &self,
         accumulators: &[Box<dyn crate::data_model::AggregateCore>],
-    ) -> Result<Box<dyn crate::data_model::AggregateCore>, String> {
+    ) -> Result<Box<dyn crate::data_model::AggregateCore>, AccumulatorError> {
         if accumulators.is_empty() {
-            return Err("merge_accumulators called with empty slice".to_string());
+            return Err(AccumulatorError::EmptySlice);
         }
 
         if accumulators.len() == 1 {

asap-query-engine/src/precompute_operators/delta_set_aggregator_accumulator.rs

@@ -5,6 +5,7 @@ use crate::data_model::{
 use asap_sketchlib::sketches::delta_set_aggregator::{deserialize_msgpack, serialize_msgpack};
 use serde_json::Value;
 use std::collections::{HashMap, HashSet};
+use tracing::warn;
 
 use promql_utilities::query_logics::enums::Statistic;
 
@@ -247,6 +248,10 @@ impl AggregateCore for DeltaSetAggregatorAccumulator {
 
     fn get_keys(&self) -> Option<Vec<KeyByLabelValues>> {
         if !self.removed.is_empty() {
+            warn!(
+                "DeltaSetAggregatorAccumulator::get_keys called with {} removed items; returning None",
+                self.removed.len()
+            );
             return None;
         }
         Some(self.added.iter().cloned().collect())

asap-query-engine/src/precompute_operators/min_max_accumulator.rs

@@ -1,3 +1,4 @@
+use super::error::AccumulatorError;
 use crate::data_model::{
     AggregateCore, AggregationType, MergeableAccumulator, SerializableToSink,
     SingleSubpopulationAggregate, SingleSubpopulationAggregateFactory,
@@ -29,17 +30,17 @@ impl MinMaxAccumulator {
         }
     }
 
-    pub fn new(sub_type: String) -> Result<Self, String> {
+    pub fn new(sub_type: String) -> Result<Self, AccumulatorError> {
         match sub_type.as_str() {
             "min" => Ok(Self::new_min()),
             "max" => Ok(Self::new_max()),
-            _ => Err(format!("sub_type must be 'min' or 'max', got '{sub_type}'")),
+            _ => Err(AccumulatorError::InvalidSubType(sub_type)),
         }
     }
 
-    pub fn with_value(value: f64, sub_type: String) -> Result<Self, String> {
+    pub fn with_value(value: f64, sub_type: String) -> Result<Self, AccumulatorError> {
         if sub_type != "min" && sub_type != "max" {
-            return Err(format!("sub_type must be 'min' or 'max', got '{sub_type}'"));
+            return Err(AccumulatorError::InvalidSubType(sub_type));
         }
         Ok(Self { value, sub_type })
     }
@@ -120,15 +121,19 @@ impl MergeableAccumulator<MinMaxAccumulator> for MinMaxAccumulator {
         accumulators: Vec<MinMaxAccumulator>,
     ) -> Result<MinMaxAccumulator, Box<dyn std::error::Error + Send + Sync>> {
         if accumulators.is_empty() {
-            return Err("No accumulators to merge".into());
+            return Err(AccumulatorError::EmptySlice.into());
         }
 
         let sub_type = &accumulators[0].sub_type;
 
         // Verify all accumulators have the same sub_type
         for acc in &accumulators {
             if acc.sub_type != *sub_type {
-                return Err("Cannot merge accumulators with different sub_types".into());
+                return Err(AccumulatorError::MergeTypeMismatch {
+                    expected: sub_type.clone(),
+                    got: acc.sub_type.clone(),
+                }
+                .into());
             }
         }
 
@@ -397,4 +402,16 @@ mod tests {
         assert!(acc.query(Statistic::Min, None).is_err());
         assert_eq!(acc.type_name(), "MinMaxAccumulator");
     }
+
+    #[test]
+    fn test_new_invalid_sub_type() {
+        let err = MinMaxAccumulator::new("mean".to_string()).unwrap_err();
+        assert!(matches!(err, AccumulatorError::InvalidSubType(s) if s == "mean"));
+    }
+
+    #[test]
+    fn test_with_value_invalid_sub_type() {
+        let err = MinMaxAccumulator::with_value(1.0, "sum".to_string()).unwrap_err();
+        assert!(matches!(err, AccumulatorError::InvalidSubType(s) if s == "sum"));
+    }
 }

asap-query-engine/src/precompute_operators/mod.rs

@@ -2,6 +2,7 @@ pub mod count_min_sketch_accumulator;
 pub mod count_min_sketch_with_heap_accumulator;
 pub mod datasketches_kll_accumulator;
 pub mod delta_set_aggregator_accumulator;
+pub mod error;
 pub mod hydra_kll_accumulator;
 pub mod increase_accumulator;
 pub mod min_max_accumulator;
@@ -15,6 +16,7 @@ pub use count_min_sketch_accumulator::*;
 pub use count_min_sketch_with_heap_accumulator::*;
 pub use datasketches_kll_accumulator::*;
 pub use delta_set_aggregator_accumulator::*;
+pub use error::AccumulatorError;
 pub use hydra_kll_accumulator::*;
 pub use increase_accumulator::*;
 pub use min_max_accumulator::*;

asap-query-engine/src/precompute_operators/multiple_min_max_accumulator.rs

@@ -1,3 +1,4 @@
+use super::error::AccumulatorError;
 use crate::data_model::{
     AggregateCore, AggregationType, KeyByLabelValues, MergeableAccumulator,
     MultipleSubpopulationAggregate, SerializableToSink,
@@ -17,9 +18,9 @@ pub struct MultipleMinMaxAccumulator {
 }
 
 impl MultipleMinMaxAccumulator {
-    pub fn new(sub_type: String) -> Result<Self, String> {
+    pub fn new(sub_type: String) -> Result<Self, AccumulatorError> {
         if sub_type != "min" && sub_type != "max" {
-            return Err(format!("sub_type must be 'min' or 'max', got '{sub_type}'"));
+            return Err(AccumulatorError::InvalidSubType(sub_type));
         }
 
         Ok(Self {
@@ -45,9 +46,9 @@ impl MultipleMinMaxAccumulator {
     pub fn new_with_values(
         values: HashMap<KeyByLabelValues, f64>,
         sub_type: String,
-    ) -> Result<Self, String> {
+    ) -> Result<Self, AccumulatorError> {
         if sub_type != "min" && sub_type != "max" {
-            return Err(format!("sub_type must be 'min' or 'max', got '{sub_type}'"));
+            return Err(AccumulatorError::InvalidSubType(sub_type));
         }
 
         Ok(Self { values, sub_type })
@@ -310,15 +311,19 @@ impl MergeableAccumulator<MultipleMinMaxAccumulator> for MultipleMinMaxAccumulat
         accumulators: Vec<MultipleMinMaxAccumulator>,
     ) -> Result<MultipleMinMaxAccumulator, Box<dyn std::error::Error + Send + Sync>> {
         if accumulators.is_empty() {
-            return Err("No accumulators to merge".into());
+            return Err(AccumulatorError::EmptySlice.into());
         }
 
         let sub_type = accumulators[0].sub_type.clone();
 
         // Verify all accumulators have the same sub_type
         for acc in &accumulators {
             if acc.sub_type != sub_type {
-                return Err("Cannot merge accumulators with different sub_types".into());
+                return Err(AccumulatorError::MergeTypeMismatch {
+                    expected: sub_type.clone(),
+                    got: acc.sub_type.clone(),
+                }
+                .into());
             }
         }
 
@@ -490,4 +495,17 @@ mod tests {
         // Test type name through trait object
         assert_eq!(trait_obj.type_name(), "MultipleMinMaxAccumulator");
     }
+
+    #[test]
+    fn test_new_invalid_sub_type() {
+        let err = MultipleMinMaxAccumulator::new("mean".to_string()).unwrap_err();
+        assert!(matches!(err, AccumulatorError::InvalidSubType(s) if s == "mean"));
+    }
+
+    #[test]
+    fn test_new_with_values_invalid_sub_type() {
+        let err = MultipleMinMaxAccumulator::new_with_values(HashMap::new(), "sum".to_string())
+            .unwrap_err();
+        assert!(matches!(err, AccumulatorError::InvalidSubType(s) if s == "sum"));
+    }
 }