diff --git a/src/engine/snapshot/query.rs b/src/engine/snapshot/query.rs index 291dc06..08d0d45 100644 --- a/src/engine/snapshot/query.rs +++ b/src/engine/snapshot/query.rs @@ -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, 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, + start_ms: Option, + end_ms: Option, + sort_missing_fallback: bool, + limit: usize, +) -> Vec { + let index = &snapshot.index; + static EMPTY_HELPER: std::sync::LazyLock> = + 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> = - 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 = 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 = BTreeMap::new(); for (sequence, timestamp, text) in snapshot.logs_range().filter(|(_, timestamp, _)| {