refactor(query-engine): replace panics with Result/None

Author: milindsrivastava1997

asap-query-engine/src/drivers/query/adapters/elastic_http.rs

@@ -91,7 +91,7 @@ impl QueryRequestAdapter for ElasticHttpAdapter {
         match self.config.language {
             QueryLanguage::elastic_sql => "/_sql",
             QueryLanguage::elastic_querydsl => "/_search",
-            _ => panic!("Invalid query language configured for Elastic"),
+            _ => unreachable!("Elastic adapter config is validated at startup"),
         }
     }
 

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

@@ -1050,16 +1050,22 @@ impl SimpleEngine {
                     debug!("  Merging accumulators (should_merge=true)");
                     #[cfg(feature = "extra_debugging")]
                     let merge_start = Instant::now();
-                    let merged_accumulator = self.merge_accumulators(&precomputes);
-                    #[cfg(feature = "extra_debugging")]
-                    let merge_duration = merge_start.elapsed();
-                    #[cfg(feature = "extra_debugging")]
-                    debug!(
-                        "  Merge completed in {:.2}ms, result type: {}",
-                        merge_duration.as_secs_f64() * 1000.0,
-                        merged_accumulator.get_accumulator_type()
-                    );
-                    merged.insert(key.clone(), merged_accumulator);
+                    match self.merge_accumulators(&precomputes) {
+                        Ok(merged_accumulator) => {
+                            #[cfg(feature = "extra_debugging")]
+                            let merge_duration = merge_start.elapsed();
+                            #[cfg(feature = "extra_debugging")]
+                            debug!(
+                                "  Merge completed in {:.2}ms, result type: {}",
+                                merge_duration.as_secs_f64() * 1000.0,
+                                merged_accumulator.get_accumulator_type()
+                            );
+                            merged.insert(key.clone(), merged_accumulator);
+                        }
+                        Err(e) => {
+                            warn!("Failed to merge accumulators for key {:?}: {}", key, e);
+                        }
+                    }
                 } else {
                     assert_eq!(
                         precomputes.len(),
@@ -1088,21 +1094,21 @@ impl SimpleEngine {
     fn merge_accumulators(
         &self,
         accumulators: &[Box<dyn crate::data_model::AggregateCore>],
-    ) -> Box<dyn crate::data_model::AggregateCore> {
+    ) -> Result<Box<dyn crate::data_model::AggregateCore>, String> {
         if accumulators.is_empty() {
-            panic!("No accumulators to merge");
+            return Err("merge_accumulators called with empty slice".to_string());
         }
 
         if accumulators.len() == 1 {
-            return accumulators[0].clone_boxed_core();
+            return Ok(accumulators[0].clone_boxed_core());
         }
 
         // Try to use optimized batch merge for KLL accumulators
         if accumulators[0].get_accumulator_type() == AggregationType::DatasketchesKLL {
             use crate::precompute_operators::datasketches_kll_accumulator::DatasketchesKLLAccumulator;
 
             match DatasketchesKLLAccumulator::merge_multiple(accumulators) {
-                Ok(merged) => return Box::new(merged),
+                Ok(merged) => return Ok(Box::new(merged)),
                 Err(e) => {
                     warn!(
                         "Batch merge failed: {}. Falling back to sequential merge.",
@@ -1118,7 +1124,7 @@ impl SimpleEngine {
             use crate::precompute_operators::count_min_sketch_accumulator::CountMinSketchAccumulator;
 
             match CountMinSketchAccumulator::merge_multiple(accumulators) {
-                Ok(merged) => return Box::new(merged),
+                Ok(merged) => return Ok(Box::new(merged)),
                 Err(e) => {
                     warn!(
                         "Batch merge failed: {}. Falling back to sequential merge.",
@@ -1145,7 +1151,7 @@ impl SimpleEngine {
             }
         }
 
-        result
+        Ok(result)
     }
 
     /// Collects results when key and value use different aggregations

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

@@ -265,9 +265,9 @@ impl SimpleEngine {
                 (QueryType::Spatial, QueryType::TemporalQuantile) => {
                     QueryPatternType::OneTemporalOneSpatial
                 }
-                _ => panic!("Unsupported query type found"),
+                _ => return None,
             },
-            _ => panic!("Unsupported query type found"),
+            _ => return None,
         };
 
         // For nested queries (spatial of temporal), the outer query has no time clause,

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

@@ -247,7 +247,7 @@ impl AggregateCore for DeltaSetAggregatorAccumulator {
 
     fn get_keys(&self) -> Option<Vec<KeyByLabelValues>> {
         if !self.removed.is_empty() {
-            panic!("DeltaSetAggregatorAccumulator does not support get_keys when removed items are present");
+            return None;
         }
         Some(self.added.iter().cloned().collect())
     }

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

@@ -29,19 +29,19 @@ impl MinMaxAccumulator {
         }
     }
 
-    pub fn new(sub_type: String) -> Self {
+    pub fn new(sub_type: String) -> Result<Self, String> {
         match sub_type.as_str() {
-            "min" => Self::new_min(),
-            "max" => Self::new_max(),
-            _ => panic!("sub_type must be 'min' or 'max'"),
+            "min" => Ok(Self::new_min()),
+            "max" => Ok(Self::new_max()),
+            _ => Err(format!("sub_type must be 'min' or 'max', got '{sub_type}'")),
         }
     }
 
-    pub fn with_value(value: f64, sub_type: String) -> Self {
+    pub fn with_value(value: f64, sub_type: String) -> Result<Self, String> {
         if sub_type != "min" && sub_type != "max" {
-            panic!("sub_type must be 'min' or 'max'");
+            return Err(format!("sub_type must be 'min' or 'max', got '{sub_type}'"));
         }
-        Self { value, sub_type }
+        Ok(Self { value, sub_type })
     }
 
     pub fn update(&mut self, value: f64) {
@@ -56,7 +56,7 @@ impl MinMaxAccumulator {
                     self.value = value;
                 }
             }
-            _ => panic!("Invalid sub_type"),
+            _ => unreachable!("MinMaxAccumulator sub_type is always 'min' or 'max'"),
         }
     }
 
@@ -73,7 +73,7 @@ impl MinMaxAccumulator {
             return Err("sub_type must be 'min' or 'max'".into());
         }
 
-        Ok(Self::with_value(value, sub_type))
+        Ok(Self::with_value(value, sub_type)?)
     }
 
     pub fn deserialize_from_bytes(buffer: &[u8]) -> Result<Self, Box<dyn std::error::Error>> {
@@ -91,7 +91,7 @@ impl MinMaxAccumulator {
             _ => return Err("Invalid sub_type byte".into()),
         };
 
-        Ok(Self::with_value(value, sub_type))
+        Ok(Self::with_value(value, sub_type)?)
     }
 }
 
@@ -108,7 +108,7 @@ impl SerializableToSink for MinMaxAccumulator {
         let sub_type_byte = match self.sub_type.as_str() {
             "min" => 0u8,
             "max" => 1u8,
-            _ => panic!("Invalid sub_type"),
+            _ => unreachable!("MinMaxAccumulator sub_type is always 'min' or 'max'"),
         };
         bytes.push(sub_type_byte);
         bytes
@@ -132,7 +132,7 @@ impl MergeableAccumulator<MinMaxAccumulator> for MinMaxAccumulator {
             }
         }
 
-        let mut result = MinMaxAccumulator::new(sub_type.clone());
+        let mut result = MinMaxAccumulator::new(sub_type.clone())?;
 
         for acc in accumulators {
             result.update(acc.value);
@@ -282,7 +282,7 @@ impl SingleSubpopulationAggregateFactory for MinMaxAccumulatorFactory {
         Ok(Box::new(MinMaxAccumulator::with_value(
             result_value,
             self.sub_type.clone(),
-        )))
+        )?))
     }
 
     fn create_default(&self) -> Box<dyn SingleSubpopulationAggregate> {
@@ -330,9 +330,9 @@ mod tests {
 
     #[test]
     fn test_merge_min_accumulators() {
-        let acc1 = MinMaxAccumulator::with_value(10.0, "min".to_string());
-        let acc2 = MinMaxAccumulator::with_value(5.0, "min".to_string());
-        let acc3 = MinMaxAccumulator::with_value(15.0, "min".to_string());
+        let acc1 = MinMaxAccumulator::with_value(10.0, "min".to_string()).unwrap();
+        let acc2 = MinMaxAccumulator::with_value(5.0, "min".to_string()).unwrap();
+        let acc3 = MinMaxAccumulator::with_value(15.0, "min".to_string()).unwrap();
 
         let merged =
             <MinMaxAccumulator as MergeableAccumulator<MinMaxAccumulator>>::merge_accumulators(
@@ -345,9 +345,9 @@ mod tests {
 
     #[test]
     fn test_merge_max_accumulators() {
-        let acc1 = MinMaxAccumulator::with_value(10.0, "max".to_string());
-        let acc2 = MinMaxAccumulator::with_value(5.0, "max".to_string());
-        let acc3 = MinMaxAccumulator::with_value(15.0, "max".to_string());
+        let acc1 = MinMaxAccumulator::with_value(10.0, "max".to_string()).unwrap();
+        let acc2 = MinMaxAccumulator::with_value(5.0, "max".to_string()).unwrap();
+        let acc3 = MinMaxAccumulator::with_value(15.0, "max".to_string()).unwrap();
 
         let merged =
             <MinMaxAccumulator as MergeableAccumulator<MinMaxAccumulator>>::merge_accumulators(
@@ -360,8 +360,8 @@ mod tests {
 
     #[test]
     fn test_merge_different_types_error() {
-        let acc1 = MinMaxAccumulator::with_value(10.0, "min".to_string());
-        let acc2 = MinMaxAccumulator::with_value(5.0, "max".to_string());
+        let acc1 = MinMaxAccumulator::with_value(10.0, "min".to_string()).unwrap();
+        let acc2 = MinMaxAccumulator::with_value(5.0, "max".to_string()).unwrap();
 
         assert!(
             <MinMaxAccumulator as MergeableAccumulator<MinMaxAccumulator>>::merge_accumulators(
@@ -373,7 +373,7 @@ mod tests {
 
     #[test]
     fn test_serialization() {
-        let acc = MinMaxAccumulator::with_value(42.5, "min".to_string());
+        let acc = MinMaxAccumulator::with_value(42.5, "min".to_string()).unwrap();
 
         // Test JSON serialization
         let json = acc.serialize_to_json();
@@ -391,7 +391,7 @@ mod tests {
     #[test]
     fn test_single_subpopulation_aggregate_trait() {
         let acc: Box<dyn SingleSubpopulationAggregate> =
-            Box::new(MinMaxAccumulator::with_value(42.0, "max".to_string()));
+            Box::new(MinMaxAccumulator::with_value(42.0, "max".to_string()).unwrap());
 
         assert_eq!(acc.query(Statistic::Max, None).unwrap(), 42.0);
         assert!(acc.query(Statistic::Min, None).is_err());

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

@@ -17,31 +17,40 @@ pub struct MultipleMinMaxAccumulator {
 }
 
 impl MultipleMinMaxAccumulator {
-    pub fn new(sub_type: String) -> Self {
+    pub fn new(sub_type: String) -> Result<Self, String> {
         if sub_type != "min" && sub_type != "max" {
-            panic!("sub_type must be 'min' or 'max'");
+            return Err(format!("sub_type must be 'min' or 'max', got '{sub_type}'"));
         }
 
-        Self {
+        Ok(Self {
             values: HashMap::new(),
             sub_type,
-        }
+        })
     }
 
     pub fn new_min() -> Self {
-        Self::new("min".to_string())
+        Self {
+            values: HashMap::new(),
+            sub_type: "min".to_string(),
+        }
     }
 
     pub fn new_max() -> Self {
-        Self::new("max".to_string())
+        Self {
+            values: HashMap::new(),
+            sub_type: "max".to_string(),
+        }
     }
 
-    pub fn new_with_values(values: HashMap<KeyByLabelValues, f64>, sub_type: String) -> Self {
+    pub fn new_with_values(
+        values: HashMap<KeyByLabelValues, f64>,
+        sub_type: String,
+    ) -> Result<Self, String> {
         if sub_type != "min" && sub_type != "max" {
-            panic!("sub_type must be 'min' or 'max'");
+            return Err(format!("sub_type must be 'min' or 'max', got '{sub_type}'"));
         }
 
-        Self { values, sub_type }
+        Ok(Self { values, sub_type })
     }
 
     pub fn update(&mut self, key: KeyByLabelValues, value: f64) {
@@ -58,7 +67,7 @@ impl MultipleMinMaxAccumulator {
                     *current = value;
                 }
             }
-            _ => panic!("Invalid sub_type"),
+            _ => unreachable!("MultipleMinMaxAccumulator sub_type is always 'min' or 'max'"),
         }
     }
 
@@ -313,7 +322,7 @@ impl MergeableAccumulator<MultipleMinMaxAccumulator> for MultipleMinMaxAccumulat
             }
         }
 
-        let mut result = MultipleMinMaxAccumulator::new(sub_type.clone());
+        let mut result = MultipleMinMaxAccumulator::new(sub_type.clone())?;
 
         for acc in accumulators {
             for (key, value) in acc.values {

asap-query-engine/src/tests/datafusion/plan_execution_tests.rs

@@ -333,8 +333,8 @@ mod tests {
     #[tokio::test]
     #[ignore = "Blocked: MinMaxAccumulator has no arroyo serde"]
     async fn test_old_vs_new_minmax_min() {
-        let mm_a = MinMaxAccumulator::with_value(10.0, "min".to_string());
-        let mm_b = MinMaxAccumulator::with_value(5.0, "min".to_string());
+        let mm_a = MinMaxAccumulator::with_value(10.0, "min".to_string()).unwrap();
+        let mm_b = MinMaxAccumulator::with_value(5.0, "min".to_string()).unwrap();
 
         let engine = create_engine_single_pop(
             "temperature",
@@ -353,8 +353,8 @@ mod tests {
     #[tokio::test]
     #[ignore = "Blocked: MinMaxAccumulator has no arroyo serde"]
     async fn test_old_vs_new_minmax_max() {
-        let mm_a = MinMaxAccumulator::with_value(90.0, "max".to_string());
-        let mm_b = MinMaxAccumulator::with_value(95.0, "max".to_string());
+        let mm_a = MinMaxAccumulator::with_value(90.0, "max".to_string()).unwrap();
+        let mm_b = MinMaxAccumulator::with_value(95.0, "max".to_string()).unwrap();
 
         let engine = create_engine_single_pop(
             "temperature",
@@ -392,7 +392,7 @@ mod tests {
     #[tokio::test]
     #[ignore = "Blocked: MultipleMinMaxAccumulator has no arroyo serde"]
     async fn test_old_vs_new_multiple_minmax() {
-        let mm = MultipleMinMaxAccumulator::new("min".to_string());
+        let mm = MultipleMinMaxAccumulator::new("min".to_string()).unwrap();
 
         let engine = create_engine_single_pop(
             "latency",
@@ -564,7 +564,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_execute_plan_not_implemented_minmax() {
-        let mm = MinMaxAccumulator::with_value(42.0, "min".to_string());
+        let mm = MinMaxAccumulator::with_value(42.0, "min".to_string()).unwrap();
 
         let engine = create_engine_single_pop(
             "temperature",

asap-query-engine/src/tests/store_correctness_tests.rs

@@ -829,7 +829,7 @@ fn test_clone_fidelity_sum(strategy: LockStrategy) {
 }
 
 fn test_clone_fidelity_min_max(strategy: LockStrategy) {
-    let acc = MinMaxAccumulator::with_value(42.0, "max".to_string());
+    let acc = MinMaxAccumulator::with_value(42.0, "max".to_string()).unwrap();
     roundtrip(strategy, acc);
 }
 
@@ -858,7 +858,7 @@ fn test_clone_fidelity_multiple_min_max(strategy: LockStrategy) {
     let mut values = HashMap::new();
     values.insert(key(&["dc", "east"]), 77.7);
     values.insert(key(&["dc", "west"]), 33.3);
-    let acc = MultipleMinMaxAccumulator::new_with_values(values, "max".to_string());
+    let acc = MultipleMinMaxAccumulator::new_with_values(values, "max".to_string()).unwrap();
     roundtrip(strategy, acc);
 }