arrow: monomorphize parser ingestion and shrink DecodeCursor

src/arrow/options.rs

@@ -37,7 +37,7 @@ pub const DEFAULT_DECODE_BUDGET_MAX_DURATION: Duration = Duration::from_millis(1
 
 /// Number of Layer 0 `Entry` items processed between wall-clock `Instant::elapsed` checks in
 /// budgeted decode steps.
-pub const DECODE_BUDGET_DURATION_CHECK_INTERVAL: usize = 16;
+pub const DECODE_BUDGET_DURATION_CHECK_INTERVAL: usize = 64;
 
 /// Cooperative step budget bounding how much entry ingestion work [`super::DecodeStream::step`],
 /// [`super::ArrowDecoder::step_decode`], [`super::ArrowDecoder::decode_cooperative`], and

src/arrow/tests.rs

@@ -444,7 +444,7 @@ fn decode_stream_splits_mid_family_preserves_metadata_and_handles_edges() {
         let mut stream = dec.decode_stream(labeled, &Format::TEXT, &zero_opts);
         assert!(stream.next().unwrap().is_ok());
     }
-    assert!(dec.prom_spans.capacity() >= 4);
+    assert!(dec.prom_state.labels.capacity() >= 4);
 
     // Consecutive assembled families (histograms/summaries) preceded by `# TYPE` flush at
     // `max_data_points_per_batch` when the next family begins, and a single assembled family
@@ -710,9 +710,11 @@ fn decode_stream_step_respects_entry_byte_and_duration_budgets() {
     assert_eq!(mid_err_stream.step(step1).unwrap(), DecodeStep::Done);
     drop(mid_err_stream);
 
-    // Non-zero `max_duration` checks elapsed time every 16 entries (`entries.is_multiple_of(16)`).
+    // Non-zero `max_duration` checks elapsed time every `DECODE_BUDGET_DURATION_CHECK_INTERVAL`
+    // entries.
+    let total_heavy_lines = DECODE_BUDGET_DURATION_CHECK_INTERVAL + 16;
     let mut heavy_lines = String::new();
-    for i in 0..20 {
+    for i in 0..total_heavy_lines {
         heavy_lines.push_str("m{");
         for k in 0..32 {
             if k > 0 {
@@ -726,7 +728,7 @@ fn decode_stream_step_respects_entry_byte_and_duration_budgets() {
     let mut dur_stream = dec.decode_stream(
         heavy_lines.as_bytes(),
         &Format::TEXT,
-        &DecodeOptions::new().with_max_data_points_per_batch(64),
+        &DecodeOptions::new().with_max_data_points_per_batch(total_heavy_lines * 2),
     );
     assert_eq!(dur_stream.step(dur_budget).unwrap(), DecodeStep::Pending);
     let consumed_lines = heavy_lines.as_bytes()[..dur_stream.pos()]
@@ -735,7 +737,7 @@ fn decode_stream_step_respects_entry_byte_and_duration_budgets() {
         .count();
     assert_eq!(consumed_lines, DECODE_BUDGET_DURATION_CHECK_INTERVAL);
     match dur_stream.step(dur_budget).unwrap() {
-        DecodeStep::Batch(batch) => assert_eq!(batch.num_data_points(), 20),
+        DecodeStep::Batch(batch) => assert_eq!(batch.num_data_points(), total_heavy_lines),
         other => panic!("expected DecodeStep::Batch, got {other:?}"),
     }
     assert_eq!(dur_stream.step(dur_budget).unwrap(), DecodeStep::Done);
@@ -850,12 +852,12 @@ fn lifetime_free_decode_cursor_supports_owned_job_across_steps() {
         DecodeStep::Pending
     );
     job.decoder.finish_decode(&mut aborted);
-    let retained_after_first = job.decoder.om_scratch.retained_memory_bytes();
+    let retained_after_first = job.decoder.om_state.retained_memory_bytes();
     assert!(aborted.is_done());
     assert!(retained_after_first > 0);
     job.decoder.finish_decode(&mut aborted);
     assert_eq!(
-        job.decoder.om_scratch.retained_memory_bytes(),
+        job.decoder.om_state.retained_memory_bytes(),
         retained_after_first
     );
 
@@ -1005,11 +1007,11 @@ fn lifetime_free_decode_cursor_supports_owned_job_across_steps() {
         DecodeStep::Batch(records) => assert_eq!(records.num_data_points(), 2),
         other => panic!("expected active cursor to finish cleanly, got {other:?}"),
     }
-    let grown_prom_cap = job.decoder.prom_spans.capacity();
+    let grown_prom_cap = job.decoder.prom_state.labels.capacity();
     assert!(grown_prom_cap >= 18);
     job.decoder.finish_decode(&mut stale_after_active);
     assert!(stale_after_active.is_done());
-    assert_eq!(job.decoder.prom_spans.capacity(), grown_prom_cap);
+    assert_eq!(job.decoder.prom_state.labels.capacity(), grown_prom_cap);
 
     let mut om_holder = job
         .decoder
@@ -1021,12 +1023,12 @@ fn lifetime_free_decode_cursor_supports_owned_job_across_steps() {
         .unwrap();
     job.decoder.finish_decode(&mut om_holder);
     assert_eq!(
-        job.decoder.om_scratch.retained_memory_bytes(),
+        job.decoder.om_state.retained_memory_bytes(),
         retained_after_first
     );
     job.decoder.finish_decode(&mut stale_om_cursor);
     assert_eq!(
-        job.decoder.om_scratch.retained_memory_bytes(),
+        job.decoder.om_state.retained_memory_bytes(),
         retained_after_first
     );
 
@@ -1153,7 +1155,7 @@ fn cooperative_async_helpers_respect_fast_path_and_yield_budgets() {
         let mut cx = Context::from_waker(Waker::noop());
         assert!(matches!(pinned.as_mut().poll(&mut cx), Poll::Pending));
     }
-    assert!(dec.prom_spans.capacity() > 0);
+    assert!(dec.prom_state.labels.capacity() > 0);
     assert!(dec.metrics.is_empty());
 
     let bad_coop = b"m{a=\"1\",b=\"2\",c=\"3\",d=\"4\"} 1\nbad_line\n";

src/parse.rs

@@ -321,6 +321,7 @@ pub trait Parser<'a> {
     fn comment(&self) -> &'a [u8];
 
     /// Returns the metric name and unit of the current [`Entry::Unit`].
+    #[inline]
     fn unit(&self) -> (&[u8], &[u8]) {
         (b"", b"")
     }

src/parse/omlex.rs

@@ -17,8 +17,9 @@
 use super::ByteSpan;
 
 /// Lexer states, named after the start conditions in `openmetricslex.l`.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 enum State {
+    #[default]
     Init,
     Comment,
     /// After `# HELP`, `# TYPE` or `# UNIT`, before the metric name.
@@ -35,7 +36,7 @@ enum State {
 }
 
 /// Lifetime-free cursor state of [`Lexer`].
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub(crate) struct LexerState {
     pub(super) pos: usize,
     pub(super) token_start: usize,

src/parse/omparse.rs

@@ -64,6 +64,7 @@ impl OmScratch {
 
 /// Lifetime-free state of [`OpenMetricsParser`] that can be stored across suspension points
 /// alongside an owned input buffer without self-referential lifetimes.
+#[derive(Default)]
 pub(crate) struct OpenMetricsParserState {
     lexer: LexerState,
     done: bool,
@@ -85,7 +86,28 @@ pub(crate) struct OpenMetricsParserState {
 }
 
 impl OpenMetricsParserState {
-    /// Takes the reusable scratch buffers in place for `Drop` handlers.
+    /// Resets cursor state for a new scrape while preserving internal scratch capacities in place.
+    #[inline]
+    pub(crate) fn reset(&mut self) {
+        self.lexer = LexerState::default();
+        self.done = false;
+        self.peeking = false;
+        self.line_start = 0;
+        self.series = ByteSpan::default();
+        self.text = ByteSpan::default();
+        self.name = None;
+        self.family_name_len = 0;
+        self.metric_type = MetricType::Unknown;
+        self.value = 0.0;
+        self.timestamp = None;
+        self.labels.clear();
+        self.exemplar = None;
+        self.exemplar_labels.clear();
+        self.peek_labels.clear();
+        self.start_timestamp_key.clear();
+        self.start_timestamp = None;
+    }
+
     pub(crate) fn take_scratch(&mut self) -> OmScratch {
         self.labels.clear();
         self.exemplar_labels.clear();
@@ -98,6 +120,20 @@ impl OpenMetricsParserState {
             start_timestamp_key: std::mem::take(&mut self.start_timestamp_key),
         }
     }
+
+    pub(crate) fn put_scratch(&mut self, mut scratch: OmScratch) {
+        scratch.clear();
+        self.labels = scratch.labels;
+        self.exemplar_labels = scratch.exemplar_labels;
+        self.peek_labels = scratch.peek_labels;
+        self.start_timestamp_key = scratch.start_timestamp_key;
+    }
+
+    pub(crate) fn retained_memory_bytes(&self) -> usize {
+        (self.labels.capacity() + self.exemplar_labels.capacity() + self.peek_labels.capacity())
+            * size_of::<LabelSpan>()
+            + self.start_timestamp_key.capacity()
+    }
 }
 
 /// Zero-copy parser for the OpenMetrics 1.0 text exposition format.
@@ -140,6 +176,7 @@ impl<'a> OpenMetricsParser<'a> {
     }
 
     /// Creates a parser over `input` reusing pre-allocated scratch buffers.
+    #[inline]
     pub(crate) fn with_scratch(input: &'a [u8], mut scratch: OmScratch) -> Self {
         scratch.clear();
         Self {
@@ -217,6 +254,7 @@ impl<'a> OpenMetricsParser<'a> {
     }
 
     /// Takes the reusable scratch buffers in place for `Drop` handlers.
+    #[inline]
     pub(crate) fn take_scratch(&mut self) -> OmScratch {
         self.labels.clear();
         self.exemplar_labels.clear();
@@ -243,12 +281,14 @@ impl<'a> OpenMetricsParser<'a> {
     }
 
     /// Returns the raw metric name and raw (escaped) help text for `Entry::Help`.
+    #[inline]
     pub(crate) fn raw_help(&self) -> (&'a [u8], &'a [u8]) {
         (self.metric_name(), &self.lexer.input[self.text])
     }
 
     /// Returns the raw metric name, validated UTF-8 label spans, full input slice, optional
     /// timestamp in milliseconds, and sample value for `Entry::Series`.
+    #[inline]
     pub(crate) fn raw_series_parts(&self) -> (&'a [u8], &[LabelSpan], &'a [u8], Option<i64>, f64) {
         (
             self.metric_name(),
@@ -261,6 +301,7 @@ impl<'a> OpenMetricsParser<'a> {
 
     /// Returns the exemplar value, optional timestamp in milliseconds, and label spans for the
     /// current `Entry::Series`, if present.
+    #[inline]
     pub(crate) fn raw_exemplar_parts(&self) -> Option<(f64, Option<i64>, &[LabelSpan])> {
         self.exemplar
             .map(|(val, ts)| (val, ts, self.exemplar_labels.as_slice()))
@@ -557,6 +598,7 @@ impl<'a> OpenMetricsParser<'a> {
 }
 
 impl<'a> Parser<'a> for OpenMetricsParser<'a> {
+    #[inline]
     fn next_entry(&mut self) -> Result<Option<Entry>, ParseError> {
         if self.done {
             return Ok(None);
@@ -599,10 +641,12 @@ impl<'a> Parser<'a> for OpenMetricsParser<'a> {
         (self.metric_name(), unescape(&self.lexer.input[self.text]))
     }
 
+    #[inline]
     fn metric_type(&self) -> (&'a [u8], MetricType) {
         (self.metric_name(), self.metric_type)
     }
 
+    #[inline]
     fn unit(&self) -> (&'a [u8], &'a [u8]) {
         (self.metric_name(), &self.lexer.input[self.text])
     }

src/parse/promlex.rs

@@ -17,8 +17,9 @@
 use super::ByteSpan;
 
 /// Lexer states, named after the start conditions in `promlex.l`.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 enum State {
+    #[default]
     Init,
     Comment,
     /// After `# HELP` or `# TYPE`, before the metric name.
@@ -32,7 +33,7 @@ enum State {
 }
 
 /// Lifetime-free cursor state of [`Lexer`].
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub(crate) struct LexerState {
     pub(super) pos: usize,
     pub(super) token_start: usize,

src/parse/promparse.rs

@@ -71,6 +71,7 @@ pub(crate) struct LabelSpan {
 
 /// Lifetime-free state of [`PromParser`] that can be stored across suspension points alongside an
 /// owned input buffer without self-referential lifetimes.
+#[derive(Default)]
 pub(crate) struct PromParserState {
     lexer: LexerState,
     series: ByteSpan,
@@ -80,14 +81,33 @@ pub(crate) struct PromParserState {
     timestamp: Option<i64>,
     line_start: usize,
     name: Option<ByteSpan>,
-    labels: Vec<LabelSpan>,
+    pub(crate) labels: Vec<LabelSpan>,
 }
 
 impl PromParserState {
-    /// Takes the reusable label span buffer in place for `Drop` handlers.
-    pub(crate) fn take_scratch(&mut self) -> Vec<LabelSpan> {
+    pub(crate) fn with_capacity(label_cap: usize) -> Self {
+        Self {
+            labels: Vec::with_capacity(label_cap),
+            ..Self::default()
+        }
+    }
+
+    /// Resets cursor state for a new scrape while preserving `labels` capacity in place.
+    #[inline]
+    pub(crate) fn reset(&mut self) {
+        self.lexer = LexerState::default();
+        self.series = ByteSpan::default();
+        self.text = ByteSpan::default();
+        self.last_type = MetricType::Unknown;
+        self.value = 0.0;
+        self.timestamp = None;
+        self.line_start = 0;
+        self.name = None;
         self.labels.clear();
-        std::mem::take(&mut self.labels)
+    }
+
+    pub(crate) fn retained_memory_bytes(&self) -> usize {
+        self.labels.capacity() * size_of::<LabelSpan>()
     }
 }
 
@@ -115,6 +135,7 @@ impl<'a> PromParser<'a> {
     }
 
     /// Creates a parser over `input` reusing a caller-owned `labels` span buffer.
+    #[inline]
     pub(crate) fn with_scratch(input: &'a [u8], mut labels: Vec<LabelSpan>) -> Self {
         labels.clear();
         Self {
@@ -169,6 +190,7 @@ impl<'a> PromParser<'a> {
     }
 
     /// Takes the reusable label span buffer in place for `Drop` handlers.
+    #[inline]
     pub(crate) fn take_scratch(&mut self) -> Vec<LabelSpan> {
         self.labels.clear();
         std::mem::take(&mut self.labels)
@@ -188,12 +210,14 @@ impl<'a> PromParser<'a> {
 
     /// Returns the raw metric name and raw (escaped) help text for `Entry::Help` without
     /// allocating a `Cow::Owned` buffer.
+    #[inline]
     pub(crate) fn raw_help(&self) -> (&'a [u8], &'a [u8]) {
         (self.metric_name(), &self.lexer.input[self.text])
     }
 
     /// Returns the raw metric name, validated UTF-8 label spans, full input slice, optional
     /// timestamp in milliseconds, and sample value for `Entry::Series`.
+    #[inline]
     pub(crate) fn raw_series_parts(&self) -> (&'a [u8], &[LabelSpan], &'a [u8], Option<i64>, f64) {
         (
             self.metric_name(),
@@ -395,6 +419,7 @@ impl<'a> PromParser<'a> {
 }
 
 impl<'a> Parser<'a> for PromParser<'a> {
+    #[inline]
     fn next_entry(&mut self) -> Result<Option<Entry>, ParseError> {
         loop {
             self.line_start = self.lexer.pos;
@@ -424,6 +449,7 @@ impl<'a> Parser<'a> for PromParser<'a> {
         )
     }
 
+    #[inline]
     fn metric_type(&self) -> (&'a [u8], MetricType) {
         (self.metric_name(), self.last_type)
     }