arrow: implement otel-arrow-dfe-pdata-views MetricsView over ArrowDecoder

Cargo.toml

@@ -14,13 +14,19 @@ exclude = ["tests/otap-compat"]
 
 [features]
 default = ["arrow", "dto"]
-arrow = ["dep:arrow-array", "dep:arrow-buffer", "dep:arrow-schema"]
+arrow = [
+    "dep:arrow-array",
+    "dep:arrow-buffer",
+    "dep:arrow-schema",
+    "dep:otel-arrow-dfe-pdata-views",
+]
 dto = ["dep:prost", "dep:prost-types"]
 
 [dependencies]
 arrow-array = { version = "59", default-features = false, optional = true }
 arrow-buffer = { version = "59", default-features = false, optional = true }
 arrow-schema = { version = "59", default-features = false, optional = true }
+otel-arrow-dfe-pdata-views = { version = "0.54", optional = true }
 prost = { version = "0.14", optional = true }
 prost-types = { version = "0.14", optional = true }
 thiserror = "2"

benches/decode.rs

@@ -31,7 +31,7 @@ use expfmt::format::Format;
 static ALLOC: CountingAlloc = CountingAlloc::new();
 
 /// Filters `alltypes.237mfs.prom.txt` down to its 209 counter, gauge, and untyped metric families
-/// until `#8` adds classic histogram and summary assembly.
+/// to isolate simple scalar family decoding from histogram and summary assembly.
 fn filter_counter_gauge_families(raw: &str) -> Vec<u8> {
     let mut out = String::with_capacity(raw.len());
     let mut skip_family: Option<String> = None;
@@ -78,6 +78,7 @@ fn main() {
     let nometa = fs::read(root.join("alltypes.237mfs.nometa.prom.txt")).unwrap();
     let full = fs::read_to_string(root.join("alltypes.237mfs.prom.txt")).unwrap();
     let typed_simple = filter_counter_gauge_families(&full);
+    let full_bytes = full.as_bytes();
 
     let opts = DecodeOptions::new().with_timestamp_ms(0);
     let stream_256_opts = opts.with_max_data_points_per_batch(256);
@@ -87,6 +88,7 @@ fn main() {
     for (case, payload) in [
         ("alltypes.237mfs.nometa", nometa.as_slice()),
         ("alltypes.209mfs.counter_gauge", typed_simple.as_slice()),
+        ("alltypes.237mfs", full_bytes),
     ] {
         let bytes = payload.len() as u64;
 
@@ -106,6 +108,19 @@ fn main() {
                     black_box(records);
                 }
             })
+            .bench(format!("Decode/case={case}/mode=decode_view"), move |b| {
+                let mut decoder = ArrowDecoder::new();
+                black_box(decoder.decode_view(payload, &Format::TEXT, &opts).unwrap());
+
+                b.set_bytes(bytes);
+                b.report_allocs();
+                while b.keep_running() {
+                    let view = decoder
+                        .decode_view(black_box(payload), &Format::TEXT, &opts)
+                        .unwrap();
+                    black_box(view);
+                }
+            })
             .bench(format!("Decode/case={case}/mode=stream_16k"), move |b| {
                 let mut decoder = ArrowDecoder::new();
                 for chunk in decoder.decode_stream(payload, &Format::TEXT, &opts) {

specs/expfmt/SPEC.md

@@ -113,7 +113,7 @@ scrape bytes
   -> direct OTAP metric records
 ```
 
-- `ArrowDecoder` appends parsed metrics, attributes, number data points, summary data points, classic and exponential histogram data points, and exemplars directly into Arrow array builders (`arrow-array` and `arrow-schema`), matching the relational `id` and `parent_id` tables in [`rust/otap-dataflow/crates/pdata`](https://github.com/open-telemetry/otel-arrow/tree/main/rust/otap-dataflow/crates/pdata). Callers can either materialize one batch set per scrape with `ArrowDecoder::decode` or pull bounded chunks on demand with `ArrowDecoder::decode_stream` to cap scratch memory and stay within OTAP's `u16` per-batch metric ID limit.
+- `ArrowDecoder` appends parsed metrics, attributes, number data points, summary data points, classic and exponential histogram data points, and exemplars directly into Arrow array builders (`arrow-array` and `arrow-schema`), matching the relational `id` and `parent_id` tables in [`rust/otap-dataflow/crates/pdata`](https://github.com/open-telemetry/otel-arrow/tree/main/rust/otap-dataflow/crates/pdata). Callers can materialize one batch set per scrape with `ArrowDecoder::decode`, pull bounded chunks on demand with `ArrowDecoder::decode_stream` to cap scratch memory and stay within OTAP's `u16` per-batch metric ID limit, or traverse zero-copy [`otel-arrow-dfe-pdata-views`](https://crates.io/crates/otel-arrow-dfe-pdata-views) `MetricsView` wrappers over the staged buffers with `ArrowDecoder::decode_view` and `DecodeStream::next_view`.
 - Metric families in text exposition are decoded as contiguous runs, with `# HELP`, `# TYPE`, and `# UNIT` metadata lines preceding that family's samples. Like OpenMetrics 1.0/2.0 and Prometheus's scrape loop, `ArrowDecoder` tracks only the active family cursor instead of maintaining a whole-scrape hash table to reassemble or reject previously seen names, so a non-contiguous repetition of an earlier metric name starts a fresh untyped run unless preceded by its own metadata.
 - A `LabelFilter` / `Relabeler` hook runs on the borrowed `LabelsBuffer` from Layer 0 so dropped series never allocate in Arrow builders or metric family state.
 - The metric family assembler groups classic histogram `_bucket`, `_sum`, and `_count` lines and summary `quantile`, `_sum`, and `_count` lines (plus OpenMetrics `_created` start timestamps) into complete metric families.

src/arrow.rs

@@ -27,9 +27,12 @@ mod records;
 mod schema;
 #[cfg(test)]
 mod tests;
+pub mod views;
 
 use std::iter::FusedIterator;
 
+pub use otel_arrow_dfe_pdata_views as pdata_views;
+
 use crate::format::{Format, FormatType};
 use crate::parse::{
     Entry, LabelSpan, METRIC_NAME_LABEL, MetricType, OmScratch, OpenMetricsParser, ParseError,
@@ -48,6 +51,7 @@ pub use self::options::{
 };
 pub use self::records::{MetricPayloadType, OtapMetricRecords};
 pub use self::schema::PROMETHEUS_TYPE_METADATA_KEY;
+pub use self::views::DecodedMetricsView;
 
 /// Maximum distinct `UnivariateMetrics` rows representable in a single OTAP batch (`u16` IDs).
 const MAX_METRICS_PER_BATCH: usize = (u16::MAX as usize) + 1;
@@ -254,12 +258,51 @@ impl ArrowDecoder {
         format: &Format,
         opts: &DecodeOptions,
     ) -> Result<OtapMetricRecords, DecodeError> {
+        let has_default_ts = self.stage_scrape(input, format, opts)?;
+        self.finish_batches(has_default_ts)
+    }
+
+    /// Decodes a single exposition scrape buffer of `format` into [`ArrowDecoder`]'s reusable
+    /// staging buffers and returns a zero-copy [`DecodedMetricsView`] implementing
+    /// [`otel_arrow_dfe_pdata_views::views::metrics::MetricsView`].
+    ///
+    /// Unlike [`ArrowDecoder::decode`], `decode_view` skips materializing Arrow
+    /// [`arrow_array::RecordBatch`]es, allowing downstream OTAP encoders such as
+    /// `otel_arrow_dfe_pdata::encode::encode_metrics_otap_batch` to traverse the staged metric
+    /// families, data points, exemplars, and interned attributes directly.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`DecodeError`] under the same conditions as [`ArrowDecoder::decode`].
+    pub fn decode_view(
+        &mut self,
+        input: &[u8],
+        format: &Format,
+        opts: &DecodeOptions,
+    ) -> Result<DecodedMetricsView<'_>, DecodeError> {
+        self.stage_scrape(input, format, opts)?;
+        Ok(self.view())
+    }
+
+    /// Returns a zero-copy [`DecodedMetricsView`] over the metric batch currently staged in `self`.
+    #[inline]
+    #[must_use]
+    pub fn view(&self) -> DecodedMetricsView<'_> {
+        DecodedMetricsView::new(self)
+    }
+
+    fn stage_scrape(
+        &mut self,
+        input: &[u8],
+        format: &Format,
+        opts: &DecodeOptions,
+    ) -> Result<bool, DecodeError> {
         self.reset_scrape();
         let (mut parser, default_ts_ns) = self.init_scrape_parser(input, format, opts)?;
 
         let mut pending_series = false;
         let mut done = false;
-        let res = self.ingest_entries::<false>(
+        let res = self.stage_entries::<false>(
             &mut parser,
             opts,
             default_ts_ns,
@@ -268,7 +311,8 @@ impl ArrowDecoder {
             &mut done,
         );
         parser.take_scratch_into(self);
-        Ok(res?.unwrap_or_default())
+        res?;
+        Ok(default_ts_ns.is_some())
     }
 
     /// Streams `input` as a sequence of non-empty [`OtapMetricRecords`] chunks bounded by
@@ -417,18 +461,18 @@ impl ArrowDecoder {
         Ok(())
     }
 
-    // Monomorphizing on `CHUNKED` shares one entry dispatch loop between `decode` (`false`) and
-    // `decode_stream` (`true`) while compiling out chunk-boundary flush checks on single-batch
-    // calls.
-    fn ingest_entries<const CHUNKED: bool>(
+    // Monomorphizing on `CHUNKED` shares one entry dispatch loop between single-batch (`false`)
+    // and streaming (`true`) decoding while compiling out chunk-boundary checks on single-batch
+    // calls. Returns `Ok(true)` when a non-empty chunk (or single batch) is staged.
+    fn stage_entries<const CHUNKED: bool>(
         &mut self,
         parser: &mut ActiveParser<'_>,
         opts: &DecodeOptions,
         default_ts_ns: Option<i64>,
         max_data_points: usize,
         pending_series: &mut bool,
         done: &mut bool,
-    ) -> Result<Option<OtapMetricRecords>, DecodeError> {
+    ) -> Result<bool, DecodeError> {
         loop {
             let entry = if CHUNKED && *pending_series {
                 *pending_series = false;
@@ -440,14 +484,9 @@ impl ArrowDecoder {
                         self.flush_pending_family(opts, default_ts_ns)?;
                         *done = true;
                         if CHUNKED {
-                            if self.metrics.is_empty() {
-                                return Ok(None);
-                            }
-                            let batch = self.finish_batches(default_ts_ns.is_some())?;
-                            self.reset_batch();
-                            return Ok(Some(batch));
+                            return Ok(!self.metrics.is_empty());
                         }
-                        return Ok(Some(self.finish_batches(default_ts_ns.is_some())?));
+                        return Ok(true);
                     }
                 }
             };
@@ -490,9 +529,7 @@ impl ArrowDecoder {
                                 self.current_family
                                     .start_new(raw_name, &self.series_name_buf);
                                 *pending_series = true;
-                                let batch = self.finish_batches(default_ts_ns.is_some())?;
-                                self.reset_batch();
-                                return Ok(Some(batch));
+                                return Ok(true);
                             }
                             self.current_family
                                 .start_new(raw_name, &self.series_name_buf);
@@ -513,9 +550,7 @@ impl ArrowDecoder {
                                 self.flush_pending_family(opts, default_ts_ns)?;
                             }
                             *pending_series = true;
-                            let batch = self.finish_batches(default_ts_ns.is_some())?;
-                            self.reset_batch();
-                            return Ok(Some(batch));
+                            return Ok(true);
                         }
                     }
 
@@ -525,9 +560,7 @@ impl ArrowDecoder {
                         && self.assembly.is_empty()
                         && self.total_data_points() >= max_data_points
                     {
-                        let batch = self.finish_batches(default_ts_ns.is_some())?;
-                        self.reset_batch();
-                        return Ok(Some(batch));
+                        return Ok(true);
                     }
                 }
             }
@@ -689,7 +722,7 @@ impl ArrowDecoder {
 
         let parent_id = self
             .current_family
-            .emit_into_batch(&mut self.metrics, opts)?;
+            .emit_into_batch(&mut self.metrics, opts, dp_id)?;
         let start_ts_ns = start_ts_ms.and_then(|ms| ms.checked_mul(1_000_000));
         let resolved_ts_ns = match sample_ts_ms {
             Some(ms) => Some(ms_to_ns(ms)?),
@@ -699,6 +732,7 @@ impl ArrowDecoder {
             .base
             .push(parent_id, start_ts_ns, resolved_ts_ns);
         self.num_dp.values.push(value);
+        self.metrics.dp_ranges[parent_id as usize].1 = dp_id + 1;
         Ok(())
     }
 
@@ -1211,27 +1245,39 @@ pub struct DecodeStream<'dec, 'input> {
     done: bool,
 }
 
-impl Iterator for DecodeStream<'_, '_> {
-    type Item = Result<OtapMetricRecords, DecodeError>;
+impl DecodeStream<'_, '_> {
+    /// Advances the stream to the next bounded chunk and returns a zero-copy
+    /// [`DecodedMetricsView`] borrowing the staged batch without materializing Arrow
+    /// [`arrow_array::RecordBatch`]es.
+    pub fn next_view(&mut self) -> Option<Result<DecodedMetricsView<'_>, DecodeError>> {
+        match self.step_chunk() {
+            Some(Ok(())) => Some(Ok(self.decoder.view())),
+            Some(Err(err)) => Some(Err(err)),
+            None => None,
+        }
+    }
 
-    fn next(&mut self) -> Option<Self::Item> {
+    fn step_chunk(&mut self) -> Option<Result<(), DecodeError>> {
         if self.done {
             return None;
         }
         if let Some(err) = self.init_err.take() {
             self.done = true;
             return Some(Err(err));
         }
-        match self.decoder.ingest_entries::<true>(
+        if !self.decoder.metrics.is_empty() {
+            self.decoder.reset_batch();
+        }
+        match self.decoder.stage_entries::<true>(
             &mut self.parser,
             &self.opts,
             self.default_ts_ns,
             self.max_data_points,
             &mut self.pending_series,
             &mut self.done,
         ) {
-            Ok(Some(batch)) => Some(Ok(batch)),
-            Ok(None) => None,
+            Ok(true) => Some(Ok(())),
+            Ok(false) => None,
             Err(err) => {
                 self.done = true;
                 self.decoder.reset_scrape();
@@ -1241,6 +1287,24 @@ impl Iterator for DecodeStream<'_, '_> {
     }
 }
 
+impl Iterator for DecodeStream<'_, '_> {
+    type Item = Result<OtapMetricRecords, DecodeError>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.step_chunk()? {
+            Ok(()) => {
+                let res = self.decoder.finish_batches(self.default_ts_ns.is_some());
+                if res.is_err() {
+                    self.done = true;
+                    self.decoder.reset_scrape();
+                }
+                Some(res)
+            }
+            Err(err) => Some(Err(err)),
+        }
+    }
+}
+
 impl FusedIterator for DecodeStream<'_, '_> {}
 
 impl Drop for DecodeStream<'_, '_> {