expose RRD manifests all through the CLI

Author: teh-cmc

crates/top/rerun/src/commands/rrd/filter.rs

@@ -176,7 +176,7 @@ impl FilterCommand {
             .join()
             .map_err(|err| anyhow::anyhow!("Unknown error: {err:?}"))??; // NOLINT: there is no `Display` for this `err`
 
-        let rrds_in_size = rx_size_bytes.recv().ok();
+        let rrds_in_size = rx_size_bytes.recv().ok().map(|(size, _footers)| size);
         let size_reduction =
             if let (Some(rrds_in_size), rrd_out_size) = (rrds_in_size, rrd_out_size) {
                 format!(

crates/top/rerun/src/commands/rrd/merge_compact.rs

@@ -321,7 +321,7 @@ fn merge_and_compact(
 
     rrd_out.flush().context("couldn't flush output")?;
 
-    let rrds_in_size = rx_size_bytes.recv().ok();
+    let rrds_in_size = rx_size_bytes.recv().ok().map(|(size, _footers)| size);
     let num_chunks_reduction = format!(
         "-{:3.3}%",
         100.0 - num_chunks_after as f64 / (num_chunks_before as f64 + f64::EPSILON) * 100.0

crates/top/rerun/src/commands/rrd/print.rs

@@ -2,6 +2,7 @@ use anyhow::Context as _;
 use arrow::array::RecordBatch;
 use itertools::Itertools as _;
 
+use re_arrow_util::RecordBatchExt as _;
 use re_byte_size::SizeBytes as _;
 use re_log_types::{LogMsg, SetStoreInfo};
 use re_sdk::EntityPath;
@@ -43,13 +44,17 @@ pub struct PrintCommand {
     #[clap(long, default_missing_value="true", num_args=0..=1)]
     full_metadata: Option<bool>,
 
-    /// Transpose record batches before printing them?
-    #[clap(long, default_missing_value="true", num_args=0..=1)]
-    transposed: Option<bool>,
-
     /// Show only chunks belonging to this entity.
     #[clap(long)]
     entity: Option<String>,
+
+    /// If true, displays all the parsed footers at the end.
+    #[clap(long, default_missing_value="true", num_args=0..=1)]
+    footers: Option<bool>,
+
+    /// Transpose record batches before printing them?
+    #[clap(long, default_missing_value="true", num_args=0..=1)]
+    transposed: Option<bool>,
 }
 
 impl PrintCommand {
@@ -60,15 +65,17 @@ impl PrintCommand {
             verbose,
             migrate,
             full_metadata,
-            transposed,
             entity,
+            footers,
+            transposed,
         } = self;
         let continue_on_error = continue_on_error.unwrap_or(true);
 
         let migrate = migrate.unwrap_or(true);
-        let transposed = transposed.unwrap_or(false);
         let full_metadata = full_metadata.unwrap_or(false);
         let entity = entity.map(|e| EntityPath::parse_forgiving(&e));
+        let footers = footers.unwrap_or(false);
+        let transposed = transposed.unwrap_or(false);
 
         let options = Options {
             verbose,
@@ -87,7 +94,7 @@ impl PrintCommand {
             );
         }
 
-        let (rx, _) = read_rrd_streams_from_file_or_stdin(&path_to_input_rrds);
+        let (rx, rx_done) = read_rrd_streams_from_file_or_stdin(&path_to_input_rrds);
 
         for (_source, res) in rx {
             let mut is_success = true;
@@ -113,6 +120,35 @@ impl PrintCommand {
             }
         }
 
+        if footers {
+            for (_, rrd_manifests) in rx_done {
+                for (source, mut rrd_manifest) in rrd_manifests? {
+                    // Just to be nice: this will display the origin of the data in the header.
+                    rrd_manifest
+                        .data
+                        .schema_metadata_mut()
+                        .insert("rerun:source".to_owned(), source.to_string());
+
+                    // Drop all per-entity and/or per-component columns to keep things readable.
+                    //
+                    // TODO(cmc): more config flags for columns to show etc.
+                    let filtered = rrd_manifest
+                        .data
+                        .filter_columns_by(|f| f.name().starts_with("chunk_"))?;
+
+                    let formatted = re_arrow_util::format_record_batch_opts(
+                        &filtered,
+                        &re_arrow_util::RecordBatchFormatOpts {
+                            max_cell_content_width: 32,
+                            ..Default::default()
+                        },
+                    );
+
+                    println!("{formatted}");
+                }
+            }
+        }
+
         Ok(())
     }
 }

crates/top/rerun/src/commands/stdio.rs

@@ -5,6 +5,7 @@ use crossbeam::channel;
 use itertools::Itertools as _;
 
 use re_chunk::external::crossbeam;
+use re_log_encoding::RrdManifest;
 
 // ---
 
@@ -30,17 +31,19 @@ impl std::fmt::Display for InputSource {
 /// * The first channel contains both the successfully decoded data, if any, as well as any
 ///   errors faced during processing.
 /// * The second channel, which will fire only once, after all processing is done, indicates the
-///   total number of bytes processed.
+///   total number of bytes processed, and returns all RRD manifests that were parsed from footers
+///   in the underlying stream.
 ///
 /// This function is best-effort: it will try to make progress even in the face of errors.
 /// It is up to the user to decide whether and when to stop.
 ///
 /// This function is capable of decoding multiple independent recordings from a single stream.
+#[expect(clippy::type_complexity)] // internal private API for the CLI impl
 pub fn read_rrd_streams_from_file_or_stdin(
     paths: &[String],
 ) -> (
     channel::Receiver<(InputSource, anyhow::Result<re_log_types::LogMsg>)>,
-    channel::Receiver<u64>,
+    channel::Receiver<(u64, anyhow::Result<Vec<(InputSource, RrdManifest)>>)>,
 ) {
     read_any_rrd_streams_from_file_or_stdin::<re_log_types::LogMsg>(paths)
 }
@@ -55,7 +58,8 @@ pub fn read_rrd_streams_from_file_or_stdin(
 /// * The first channel contains both the successfully decoded data, if any, as well as any
 ///   errors faced during processing.
 /// * The second channel, which will fire only once, after all processing is done, indicates the
-///   total number of bytes processed.
+///   total number of bytes processed, and returns all RRD manifests that were parsed from footers
+///   in the underlying stream.
 ///
 /// This function is best-effort: it will try to make progress even in the face of errors.
 /// It is up to the user to decide whether and when to stop.
@@ -68,25 +72,27 @@ pub fn read_rrd_streams_from_file_or_stdin(
 // TODO(ab): For pre-0.25 legacy data with `StoreId` missing their application id, the migration
 // in `Decoder` requires `SetStoreInfo` to arrive before the corresponding `ArrowMsg`. Ideally
 // this tool would cache orphan `ArrowMsg` until a matching `SetStoreInfo` arrives.
+#[expect(clippy::type_complexity)] // internal private API for the CLI impl
 pub fn read_raw_rrd_streams_from_file_or_stdin(
     paths: &[String],
 ) -> (
     channel::Receiver<(
         InputSource,
         anyhow::Result<re_protos::log_msg::v1alpha1::log_msg::Msg>,
     )>,
-    channel::Receiver<u64>,
+    channel::Receiver<(u64, anyhow::Result<Vec<(InputSource, RrdManifest)>>)>,
 ) {
     read_any_rrd_streams_from_file_or_stdin::<re_protos::log_msg::v1alpha1::log_msg::Msg>(paths)
 }
 
+#[expect(clippy::type_complexity)] // internal private API for the CLI impl
 fn read_any_rrd_streams_from_file_or_stdin<
     T: re_log_encoding::DecoderEntrypoint + Send + 'static,
 >(
     paths: &[String],
 ) -> (
     channel::Receiver<(InputSource, anyhow::Result<T>)>,
-    channel::Receiver<u64>,
+    channel::Receiver<(u64, anyhow::Result<Vec<(InputSource, RrdManifest)>>)>,
 ) {
     let path_to_input_rrds = paths
         .iter()
@@ -95,26 +101,32 @@ fn read_any_rrd_streams_from_file_or_stdin<
         .collect_vec();
 
     // TODO(cmc): might want to make this configurable at some point.
-    let (tx, rx) = crossbeam::channel::bounded(100);
-    let (tx_size_bytes, rx_size_bytes) = crossbeam::channel::bounded(1);
+    let (tx_msgs, rx_msgs) = crossbeam::channel::bounded(100);
+    let (tx_metadata, rx_metadata) = crossbeam::channel::bounded(1);
 
     _ = std::thread::Builder::new()
         .name("rerun-rrd-in".to_owned())
         .spawn(move || {
+            let mut rrd_manifests = Ok(Vec::new());
             let mut size_bytes = 0;
 
             if path_to_input_rrds.is_empty() {
                 // stdin
 
+                let source = InputSource::Stdin;
                 let stdin = std::io::BufReader::new(std::io::stdin().lock());
                 let mut decoder = re_log_encoding::Decoder::decode_lazy(stdin);
 
                 for res in &mut decoder {
                     let res = res.context("couldn't decode message from stdin -- skipping");
-                    tx.send((InputSource::Stdin, res)).ok();
+                    tx_msgs.send((source.clone(), res)).ok();
                 }
 
                 size_bytes += decoder.num_bytes_processed();
+                rrd_manifests = decoder
+                    .rrd_manifests()
+                    .context("couldn't decode footers")
+                    .map(|manifests| manifests.into_iter().map(|m| (source.clone(), m)).collect());
             } else {
                 // file(s)
 
@@ -124,28 +136,35 @@ fn read_any_rrd_streams_from_file_or_stdin<
                     {
                         Ok(file) => file,
                         Err(err) => {
-                            tx.send((InputSource::File(rrd_path.clone()), Err(err)))
+                            tx_msgs
+                                .send((InputSource::File(rrd_path.clone()), Err(err)))
                                 .ok();
                             continue;
                         }
                     };
 
+                    let source = InputSource::File(rrd_path.clone());
                     let rrd_file = std::io::BufReader::new(rrd_file);
-                    let mut messages = re_log_encoding::Decoder::decode_lazy(rrd_file);
-
-                    for res in &mut messages {
+                    let mut decoder = re_log_encoding::Decoder::decode_lazy(rrd_file);
+                    for res in &mut decoder {
                         let res = res.context("decode rrd message").with_context(|| {
                             format!("couldn't decode message {rrd_path:?} -- skipping")
                         });
-                        tx.send((InputSource::File(rrd_path.clone()), res)).ok();
+                        tx_msgs.send((source.clone(), res)).ok();
                     }
 
-                    size_bytes += messages.num_bytes_processed();
+                    size_bytes += decoder.num_bytes_processed();
+                    rrd_manifests = decoder
+                        .rrd_manifests()
+                        .context("couldn't decode footers")
+                        .map(|manifests| {
+                            manifests.into_iter().map(|m| (source.clone(), m)).collect()
+                        });
                 }
             }
 
-            tx_size_bytes.send(size_bytes).ok();
+            tx_metadata.send((size_bytes, rrd_manifests)).ok();
         });
 
-    (rx, rx_size_bytes)
+    (rx_msgs, rx_metadata)
 }

docs/content/reference/cli.md

@@ -537,12 +537,15 @@ Example: `rerun rrd print /my/recordings/*.rrd`
 * `--full-metadata <FULL_METADATA>`
 > If true, includes `rerun.` prefixes on keys.
 
-* `--transposed <TRANSPOSED>`
-> Transpose record batches before printing them?
-
 * `--entity <ENTITY>`
 > Show only chunks belonging to this entity.
 
+* `--footers <FOOTERS>`
+> If true, displays all the parsed footers at the end.
+
+* `--transposed <TRANSPOSED>`
+> Transpose record batches before printing them?
+
 ## rerun rrd route
 
 Manipulates the metadata of log message streams without decoding the payloads.
@@ -576,6 +579,13 @@ Note: Because the payload of the messages is never decoded, no migration or veri
 >
 > When this flag is set and multiple input .rdd files are specified, blueprint activation commands will be dropped from the resulting output.
 
+* `--recompute-manifests <RECOMPUTE_MANIFESTS>`
+> If set, this will compute an RRD footer with the appropriate manifest for the routed data.
+>
+> By default, `rerun rrd route` will always drop all existing RRD manifests when routing data, as doing so invalidates their contents. This flag makes it possible to recompute an RRD manifest for the routed data, but beware that it has to decode the data, which means it is A) much slower and B) will migrate the data to the latest Sorbet specification automatically.
+>
+> [Default: `false`]
+
 ## rerun rrd stats
 
 Compute important statistics for one or more .rrd/.rbl files/streams.