feat: make metric bucket cap configurable (#14)

Author: everpcpc

README.md

@@ -22,6 +22,7 @@ The adapter listens on `--bind` (default `0.0.0.0:3100`) and exposes a minimal s
 | `--timestamp-column` | `TIMESTAMP_COLUMN` | auto-detect             | Override the timestamp column name.                               |
 | `--line-column`      | `LINE_COLUMN`      | auto-detect             | Override the log line column name.                                |
 | `--labels-column`    | `LABELS_COLUMN`    | auto-detect (loki only) | Override the labels column name.                                  |
+| `--max-metric-buckets` | `MAX_METRIC_BUCKETS` | `240`                 | Maximum bucket count per metric range query before clamping `step`. |
 
 ## Schema Support
 
@@ -122,7 +123,7 @@ All endpoints return Loki-compatible JSON responses and reuse the same error sha
 | Endpoint                                | Description                                                                                                                                                                                                         |
 | --------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | `GET /loki/api/v1/query`                | Instant query. Supports the same LogQL used by Grafana's Explore panel. An optional `time` parameter (nanoseconds) defaults to "now", and the adapter automatically looks back 5 minutes when computing SQL bounds. |
-| `GET /loki/api/v1/query_range`          | Range query. Requires `start`/`end` nanoseconds and accepts `limit`/`step`. Log queries stream raw lines; metric queries return Loki matrix results and require `step` to match the range selector duration.          |
+| `GET /loki/api/v1/query_range`          | Range query. Requires `start`/`end` nanoseconds and accepts `limit`/`step`. Log queries stream raw lines; metric queries return Loki matrix results and require a `step` value (the adapter may clamp it to keep bucket counts bounded, default cap 240 buckets).          |
 | `GET /loki/api/v1/labels`               | Lists known label keys for the selected schema. Optional `start`/`end` parameters (nanoseconds) fence the search window; unspecified values default to the last 5 minutes, matching Grafana's Explore defaults.     |
 | `GET /loki/api/v1/label/{label}/values` | Lists distinct values for a specific label key using the same optional `start`/`end` bounds as `/labels`. Works for both `loki` and `flat` schemas and automatically filters out empty strings.                     |
 | `GET /loki/api/v1/index/stats`          | Returns approximate `streams`, `chunks`, `entries`, and `bytes` counters for a selector over a `[start, end]` window. `chunks` are estimated via unique stream keys because Databend does not store Loki chunks.    |
@@ -148,7 +149,7 @@ The adapter currently supports a narrow LogQL metric surface area:
 - Range functions: `count_over_time` and `rate`. The latter reports per-second values (`COUNT / window_seconds`).
 - Optional outer aggregations: `sum`, `avg`, `min`, `max`, `count`, each with `by (...)`. `without` or other modifiers return `errorType:bad_data`.
 - Pipelines: only `drop` stages are honored (labels are removed after aggregation to match Loki semantics). Any other stage still results in `errorType:bad_data`.
-- `/loki/api/v1/query_range` metric calls must provide `step`, and `step` must equal the range selector duration so Databend can materialize every bucket in a single SQL statement; the adapter never fans out multiple queries or aggregates in memory.
+- `/loki/api/v1/query_range` metric calls must provide `step`. When the requested `(end - start) / step` would exceed the configured bucket cap (default 240, tweak via `--max-metric-buckets`), the adapter automatically increases the effective step to keep the SQL result size manageable; the adapter never fans out multiple queries or aggregates in memory.
 - `/loki/api/v1/query` metric calls reuse the same expressions but evaluate them over `[time - range, time]`.
 
 Both schema adapters (loki VARIANT labels and flat wide tables) translate the metric expression into one SQL statement that joins generated buckets with the raw rows via `generate_series`, so all aggregation happens inside Databend. Non-metric queries continue to stream raw logs.

src/app/handlers.rs

@@ -197,12 +197,18 @@ async fn range_query(
             .as_deref()
             .ok_or_else(|| AppError::BadRequest("step is required for metric queries".into()))?;
         let step_duration = parse_step_duration(step_raw)?;
-        let step_ns = step_duration.as_nanoseconds();
+        let requested_step_ns = step_duration.as_nanoseconds();
         let window_ns = metric.range.duration.as_nanoseconds();
-        if step_ns != window_ns {
-            return Err(AppError::BadRequest(
-                "metric range queries require step to match the range selector duration".into(),
-            ));
+        let range_ns = end - start;
+        let step_ns = clamp_metric_step_ns(range_ns, requested_step_ns, state.max_metric_buckets());
+        if step_ns != requested_step_ns {
+            log::info!(
+                "metric step clamped to limit buckets: range={:.3}s requested_step={:.3}s effective_step={:.3}s max_buckets={}",
+                (range_ns as f64) / 1_000_000_000_f64,
+                (requested_step_ns as f64) / 1_000_000_000_f64,
+                (step_ns as f64) / 1_000_000_000_f64,
+                state.max_metric_buckets()
+            );
         }
         let plan = state.schema().build_metric_range_query(
             state.table(),
@@ -340,6 +346,16 @@ fn parse_numeric_step_seconds(step_raw: &str) -> Result<DurationValue, String> {
         .map_err(|err| format!("failed to convert numeric seconds to duration: {err}"))
 }
 
+fn clamp_metric_step_ns(range_ns: i64, requested_step_ns: i64, max_buckets: i64) -> i64 {
+    if range_ns <= 0 || requested_step_ns <= 0 {
+        return requested_step_ns;
+    }
+    let max_buckets = max_buckets.max(1);
+    let numerator = i128::from(range_ns) + i128::from(max_buckets - 1);
+    let min_step_ns = (numerator / i128::from(max_buckets)) as i64;
+    requested_step_ns.max(min_step_ns.max(1))
+}
+
 async fn label_names(
     State(state): State<AppState>,
     Query(params): Query<LabelsQueryParams>,
@@ -633,7 +649,10 @@ impl TailRequest {
 
 #[cfg(test)]
 mod tests {
-    use super::{ProcessedEntry, TailCursor, filter_tail_entries, parse_constant_vector_expr};
+    use super::{
+        ProcessedEntry, TailCursor, clamp_metric_step_ns, filter_tail_entries,
+        parse_constant_vector_expr,
+    };
     use std::collections::BTreeMap;
 
     #[test]
@@ -694,4 +713,31 @@ mod tests {
                 .is_empty()
         );
     }
+
+    #[test]
+    fn clamps_when_bucket_count_exceeds_limit() {
+        let range_ns = 3_600_000_000_000;
+        let requested_step_ns = 1_000_000_000;
+        assert_eq!(
+            clamp_metric_step_ns(range_ns, requested_step_ns, 600),
+            6_000_000_000
+        );
+    }
+
+    #[test]
+    fn leaves_large_steps_unchanged() {
+        let range_ns = 10_800_000_000_000;
+        let requested_step_ns = 60_000_000_000;
+        assert_eq!(
+            clamp_metric_step_ns(range_ns, requested_step_ns, 600),
+            requested_step_ns
+        );
+    }
+
+    #[test]
+    fn handles_zero_or_negative_ranges() {
+        assert_eq!(clamp_metric_step_ns(0, 1_000_000, 600), 1_000_000);
+        assert_eq!(clamp_metric_step_ns(-10, 1_000_000, 600), 1_000_000);
+        assert_eq!(clamp_metric_step_ns(1_000, 0, 600), 0);
+    }
 }

src/app/state.rs

@@ -34,6 +34,7 @@ pub struct AppState {
     table: TableRef,
     parser: LogqlParser,
     schema: SchemaAdapter,
+    max_metric_buckets: i64,
 }
 
 impl AppState {
@@ -43,6 +44,7 @@ impl AppState {
             table,
             schema_type,
             schema_config,
+            max_metric_buckets,
         } = config;
         info!("resolving table reference for `{table}`");
         let table = resolve_table_ref(&dsn, &table)?;
@@ -65,6 +67,7 @@ impl AppState {
             table,
             parser: LogqlParser,
             schema,
+            max_metric_buckets: i64::from(max_metric_buckets.max(1)),
         })
     }
 
@@ -80,6 +83,10 @@ impl AppState {
         &self.schema
     }
 
+    pub fn max_metric_buckets(&self) -> i64 {
+        self.max_metric_buckets
+    }
+
     pub fn parse(&self, query: &str) -> Result<LogqlExpr, AppError> {
         self.parser.parse(query).map_err(AppError::from)
     }
@@ -117,6 +124,7 @@ pub struct AppConfig {
     pub table: String,
     pub schema_type: SchemaType,
     pub schema_config: SchemaConfig,
+    pub max_metric_buckets: u32,
 }
 
 async fn verify_connection(client: &Client) -> Result<(), AppError> {

src/main.rs

@@ -25,6 +25,8 @@ mod databend;
 mod error;
 mod logql;
 
+const DEFAULT_MAX_METRIC_BUCKETS: u32 = 240;
+
 #[derive(Debug, Parser)]
 #[command(author, version, about, disable_help_subcommand = true)]
 struct Args {
@@ -54,6 +56,13 @@ struct Args {
     /// Override the column storing labels (loki schema only)
     #[arg(long = "labels-column", env = "LABELS_COLUMN")]
     labels_column: Option<String>,
+    /// Maximum number of buckets per metric range query
+    #[arg(
+        long = "max-metric-buckets",
+        env = "MAX_METRIC_BUCKETS",
+        default_value_t = DEFAULT_MAX_METRIC_BUCKETS
+    )]
+    max_metric_buckets: u32,
 }
 
 #[derive(Copy, Clone, Debug, ValueEnum)]
@@ -97,6 +106,7 @@ async fn main() -> Result<(), AppError> {
             line_column: args.line_column.clone(),
             labels_column: args.labels_column.clone(),
         },
+        max_metric_buckets: args.max_metric_buckets,
     };
     info!("bootstrapping application state");
     let state = AppState::bootstrap(config).await?;