Extract filter-compile and sort-by-param helpers out of handle_query_inner
The Logs and Timeline arms of handle_query_inner each duplicated the same "compile filter, bail out on error" block; the Logs arm also inlined its sort-helper-index walk. Pull both into named helpers (compile_optional_filter, logs_sorted_by_param) so the Logs arm reads as "compile filter, then dispatch to sorted or unsorted." Pure decomposition, no behavior change. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
f96de8f38d
commit
f67cfc5d94
@@ -12,9 +12,12 @@ use crate::engine::message::{EngineQuery, EngineResponse};
|
|||||||
use crate::engine::metric_primitive::{MetricPrimitive, Number};
|
use crate::engine::metric_primitive::{MetricPrimitive, Number};
|
||||||
use crate::engine::promql::{LabelMatcher, PromqlQuery};
|
use crate::engine::promql::{LabelMatcher, PromqlQuery};
|
||||||
use crate::engine::rate;
|
use crate::engine::rate;
|
||||||
|
use crate::engine::snapshot::filter::CompiledFilter;
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
use crate::engine::snapshot::sort::SortHelperIndexEntry;
|
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::snapshot::EngineSnapshot;
|
||||||
use crate::engine::timestamp::Timestamp;
|
use crate::engine::timestamp::Timestamp;
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
@@ -30,23 +33,94 @@ pub(crate) fn handle_query(
|
|||||||
handle_query_inner(snapshot, query, SortHelperAccess::Mutable(sort_helper_indexes))
|
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(
|
pub(super) fn handle_query_inner(
|
||||||
snapshot: &EngineSnapshot,
|
snapshot: &EngineSnapshot,
|
||||||
query: EngineQuery,
|
query: EngineQuery,
|
||||||
sort_helper_access: SortHelperAccess<'_>,
|
sort_helper_access: SortHelperAccess<'_>,
|
||||||
) -> EngineResponse {
|
) -> EngineResponse {
|
||||||
let index = &snapshot.index;
|
|
||||||
let approx_bytes_total = snapshot.approx_bytes_total;
|
let approx_bytes_total = snapshot.approx_bytes_total;
|
||||||
let ingest_lag_high_watermark_ms = snapshot.ingest_lag_high_watermark_ms;
|
let ingest_lag_high_watermark_ms = snapshot.ingest_lag_high_watermark_ms;
|
||||||
match query {
|
match query {
|
||||||
EngineQuery::Logs(params) => {
|
EngineQuery::Logs(params) => {
|
||||||
let compiled_filter = match params.filter.as_deref() {
|
let compiled_filter =
|
||||||
Some(filter) => match snapshot.compile_filter_query(filter, Timestamp::now()) {
|
match compile_optional_filter(snapshot, params.filter.as_deref(), Timestamp::now()) {
|
||||||
Ok(compiled) => Some(compiled),
|
Ok(compiled) => compiled,
|
||||||
Err(err) => return EngineResponse::Error(err),
|
Err(err) => return EngineResponse::Error(err),
|
||||||
},
|
};
|
||||||
None => None,
|
|
||||||
};
|
|
||||||
if let Some(sort_by) = params.sort_by.as_deref() {
|
if let Some(sort_by) = params.sort_by.as_deref() {
|
||||||
if !snapshot.param_declarations_range().any(|(name, ..)| name == sort_by) {
|
if !snapshot.param_declarations_range().any(|(name, ..)| name == sort_by) {
|
||||||
return EngineResponse::Error(format!("unknown param '{sort_by}' for 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),
|
params: log_entry_params(text, &compiled_params),
|
||||||
})
|
})
|
||||||
.collect(),
|
.collect(),
|
||||||
Some(sort_by) => {
|
Some(sort_by) => logs_sorted_by_param(
|
||||||
// Not range-bounded by since_sequence/start_ms: SortHelperKey::WithParam
|
snapshot,
|
||||||
// sorts by the param's value first (that's the point — it's what gives
|
sort_by,
|
||||||
// sort-by-param order), so sequence/time bounds can't narrow a BTreeMap
|
sort_helper_access,
|
||||||
// range seek here the way they do for the unsorted path above.
|
compiled_filter.as_ref(),
|
||||||
static EMPTY_HELPER: std::sync::LazyLock<BTreeMap<SortHelperKey, Timestamp>> =
|
&compiled_params,
|
||||||
std::sync::LazyLock::new(BTreeMap::new);
|
params.since_sequence,
|
||||||
let helper_index =
|
params.start_ms,
|
||||||
sort_helper_access.resolve(index, sort_by).unwrap_or(&EMPTY_HELPER);
|
params.end_ms,
|
||||||
|
params.sort_missing_fallback.unwrap_or(true),
|
||||||
let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true);
|
limit,
|
||||||
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
|
|
||||||
}
|
|
||||||
};
|
};
|
||||||
let entries: Vec<LogEntryDto> = entries.into_iter().take(limit).collect();
|
let entries: Vec<LogEntryDto> = entries.into_iter().take(limit).collect();
|
||||||
EngineResponse::Logs(entries)
|
EngineResponse::Logs(entries)
|
||||||
}
|
}
|
||||||
EngineQuery::Timeline(params) => {
|
EngineQuery::Timeline(params) => {
|
||||||
let compiled_filter = match params.filter.as_deref() {
|
let compiled_filter =
|
||||||
Some(filter) => match snapshot.compile_filter_query(filter, Timestamp::now()) {
|
match compile_optional_filter(snapshot, params.filter.as_deref(), Timestamp::now()) {
|
||||||
Ok(compiled) => Some(compiled),
|
Ok(compiled) => compiled,
|
||||||
Err(err) => return EngineResponse::Error(err),
|
Err(err) => return EngineResponse::Error(err),
|
||||||
},
|
};
|
||||||
None => None,
|
|
||||||
};
|
|
||||||
let bucket_ms = params.bucket_ms.unwrap_or(DEFAULT_TIMELINE_BUCKET_MS).max(1);
|
let bucket_ms = params.bucket_ms.unwrap_or(DEFAULT_TIMELINE_BUCKET_MS).max(1);
|
||||||
let mut buckets: BTreeMap<i64, u64> = BTreeMap::new();
|
let mut buckets: BTreeMap<i64, u64> = BTreeMap::new();
|
||||||
for (sequence, timestamp, text) in snapshot.logs_range().filter(|(_, timestamp, _)| {
|
for (sequence, timestamp, text) in snapshot.logs_range().filter(|(_, timestamp, _)| {
|
||||||
|
|||||||
Reference in New Issue
Block a user