Implement fallback mode for raw MCAP layer (#11136)

Co-authored-by: Jochen Görtler <[email protected]>

Author: oxkitsune

crates/store/re_data_loader/src/loader_mcap.rs

@@ -2,7 +2,6 @@
 
 use std::{io::Cursor, path::Path, sync::mpsc::Sender};
 
-use anyhow::Context as _;
 use re_chunk::RowId;
 use re_log_types::{SetStoreInfo, StoreId, StoreInfo};
 use re_mcap::{LayerRegistry, SelectedLayers};
@@ -25,20 +24,33 @@ const MCAP_LOADER_NAME: &str = "McapLoader";
 /// - [`re_mcap::layers::McapRawLayer`]
 pub struct McapLoader {
     selected_layers: SelectedLayers,
+    raw_fallback_enabled: bool,
 }
 
 impl Default for McapLoader {
     fn default() -> Self {
         Self {
             selected_layers: SelectedLayers::All,
+            raw_fallback_enabled: true,
         }
     }
 }
 
 impl McapLoader {
     /// Creates a new [`McapLoader`] that only extracts the specified `layers`.
     pub fn new(selected_layers: SelectedLayers) -> Self {
-        Self { selected_layers }
+        Self {
+            selected_layers,
+            raw_fallback_enabled: true,
+        }
+    }
+
+    /// Creates a new [`McapLoader`] with configurable raw fallback.
+    pub fn with_raw_fallback(selected_layers: SelectedLayers, raw_fallback_enabled: bool) -> Self {
+        Self {
+            selected_layers,
+            raw_fallback_enabled,
+        }
     }
 }
 
@@ -67,16 +79,20 @@ impl DataLoader for McapLoader {
         // common rayon thread pool.
         let settings = settings.clone();
         let selected_layers = self.selected_layers.clone();
+        let raw_fallback_enabled = self.raw_fallback_enabled;
         std::thread::Builder::new()
             .name(format!("load_mcap({path:?}"))
-            .spawn(
-                move || match load_mcap_mmap(&path, &settings, &tx, selected_layers) {
-                    Ok(_) => {}
-                    Err(err) => {
-                        re_log::error!("Failed to load MCAP file: {err}");
-                    }
-                },
-            )
+            .spawn(move || {
+                if let Err(err) = load_mcap_mmap(
+                    &path,
+                    &settings,
+                    &tx,
+                    &selected_layers,
+                    raw_fallback_enabled,
+                ) {
+                    re_log::error!("Failed to load MCAP file: {err}");
+                }
+            })
             .map_err(|err| DataLoaderError::Other(err.into()))?;
 
         Ok(())
@@ -98,6 +114,7 @@ impl DataLoader for McapLoader {
 
         let settings = settings.clone();
         let selected_layers = self.selected_layers.clone();
+        let raw_fallback_enabled = self.raw_fallback_enabled;
 
         // NOTE(1): `spawn` is fine, this whole function is native-only.
         // NOTE(2): this must spawned on a dedicated thread to avoid a deadlock!
@@ -106,14 +123,17 @@ impl DataLoader for McapLoader {
         // common rayon thread pool.
         std::thread::Builder::new()
             .name(format!("load_mcap({filepath:?}"))
-            .spawn(
-                move || match load_mcap_mmap(&filepath, &settings, &tx, selected_layers) {
-                    Ok(_) => {}
-                    Err(err) => {
-                        re_log::error!("Failed to load MCAP file: {err}");
-                    }
-                },
-            )
+            .spawn(move || {
+                if let Err(err) = load_mcap_mmap(
+                    &filepath,
+                    &settings,
+                    &tx,
+                    &selected_layers,
+                    raw_fallback_enabled,
+                ) {
+                    re_log::error!("Failed to load MCAP file: {err}");
+                }
+            })
             .map_err(|err| DataLoaderError::Other(err.into()))?;
 
         Ok(())
@@ -133,7 +153,13 @@ impl DataLoader for McapLoader {
 
         let contents = contents.into_owned();
 
-        load_mcap(&contents, settings, &tx, self.selected_layers.clone())
+        load_mcap(
+            &contents,
+            settings,
+            &tx,
+            &self.selected_layers,
+            self.raw_fallback_enabled,
+        )
     }
 }
 
@@ -142,7 +168,8 @@ fn load_mcap_mmap(
     filepath: &std::path::PathBuf,
     settings: &DataLoaderSettings,
     tx: &Sender<LoadedData>,
-    selected_layers: SelectedLayers,
+    selected_layers: &SelectedLayers,
+    raw_fallback_enabled: bool,
 ) -> std::result::Result<(), DataLoaderError> {
     use std::fs::File;
     let file = File::open(filepath)?;
@@ -151,14 +178,15 @@ fn load_mcap_mmap(
     #[allow(unsafe_code)]
     let mmap = unsafe { memmap2::Mmap::map(&file)? };
 
-    load_mcap(&mmap, settings, tx, selected_layers)
+    load_mcap(&mmap, settings, tx, selected_layers, raw_fallback_enabled)
 }
 
 fn load_mcap(
     mcap: &[u8],
     settings: &DataLoaderSettings,
     tx: &Sender<LoadedData>,
-    selected_layers: SelectedLayers,
+    selected_layers: &SelectedLayers,
+    raw_fallback_enabled: bool,
 ) -> Result<(), DataLoaderError> {
     re_tracing::profile_function!();
 
@@ -200,21 +228,11 @@ fn load_mcap(
     let summary = re_mcap::read_summary(reader)?
         .ok_or_else(|| anyhow::anyhow!("MCAP file does not contain a summary"))?;
 
-    let registry = LayerRegistry::all();
-
     // TODO(#10862): Add warning for channel that miss semantic information.
-
-    let mut empty = true;
-    for mut layer in registry.layers(selected_layers) {
-        re_tracing::profile_scope!("process-layer");
-        empty = false;
-        layer
-            .process(mcap, &summary, &mut send_chunk)
-            .with_context(|| "processing layers")?;
-    }
-    if empty {
-        re_log::warn_once!("No layers were selected");
-    }
+    LayerRegistry::all_builtin(raw_fallback_enabled)
+        .select(selected_layers)
+        .plan(&summary)?
+        .run(mcap, &summary, &mut send_chunk)?;
 
     Ok(())
 }

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

@@ -25,6 +25,12 @@ pub struct ConvertCommand {
     #[clap(short = 'l', long = "layer")]
     selected_layers: Vec<String>,
 
+    /// Disable using the raw layer as a fallback for unsupported channels.
+    /// By default, channels that cannot be handled by semantic layers (protobuf, ROS2)
+    /// will be processed by the raw layer.
+    #[clap(long = "disable-raw-fallback")]
+    disable_raw_fallback: bool,
+
     /// If set, specifies the recording id of the output.
     ///
     /// When this flag is set and multiple input .rdd files are specified,
@@ -42,6 +48,7 @@ impl ConvertCommand {
             application_id,
             recording_id,
             selected_layers,
+            disable_raw_fallback,
         } = self;
 
         let start_time = std::time::Instant::now();
@@ -68,7 +75,8 @@ impl ConvertCommand {
             )
         };
 
-        let loader: &dyn DataLoader = &McapLoader::new(selected_layers);
+        let loader: &dyn DataLoader =
+            &McapLoader::with_raw_fallback(selected_layers, !*disable_raw_fallback);
 
         // TODO(#10862): This currently loads the entire file into memory.
         let (tx, rx) = std::sync::mpsc::channel::<LoadedData>();