Merge ticket branch t083

2026-09-07 10:31:23 +01:00
parent d987663aaa f67cfc5d94
commit cdd781c904
1 changed files with 98 additions and 65 deletions
@@ -12,9 +12,12 @@ use crate::engine::message::{EngineQuery, EngineResponse};
 use crate::engine::metric_primitive::{MetricPrimitive, Number};
 use crate::engine::promql::{LabelMatcher, PromqlQuery};
 use crate::engine::rate;
+use crate::engine::snapshot::filter::CompiledFilter;
 #[cfg(test)]
 use crate::engine::snapshot::sort::SortHelperIndexEntry;
-use crate::engine::snapshot::sort::{log_entry_params, SortHelperAccess, SortHelperKey};
+use crate::engine::snapshot::sort::{
+    log_entry_params, CompiledParamDeclaration, SortHelperAccess, SortHelperKey,
+};
 use crate::engine::snapshot::EngineSnapshot;
 use crate::engine::timestamp::Timestamp;
 #[cfg(test)]
@@ -30,23 +33,94 @@ pub(crate) fn handle_query(
    handle_query_inner(snapshot, query, SortHelperAccess::Mutable(sort_helper_indexes))
 }

+/// Compiles `filter` against `snapshot`, or returns `Ok(None)` when there's no filter string
+/// to compile. Shared by the `Logs` and `Timeline` query arms, which both bail out early with
+/// `EngineResponse::Error` on a compile failure.
+fn compile_optional_filter(
+    snapshot: &EngineSnapshot,
+    filter: Option<&str>,
+    now: Timestamp,
+) -> Result<Option<CompiledFilter>, String> {
+    filter.map(|filter| snapshot.compile_filter_query(filter, now)).transpose()
+}
+
+/// Walks the sort-helper index for `sort_by` — rather than `snapshot.logs_range()`, which is
+/// ordered by sequence, not by the param's value — collecting up to `limit` matching entries
+/// in sort-by-param order.
+///
+/// Not range-bounded by since_sequence/start_ms the way the unsorted path is: `SortHelperKey::
+/// WithParam` sorts by the param's value first (that's the point — it's what gives sort-by-param
+/// order), so sequence/time bounds can't narrow a BTreeMap range seek here.
+#[allow(clippy::too_many_arguments)]
+fn logs_sorted_by_param(
+    snapshot: &EngineSnapshot,
+    sort_by: &str,
+    sort_helper_access: SortHelperAccess<'_>,
+    compiled_filter: Option<&CompiledFilter>,
+    compiled_params: &[CompiledParamDeclaration],
+    since_sequence: Option<u64>,
+    start_ms: Option<i64>,
+    end_ms: Option<i64>,
+    sort_missing_fallback: bool,
+    limit: usize,
+) -> Vec<LogEntryDto> {
+    let index = &snapshot.index;
+    static EMPTY_HELPER: std::sync::LazyLock<BTreeMap<SortHelperKey, Timestamp>> =
+        std::sync::LazyLock::new(BTreeMap::new);
+    let helper_index = sort_helper_access.resolve(index, sort_by).unwrap_or(&EMPTY_HELPER);
+
+    let mut entries = Vec::new();
+    for (helper_key, timestamp) in helper_index.iter() {
+        let sequence = match helper_key {
+            SortHelperKey::WithParam(_, sequence) => *sequence,
+            SortHelperKey::WithoutParam(sequence) => {
+                if !sort_missing_fallback {
+                    continue;
+                }
+                *sequence
+            }
+        };
+        if !(since_sequence.is_none_or(|since| sequence > since)
+            && start_ms.is_none_or(|start| timestamp.as_millis() >= start)
+            && end_ms.is_none_or(|end| timestamp.as_millis() <= end))
+        {
+            continue;
+        }
+        let Some(DataValue::LogLiteral { text }) =
+            index.get(&IndexKey::IngestedLog { sequence, timestamp: *timestamp })
+        else {
+            continue;
+        };
+        if !compiled_filter.is_none_or(|filter| filter.eval(sequence, *timestamp, text)) {
+            continue;
+        }
+        entries.push(LogEntryDto {
+            sequence,
+            timestamp_ms: timestamp.as_millis(),
+            text: text.clone(),
+            params: log_entry_params(text, compiled_params),
+        });
+        if entries.len() >= limit {
+            break;
+        }
+    }
+    entries
+}
+
 pub(super) fn handle_query_inner(
    snapshot: &EngineSnapshot,
    query: EngineQuery,
    sort_helper_access: SortHelperAccess<'_>,
 ) -> EngineResponse {
-    let index = &snapshot.index;
    let approx_bytes_total = snapshot.approx_bytes_total;
    let ingest_lag_high_watermark_ms = snapshot.ingest_lag_high_watermark_ms;
    match query {
        EngineQuery::Logs(params) => {
-            let compiled_filter = match params.filter.as_deref() {
-                Some(filter) => match snapshot.compile_filter_query(filter, Timestamp::now()) {
-                    Ok(compiled) => Some(compiled),
+            let compiled_filter =
+                match compile_optional_filter(snapshot, params.filter.as_deref(), Timestamp::now()) {
+                    Ok(compiled) => compiled,
                    Err(err) => return EngineResponse::Error(err),
-                },
-                None => None,
-            };
+                };
            if let Some(sort_by) = params.sort_by.as_deref() {
                if !snapshot.param_declarations_range().any(|(name, ..)| name == sort_by) {
                    return EngineResponse::Error(format!("unknown param '{sort_by}' for sort_by"));
@@ -80,69 +154,28 @@ pub(super) fn handle_query_inner(
                        params: log_entry_params(text, &compiled_params),
                    })
                    .collect(),
-                Some(sort_by) => {
-                    // Not range-bounded by since_sequence/start_ms: SortHelperKey::WithParam
-                    // sorts by the param's value first (that's the point — it's what gives
-                    // sort-by-param order), so sequence/time bounds can't narrow a BTreeMap
-                    // range seek here the way they do for the unsorted path above.
-                    static EMPTY_HELPER: std::sync::LazyLock<BTreeMap<SortHelperKey, Timestamp>> =
-                        std::sync::LazyLock::new(BTreeMap::new);
-                    let helper_index =
-                        sort_helper_access.resolve(index, sort_by).unwrap_or(&EMPTY_HELPER);
-
-                    let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true);
-                    let mut entries = Vec::new();
-                    for (helper_key, timestamp) in helper_index.iter() {
-                        let sequence = match helper_key {
-                            SortHelperKey::WithParam(_, sequence) => *sequence,
-                            SortHelperKey::WithoutParam(sequence) => {
-                                if !sort_missing_fallback {
-                                    continue;
-                                }
-                                *sequence
-                            }
-                        };
-                        if !(params.since_sequence.is_none_or(|since| sequence > since)
-                            && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
-                            && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end))
-                        {
-                            continue;
-                        }
-                        let Some(DataValue::LogLiteral { text }) =
-                            index.get(&IndexKey::IngestedLog { sequence, timestamp: *timestamp })
-                        else {
-                            continue;
-                        };
-                        if !compiled_filter
-                            .as_ref()
-                            .is_none_or(|filter| filter.eval(sequence, *timestamp, text))
-                        {
-                            continue;
-                        }
-                        entries.push(LogEntryDto {
-                            sequence,
-                            timestamp_ms: timestamp.as_millis(),
-                            text: text.clone(),
-                            params: log_entry_params(text, &compiled_params),
-                        });
-                        if entries.len() >= limit {
-                            break;
-                        }
-                    }
-                    entries
-                }
+                Some(sort_by) => logs_sorted_by_param(
+                    snapshot,
+                    sort_by,
+                    sort_helper_access,
+                    compiled_filter.as_ref(),
+                    &compiled_params,
+                    params.since_sequence,
+                    params.start_ms,
+                    params.end_ms,
+                    params.sort_missing_fallback.unwrap_or(true),
+                    limit,
+                ),
            };
            let entries: Vec<LogEntryDto> = entries.into_iter().take(limit).collect();
            EngineResponse::Logs(entries)
        }
        EngineQuery::Timeline(params) => {
-            let compiled_filter = match params.filter.as_deref() {
-                Some(filter) => match snapshot.compile_filter_query(filter, Timestamp::now()) {
-                    Ok(compiled) => Some(compiled),
+            let compiled_filter =
+                match compile_optional_filter(snapshot, params.filter.as_deref(), Timestamp::now()) {
+                    Ok(compiled) => compiled,
                    Err(err) => return EngineResponse::Error(err),
-                },
-                None => None,
-            };
+                };
            let bucket_ms = params.bucket_ms.unwrap_or(DEFAULT_TIMELINE_BUCKET_MS).max(1);
            let mut buckets: BTreeMap<i64, u64> = BTreeMap::new();
            for (sequence, timestamp, text) in snapshot.logs_range().filter(|(_, timestamp, _)| {