Fix captool analyze for dictionary labels; bind it to the writer in CI

The `Dictionary(Int32, Utf8)` label typing broke a second in-repo reader that
the first pass missed: `captool analyze` decoded the label `key`/`value` columns
with `downcast_ref::<StringArray>()`, which returns `None` once those columns are
`DictionaryArray<Int32Type>`, so `analyze_metric` failed on every capture written
by the new writer. It compiles (auto-detected bin target), so there was no
compile-time signal, and nothing exercised the path.

- Make `resolve_label_dictionary` `pub` so `captool` (in the `lading` crate) uses
  the same canonical decode path as the `lading-capture` reader/validator instead
  of re-rolling the two-step downcast.
- Route `captool analyze` label decoding through `resolve_label_dictionary`.
- Add a `captool` test that writes a capture with the production `Format` writer
  and runs `list_metrics`/`analyze_metric` over it, so any future divergence
  between the writer's on-disk schema and the analyzer fails in CI.
- Add a `lading-capture` roundtrip test that flushes between chunks, forcing
  shared label strings into independent per-batch dictionaries.
- Add a `validate_parquet` test covering its own dictionary-decode and
  per-series label-reconstruction path.
- Fix doc/import/casing nits from review.

Author: preinlein

Cargo.lock

@@ -107,6 +107,7 @@ async-pidfd = { version = "0.1" }
 tempfile = { workspace = true }
 proptest = { workspace = true }
 tower-test = { version = "0.4" }
+uuid = { workspace = true }
 
 [build-dependencies]
 prost-build = { workspace = true }

lading/Cargo.toml

@@ -7,7 +7,8 @@ use std::collections::BTreeSet;
 use std::fs::File;
 use std::path::Path;
 
-use arrow_array::{Array, Float64Array, MapArray, StringArray, UInt64Array};
+use arrow_array::{Array, ArrayAccessor, Float64Array, MapArray, StringArray, UInt64Array};
+use lading_capture::formats::parquet::resolve_label_dictionary;
 use lading_capture_schema::columns;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
 use rustc_hash::FxHashMap;
@@ -170,16 +171,10 @@ pub(crate) fn analyze_metric<P: AsRef<Path>>(
             // Extract labels
             let mut sorted_labels = BTreeSet::new();
             let labels_slice = labels_array.value(row);
-            let key_array = labels_slice
-                .column(0)
-                .as_any()
-                .downcast_ref::<StringArray>()
-                .ok_or_else(|| Error::InvalidColumnType("Label keys not String".to_string()))?;
-            let value_array = labels_slice
-                .column(1)
-                .as_any()
-                .downcast_ref::<StringArray>()
-                .ok_or_else(|| Error::InvalidColumnType("Label values not String".to_string()))?;
+            let key_array = resolve_label_dictionary(labels_slice.column(0), "Labels keys")
+                .map_err(Error::InvalidColumnType)?;
+            let value_array = resolve_label_dictionary(labels_slice.column(1), "Labels values")
+                .map_err(Error::InvalidColumnType)?;
 
             for i in 0..key_array.len() {
                 let key = key_array.value(i);
@@ -231,3 +226,100 @@ pub(crate) fn analyze_metric<P: AsRef<Path>>(
 
     Ok(results)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use lading_capture::formats::parquet::Format;
+    use lading_capture::line::{Line, LineValue, MetricKind};
+    use rustc_hash::FxHashMap;
+    use tempfile::NamedTempFile;
+    use uuid::Uuid;
+
+    fn labels(pairs: &[(&str, &str)]) -> FxHashMap<String, String> {
+        pairs
+            .iter()
+            .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
+            .collect()
+    }
+
+    /// Write the given lines to a parquet capture file using the production
+    /// [`Format`] writer and return the backing temp file.
+    fn write_capture(lines: &[Line]) -> NamedTempFile {
+        let tmp = NamedTempFile::new().expect("create temp file");
+        let file = tmp.reopen().expect("reopen temp file");
+        let mut writer = Format::new(file, 3).expect("create parquet writer");
+        for line in lines {
+            writer.write_metric(line).expect("write metric");
+        }
+        writer.flush().expect("flush");
+        writer.close().expect("close");
+        tmp
+    }
+
+    /// `analyze_metric` reads captures written by the production parquet writer.
+    ///
+    /// This binds `captool analyze` to the capture writer: the file is produced
+    /// by `lading_capture`'s [`Format`], so any change to the on-disk label
+    /// encoding (e.g. the `Dictionary(Int32, Utf8)` label columns) that the
+    /// analyzer fails to track breaks this test in CI rather than shipping a
+    /// silently broken tool.
+    #[test]
+    #[expect(
+        clippy::float_cmp,
+        reason = "stats are computed from exact integer-valued inputs"
+    )]
+    fn analyze_metric_reads_writer_output() {
+        let run_id = Uuid::from_u128(0x1234_5678_9abc_def0_1234_5678_9abc_def0);
+        let prod = labels(&[("env", "prod"), ("service", "api")]);
+        let staging = labels(&[("env", "staging")]);
+
+        let mut lines: Vec<Line> = (0u64..3)
+            .map(|i| Line {
+                run_id,
+                time: 1_000 + u128::from(i),
+                fetch_index: i,
+                metric_name: "requests".to_string(),
+                metric_kind: MetricKind::Counter,
+                value: LineValue::Int((i + 1) * 10),
+                labels: prod.clone(),
+                value_histogram: Vec::new(),
+            })
+            .collect();
+        lines.extend((0u64..2).map(|i| Line {
+            run_id,
+            time: 2_000 + u128::from(i),
+            fetch_index: i,
+            metric_name: "requests".to_string(),
+            metric_kind: MetricKind::Counter,
+            value: LineValue::Int((i + 1) * 100),
+            labels: staging.clone(),
+            value_histogram: Vec::new(),
+        }));
+
+        let tmp = write_capture(&lines);
+
+        let metrics = list_metrics(tmp.path()).expect("list metrics");
+        assert!(
+            metrics.iter().any(|m| m.name == "requests"),
+            "requests metric should be listed, got {metrics:?}"
+        );
+
+        let series = analyze_metric(tmp.path(), "requests").expect("analyze requests");
+        assert_eq!(series.len(), 2, "two distinct label sets expected");
+
+        let prod_key: BTreeSet<String> = ["env:prod".to_string(), "service:api".to_string()].into();
+        let staging_key: BTreeSet<String> = ["env:staging".to_string()].into();
+
+        let prod_stats = series.get(&prod_key).expect("prod series present");
+        assert_eq!(prod_stats.min, 10.0);
+        assert_eq!(prod_stats.max, 30.0);
+        assert_eq!(prod_stats.mean, 20.0);
+        assert!(prod_stats.is_monotonic);
+
+        let staging_stats = series.get(&staging_key).expect("staging series present");
+        assert_eq!(staging_stats.min, 100.0);
+        assert_eq!(staging_stats.max, 200.0);
+    }
+}

lading/src/bin/captool/analyze/parquet.rs

@@ -465,6 +465,68 @@ mod tests {
         }
     }
 
+    /// Roundtrip labels across multiple flushed batches.
+    ///
+    /// The writer builds a fresh dictionary per `flush()`, so each batch encodes
+    /// its label strings in an independent dictionary whose indices restart at 0.
+    /// Flushing between chunks that share the same strings (`env=prod`) forces
+    /// those strings into separate per-batch dictionaries, exercising the reader's
+    /// per-batch dictionary reconciliation — a case a single-flush test cannot
+    /// reach. A reader that assumed one global dictionary would return wrong
+    /// labels here while passing the single-batch tests.
+    #[test]
+    fn parquet_round_trip_multiple_batches() {
+        let run_id = Uuid::from_u128(0x0fed_cba9_8765_4321_0fed_cba9_8765_4321);
+        let label_sets = [
+            [("env".to_string(), "prod".to_string())],
+            [("env".to_string(), "staging".to_string())],
+            [("env".to_string(), "prod".to_string())],
+        ];
+
+        let mut buffer = Cursor::new(Vec::new());
+        let mut input_lines: Vec<Line> = Vec::new();
+        {
+            let mut writer = super::parquet::Format::new(&mut buffer, 3).expect("create writer");
+            for (chunk_idx, labels) in label_sets.iter().enumerate() {
+                for row in 0u64..4 {
+                    let fetch_index = chunk_idx as u64 * 4 + row;
+                    let line = Line {
+                        run_id,
+                        time: 1_000 + u128::from(fetch_index),
+                        fetch_index,
+                        metric_name: "requests".to_string(),
+                        metric_kind: MetricKind::Counter,
+                        value: LineValue::Int(fetch_index),
+                        labels: labels.iter().cloned().collect(),
+                        value_histogram: Vec::new(),
+                    };
+                    writer.write_metric(&line).expect("write");
+                    input_lines.push(line);
+                }
+                // Flush between chunks so each lands in its own dictionary-encoded
+                // batch rather than one coalesced batch.
+                writer.flush().expect("flush");
+            }
+            writer.close().expect("close");
+        }
+        let bytes = buffer.into_inner();
+
+        let deserialized_lines = read_parquet_lines(&bytes).expect("read parquet");
+
+        assert_eq!(input_lines.len(), deserialized_lines.len());
+        let mut by_fetch_index: FxHashMap<u64, &Line> = deserialized_lines
+            .iter()
+            .map(|l| (l.fetch_index, l))
+            .collect();
+        for input in &input_lines {
+            let output = by_fetch_index
+                .remove(&input.fetch_index)
+                .expect("row round-trips");
+            assert_eq!(input.labels, output.labels);
+            assert_eq!(input.metric_name, output.metric_name);
+        }
+    }
+
     #[expect(clippy::too_many_lines)]
     fn read_parquet_lines(bytes: &[u8]) -> Result<Vec<Line>, Box<dyn std::error::Error>> {
         let bytes_buf = Bytes::copy_from_slice(bytes);
@@ -568,9 +630,9 @@ mod tests {
                 };
 
                 let labels_slice: StructArray = labels_array.value(row_idx);
-                let keys = resolve_label_dictionary(labels_slice.column(0), "label keys")
+                let keys = resolve_label_dictionary(labels_slice.column(0), "Labels keys")
                     .expect("label keys are Dictionary(Int32,Utf8)");
-                let values = resolve_label_dictionary(labels_slice.column(1), "label values")
+                let values = resolve_label_dictionary(labels_slice.column(1), "Labels values")
                     .expect("label values are Dictionary(Int32,Utf8)");
 
                 let mut labels = FxHashMap::default();

lading_capture/src/formats.rs

@@ -30,10 +30,17 @@ use crate::line;
 /// dictionary view of its `Utf8` values.
 ///
 /// Both the label `key` and `value` columns are written as
-/// `Dictionary(Int32, Utf8)`, so every reader resolves them the same way. On
-/// failure returns a message naming `column` (e.g. `"Labels keys"`) so callers
-/// can surface distinct errors for the key and value columns.
-pub(crate) fn resolve_label_dictionary<'a>(
+/// `Dictionary(Int32, Utf8)`, so every reader resolves them the same way. This
+/// is the single canonical decode path: readers in other crates (e.g. `captool`)
+/// call it too, so a schema change here surfaces in every reader at once. On
+/// failure returns a message prefixed with `column_name` (e.g. `"Labels keys"`)
+/// so callers can surface distinct errors for the key and value columns.
+///
+/// # Errors
+///
+/// Returns an error message if `column` is not a `Dictionary(Int32, _)` or its
+/// dictionary values are not `Utf8`.
+pub fn resolve_label_dictionary<'a>(
     column: &'a dyn Array,
     column_name: &str,
 ) -> Result<TypedDictionaryArray<'a, Int32Type, StringArray>, String> {

lading_capture/src/formats/parquet.rs

@@ -17,11 +17,9 @@ use arrow_array::{
     UInt64Array,
 };
 use lading_capture_schema::columns;
-
-use crate::formats::parquet::resolve_label_dictionary;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
 
-use crate::validate::ValidationResult;
+use crate::{formats::parquet::resolve_label_dictionary, validate::ValidationResult};
 
 /// Errors for parquet validation
 #[derive(thiserror::Error, Debug)]
@@ -311,3 +309,66 @@ pub fn validate_parquet<P: AsRef<Path>>(
         first_error,
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use crate::formats::parquet::Format;
+    use crate::line::{Line, LineValue, MetricKind};
+    use tempfile::NamedTempFile;
+    use uuid::Uuid;
+
+    /// A well-formed capture with dictionary-encoded labels validates cleanly.
+    ///
+    /// This drives the validator's own decode path — `resolve_label_dictionary`
+    /// plus the `sorted_labels` reconstruction that builds per-series keys from
+    /// the `Dictionary(Int32, Utf8)` label columns — which is distinct from the
+    /// reader path exercised elsewhere. Two label sets sharing a key (`env`) must
+    /// resolve to two separate series.
+    #[test]
+    fn validate_parquet_decodes_dictionary_labels() {
+        let run_id = Uuid::from_u128(0x1111_2222_3333_4444_5555_6666_7777_8888);
+
+        // fetch_index -> time is 1:1 across all rows (required invariant); each
+        // series' fetch_index and time are strictly increasing.
+        let mut lines: Vec<Line> = Vec::new();
+        for (env, count) in [("prod", 3u64), ("staging", 2u64)] {
+            for fetch_index in 0..count {
+                lines.push(Line {
+                    run_id,
+                    time: 1_000 + u128::from(fetch_index),
+                    fetch_index,
+                    metric_name: "requests".to_string(),
+                    metric_kind: MetricKind::Counter,
+                    value: LineValue::Int(fetch_index),
+                    labels: [("env".to_string(), env.to_string())].into_iter().collect(),
+                    value_histogram: Vec::new(),
+                });
+            }
+        }
+
+        let tmp = NamedTempFile::new().expect("create temp file");
+        {
+            let file = tmp.reopen().expect("reopen temp file");
+            let mut writer = Format::new(file, 3).expect("create writer");
+            for line in &lines {
+                writer.write_metric(line).expect("write");
+            }
+            writer.flush().expect("flush");
+            writer.close().expect("close");
+        }
+
+        let result = validate_parquet(tmp.path(), None).expect("validate");
+
+        assert_eq!(result.line_count, 5);
+        assert_eq!(
+            result.unique_series, 2,
+            "prod and staging are distinct series"
+        );
+        assert_eq!(result.unique_fetch_indices, 3);
+        assert_eq!(result.fetch_index_errors, 0);
+        assert_eq!(result.per_series_errors, 0);
+        assert!(result.first_error.is_none(), "{:?}", result.first_error);
+    }
+}