chore: reuse time bin logic across apis

remove binning logic in counts api
make reusable component
reuse in counts api, errors api, agent-observability related apis

Author: nikhilsinhaparseable

src/query/mod.rs

@@ -73,7 +73,7 @@ use crate::metrics::increment_bytes_scanned_in_query_by_date;
 use crate::option::Mode;
 use crate::parseable::{DEFAULT_TENANT, PARSEABLE};
 use crate::storage::{ObjectStorageProvider, ObjectStoreFormat};
-use crate::utils::time::TimeRange;
+use crate::utils::time::{DATE_BIN_EPOCH_ANCHOR, TimeRange, count_api_bin_interval};
 
 /// Boxed record-batch stream used as the streaming half of query results.
 type BoxedBatchStream = SendableRecordBatchStream;
@@ -638,32 +638,13 @@ impl CountsRequest {
 
         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 table_name = &self.stream;
         let start_time_col_name = "_bin_start_time_";
         let end_time_col_name = "_bin_end_time_";
-        let date_bin = if dur.num_minutes() <= 60 * 5 {
-            // less than 5 hour = 1 min bin
-            format!(
-                "CAST(DATE_BIN('1m', \"{table_name}\".\"{time_column}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as {start_time_col_name}, DATE_BIN('1m', \"{table_name}\".\"{time_column}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1m' as {end_time_col_name}"
-            )
-        } else if dur.num_minutes() <= 60 * 24 {
-            // 1 day = 5 min bin
-            format!(
-                "CAST(DATE_BIN('5m', \"{table_name}\".\"{time_column}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as {start_time_col_name}, DATE_BIN('5m', \"{table_name}\".\"{time_column}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '5m' as {end_time_col_name}"
-            )
-        } else if dur.num_minutes() < 60 * 24 * 10 {
-            // 10 days = 1 hour bin
-            format!(
-                "CAST(DATE_BIN('1h', \"{table_name}\".\"{time_column}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as {start_time_col_name}, DATE_BIN('1h', \"{table_name}\".\"{time_column}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1h' as {end_time_col_name}"
-            )
-        } else {
-            // 1 day
-            format!(
-                "CAST(DATE_BIN('1d', \"{table_name}\".\"{time_column}\", TIMESTAMP '1970-01-01 00:00:00+00') AS TEXT) as {start_time_col_name}, DATE_BIN('1d', \"{table_name}\".\"{time_column}\", TIMESTAMP '1970-01-01 00:00:00+00') + INTERVAL '1d' as {end_time_col_name}"
-            )
-        };
+        let bin_interval = count_api_bin_interval(&time_range.start, &time_range.end);
+        let date_bin = format!(
+            "CAST(DATE_BIN('{bin_interval}', \"{table_name}\".\"{time_column}\", TIMESTAMP '{DATE_BIN_EPOCH_ANCHOR}') AS TEXT) as {start_time_col_name}, DATE_BIN('{bin_interval}', \"{table_name}\".\"{time_column}\", TIMESTAMP '{DATE_BIN_EPOCH_ANCHOR}') + INTERVAL '{bin_interval}' as {end_time_col_name}"
+        );
 
         let group_by_cols = count_conditions
             .group_by

src/utils/time.rs

@@ -18,6 +18,8 @@
 
 use chrono::{DateTime, Datelike, NaiveDate, NaiveDateTime, TimeDelta, TimeZone, Timelike, Utc};
 
+pub const DATE_BIN_EPOCH_ANCHOR: &str = "1970-01-01 00:00:00+00";
+
 #[derive(Debug, thiserror::Error)]
 pub enum TimeParseError {
     #[error("Parsing humantime")]
@@ -55,6 +57,69 @@ pub struct TimeRange {
     pub end: DateTime<Utc>,
 }
 
+pub fn count_api_bin_interval(start: &DateTime<Utc>, end: &DateTime<Utc>) -> &'static str {
+    let dur = end.signed_duration_since(*start);
+
+    if dur.num_minutes() <= 60 * 5 {
+        "1m"
+    } else if dur.num_minutes() <= 60 * 24 {
+        "5m"
+    } else if dur.num_minutes() < 60 * 24 * 10 {
+        "1h"
+    } else {
+        "1d"
+    }
+}
+
+pub fn interval_for_num_bins(start: &DateTime<Utc>, end: &DateTime<Utc>, num_bins: u64) -> String {
+    let total_seconds = end.signed_duration_since(*start).num_seconds().max(1) as u64;
+    let bin_seconds = (total_seconds / num_bins).max(1);
+    format!("{bin_seconds} seconds")
+}
+
+pub fn expected_time_bins(time_range: &TimeRange, num_bins: u64) -> Vec<(String, String)> {
+    let total_seconds = time_range
+        .end
+        .signed_duration_since(time_range.start)
+        .num_seconds()
+        .max(1) as u64;
+    let bin_seconds = (total_seconds / num_bins).max(1);
+
+    (0..num_bins)
+        .map(|i| {
+            let bin_start = time_range.start + chrono::Duration::seconds((i * bin_seconds) as i64);
+            let bin_end = if i == num_bins - 1 {
+                time_range.end
+            } else {
+                time_range.start + chrono::Duration::seconds(((i + 1) * bin_seconds) as i64)
+            };
+            (bin_start.to_rfc3339(), bin_end.to_rfc3339())
+        })
+        .collect()
+}
+
+pub fn match_time_bin_key(sql_bin_start: &str, expected_bins: &[(String, String)]) -> String {
+    let normalized = if !sql_bin_start.contains('+') && !sql_bin_start.ends_with('Z') {
+        format!("{sql_bin_start}+00:00")
+    } else {
+        sql_bin_start.to_string()
+    };
+
+    if let Ok(sql_ts) = DateTime::parse_from_rfc3339(&normalized) {
+        let sql_ts = sql_ts.with_timezone(&Utc);
+        for (bs, _) in expected_bins {
+            if let Ok(expected_ts) = DateTime::parse_from_rfc3339(bs) {
+                let expected_ts = expected_ts.with_timezone(&Utc);
+                if (sql_ts - expected_ts).num_seconds().abs() <= 1 {
+                    return bs.clone();
+                }
+            }
+        }
+    }
+
+    sql_bin_start.to_string()
+}
+
 impl TimeRange {
     pub fn new(start: DateTime<Utc>, end: DateTime<Utc>) -> Self {
         TimeRange { start, end }