Add rerun rrd verify (#9128)

### Related
* Part of #9110

### What
Add `rerun rrd verify some.rrd` which verifies that the current rerun
version can load and understand the given .rrd file.

It goes through each component column in each record batch, find the
corresponding component, and then tries to deserialize the arrow data
within.

Author: emilk

crates/build/re_types_builder/src/codegen/rust/reflection.rs

@@ -147,6 +147,7 @@ fn generate_component_reflection(
                 docstring_md: #docstring_md,
                 custom_placeholder: #custom_placeholder,
                 datatype: #type_name::arrow_datatype(),
+                verify_arrow_array: #type_name::verify_arrow_array,
             }
         };
         quoted_pairs.push(quote! { (#quoted_name, #quoted_reflection) });

crates/store/re_log_encoding/src/decoder/mod.rs

@@ -310,7 +310,7 @@ impl<R: std::io::Read> Iterator for Decoder<R> {
                 return self.next();
             }
 
-            re_log::debug!("Reached end of stream, iterator complete");
+            re_log::trace!("Reached end of stream, iterator complete");
             return None;
         };
 

crates/store/re_log_encoding/src/decoder/streaming.rs

@@ -209,7 +209,7 @@ impl<R: AsyncBufRead + Unpin> Stream for StreamingDecoder<R> {
                     continue;
                 }
 
-                re_log::debug!("Reached end of stream, iterator complete");
+                re_log::trace!("Reached end of stream, iterator complete");
                 return std::task::Poll::Ready(None);
             };
 

crates/store/re_log_encoding/src/lib.rs

@@ -123,7 +123,9 @@ pub enum OptionsError {
     UnknownCompression(u8),
 
     // TODO(jan): Remove this at some point, realistically 1-2 releases after 0.23
-    #[error("Attempted to use the removed MsgPack serializer, which is no longer supported")]
+    #[error(
+        "You are trying to load an old .rrd file that's not supported by this version of Rerun."
+    )]
     RemovedMsgPackSerializer,
 
     #[error("Unknown serializer: {0}")]

crates/store/re_types/src/reflection/mod.rs

@@ -75,6 +75,13 @@ pub trait Loggable: 'static + Send + Sync + Clone + Sized + SizeBytes {
     ) -> crate::DeserializationResult<Vec<Option<Self>>> {
         Self::from_arrow(data).map(|v| v.into_iter().map(Some).collect())
     }
+
+    /// Verifies that the given Arrow array can be deserialized into a collection of [`Self`]s.
+    ///
+    /// Calls [`Self::from_arrow`] and returns an error if it fails.
+    fn verify_arrow_array(data: &dyn arrow::array::Array) -> crate::DeserializationResult<()> {
+        Self::from_arrow(data).map(|_| ())
+    }
 }
 
 /// A [`Component`] describes semantic data that can be used by any number of [`Archetype`]s.

crates/store/re_types_core/src/loggable.rs

@@ -240,6 +240,9 @@ pub struct ComponentReflection {
 
     /// Datatype of the component.
     pub datatype: arrow::datatypes::DataType,
+
+    /// Checks that the given Arrow array can be deserialized into a collection of [`Self`]s.
+    pub verify_arrow_array: fn(&dyn arrow::array::Array) -> crate::DeserializationResult<()>,
 }
 
 /// Runtime reflection about archetypes.
@@ -271,6 +274,10 @@ impl ArchetypeReflection {
     pub fn required_fields(&self) -> impl Iterator<Item = &ArchetypeFieldReflection> {
         self.fields.iter().filter(|field| field.is_required)
     }
+
+    pub fn get_field(&self, field_name: &str) -> Option<&ArchetypeFieldReflection> {
+        self.fields.iter().find(|field| field.name == field_name)
+    }
 }
 
 /// Additional information about an archetype's field.

crates/store/re_types_core/src/reflection.rs

@@ -104,7 +104,7 @@ impl FilterCommand {
                 Ok(size_bytes)
             });
 
-        for res in rx_decoder {
+        for (_source, res) in rx_decoder {
             let mut is_success = true;
 
             match res {

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

@@ -163,7 +163,7 @@ fn merge_and_compact(
 
     let mut entity_dbs: std::collections::HashMap<StoreId, EntityDb> = Default::default();
 
-    for res in rx {
+    for (_source, res) in rx {
         let mut is_success = true;
 
         match res {

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

@@ -2,11 +2,13 @@ mod compare;
 mod filter;
 mod merge_compact;
 mod print;
+mod verify;
 
 use self::compare::CompareCommand;
 use self::filter::FilterCommand;
 use self::merge_compact::{CompactCommand, MergeCommand};
 use self::print::PrintCommand;
+use self::verify::VerifyCommand;
 
 // ---
 
@@ -29,6 +31,11 @@ pub enum RrdCommands {
     /// Example: `rerun rrd print /my/recordings/*.rrd`
     Print(PrintCommand),
 
+    /// Verify the that the .rrd file can be loaded and correctly interpreted.
+    ///
+    /// Can be used to ensure that the current Rerun version can load the data.
+    Verify(VerifyCommand),
+
     /// Compacts the contents of one or more .rrd/.rbl files/streams and writes the result standard output.
     ///
     /// Reads from standard input if no paths are specified.
@@ -76,6 +83,7 @@ impl RrdCommands {
                     .with_context(|| format!("current directory {:?}", std::env::current_dir()))
             }
             Self::Print(print_command) => print_command.run(),
+            Self::Verify(verify_command) => verify_command.run(),
             Self::Compact(compact_command) => compact_command.run(),
             Self::Merge(merge_command) => merge_command.run(),
             Self::Filter(drop_command) => drop_command.run(),

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

@@ -44,7 +44,7 @@ impl PrintCommand {
         let version_policy = re_log_encoding::VersionPolicy::Warn;
         let (rx, _) = read_rrd_streams_from_file_or_stdin(version_policy, path_to_input_rrds);
 
-        for res in rx {
+        for (_source, res) in rx {
             let mut is_success = true;
 
             match res {

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

@@ -0,0 +1,174 @@
+use std::collections::HashSet;
+
+use arrow::array::AsArray as _;
+
+use re_log_types::LogMsg;
+use re_types::reflection::Reflection;
+
+use crate::commands::read_rrd_streams_from_file_or_stdin;
+
+// ---
+
+#[derive(Debug, Clone, clap::Parser)]
+pub struct VerifyCommand {
+    /// Paths to read from. Reads from standard input if none are specified.
+    path_to_input_rrds: Vec<String>,
+}
+
+impl VerifyCommand {
+    pub fn run(&self) -> anyhow::Result<()> {
+        let mut verifier = Verifier::new()?;
+
+        let Self { path_to_input_rrds } = self;
+
+        // TODO(cmc): might want to make this configurable at some point.
+        let version_policy = re_log_encoding::VersionPolicy::Warn;
+        let (rx, _) = read_rrd_streams_from_file_or_stdin(version_policy, path_to_input_rrds);
+
+        let mut seen_files = std::collections::HashSet::new();
+
+        for (source, res) in rx {
+            verifier.verify_log_msg(&source.to_string(), res?);
+            seen_files.insert(source);
+        }
+
+        if verifier.errors.is_empty() {
+            if seen_files.len() == 1 {
+                eprintln!("1 file verified without error.");
+            } else {
+                eprintln!("{} files verified without error.", seen_files.len());
+            }
+            Ok(())
+        } else {
+            for err in &verifier.errors {
+                eprintln!("{err}");
+            }
+            Err(anyhow::anyhow!(
+                "Verification failed with {} errors",
+                verifier.errors.len()
+            ))
+        }
+    }
+}
+
+struct Verifier {
+    reflection: Reflection,
+    errors: HashSet<String>,
+}
+
+impl Verifier {
+    fn new() -> anyhow::Result<Self> {
+        Ok(Self {
+            reflection: re_types::reflection::generate_reflection()?,
+            errors: HashSet::new(),
+        })
+    }
+
+    fn verify_log_msg(&mut self, source: &str, msg: LogMsg) {
+        match msg {
+            LogMsg::SetStoreInfo { .. } | LogMsg::BlueprintActivationCommand { .. } => {}
+
+            LogMsg::ArrowMsg(_store_id, arrow_msg) => {
+                self.verify_record_batch(source, &arrow_msg.batch);
+            }
+        }
+    }
+
+    fn verify_record_batch(&mut self, source: &str, batch: &arrow::array::RecordBatch) {
+        match re_sorbet::ChunkBatch::try_from(batch) {
+            Ok(chunk_batch) => self.verify_chunk_batch(source, &chunk_batch),
+            Err(err) => {
+                self.errors
+                    .insert(format!("{source}: Failed to parse batch: {err}"));
+            }
+        }
+    }
+
+    fn verify_chunk_batch(&mut self, source: &str, chunk_batch: &re_sorbet::ChunkBatch) {
+        for (component_descriptor, column) in chunk_batch.component_columns() {
+            if let Err(err) = self.verify_component_column(component_descriptor, column) {
+                self.errors.insert(format!(
+                    "{source}: Failed to deserialize column {}: {}",
+                    component_descriptor.component_name,
+                    re_error::format(err)
+                ));
+            }
+        }
+    }
+
+    fn verify_component_column(
+        &self,
+        component_descriptor: &re_sorbet::ComponentColumnDescriptor,
+        column: &dyn arrow::array::Array,
+    ) -> anyhow::Result<()> {
+        let re_sorbet::ComponentColumnDescriptor {
+            component_name,
+            archetype_name,
+            archetype_field_name,
+            ..
+        } = component_descriptor;
+
+        if !component_name.full_name().starts_with("rerun.") {
+            re_log::debug_once!("Ignoring non-Rerun component {component_name:?}");
+            return Ok(());
+        }
+
+        if component_name.is_indicator_component() {
+            // Lacks reflection and data
+        } else {
+            // Verify data
+            let component_reflection = self
+                .reflection
+                .components
+                .get(component_name)
+                .ok_or_else(|| anyhow::anyhow!("Unknown component"))?;
+
+            let list_array = column.as_list_opt::<i32>().ok_or_else(|| {
+                anyhow::anyhow!("Expected list array, found {:?}", column.data_type())
+            })?;
+
+            assert_eq!(column.len() + 1, list_array.offsets().len());
+
+            for i in 0..column.len() {
+                let cell = list_array.value(i);
+                (component_reflection.verify_arrow_array)(cell.as_ref())?;
+            }
+        }
+
+        if let Some(archetype_name) = archetype_name {
+            if archetype_name.full_name().starts_with("rerun.") {
+                // Verify archetype.
+                // We may want to have a flag to allow some of this?
+                let archetype_reflection = self
+                    .reflection
+                    .archetypes
+                    .get(archetype_name)
+                    .ok_or_else(|| anyhow::anyhow!("Unknown archetype: {archetype_name:?}"))?;
+
+                if let Some(archetype_field_name) = archetype_field_name {
+                    // Verify archetype field.
+                    // We may want to have a flag to allow some of this?
+                    let archetype_field_reflection = archetype_reflection
+                        .get_field(archetype_field_name)
+                        .ok_or_else(|| {
+                            anyhow::anyhow!(
+                                "Input column referred to the archetype field name {archetype_field_name:?} of {archetype_name:?}, which only has the fields: {:?}",
+                                archetype_reflection.fields.iter().map(|field| field.name)
+                            )
+                        })?;
+
+                    let expected_component_name = &archetype_field_reflection.component_name;
+                    if component_name != expected_component_name {
+                        return Err(anyhow::anyhow!(
+                            "Archetype field {archetype_field_name:?} of {archetype_name:?} has component {expected_component_name:?} in this version of Rerun, but the data column has component {component_name:?}"
+                        ));
+                    }
+                }
+            } else {
+                re_log::debug_once!("Ignoring non-Rerun archetype {archetype_name:?}");
+            }
+        }
+
+        Ok(())
+    }
+}

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

@@ -9,6 +9,21 @@ use re_log_types::LogMsg;
 
 // ---
 
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum InputSource {
+    Stdin,
+    File(PathBuf),
+}
+
+impl std::fmt::Display for InputSource {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Stdin => write!(f, "stdin"),
+            Self::File(path) => write!(f, "{path:?}"),
+        }
+    }
+}
+
 /// Asynchronously decodes potentially multiplexed RRD streams from the given `paths`, or standard
 /// input if none are specified.
 ///
@@ -26,7 +41,7 @@ pub fn read_rrd_streams_from_file_or_stdin(
     version_policy: re_log_encoding::VersionPolicy,
     paths: &[String],
 ) -> (
-    channel::Receiver<anyhow::Result<LogMsg>>,
+    channel::Receiver<(InputSource, anyhow::Result<LogMsg>)>,
     channel::Receiver<u64>,
 ) {
     let path_to_input_rrds = paths.iter().map(PathBuf::from).collect_vec();
@@ -53,14 +68,14 @@ pub fn read_rrd_streams_from_file_or_stdin(
                 {
                     Ok(decoder) => decoder,
                     Err(err) => {
-                        tx.send(Err(err)).ok();
+                        tx.send((InputSource::Stdin, Err(err))).ok();
                         return;
                     }
                 };
 
                 for res in &mut decoder {
                     let res = res.context("couldn't decode message from stdin -- skipping");
-                    tx.send(res).ok();
+                    tx.send((InputSource::Stdin, res)).ok();
                 }
 
                 size_bytes += decoder.size_bytes();
@@ -73,7 +88,8 @@ pub fn read_rrd_streams_from_file_or_stdin(
                     {
                         Ok(file) => file,
                         Err(err) => {
-                            tx.send(Err(err)).ok();
+                            tx.send((InputSource::File(rrd_path.clone()), Err(err)))
+                                .ok();
                             continue;
                         }
                     };
@@ -84,7 +100,8 @@ pub fn read_rrd_streams_from_file_or_stdin(
                         {
                             Ok(decoder) => decoder,
                             Err(err) => {
-                                tx.send(Err(err)).ok();
+                                tx.send((InputSource::File(rrd_path.clone()), Err(err)))
+                                    .ok();
                                 continue;
                             }
                         };
@@ -93,7 +110,7 @@ pub fn read_rrd_streams_from_file_or_stdin(
                         let res = res.context("decode rrd message").with_context(|| {
                             format!("couldn't decode message {rrd_path:?} -- skipping")
                         });
-                        tx.send(res).ok();
+                        tx.send((InputSource::File(rrd_path.clone()), res)).ok();
                     }
 
                     size_bytes += decoder.size_bytes();