coderabbit changes

Author: parmesant

src/alerts/alert_structs.rs

@@ -273,7 +273,7 @@ impl AlertRequest {
 
         if datasets.len() != 1 {
             return Err(AlertError::ValidationFailure(format!(
-                "Query should include only one dataset. Found- {datasets:?}"
+                "Query should include only one dataset. Found: {datasets:?}"
             )));
         }
 

src/handlers/http/cluster/utils.rs

@@ -135,7 +135,7 @@ impl StorageStats {
     }
 }
 
-pub fn merge_quried_stats(stats: Vec<QueriedStats>) -> QueriedStats {
+pub fn merge_queried_stats(stats: Vec<QueriedStats>) -> QueriedStats {
     // get the stream name
     let stream_name = stats[1].stream.clone();
 

src/handlers/http/modal/query/querier_logstream.rs

@@ -39,7 +39,7 @@ use crate::{
             base_path_without_preceding_slash,
             cluster::{
                 self, fetch_daily_stats, fetch_stats_from_ingestors, sync_streams_with_ingestors,
-                utils::{IngestionStats, QueriedStats, StorageStats, merge_quried_stats},
+                utils::{IngestionStats, QueriedStats, StorageStats, merge_queried_stats},
             },
             logstream::error::StreamError,
             modal::{NodeMetadata, NodeType},
@@ -118,7 +118,7 @@ pub async fn put_stream(
     body: Bytes,
 ) -> Result<impl Responder, StreamError> {
     let stream_name = stream_name.into_inner();
-    let _ = CREATE_STREAM_LOCK.lock().await;
+    let _guard = CREATE_STREAM_LOCK.lock().await;
     let headers = PARSEABLE
         .create_update_stream(req.headers(), &body, &stream_name)
         .await?;
@@ -134,7 +134,7 @@ pub async fn put_stream(
     if is_update {
         Ok(("Log stream updated", StatusCode::OK))
     } else {
-        Ok(("Log stream created", StatusCode::OK))
+        Ok(("Log stream created", StatusCode::CREATED))
     }
 }
 
@@ -231,7 +231,7 @@ pub async fn get_stats(
 
     let stats = if let Some(mut ingestor_stats) = ingestor_stats {
         ingestor_stats.push(stats);
-        merge_quried_stats(ingestor_stats)
+        merge_queried_stats(ingestor_stats)
     } else {
         stats
     };

src/prism/logstream/mod.rs

@@ -30,7 +30,7 @@ use crate::{
     handlers::http::{
         cluster::{
             fetch_stats_from_ingestors,
-            utils::{IngestionStats, QueriedStats, StorageStats, merge_quried_stats},
+            utils::{IngestionStats, QueriedStats, StorageStats, merge_queried_stats},
         },
         logstream::error::StreamError,
         query::{QueryError, update_schema_when_distributed},
@@ -136,7 +136,7 @@ async fn get_stats(stream_name: &str) -> Result<QueriedStats, PrismLogstreamErro
 
     let stats = if let Some(mut ingestor_stats) = ingestor_stats {
         ingestor_stats.push(stats);
-        merge_quried_stats(ingestor_stats)
+        merge_queried_stats(ingestor_stats)
     } else {
         stats
     };

src/query/mod.rs

@@ -56,7 +56,7 @@ use crate::catalog::Snapshot as CatalogSnapshot;
 use crate::catalog::column::{Int64Type, TypedStatistics};
 use crate::catalog::manifest::Manifest;
 use crate::catalog::snapshot::Snapshot;
-use crate::event::{self, DEFAULT_TIMESTAMP_KEY};
+use crate::event::DEFAULT_TIMESTAMP_KEY;
 use crate::handlers::http::query::QueryError;
 use crate::option::Mode;
 use crate::parseable::PARSEABLE;
@@ -345,7 +345,7 @@ impl CountsRequest {
             .get_stream(&self.stream)
             .map_err(|err| anyhow::Error::msg(err.to_string()))?
             .get_time_partition()
-            .unwrap_or(event::DEFAULT_TIMESTAMP_KEY.to_owned());
+            .unwrap_or_else(|| DEFAULT_TIMESTAMP_KEY.to_owned());
 
         // get time range
         let time_range = TimeRange::parse_human_time(&self.start_time, &self.end_time)?;
@@ -449,7 +449,7 @@ impl CountsRequest {
         let time_partition = PARSEABLE
             .get_stream(&self.stream)?
             .get_time_partition()
-            .unwrap_or(DEFAULT_TIMESTAMP_KEY.into());
+            .unwrap_or_else(|| DEFAULT_TIMESTAMP_KEY.into());
 
         let time_range = TimeRange::parse_human_time(&self.start_time, &self.end_time)?;
 
@@ -458,20 +458,20 @@ impl CountsRequest {
         let date_bin = if dur.num_minutes() <= 60 * 10 {
             // date_bin 1 minute
             format!(
-                "CAST(DATE_BIN('1 minute', \"{}\".\"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as start_time, DATE_BIN('1 minute', \"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1 minute' as end_time",
-                self.stream
+                "CAST(DATE_BIN('1 minute', \"{}\".\"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as start_time, DATE_BIN('1 minute', \"{}\".\"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1 minute' as end_time",
+                self.stream, self.stream
             )
         } else if dur.num_minutes() > 60 * 10 && dur.num_minutes() < 60 * 240 {
             // date_bin 1 hour
             format!(
-                "CAST(DATE_BIN('1 hour', \"{}\".\"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as start_time, DATE_BIN('1 hour', \"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1 hour' as end_time",
-                self.stream
+                "CAST(DATE_BIN('1 hour', \"{}\".\"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as start_time, DATE_BIN('1 hour', \"{}\".\"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1 hour' as end_time",
+                self.stream, self.stream
             )
         } else {
             // date_bin 1 day
             format!(
-                "CAST(DATE_BIN('1 day', \"{}\".\"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as start_time, DATE_BIN('1 day', \"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1 day' as end_time",
-                self.stream
+                "CAST(DATE_BIN('1 day', \"{}\".\"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as start_time, DATE_BIN('1 day', \"{}\".\"{time_partition}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1 day' as end_time",
+                self.stream, self.stream
             )
         };