Improve ingestion performance

src/engine/handle.rs

@@ -9,7 +9,7 @@ use crate::engine::data_value::DataValue;
 use crate::engine::index_key::IndexKey;
 use crate::engine::message::EngineMessage;
 use crate::engine::metric_primitive::{MetricPrimitive, Number};
-use crate::engine::snapshot::params::{derive_log_stream_params_from_row, log_stream_header_transition};
+use crate::engine::snapshot::params::log_stream_header_transition;
 use crate::engine::snapshot::{approx_log_bytes, EngineSnapshot};
 use crate::engine::timestamp::Timestamp;
 use crate::engine::{ClearParamsResult, IngestedLine};
@@ -144,7 +144,7 @@ async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle, mu
                     );
                     snap.log_count += 1;
                     if let Some(header) = row_header.as_ref() {
-                        derive_log_stream_params_from_row(&mut snap.index, header, &text_for_op);
+                        snap.derive_log_stream_params_from_row(header, &text_for_op);
                     }
                     snap.auto_declare_params(&text_for_op);
                     snap.apply_all_metrics(sequence, timestamp, &text_for_op);
@@ -209,6 +209,7 @@ async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle, mu
                 write.mutate(move |snap: &mut EngineSnapshot| {
                     snap.index.remove(&IndexKey::ParamDeclaration { name: name_for_op.clone() });
                     snap.sort_helper_indexes.remove(&name_for_op);
+                    snap.param_regex_cache.remove(&name_for_op);
                 });
                 let _ = respond_to.send(existed);
             }

src/engine/snapshot/metrics.rs

@@ -40,6 +40,7 @@ impl EngineSnapshot {
 
     pub(crate) fn declare_metric(&mut self, request: DeclareMetricRequest) -> Result<(), String> {
         let re = self.validate_metric_request(&request)?;
+        self.metric_regex_cache.insert(request.name.clone(), re.clone());
         let resolved_params = self.resolved_param_declarations();
 
         self.backfill_metric(
@@ -87,7 +88,9 @@ impl EngineSnapshot {
         self.clear_metric_data(existing_name);
         if request.name != existing_name {
             self.index.remove(&IndexKey::MetricDeclaration { name: existing_name.to_string() });
+            self.metric_regex_cache.remove(existing_name);
         }
+        self.metric_regex_cache.insert(request.name.clone(), re.clone());
 
         self.backfill_metric(
             &request.name,
@@ -121,6 +124,7 @@ impl EngineSnapshot {
 
     pub(crate) fn remove_metric_declaration(&mut self, name: &str) -> bool {
         let existed = self.index.remove(&IndexKey::MetricDeclaration { name: name.to_string() }).is_some();
+        self.metric_regex_cache.remove(name);
         self.clear_metric_data(name);
         existed
     }
@@ -251,7 +255,6 @@ impl EngineSnapshot {
 
     pub(crate) fn apply_all_metrics(&mut self, sequence: u64, timestamp: Timestamp, text: &str) {
         type SeededMetric = (
-            String,
             String,
             Vec<String>,
             Option<String>,
@@ -260,10 +263,9 @@ impl EngineSnapshot {
         );
         let metrics: Vec<SeededMetric> = self
             .metric_declarations_range()
-            .map(|(name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels)| {
+            .map(|(name, _, label_params, value_param, matched_labels, unmatched_labels)| {
                 (
                     name.to_string(),
-                    message_match_regex.to_string(),
                     label_params.to_vec(),
                     value_param.map(|s| s.to_string()),
                     matched_labels.cloned(),
@@ -272,8 +274,8 @@ impl EngineSnapshot {
             })
             .collect();
         let resolved_params = self.resolved_param_declarations();
-        for (name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels) in metrics {
+        for (name, label_params, value_param, matched_labels, unmatched_labels) in metrics {
-            if let Ok(re) = Regex::new(&message_match_regex) {
+            if let Some(re) = self.metric_regex_cache.get(&name).cloned() {
                 self.apply_metric_to_log(
                     &name,
                     &re,

src/engine/snapshot/mod.rs

@@ -17,6 +17,7 @@ use crate::engine::index_key::IndexKey;
 use crate::engine::message::{EngineQuery, EngineResponse};
 use crate::engine::timestamp::Timestamp;
 use query::handle_query_inner;
+use regex::Regex;
 use sort::{SortHelperAccess, SortHelperIndexEntry};
 use std::collections::{BTreeMap, HashMap};
 
@@ -24,6 +25,13 @@ use std::collections::{BTreeMap, HashMap};
 pub(crate) struct EngineSnapshot {
     pub(crate) index: BTreeMap<IndexKey, DataValue>,
     pub(crate) sort_helper_indexes: HashMap<String, SortHelperIndexEntry>,
+    /// Compiled regexes for param/metric declarations, keyed by declaration name. Compiling a
+    /// regex is far more expensive than matching with one, and `apply_all_metrics`/
+    /// `update_sort_helper_indexes_on_ingest` run per ingested line, so every insertion or
+    /// removal of a declaration (see `params.rs`/`metrics.rs`) must keep these in sync rather
+    /// than recompiling from the stored pattern string on every line.
+    pub(crate) param_regex_cache: HashMap<String, Regex>,
+    pub(crate) metric_regex_cache: HashMap<String, Regex>,
     pub(crate) log_count: usize,
     pub(crate) approx_bytes_total: u64,
     pub(crate) ingest_lag_high_watermark_ms: i64,
@@ -105,7 +113,28 @@ pub(crate) fn approx_log_bytes(text: &str) -> u64 {
 
 impl EngineSnapshot {
     pub(crate) fn from_index(index: BTreeMap<IndexKey, DataValue>) -> Self {
-        EngineSnapshot { index, ..Default::default() }
+        let mut snapshot = EngineSnapshot { index, ..Default::default() };
+        snapshot.rebuild_regex_caches();
+        snapshot
+    }
+
+    /// Recompiles `param_regex_cache`/`metric_regex_cache` from the declarations currently in
+    /// `index`. Needed whenever a snapshot's declarations are populated by inserting straight
+    /// into `index` rather than through the normal declare/insert methods that keep the caches
+    /// in sync incrementally (e.g. `from_index` building a snapshot from a bare index).
+    pub(crate) fn rebuild_regex_caches(&mut self) {
+        self.param_regex_cache = self
+            .param_declarations_range()
+            .filter_map(|(name, message_match_regex, ..)| {
+                regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re))
+            })
+            .collect();
+        self.metric_regex_cache = self
+            .metric_declarations_range()
+            .filter_map(|(name, message_match_regex, ..)| {
+                regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re))
+            })
+            .collect();
     }
 
     pub(crate) fn evict_excess_logs(&mut self, max_logs: Option<usize>) {

src/engine/snapshot/params.rs

@@ -51,6 +51,9 @@ impl EngineSnapshot {
     }
 
     pub(crate) fn insert_param_declaration(&mut self, request: DeclareParamRequest) {
+        if let Ok(re) = Regex::new(&request.message_match_regex) {
+            self.param_regex_cache.insert(request.name.clone(), re);
+        }
         self.index.insert(
             IndexKey::ParamDeclaration { name: request.name.clone() },
             DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }),
@@ -146,6 +149,9 @@ impl EngineSnapshot {
             if self.index.contains_key(&index_key) {
                 continue;
             }
+            if let Ok(re) = Regex::new(&candidate.message_match_regex) {
+                self.param_regex_cache.insert(candidate.name.clone(), re);
+            }
             self.index.insert(
                 index_key,
                 DataValue::ParamDeclaration(ParamDeclaration {
@@ -172,56 +178,70 @@ pub(crate) fn log_stream_header_transition(
     crate::engine::logfmt::LogStreamHeader::detect(text).or_else(|| current.clone())
 }
 
-pub(crate) fn derive_log_stream_params_from_row(
+impl EngineSnapshot {
-    index: &mut BTreeMap<IndexKey, DataValue>,
+    pub(crate) fn derive_log_stream_params_from_row(
-    header: &crate::engine::logfmt::LogStreamHeader,
+        &mut self,
-    text: &str,
+        header: &crate::engine::logfmt::LogStreamHeader,
-) {
+        text: &str,
-    if !crate::engine::logfmt::is_log_stream_line(text) {
+    ) {
-        return;
+        if !crate::engine::logfmt::is_log_stream_line(text) {
-    }
+            return;
-    let row = header.split_row(text);
-
-    for (i, column) in header.columns.iter().enumerate() {
-        let Some(value) = row.get(i) else { continue };
-        let index_key = IndexKey::ParamDeclaration { name: column.name.clone() };
-        if index.contains_key(&index_key) {
-            continue;
         }
-        index.insert(
+        let row = header.split_row(text);
-            index_key,
-            DataValue::ParamDeclaration(ParamDeclaration {
-                name: column.name.clone(),
-                message_match_regex: header.column_pattern(i, &column.name),
-                capture_group: column.name.clone(),
-                json_field_path: None,
-                kind: infer_kind(value.trim()),
-                elapsed_since_last: false,
-                auto_derived: true,
-                log_stream_gated: true,
-            }),
-        );
-    }
 
-    if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) {
+        for (i, column) in header.columns.iter().enumerate() {
-        index.entry(IndexKey::ParamDeclaration { name: "message".to_string() }).or_insert_with(|| {
+            let Some(value) = row.get(i) else { continue };
-            DataValue::ParamDeclaration(ParamDeclaration {
+            let index_key = IndexKey::ParamDeclaration { name: column.name.clone() };
-                name: "message".to_string(),
+            if self.index.contains_key(&index_key) {
-                message_match_regex: header.message_pattern("message"),
+                continue;
-                capture_group: "message".to_string(),
+            }
-                json_field_path: None,
+            let message_match_regex = header.column_pattern(i, &column.name);
-                kind: infer_kind(message.trim()),
+            if let Ok(re) = Regex::new(&message_match_regex) {
-                elapsed_since_last: false,
+                self.param_regex_cache.insert(column.name.clone(), re);
-                auto_derived: true,
+            }
-                log_stream_gated: true,
+            self.index.insert(
-            })
+                index_key,
-        });
+                DataValue::ParamDeclaration(ParamDeclaration {
+                    name: column.name.clone(),
+                    message_match_regex,
+                    capture_group: column.name.clone(),
+                    json_field_path: None,
+                    kind: infer_kind(value.trim()),
+                    elapsed_since_last: false,
+                    auto_derived: true,
+                    log_stream_gated: true,
+                }),
+            );
+        }
+
+        if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) {
+            let index_key = IndexKey::ParamDeclaration { name: "message".to_string() };
+            if !self.index.contains_key(&index_key) {
+                let message_match_regex = header.message_pattern("message");
+                if let Ok(re) = Regex::new(&message_match_regex) {
+                    self.param_regex_cache.insert("message".to_string(), re);
+                }
+                self.index.insert(
+                    index_key,
+                    DataValue::ParamDeclaration(ParamDeclaration {
+                        name: "message".to_string(),
+                        message_match_regex,
+                        capture_group: "message".to_string(),
+                        json_field_path: None,
+                        kind: infer_kind(message.trim()),
+                        elapsed_since_last: false,
+                        auto_derived: true,
+                        log_stream_gated: true,
+                    }),
+                );
+            }
+        }
     }
 }
 
 #[cfg(test)]
 pub(crate) fn auto_declare_log_stream_params(
-    index: &mut BTreeMap<IndexKey, DataValue>,
+    snapshot: &mut EngineSnapshot,
     header: &mut Option<crate::engine::logfmt::LogStreamHeader>,
     text: &str,
 ) {
@@ -230,8 +250,8 @@ pub(crate) fn auto_declare_log_stream_params(
     if is_new_header {
         return;
     }
-    if let Some(header) = header.as_ref() {
+    if let Some(header) = header.clone() {
-        derive_log_stream_params_from_row(index, header, text);
+        snapshot.derive_log_stream_params_from_row(&header, text);
     }
 }
 
@@ -241,7 +261,8 @@ impl EngineSnapshot {
             self.param_declarations_range().map(|(name, ..)| name.to_string()).collect();
         let removed = names.len();
         for name in names {
-            self.index.remove(&IndexKey::ParamDeclaration { name });
+            self.index.remove(&IndexKey::ParamDeclaration { name: name.clone() });
+            self.param_regex_cache.remove(&name);
         }
         removed
     }
@@ -269,12 +290,12 @@ pub(super) struct ResolvedParam {
 impl EngineSnapshot {
     pub(super) fn resolved_param_declarations(&self) -> HashMap<String, ResolvedParam> {
         self.param_declarations_range()
-            .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| {
+            .filter_map(|(name, _, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| {
-                Regex::new(message_match_regex).ok().map(|regex| {
+                self.param_regex_cache.get(name).map(|regex| {
                     (
                         name.to_string(),
                         ResolvedParam {
-                            regex,
+                            regex: regex.clone(),
                             capture_group: capture_group.to_string(),
                             json_field_path: json_field_path.map(|path| path.to_string()),
                             kind,

src/engine/snapshot/sort.rs

@@ -97,11 +97,11 @@ impl EngineSnapshot {
     pub(crate) fn compiled_param_declarations(&self) -> Vec<CompiledParamDeclaration> {
         self.param_declarations_range()
             .filter(|(_, _, _, _, _, elapsed_since_last, _, _)| !elapsed_since_last)
-            .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _, _, log_stream_gated)| {
+            .filter_map(|(name, _, capture_group, json_field_path, kind, _, _, log_stream_gated)| {
-                Regex::new(message_match_regex).ok().map(|re| {
+                self.param_regex_cache.get(name).map(|re| {
                     (
                         name.to_string(),
-                        re,
+                        re.clone(),
                         capture_group.to_string(),
                         json_field_path.map(|path| path.to_string()),
                         kind,

src/engine/tests.rs

@@ -119,7 +119,7 @@ use std::collections::{BTreeMap, HashMap, HashSet};
     fn ingest_log_stream_fixture(index: &mut EngineSnapshot) {
         let mut header = None;
         for line in LOG_STREAM_FIXTURE.lines() {
-            auto_declare_log_stream_params(&mut index.index, &mut header, line);
+            auto_declare_log_stream_params(index, &mut header, line);
             index.auto_declare_params(line);
         }
     }
@@ -197,7 +197,7 @@ use std::collections::{BTreeMap, HashMap, HashSet};
         for (i, line) in LOG_STREAM_FIXTURE.lines().enumerate() {
             let sequence = i as u64;
             let timestamp = Timestamp::from_millis(sequence as i64);
-            auto_declare_log_stream_params(&mut index.index, &mut header, line);
+            auto_declare_log_stream_params(&mut index, &mut header, line);
             index.auto_declare_params(line);
             index.index.insert(
                 IndexKey::IngestedLog { sequence, timestamp },
@@ -232,7 +232,7 @@ use std::collections::{BTreeMap, HashMap, HashSet};
     fn non_log_stream_lines_do_not_get_any_columns_declared_without_a_cached_header() {
         let mut index = EngineSnapshot::default();
         let mut header = None;
-        auto_declare_log_stream_params(&mut index.index, &mut header, r#"ts=1788648105773 msg="entity_moved" x1=-4"#);
+        auto_declare_log_stream_params(&mut index, &mut header, r#"ts=1788648105773 msg="entity_moved" x1=-4"#);
 
         assert!(index.param_declarations_range().next().is_none());
         assert!(header.is_none());
@@ -247,7 +247,7 @@ use std::collections::{BTreeMap, HashMap, HashSet};
         for _ in 0..repeats {
             let mut header = None;
             for line in LOG_STREAM_FIXTURE.lines() {
-                auto_declare_log_stream_params(&mut index.index, &mut header, line);
+                auto_declare_log_stream_params(&mut index, &mut header, line);
                 index.auto_declare_params(line);
             }
         }