fix(timeseries): repair numeric aggregates and time_bucket() after nodedb-sql migration

Fixes #15.

The aggregate executor receives (op, field) pairs but all three
PhysicalTask construction sites were passing the alias instead of the
source field name, so the executor could never locate the column and
returned null for every MIN/MAX/SUM/AVG result.

parse_interval_to_ms only handled compact suffixes ("1h", "15m") and
rejected the word-form intervals that DataFusion normalises SQL INTERVAL
literals into ("1 hour", "15 minutes"), causing time_bucket() to bucket
everything into epoch-zero and return an empty result set.

Changes:
- sql_plan_convert: extract agg_expr_to_pair() helper that reads the
  first argument expression (Column name or Wildcard) and use it at all
  three PhysicalTask construction sites instead of a.alias
- aggregate: extend parse_interval_to_ms to accept singular and plural
  word-form units (hour/hours, minute/minutes, second/seconds, day/days)
  alongside the existing compact forms; bare numbers are treated as
  seconds; unrecognised units return 0 instead of silently mis-routing
- aggregate: add unit tests covering word-form parsing

Author: farhan-syah

nodedb-sql/src/planner/aggregate.rs

@@ -133,23 +133,33 @@ fn extract_bucket_interval(func: &ast::Function) -> Result<i64> {
     Ok(parse_interval_to_ms(&interval_str))
 }
 
-/// Parse an interval string like "1h", "15m", "30s", "1d" to milliseconds.
+/// Parse an interval string to milliseconds.
+///
+/// Accepted forms: `"1h"`, `"15m"`, `"30s"`, `"1d"`, `"1 hour"`, `"15 minutes"`,
+/// `"30 seconds"`, `"7 days"`. Plural and singular word forms both work.
 fn parse_interval_to_ms(s: &str) -> i64 {
     let s = s.trim();
     if s.is_empty() {
         return 0;
     }
-    let (num_str, suffix) = s.split_at(s.len() - 1);
-    let num: i64 = num_str.trim().parse().unwrap_or(0);
-    match suffix {
-        "s" => num * 1_000,
-        "m" => num * 60_000,
-        "h" => num * 3_600_000,
-        "d" => num * 86_400_000,
-        _ => {
-            // Try full string as seconds.
-            s.parse::<i64>().unwrap_or(0) * 1_000
+
+    // Split into numeric part and unit part (handles both "1h" and "1 hour").
+    let num_end = s
+        .find(|c: char| !c.is_ascii_digit() && c != '.')
+        .unwrap_or(s.len());
+    let num: i64 = s[..num_end].trim().parse().unwrap_or(0);
+    let unit = s[num_end..].trim();
+
+    match unit {
+        "s" | "sec" | "second" | "seconds" => num * 1_000,
+        "m" | "min" | "minute" | "minutes" => num * 60_000,
+        "h" | "hr" | "hour" | "hours" => num * 3_600_000,
+        "d" | "day" | "days" => num * 86_400_000,
+        "" => {
+            // Bare number — treat as seconds.
+            num * 1_000
         }
+        _ => 0,
     }
 }
 
@@ -260,5 +270,12 @@ mod tests {
         assert_eq!(parse_interval_to_ms("15m"), 900_000);
         assert_eq!(parse_interval_to_ms("30s"), 30_000);
         assert_eq!(parse_interval_to_ms("7d"), 604_800_000);
+        // Word-form intervals.
+        assert_eq!(parse_interval_to_ms("1 hour"), 3_600_000);
+        assert_eq!(parse_interval_to_ms("2 hours"), 7_200_000);
+        assert_eq!(parse_interval_to_ms("15 minutes"), 900_000);
+        assert_eq!(parse_interval_to_ms("30 seconds"), 30_000);
+        assert_eq!(parse_interval_to_ms("1 day"), 86_400_000);
+        assert_eq!(parse_interval_to_ms("5 min"), 300_000);
     }
 }

nodedb/src/control/planner/sql_plan_convert.rs

@@ -249,10 +249,8 @@ fn convert_one(
             tiered,
         } => {
             let filter_bytes = serialize_filters(filters)?;
-            let agg_pairs: Vec<(String, String)> = aggregates
-                .iter()
-                .map(|a| (a.function.clone(), a.alias.clone()))
-                .collect();
+            let agg_pairs: Vec<(String, String)> =
+                aggregates.iter().map(agg_expr_to_pair).collect();
 
             // AUTO_TIER: split query across retention tiers if enabled.
             if *tiered
@@ -667,10 +665,7 @@ fn convert_aggregate(
         let vshard = VShardId::from_collection(&left_collection);
 
         let group_strs = group_by_to_strings(group_by);
-        let agg_pairs = aggregates
-            .iter()
-            .map(|a| (a.function.clone(), a.alias.clone()))
-            .collect();
+        let agg_pairs = aggregates.iter().map(agg_expr_to_pair).collect();
 
         return Ok(vec![PhysicalTask {
             tenant_id,
@@ -700,10 +695,7 @@ fn convert_aggregate(
     let having_bytes = serialize_filters(having)?;
 
     let group_strs = group_by_to_strings(group_by);
-    let agg_pairs = aggregates
-        .iter()
-        .map(|a| (a.function.clone(), a.alias.clone()))
-        .collect();
+    let agg_pairs = aggregates.iter().map(agg_expr_to_pair).collect();
 
     Ok(vec![PhysicalTask {
         tenant_id,
@@ -734,6 +726,24 @@ fn extract_collection_name(plan: &SqlPlan) -> String {
     }
 }
 
+/// Convert an `AggregateExpr` to the `(op, field)` pair the executor expects.
+///
+/// The field is extracted from the first argument (e.g. `elapsed_ms` from
+/// `AVG(elapsed_ms)`). Wildcard args produce `"*"`. Falls back to `"*"` when
+/// there are no arguments (bare `COUNT`).
+fn agg_expr_to_pair(a: &AggregateExpr) -> (String, String) {
+    let field = a
+        .args
+        .first()
+        .map(|arg| match arg {
+            SqlExpr::Column { name, .. } => name.clone(),
+            SqlExpr::Wildcard => "*".into(),
+            _ => format!("{arg:?}"),
+        })
+        .unwrap_or_else(|| "*".into());
+    (a.function.clone(), field)
+}
+
 fn group_by_to_strings(exprs: &[SqlExpr]) -> Vec<String> {
     exprs
         .iter()