Correct multi-batch test coverage and drop redundant validate test

Addresses a second review pass over the dictionary-label changes:

- Rewrite the `formats` multi-batch test. The old
  `parquet_round_trip_multiple_batches` documented that flushing between
  chunks forces separate per-batch dictionaries, but `ArrowWriter`
  coalesces per-`write()` batches into one row group, so three flushes of
  four rows produced a single read batch — the doc claim was false and the
  test never exercised per-batch decode. `parquet_round_trip_multiple_read_batches`
  reads with `with_batch_size(2)` to genuinely emit several batches, decodes
  each batch's `Dictionary(Int32, Utf8)` labels via `resolve_label_dictionary`,
  and asserts `batch_count > 1` so the coverage claim cannot silently lapse.
- Assert `mean` and `is_monotonic` on the staging series in the captool
  `analyze_metric` test, matching the prod series assertions.
- Remove a redundant dictionary-label validate test; `validate.rs` already
  covers the `validate_parquet` decode path across happy and error cases.

Author: preinlein

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

@@ -321,5 +321,7 @@ mod tests {
         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);
+        assert_eq!(staging_stats.mean, 150.0);
+        assert!(staging_stats.is_monotonic);
     }
 }

lading_capture/src/formats.rs

@@ -465,66 +465,108 @@ mod tests {
         }
     }
 
-    /// Roundtrip labels across multiple flushed batches.
+    /// Decode dictionary labels across multiple read 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.
+    /// The parquet reader emits one `RecordBatch` per `batch_size` rows, and each
+    /// batch decodes its own `Dictionary(Int32, Utf8)` label columns independently
+    /// (indices are batch-local). Production files exceed the default batch size,
+    /// so they are always read as several such batches — a path a small
+    /// single-batch roundtrip never reaches. Here a small `batch_size` forces
+    /// multiple batches over rows that share label strings (`env=prod`) so each
+    /// batch resolves the shared strings from its own dictionary. A reader that
+    /// mishandled per-batch dictionaries (stale indices, assuming one global
+    /// dictionary) would return wrong labels for some batch. The test asserts it
+    /// actually produced more than one batch so the coverage claim cannot silently
+    /// lapse.
     #[test]
-    fn parquet_round_trip_multiple_batches() {
+    fn parquet_round_trip_multiple_read_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 envs = ["prod", "staging", "prod", "staging", "prod", "staging"];
 
         let mut buffer = Cursor::new(Vec::new());
-        let mut input_lines: Vec<Line> = Vec::new();
+        let mut expected_labels: FxHashMap<u64, FxHashMap<String, String>> = FxHashMap::default();
         {
             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");
+            for (i, env) in envs.iter().enumerate() {
+                let fetch_index = i as u64;
+                let labels: FxHashMap<String, String> = [("env".to_string(), (*env).to_string())]
+                    .into_iter()
+                    .collect();
+                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.clone(),
+                    value_histogram: Vec::new(),
+                };
+                writer.write_metric(&line).expect("write");
+                expected_labels.insert(fetch_index, labels);
             }
+            writer.flush().expect("flush");
             writer.close().expect("close");
         }
         let bytes = buffer.into_inner();
 
-        let deserialized_lines = read_parquet_lines(&bytes).expect("read parquet");
+        // A batch size below the row count forces the reader to emit several
+        // batches, each decoding its labels from an independent dictionary.
+        let bytes_buf = Bytes::copy_from_slice(&bytes);
+        let reader = ParquetRecordBatchReaderBuilder::try_new(bytes_buf)
+            .expect("reader builder")
+            .with_batch_size(2)
+            .build()
+            .expect("reader");
+
+        let mut batch_count = 0usize;
+        let mut seen = 0usize;
+        for batch_result in reader {
+            let batch = batch_result.expect("batch");
+            if batch.num_rows() == 0 {
+                continue;
+            }
+            batch_count += 1;
 
-        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);
+            let fetch_index_array = batch
+                .column_by_name("fetch_index")
+                .expect("fetch_index column")
+                .as_any()
+                .downcast_ref::<UInt64Array>()
+                .expect("fetch_index is UInt64Array");
+            let labels_array = batch
+                .column_by_name("labels")
+                .expect("labels column")
+                .as_any()
+                .downcast_ref::<MapArray>()
+                .expect("labels is MapArray");
+
+            for row in 0..batch.num_rows() {
+                let fetch_index = fetch_index_array.value(row);
+                let labels_slice: StructArray = labels_array.value(row);
+                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), "Labels values")
+                    .expect("label values are Dictionary(Int32,Utf8)");
+                let decoded: FxHashMap<String, String> = (0..keys.len())
+                    .map(|i| (keys.value(i).to_string(), values.value(i).to_string()))
+                    .collect();
+                assert_eq!(
+                    &decoded,
+                    expected_labels
+                        .get(&fetch_index)
+                        .expect("known fetch_index"),
+                    "labels mismatch at fetch_index {fetch_index}"
+                );
+                seen += 1;
+            }
         }
+
+        assert!(
+            batch_count > 1,
+            "expected multiple read batches, got {batch_count}"
+        );
+        assert_eq!(seen, envs.len(), "every row decoded exactly once");
     }
 
     #[expect(clippy::too_many_lines)]

lading_capture/src/validate/parquet.rs

@@ -309,66 +309,3 @@ 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);
-    }
-}