Observe points without per-point allocation

asap-precompute-rs/src/config.rs

@@ -347,18 +347,27 @@ impl PrecomputeConfig {
     /// Thin wrapper over [`crate::matchers::series_key`], which owns
     /// the full canonical-byte formatting.
     pub fn series_key_for(&self, obs: &Observation) -> String {
+        let mut buf = String::new();
+        self.write_series_key_for(&mut buf, obs);
+        buf
+    }
+
+    /// Writes [`Self::series_key_for`] into `buf`, replacing its contents.
+    pub(crate) fn write_series_key_for(&self, buf: &mut String, obs: &Observation) {
+        use crate::matchers::write_series_key;
         if self.global_aggregation {
-            return crate::matchers::series_key(self.agg_id, &[], &[], &[]);
-        }
-        if self.omit_resource_attrs {
-            return crate::matchers::series_key(self.agg_id, &[], &obs.labels, &self.aggregate_by);
+            write_series_key(buf, self.agg_id, &[], &[], &[]);
+        } else if self.omit_resource_attrs {
+            write_series_key(buf, self.agg_id, &[], &obs.labels, &self.aggregate_by);
+        } else {
+            write_series_key(
+                buf,
+                self.agg_id,
+                &obs.resource_labels,
+                &obs.labels,
+                &self.aggregate_by,
+            );
         }
-        crate::matchers::series_key(
-            self.agg_id,
-            &obs.resource_labels,
-            &obs.labels,
-            &self.aggregate_by,
-        )
     }
 
     /// Rebuilds the same key from a series entry's stored labels.

asap-precompute-rs/src/matchers.rs

@@ -137,12 +137,26 @@ pub fn series_key(
     aggregate_by: &[String],
 ) -> String {
     let mut buf = String::new();
-    buf.push_str(&agg_id.to_string());
+    write_series_key(&mut buf, agg_id, resource_labels, labels, aggregate_by);
+    buf
+}
+
+/// Writes [`series_key`] into `buf`, replacing its contents, so a hot path
+/// can reuse one buffer.
+pub(crate) fn write_series_key(
+    buf: &mut String,
+    agg_id: AggId,
+    resource_labels: &[KeyValue],
+    labels: &[KeyValue],
+    aggregate_by: &[String],
+) {
+    use std::fmt::Write;
+    buf.clear();
+    let _ = write!(buf, "{agg_id}");
     buf.push('|');
-    write_attributes_key(&mut buf, resource_labels, &[]);
+    write_attributes_key(buf, resource_labels, &[]);
     buf.push('|');
-    write_attributes_key(&mut buf, labels, aggregate_by);
-    buf
+    write_attributes_key(buf, labels, aggregate_by);
 }
 
 /// Returns just the per-label-set portion of [`series_key`].
@@ -160,6 +174,17 @@ pub fn attributes_key(labels: &[KeyValue], aggregate_by: &[String]) -> String {
 
 fn write_attributes_key(buf: &mut String, labels: &[KeyValue], aggregate_by: &[String]) {
     if aggregate_by.is_empty() {
+        // Already sorted with unique keys (e.g. one label): same output as
+        // the sorted walk below, without collecting the keys.
+        if labels.windows(2).all(|w| w[0].key < w[1].key) {
+            for kv in labels {
+                buf.push_str(&kv.key);
+                buf.push('=');
+                buf.push_str(&kv.value);
+                buf.push(';');
+            }
+            return;
+        }
         // Sort by key for stable output.
         let mut keys: Vec<&str> = labels.iter().map(|kv| kv.key.as_str()).collect();
         keys.sort();
@@ -213,19 +238,35 @@ pub const ATTRIBUTE_KEY_LABEL: &str = "attribute.key";
 /// the attribute's value as bytes. A cardinality or frequency sketch over
 /// these series answers "how many distinct values" and "which values
 /// dominate" per attribute key.
-pub fn per_attribute(obs: &Observation, keys: &[String]) -> Vec<Observation> {
+///
+/// Each split is written into `scratch` and passed to `f`, so a stream of
+/// points reuses one observation's buffers instead of allocating per point.
+pub fn per_attribute<E>(
+    obs: &Observation,
+    keys: &[String],
+    scratch: &mut Observation,
+    mut f: impl FnMut(&Observation) -> Result<(), E>,
+) -> Result<(), E> {
     let all = keys.iter().any(|k| k == "*");
-    obs.labels
-        .iter()
-        .filter(|kv| all || keys.contains(&kv.key))
-        .map(|kv| Observation {
-            timestamp_ms: obs.timestamp_ms,
-            metric: obs.metric.clone(),
-            resource_labels: obs.resource_labels.clone(),
-            labels: vec![KeyValue::new(ATTRIBUTE_KEY_LABEL, kv.key.clone())],
-            value: crate::observation::ObservationValue::bytes(kv.value.as_bytes().to_vec()),
-        })
-        .collect()
+    for kv in &obs.labels {
+        if !all && !keys.contains(&kv.key) {
+            continue;
+        }
+        scratch.timestamp_ms = obs.timestamp_ms;
+        scratch.metric.clone_from(&obs.metric);
+        scratch.resource_labels.clone_from(&obs.resource_labels);
+        scratch.labels.resize_with(1, KeyValue::default);
+        scratch.labels[0].key.clear();
+        scratch.labels[0].key.push_str(ATTRIBUTE_KEY_LABEL);
+        scratch.labels[0].value.clone_from(&kv.key);
+        let value = &mut scratch.value;
+        value.kind = crate::observation::ObservationValueKind::Bytes;
+        value.envelope = None;
+        value.bytes.clear();
+        value.bytes.extend_from_slice(kv.value.as_bytes());
+        f(scratch)?;
+    }
+    Ok(())
 }
 
 #[cfg(test)]
@@ -264,4 +305,43 @@ mod tests {
         let matchers = vec![LabelMatcher::equal("region", "us-east")];
         assert!(LabelMatcher::matches_all(&matchers, &obs));
     }
+
+    #[test]
+    fn sorted_fast_path_writes_the_same_key_as_the_sorted_walk() {
+        let sorted = vec![KeyValue::new("a", "1"), KeyValue::new("b", "2")];
+        let unsorted = vec![KeyValue::new("b", "2"), KeyValue::new("a", "1")];
+        assert_eq!(attributes_key(&sorted, &[]), "a=1;b=2;");
+        assert_eq!(attributes_key(&unsorted, &[]), "a=1;b=2;");
+        let mut buf = "stale".to_string();
+        write_series_key(&mut buf, 7, &sorted, &unsorted, &[]);
+        assert_eq!(buf, series_key(7, &sorted, &unsorted, &[]));
+    }
+
+    #[test]
+    fn per_attribute_reuses_the_scratch_for_each_selected_key() {
+        let obs = Observation::new(
+            5,
+            "m",
+            vec![KeyValue::new("service", "s")],
+            vec![KeyValue::new("route", "/a"), KeyValue::new("user", "u1")],
+            ObservationValue::float(1.0),
+        );
+        let mut scratch = Observation::default();
+        let mut seen = Vec::new();
+        per_attribute::<()>(&obs, &["*".into()], &mut scratch, |o| {
+            seen.push(o.clone());
+            Ok(())
+        })
+        .unwrap();
+        let attr = |key: &str, value: &str| {
+            Observation::new(
+                5,
+                "m",
+                vec![KeyValue::new("service", "s")],
+                vec![KeyValue::new(ATTRIBUTE_KEY_LABEL, key)],
+                ObservationValue::bytes(value.as_bytes().to_vec()),
+            )
+        };
+        assert_eq!(seen, vec![attr("route", "/a"), attr("user", "u1")]);
+    }
 }

asap-precompute-rs/src/otap/baseline_nodes.rs

@@ -68,7 +68,7 @@ use super::answer_nodes::{
     default_grace, default_k, default_novelty_windows, default_quantiles, default_window, emit,
     gauges_pdata, invalid, now_ms, observations, series_key,
 };
-use super::codec::OtapSketchEncoder;
+use super::codec::{decode_each, OtapSketchEncoder};
 use crate::answers::{
     exact_answers, gauge, ks_distance, topk_of, CardinalityClassifier, CardinalityRules,
     DriftRules, DriftTracker, Event, ExactWindow, NoveltyTracker,
@@ -273,14 +273,20 @@ impl local::Processor<OtapPdata> for ExactBranch {
             Message::PData(pdata) => {
                 // Consumed into state here: ack so upstream can move on.
                 effects.notify_ack(AckMsg::new(pdata.clone())).await?;
-                let points = observations(pdata, effects).await;
-                // A window closes as soon as a point's event time passes its end.
-                let out = match points.first() {
-                    Some(first) => self.close(first.timestamp_ms),
-                    None => Vec::new(),
-                };
-                points.iter().for_each(|obs| self.observe(obs));
-                out
+                let mut out = None;
+                let decoded = decode_each(pdata, |obs| {
+                    // A window closes as soon as a point's event time passes its end.
+                    if out.is_none() {
+                        out = Some(self.close(obs.timestamp_ms));
+                    }
+                    self.observe(obs);
+                });
+                if let Err(e) = decoded {
+                    effects
+                        .info(&format!("exact: undecodable input: {e}"))
+                        .await;
+                }
+                out.unwrap_or_default()
             }
             Message::Control(NodeControlMsg::TimerTick { .. }) => {
                 self.close(now_ms().saturating_sub(IDLE_CLOSE_GRACE.as_millis() as u64))

asap-precompute-rs/src/otap/codec.rs

@@ -1310,19 +1310,35 @@ fn pretty_arrow_type(data_type: &arrow_schema::DataType) -> String {
 /// match arm already owns it and has no further use for the original
 /// context once this conversion runs.
 pub fn decode_pdata_to_observations(pdata: OtapPdata) -> Result<DecodeOutcome, CodecError> {
+    let mut observations = Vec::new();
+    let skipped_non_scalar = decode_each(pdata, |obs| observations.push(obs.clone()))?;
+    Ok(DecodeOutcome {
+        observations,
+        skipped_non_scalar,
+    })
+}
+
+/// Walks `pdata` like [`decode_pdata_to_observations`], passing each data
+/// point to `f` instead of collecting them. The observation is one buffer
+/// rewritten per point, so a hot path decodes without allocating per point.
+/// Returns the number of non-scalar data points skipped.
+pub fn decode_each(pdata: OtapPdata, f: impl FnMut(&Observation)) -> Result<usize, CodecError> {
     let (_context, payload) = pdata.into_parts();
     let arrow_records: OtapArrowRecords = payload
         .try_into_with_default()
         .map_err(|e| CodecError::Decode(format!("{e}")))?;
 
     let OtapArrowRecords::Metrics(_) = &arrow_records else {
-        return Ok(DecodeOutcome::default());
+        return Ok(0);
     };
 
     let view = OtapMetricsView::try_from(&arrow_records)
         .map_err(|e| CodecError::Decode(format!("{e}")))?;
 
-    let mut acc = DecodeAccumulator::default();
+    let mut acc = DecodeAccumulator {
+        scratch: Observation::default(),
+        sink: f,
+    };
     let mut skipped_non_scalar = 0usize;
 
     for resource in view.resources() {
@@ -1418,13 +1434,23 @@ pub fn decode_pdata_to_observations(pdata: OtapPdata) -> Result<DecodeOutcome, C
         }
     }
 
-    Ok(DecodeOutcome {
-        observations: acc.observations,
-        skipped_non_scalar,
-    })
+    Ok(skipped_non_scalar)
 }
 
-/// Accumulates one `Observation` per data point while walking the real
+/// Overwrites `labels[index]` (appending if needed) with `key=value`,
+/// reusing the label's string buffers.
+fn set_label(labels: &mut Vec<KeyValue>, index: usize, key: &[u8], value: &[u8]) {
+    if index == labels.len() {
+        labels.push(KeyValue::default());
+    }
+    let label = &mut labels[index];
+    label.key.clear();
+    label.key.push_str(&String::from_utf8_lossy(key));
+    label.value.clear();
+    label.value.push_str(&String::from_utf8_lossy(value));
+}
+
+/// Passes one `Observation` per data point to `sink` while walking the real
 /// `OtapMetricsView` — factored out of [`decode_pdata_to_observations`]
 /// so the Gauge and Sum branches (whose `NumberDataPointView`
 /// implementations are different concrete types, both borrowing from
@@ -1433,9 +1459,10 @@ pub fn decode_pdata_to_observations(pdata: OtapPdata) -> Result<DecodeOutcome, C
 /// point body twice, or trying to unify their different concrete
 /// iterator types into one `Vec` first (which doesn't compile —
 /// `gauge`/`sum` would need to outlive the branch that produced them).
-#[derive(Default)]
-struct DecodeAccumulator {
-    observations: Vec<Observation>,
+struct DecodeAccumulator<F> {
+    /// Rewritten for each data point.
+    scratch: Observation,
+    sink: F,
 }
 
 /// Parses an optional `sketch.type` value; absent means `Unspecified`.
@@ -1454,7 +1481,7 @@ struct DecodeSchema {
     schema_version: Option<u32>,
 }
 
-impl DecodeAccumulator {
+impl<F: FnMut(&Observation)> DecodeAccumulator<F> {
     fn push_summary_data_point<D: SummaryDataPointView>(
         &mut self,
         dp: D,
@@ -1512,7 +1539,7 @@ impl DecodeAccumulator {
             aggregation_temporality: 0,
             value: dp.sum(),
         };
-        self.observations.push(Observation::new(
+        (self.sink)(&Observation::new(
             timestamp_ms,
             metric_name.to_string(),
             resource_labels.to_vec(),
@@ -1549,12 +1576,21 @@ impl DecodeAccumulator {
         let mut schema_version = inherited_schema.schema_version;
         let mut window_start_ms: Option<u64> = None;
         let mut window_end_ms: Option<u64> = None;
-        let mut labels: Vec<KeyValue> = series_labels.to_vec();
+        let obs = &mut self.scratch;
+        for (index, label) in series_labels.iter().enumerate() {
+            set_label(
+                &mut obs.labels,
+                index,
+                label.key.as_bytes(),
+                label.value.as_bytes(),
+            );
+        }
+        let mut label_count = series_labels.len();
 
         for attr in dp.attributes() {
             let Some(val) = attr.value() else { continue };
-            let key = String::from_utf8_lossy(attr.key()).into_owned();
-            match key.as_str() {
+            let key = attr.key();
+            match std::str::from_utf8(key).unwrap_or_default() {
                 ATTR_ENVELOPE => envelope_bytes = val.as_bytes().map(<[u8]>::to_vec),
                 ATTR_SKETCH_TYPE => {
                     sketch_type_raw = val
@@ -1576,15 +1612,17 @@ impl DecodeAccumulator {
                     // string-typed values are representable as a label;
                     // a non-string, non-reserved attribute is dropped.
                     if let Some(s) = val.as_string() {
-                        labels.push(KeyValue::new(key, String::from_utf8_lossy(s).into_owned()));
+                        set_label(&mut obs.labels, label_count, key, s);
+                        label_count += 1;
                     }
                 }
             }
         }
+        obs.labels.truncate(label_count);
 
         // An empty `sketch.envelope` from some other producer routes to
         // the scalar path rather than an empty-payload envelope.
-        let observation_value = match envelope_bytes.filter(|b| !b.is_empty()) {
+        obs.value = match envelope_bytes.filter(|b| !b.is_empty()) {
             Some(payload) => {
                 let sketch_type = parse_sketch_type(sketch_type_raw.as_deref())?;
                 validate_self_describing_payload(sketch_type, &payload)
@@ -1594,7 +1632,7 @@ impl DecodeAccumulator {
                     sketch_type,
                     agg_id: agg_id.unwrap_or(0),
                     resource_labels: Vec::new(),
-                    labels: labels.clone(),
+                    labels: obs.labels.clone(),
                     window_start_ms: window_start_ms.unwrap_or(0),
                     window_end_ms: window_end_ms.unwrap_or(0),
                     payload,
@@ -1606,14 +1644,16 @@ impl DecodeAccumulator {
             }
             None => ObservationValue::float(value),
         };
-
-        self.observations.push(Observation::new(
-            timestamp_ms,
-            metric_name.to_string(),
-            resource_labels.to_vec(),
-            labels,
-            observation_value,
-        ));
+        obs.timestamp_ms = timestamp_ms;
+        if obs.metric != metric_name {
+            obs.metric.clear();
+            obs.metric.push_str(metric_name);
+        }
+        if obs.resource_labels.as_slice() != resource_labels {
+            obs.resource_labels.clear();
+            obs.resource_labels.extend_from_slice(resource_labels);
+        }
+        (self.sink)(obs);
         Ok(())
     }
 }