Redesign parallel RRD ingestion with per-EntityDb workers

- Move ingestion worker from App-level to per-EntityDb architecture
- Each EntityDb now has its own dedicated ingestion worker with separate
  channels for commands and message batches
- Add benchmark infrastructure for ingestion performance testing
- Refactor loader_rrd.rs to support parallel message distribution

Author: joelreymont

.gitignore

@@ -92,3 +92,6 @@ justfile
 
 # Lychee link checker cache
 .lycheecache
+
+# Development notes
+notes

Cargo.lock

@@ -8659,13 +8659,15 @@ dependencies = [
  "ahash",
  "anyhow",
  "arrow",
+ "criterion",
  "crossbeam",
  "image",
  "indexmap 2.11.4",
  "insta",
  "itertools 0.14.0",
  "mcap",
  "memmap2 0.9.8",
+ "mimalloc",
  "notify",
  "parking_lot",
  "parquet",
@@ -8874,6 +8876,8 @@ version = "0.27.0-alpha.8+dev"
 dependencies = [
  "ahash",
  "anyhow",
+ "arrow",
+ "crossbeam",
  "document-features",
  "emath",
  "indexmap 2.11.4",
@@ -8894,6 +8898,7 @@ dependencies = [
  "re_smart_channel",
  "re_sorbet",
  "re_tracing",
+ "re_types",
  "re_types_core",
  "re_uri",
  "saturating_cast",
@@ -10250,6 +10255,7 @@ dependencies = [
  "re_renderer",
  "re_selection_panel",
  "re_smart_channel",
+ "re_sorbet",
  "re_test_context",
  "re_test_viewport",
  "re_time_panel",

crates/store/re_data_loader/Cargo.toml

@@ -63,3 +63,9 @@ re_chunk_store.workspace = true
 re_log_encoding = { workspace = true, features = ["decoder", "encoder"] }
 
 insta = { workspace = true, features = ["glob"] }
+criterion.workspace = true
+mimalloc.workspace = true
+
+[[bench]]
+name = "parallel_ingestion_bench"
+harness = false

crates/store/re_data_loader/benches/parallel_ingestion_bench.rs

@@ -0,0 +1,102 @@
+// Allow unwrap() in benchmarks
+#![expect(clippy::unwrap_used)]
+
+#[global_allocator]
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
+use criterion::{Criterion, criterion_group, criterion_main};
+use re_chunk::{Chunk, RowId};
+use re_log_encoding::Encoder;
+use re_log_types::{
+    LogMsg, NonMinI64, StoreId, StoreKind, TimeInt, TimePoint, Timeline, entity_path,
+};
+use re_types::archetypes::Points2D;
+use std::sync::mpsc;
+
+use re_data_loader::{DataLoader as _, DataLoaderSettings, RrdLoader};
+
+#[cfg(not(debug_assertions))]
+const NUM_MESSAGES: usize = 10_000;
+
+#[cfg(debug_assertions)]
+const NUM_MESSAGES: usize = 100;
+
+criterion_group!(benches, benchmark_load_from_file_contents);
+criterion_main!(benches);
+
+fn generate_messages(count: usize) -> Vec<LogMsg> {
+    let store_id = StoreId::random(StoreKind::Recording, "bench_app");
+    let mut messages = Vec::with_capacity(count);
+
+    for i in 0..count {
+        let chunk = Chunk::builder(entity_path!("points", i.to_string()))
+            .with_archetype(
+                RowId::new(),
+                TimePoint::default().with(
+                    Timeline::new_sequence("log_time"),
+                    TimeInt::from_millis(NonMinI64::new(i64::try_from(i).unwrap()).unwrap()),
+                ),
+                &Points2D::new([(i as f32, i as f32), ((i + 1) as f32, (i + 1) as f32)]),
+            )
+            .build()
+            .unwrap();
+
+        messages.push(LogMsg::ArrowMsg(
+            store_id.clone(),
+            chunk.to_arrow_msg().unwrap(),
+        ));
+    }
+
+    messages
+}
+
+fn encode_messages(messages: &[LogMsg]) -> Vec<u8> {
+    let mut bytes = Vec::new();
+    {
+        let mut encoder = Encoder::new_eager(
+            re_build_info::CrateVersion::LOCAL,
+            re_log_encoding::rrd::EncodingOptions::PROTOBUF_UNCOMPRESSED,
+            &mut bytes,
+        )
+        .unwrap();
+
+        for msg in messages {
+            encoder.append(msg).unwrap();
+        }
+        encoder.flush_blocking().unwrap();
+        encoder.finish().unwrap();
+    }
+    bytes
+}
+
+/// Benchmark loading from file (parallel processing)
+fn benchmark_load_from_file_contents(c: &mut Criterion) {
+    let messages = generate_messages(NUM_MESSAGES);
+    let encoded = encode_messages(&messages);
+    let filepath = std::path::PathBuf::from("bench_data.rrd");
+    let settings = DataLoaderSettings::recommended(re_log_types::RecordingId::random());
+
+    let mut group = c.benchmark_group("load_from_file_contents");
+    group.throughput(criterion::Throughput::Elements(NUM_MESSAGES as u64));
+
+    group.bench_function("rrd_loader", |b| {
+        b.iter(|| {
+            let (tx, rx) = mpsc::channel();
+            let contents = std::borrow::Cow::Borrowed(encoded.as_slice());
+            let loader = RrdLoader;
+
+            loader
+                .load_from_file_contents(&settings, filepath.clone(), contents, tx)
+                .unwrap();
+
+            let mut count = 0;
+            while rx.try_recv().is_ok() {
+                count += 1;
+            }
+
+            assert_eq!(count, NUM_MESSAGES);
+        });
+    });
+
+    group.finish();
+}