Updates

- changes related to time-partition
- general updates

Author: parmesant

src/alerts/alert_structs.rs

@@ -270,6 +270,13 @@ impl AlertRequest {
             TARGETS.get_target_by_id(id).await?;
         }
         let datasets = resolve_stream_names(&self.query)?;
+
+        if datasets.len() != 1 {
+            return Err(AlertError::ValidationFailure(format!(
+                "Query should include only one dataset. Found- {datasets:?}"
+            )));
+        }
+
         let config = AlertConfig {
             version: AlertVersion::from(CURRENT_ALERTS_VERSION),
             id: Ulid::new(),

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

@@ -33,14 +33,17 @@ use tracing::{error, warn};
 static CREATE_STREAM_LOCK: Mutex<()> = Mutex::const_new(());
 
 use crate::{
-    handlers::http::{
-        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},
+    handlers::{
+        UPDATE_STREAM_KEY,
+        http::{
+            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},
+            },
+            logstream::error::StreamError,
+            modal::{NodeMetadata, NodeType},
         },
-        logstream::error::StreamError,
-        modal::{NodeMetadata, NodeType},
     },
     hottier::HotTierManager,
     parseable::{PARSEABLE, StreamNotFound},
@@ -120,9 +123,19 @@ pub async fn put_stream(
         .create_update_stream(req.headers(), &body, &stream_name)
         .await?;
 
+    let is_update = if let Some(val) = headers.get(UPDATE_STREAM_KEY) {
+        val.to_str().unwrap() == "true"
+    } else {
+        false
+    };
+
     sync_streams_with_ingestors(headers, body, &stream_name).await?;
 
-    Ok(("Log stream created", StatusCode::OK))
+    if is_update {
+        Ok(("Log stream updated", StatusCode::OK))
+    } else {
+        Ok(("Log stream created", StatusCode::OK))
+    }
 }
 
 pub async fn get_stats(

src/query/mod.rs

@@ -345,7 +345,7 @@ impl CountsRequest {
             .get_stream(&self.stream)
             .map_err(|err| anyhow::Error::msg(err.to_string()))?
             .get_time_partition()
-            .unwrap_or_else(|| event::DEFAULT_TIMESTAMP_KEY.to_owned());
+            .unwrap_or(event::DEFAULT_TIMESTAMP_KEY.to_owned());
 
         // get time range
         let time_range = TimeRange::parse_human_time(&self.start_time, &self.end_time)?;
@@ -445,26 +445,32 @@ impl CountsRequest {
         // unwrap because we have asserted that it is some
         let count_conditions = self.conditions.as_ref().unwrap();
 
+        // get time partition column
+        let time_partition = PARSEABLE
+            .get_stream(&self.stream)?
+            .get_time_partition()
+            .unwrap_or(DEFAULT_TIMESTAMP_KEY.into());
+
         let time_range = TimeRange::parse_human_time(&self.start_time, &self.end_time)?;
 
         let dur = time_range.end.signed_duration_since(time_range.start);
 
         let date_bin = if dur.num_minutes() <= 60 * 10 {
             // date_bin 1 minute
             format!(
-                "CAST(DATE_BIN('1 minute', \"{}\".p_timestamp, TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as start_time, DATE_BIN('1 minute', p_timestamp, TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1 minute' as end_time",
+                "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
             )
         } else if dur.num_minutes() > 60 * 10 && dur.num_minutes() < 60 * 240 {
             // date_bin 1 hour
             format!(
-                "CAST(DATE_BIN('1 hour', \"{}\".p_timestamp, TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as start_time, DATE_BIN('1 hour', p_timestamp, TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1 hour' as end_time",
+                "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
             )
         } else {
             // date_bin 1 day
             format!(
-                "CAST(DATE_BIN('1 day', \"{}\".p_timestamp, TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as start_time, DATE_BIN('1 day', p_timestamp, TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1 day' as end_time",
+                "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
             )
         };
@@ -653,7 +659,7 @@ fn table_contains_any_time_filters(
         })
         .any(|expr| {
             matches!(&*expr.left, Expr::Column(Column { name, .. })
-            if name == time_column)
+            if name == &default_timestamp || name == time_column)
         })
 }