first try at remote mcap reading

Author: Eric (SPG) Fang

Cargo.lock

@@ -269,6 +269,7 @@ jsonwebtoken = { version = "9.3", default-features = false }
 lance = { version = "0.38.2", default-features = false } # When you update this, also update the list of features enabled for `datafusion` (~50 lines up)
 lance-index = { version = "0.38.2", default-features = false }
 lance-linalg = { version = "0.38.2", default-features = false }
+lance-io = { version = "0.38.2", default-features = false }
 libc = "0.2.176"
 linked-hash-map = { version = "0.5.6", default-features = false }
 log = "0.4.28"
@@ -289,6 +290,7 @@ notify = { version = "8.2", features = ["macos_kqueue"] }
 num-derive = "0.4.2"
 num-traits = "0.2.19"
 numpy = "0.25.0"
+object_store = "0.12.4"
 objc2-app-kit = "0.3.2"
 opentelemetry = { version = "0.31.0", features = ["metrics"] }
 opentelemetry-appender-tracing = "0.31.0"

Cargo.toml

@@ -46,12 +46,16 @@ indexmap.workspace = true
 itertools.workspace = true
 notify = { workspace = true, features = ["crossbeam-channel"] }
 mcap.workspace = true
+lance-io = { workspace = true, features = ["gcp", "aws"] }
+object_store = { workspace = true }
 parking_lot.workspace = true
 rayon.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
+tokio = { workspace = true, features = ["rt-multi-thread"] }
 urdf-rs.workspace = true
+url.workspace = true
 walkdir.workspace = true
 
 [target.'cfg(not(any(target_arch = "wasm32")))'.dependencies]

crates/store/re_data_loader/Cargo.toml

@@ -27,13 +27,13 @@ pub fn load_from_path(
 
     re_tracing::profile_function!(path.to_string_lossy());
 
-    if !path.exists() {
-        return Err(std::io::Error::new(
-            std::io::ErrorKind::NotFound,
-            format!("path does not exist: {path:?}"),
-        )
-        .into());
-    }
+    // if !path.exists() {
+    //     return Err(std::io::Error::new(
+    //         std::io::ErrorKind::NotFound,
+    //         format!("path does not exist: {path:?}"),
+    //     )
+    //     .into());
+    // }
 
     re_log::info!("Loading {path:?}…");
 

crates/store/re_data_loader/src/load_file.rs

@@ -1,14 +1,17 @@
 //! Rerun dataloader for MCAP files.
 
-use std::io::Cursor;
-use std::path::Path;
-use std::sync::mpsc::Sender;
-
+use crate::{DataLoader, DataLoaderError, DataLoaderSettings, LoadedData};
+use lance_io::object_store::{ObjectStoreParams, ObjectStoreRegistry};
+use mcap::sans_io::IndexedReaderOptions;
 use re_chunk::RowId;
 use re_log_types::{SetStoreInfo, StoreId, StoreInfo};
-use re_mcap::{LayerRegistry, SelectedLayers};
+use re_mcap::{AsyncSeekRead, LayerRegistry, SelectedLayers};
+use std::io::Cursor;
+use std::path::Path;
+use std::str::FromStr as _;
 
-use crate::{DataLoader, DataLoaderError, DataLoaderSettings, LoadedData};
+use std::sync::mpsc::Sender;
+use url::Url;
 
 const MCAP_LOADER_NAME: &str = "McapLoader";
 
@@ -56,6 +59,19 @@ impl McapLoader {
     }
 }
 
+fn mcap_option_from_url(url: &Url) -> anyhow::Result<IndexedReaderOptions> {
+    let mut mcap_options = IndexedReaderOptions::default();
+    for (name, val) in url.query_pairs() {
+        if name == "start" {
+            mcap_options = mcap_options.log_time_on_or_after(u64::from_str(&val)?);
+        } else if name == "end" {
+            mcap_options = mcap_options.log_time_before(u64::from_str(&val)?);
+        }
+    }
+
+    Ok(mcap_options)
+}
+
 impl DataLoader for McapLoader {
     fn name(&self) -> crate::DataLoaderName {
         MCAP_LOADER_NAME.into()
@@ -68,11 +84,31 @@ impl DataLoader for McapLoader {
         path: std::path::PathBuf,
         tx: Sender<crate::LoadedData>,
     ) -> std::result::Result<(), DataLoaderError> {
-        if !is_mcap_file(&path) {
-            return Err(DataLoaderError::Incompatible(path)); // simply not interested
+        re_tracing::profile_function!();
+        let url = path.to_string_lossy();
+        let mcap_url = match Url::parse(&url) {
+            Ok(mcap_url) => mcap_url,
+            Err(err) => match err {
+                url::ParseError::RelativeUrlWithoutBase => {
+                    match Url::parse(&format!("file://{url}")) {
+                        Ok(mcap_url) => mcap_url,
+                        Err(err) => return Err(DataLoaderError::Other(err.into())),
+                    }
+                }
+                _ => return Err(DataLoaderError::Other(err.into())),
+            },
+        };
+
+        let path = std::path::PathBuf::from_str(mcap_url.path()).unwrap();
+
+        if mcap_url.scheme() == "file" && !is_mcap_file(&path) {
+            return Err(DataLoaderError::Incompatible(path.clone())); // simply not interested
         }
 
-        re_tracing::profile_function!();
+        let mcap_options = match mcap_option_from_url(&mcap_url) {
+            Ok(mcap_options) => mcap_options,
+            Err(err) => return Err(DataLoaderError::Other(err)),
+        };
 
         // NOTE(1): `spawn` is fine, this whole function is native-only.
         // NOTE(2): this must spawned on a dedicated thread to avoid a deadlock!
@@ -85,8 +121,23 @@ impl DataLoader for McapLoader {
         std::thread::Builder::new()
             .name(format!("load_mcap({path:?}"))
             .spawn(move || {
-                if let Err(err) = load_mcap_mmap(
-                    &path,
+                // Load from local disk.
+                if mcap_url.scheme() == "file" {
+                    if let Err(err) = load_mcap_mmap(
+                        &path,
+                        &mcap_options,
+                        &settings,
+                        &tx,
+                        &selected_layers,
+                        raw_fallback_enabled,
+                    ) {
+                        re_log::error!("Failed to load MCAP file: {err}");
+                    }
+                }
+                // Load from cloud.
+                else if let Err(err) = load_mcap_cloud(
+                    &mcap_url,
+                    &mcap_options,
                     &settings,
                     &tx,
                     &selected_layers,
@@ -128,6 +179,7 @@ impl DataLoader for McapLoader {
             .spawn(move || {
                 if let Err(err) = load_mcap_mmap(
                     &filepath,
+                    &IndexedReaderOptions::default(),
                     &settings,
                     &tx,
                     &selected_layers,
@@ -151,12 +203,11 @@ impl DataLoader for McapLoader {
     ) -> std::result::Result<(), DataLoaderError> {
         if !is_mcap_file(&filepath) {
             return Err(DataLoaderError::Incompatible(filepath)); // simply not interested
-        }
-
+        };
         let contents = contents.into_owned();
-
         load_mcap(
-            &contents,
+            Box::new(Cursor::new(contents)),
+            &IndexedReaderOptions::default(),
             settings,
             &tx,
             &self.selected_layers,
@@ -165,9 +216,48 @@ impl DataLoader for McapLoader {
     }
 }
 
+#[cfg(not(target_arch = "wasm32"))]
+fn load_mcap_cloud(
+    url: &Url,
+    mcap_options: &IndexedReaderOptions,
+    settings: &DataLoaderSettings,
+    tx: &Sender<LoadedData>,
+    selected_layers: &SelectedLayers,
+    raw_fallback_enabled: bool,
+) -> anyhow::Result<()> {
+    let mcap_reader = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?
+        .block_on(async {
+            let store = ObjectStoreRegistry::default()
+                .get_store(url.clone(), &ObjectStoreParams::default())
+                .await?;
+            let meta = store
+                .inner
+                .head(&object_store::path::Path::parse(url.path())?)
+                .await?;
+            let mcap_reader = Box::from(object_store::buffered::BufReader::new(
+                store.inner.clone(),
+                &meta,
+            ));
+
+            Ok::<Box<object_store::buffered::BufReader>, anyhow::Error>(mcap_reader)
+        })?;
+
+    Ok(load_mcap(
+        mcap_reader,
+        mcap_options,
+        settings,
+        tx,
+        selected_layers,
+        raw_fallback_enabled,
+    )?)
+}
+
 #[cfg(not(target_arch = "wasm32"))]
 fn load_mcap_mmap(
     filepath: &std::path::PathBuf,
+    mcap_options: &IndexedReaderOptions,
     settings: &DataLoaderSettings,
     tx: &Sender<LoadedData>,
     selected_layers: &SelectedLayers,
@@ -180,17 +270,26 @@ fn load_mcap_mmap(
     #[expect(unsafe_code)]
     let mmap = unsafe { memmap2::Mmap::map(&file)? };
 
-    load_mcap(&mmap, settings, tx, selected_layers, raw_fallback_enabled)
+    load_mcap(
+        Box::new(Cursor::new(mmap)),
+        mcap_options,
+        settings,
+        tx,
+        selected_layers,
+        raw_fallback_enabled,
+    )
 }
 
 pub fn load_mcap(
-    mcap: &[u8],
+    mut mcap: Box<dyn AsyncSeekRead>,
+    mcap_options: &IndexedReaderOptions,
     settings: &DataLoaderSettings,
     tx: &Sender<LoadedData>,
     selected_layers: &SelectedLayers,
     raw_fallback_enabled: bool,
 ) -> Result<(), DataLoaderError> {
     re_tracing::profile_function!();
+
     let store_id = settings.recommended_store_id();
 
     if tx
@@ -223,16 +322,14 @@ pub fn load_mcap(
         }
     };
 
-    let reader = Cursor::new(&mcap);
-
-    let summary = re_mcap::read_summary(reader)?
+    let summary = re_mcap::read_summary(&mut mcap)?
         .ok_or_else(|| anyhow::anyhow!("MCAP file does not contain a summary"))?;
 
     // TODO(#10862): Add warning for channel that miss semantic information.
     LayerRegistry::all_builtin(raw_fallback_enabled)
         .select(selected_layers)
         .plan(&summary)?
-        .run(mcap, &summary, &mut send_chunk)?;
+        .run(&mut mcap, &summary, mcap_options, &mut send_chunk)?;
 
     Ok(())
 }

crates/store/re_data_loader/src/loader_mcap.rs

@@ -29,7 +29,7 @@ impl crate::DataLoader for RrdLoader {
 
         let mut extension = crate::extension(&filepath);
         if !matches!(extension.as_str(), "rbl" | "rrd") {
-            if filepath.is_file() || filepath.is_dir() {
+            if filepath.is_file() || filepath.is_dir() || !extension.is_empty() {
                 // NOTE: blueprints and recordings have the same file format
                 return Err(crate::DataLoaderError::Incompatible(filepath.clone()));
             } else {

crates/store/re_data_loader/src/loader_rrd.rs

@@ -2,21 +2,33 @@
 mod tests {
     use std::sync::Arc;
 
+    use mcap::sans_io::IndexedReaderOptions;
     use re_chunk::Chunk;
     use re_chunk_store::{ChunkStore, ChunkStoreConfig, ChunkStoreHandle};
     use re_data_loader::loader_mcap::load_mcap;
     use re_data_loader::{DataLoaderSettings, LoadedData};
     use re_log_types::StoreId;
     use re_mcap::layers::SelectedLayers;
+    use std::io::Cursor;
 
     // Load an MCAP file into a list of chunks.
     fn load_mcap_chunks(path: impl AsRef<std::path::Path>) -> Vec<Chunk> {
         let path = path.as_ref();
         println!("Loading MCAP file: {}", path.display());
-        let mcap_data = std::fs::read(path).unwrap();
+        let mcap_data = Box::new(Cursor::new(std::fs::read(path).unwrap()));
+
         let (tx, rx) = std::sync::mpsc::channel();
         let settings = DataLoaderSettings::recommended("test");
-        load_mcap(&mcap_data, &settings, &tx, &SelectedLayers::All, false).unwrap();
+        let mcap_options = IndexedReaderOptions::default();
+        load_mcap(
+            mcap_data,
+            &mcap_options,
+            &settings,
+            &tx,
+            &SelectedLayers::All,
+            false,
+        )
+        .unwrap();
         drop(tx);
 
         // Collect chunks

crates/store/re_data_loader/tests/test_mcap_loader.rs

@@ -57,8 +57,6 @@ impl LogDataSource {
     pub fn from_uri(_file_source: re_log_types::FileSource, url: &str) -> Option<Self> {
         #[cfg(not(target_arch = "wasm32"))]
         {
-            use itertools::Itertools as _;
-
             fn looks_like_windows_abs_path(path: &str) -> bool {
                 let path = path.as_bytes();
                 // "C:/" etc
@@ -79,20 +77,10 @@ impl LogDataSource {
                     true // Unix absolute path
                 } else if looks_like_windows_abs_path(uri) {
                     true
-                } else if uri.starts_with("http:") || uri.starts_with("https:") {
-                    false
                 } else {
-                    // We use a simple heuristic here: if there are multiple dots, it is likely an url,
-                    // like "example.com/foo.zip".
-                    // If there is only one dot, we treat it as an extension and look it up in a list of common
-                    // file extensions:
-
-                    let parts = uri.split('.').collect_vec();
-                    if parts.len() == 2 {
-                        true
-                    } else {
-                        false // Too many dots; assume an url
-                    }
+                    !(uri.starts_with("http:")
+                        || uri.starts_with("https:")
+                        || uri.starts_with("rerun:"))
                 }
             }
 

crates/store/re_data_source/src/data_source.rs

@@ -33,6 +33,7 @@ saturating_cast.workspace = true
 serde.workspace = true
 serde_bytes.workspace = true
 thiserror.workspace = true
+tokio = { workspace = true, features = ["rt-multi-thread"] }
 
 [dev-dependencies]
 insta = { workspace = true, features = ["filters", "redactions"] }