Merge branch 'main' into alerts-updates

Signed-off-by: parmesant <[email protected]>

Author: parmesant

src/alerts/mod.rs

@@ -766,25 +766,22 @@ fn extract_alias_from_expr(expr: &Expr) -> Option<(String, Option<String>)> {
             // This is an aliased expression
             let alias_name = alias_expr.name.clone();
 
-            match alias_expr.expr.as_ref() {
-                Expr::AggregateFunction(agg_func) => {
-                    let agg_name = format!("{:?}", agg_func.func);
-                    Some((agg_name, Some(alias_name)))
-                }
+            if let Expr::AggregateFunction(agg_func) = alias_expr.expr.as_ref() {
+                let agg_name = format!("{:?}", agg_func.func);
+                Some((agg_name, Some(alias_name)))
+            } else {
                 // Handle other aggregate expressions like Count, etc.
-                _ => {
-                    // Check if the inner expression is an aggregate
-                    let expr_str = format!("{:?}", alias_expr.expr);
-                    if expr_str.contains("count")
-                        || expr_str.contains("sum")
-                        || expr_str.contains("avg")
-                        || expr_str.contains("min")
-                        || expr_str.contains("max")
-                    {
-                        Some((expr_str, Some(alias_name)))
-                    } else {
-                        None
-                    }
+                // Check if the inner expression is an aggregate
+                let expr_str = format!("{:?}", alias_expr.expr);
+                if expr_str.contains("count")
+                    || expr_str.contains("sum")
+                    || expr_str.contains("avg")
+                    || expr_str.contains("min")
+                    || expr_str.contains("max")
+                {
+                    Some((expr_str, Some(alias_name)))
+                } else {
+                    None
                 }
             }
         }

src/catalog/mod.rs

@@ -366,8 +366,8 @@ pub async fn get_first_event(
         Mode::All | Mode::Ingest => {
             // get current snapshot
             let stream_first_event = PARSEABLE.get_stream(stream_name)?.get_first_event();
-            if let Some(event) = stream_first_event {
-                first_event_at = event;
+            if let Some(first_event) = stream_first_event {
+                first_event_at = first_event;
             } else {
                 let mut meta = storage.get_object_store_format(stream_name).await?;
                 let meta_clone = meta.clone();

src/livetail.rs

@@ -161,7 +161,8 @@ impl Stream for ReceiverPipe {
 impl Drop for ReceiverPipe {
     fn drop(&mut self) {
         if let Some(map) = self._ref.upgrade()
-            && let Some(pipes) = map.write().unwrap().get_mut(&self.stream)
+            && let Ok(mut guard) = map.write()
+            && let Some(pipes) = guard.get_mut(&self.stream)
         {
             pipes.retain(|x| x.id != self.id)
         }