Redesign ingestion using a dedicated ingestion worker

Addresses #4298 by moving CPU-intensive Arrow message processing off the
UI thread.

Key changes:

1. Ingestion Worker (crates/viewer/re_viewer/src/ingestion_worker.rs):
   - Dedicated background thread for processing Arrow messages into chunks
   - Bounded channel (capacity: 2000) provides natural backpressure

2. Viewer Integration (crates/viewer/re_viewer/src/app.rs):
   - Arrow messages submitted to worker queue instead of blocking UI thread
   - Poll worker output during update() and add chunks to EntityDb

Author: joelreymont

Cargo.lock

@@ -8622,13 +8622,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",
@@ -10222,6 +10224,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/benches/parallel_ingestion_bench.rs

@@ -8,7 +8,7 @@ use criterion::{Criterion, criterion_group, criterion_main};
 use re_chunk::{Chunk, RowId};
 use re_log_encoding::Encoder;
 use re_log_types::{
-    entity_path, LogMsg, NonMinI64, StoreId, StoreKind, TimeInt, TimePoint, Timeline,
+    LogMsg, NonMinI64, StoreId, StoreKind, TimeInt, TimePoint, Timeline, entity_path,
 };
 use re_types::archetypes::Points2D;
 use std::sync::mpsc;
@@ -21,11 +21,7 @@ const NUM_MESSAGES: usize = 10_000;
 #[cfg(debug_assertions)]
 const NUM_MESSAGES: usize = 100;
 
-criterion_group!(
-    benches,
-    benchmark_load_from_file_contents,
-    benchmark_message_processing_batches,
-);
+criterion_group!(benches, benchmark_load_from_file_contents);
 criterion_main!(benches);
 
 fn generate_messages(count: usize) -> Vec<LogMsg> {
@@ -70,7 +66,7 @@ fn encode_messages(messages: &[LogMsg]) -> Vec<u8> {
     bytes
 }
 
-/// Benchmark loading from file contents (parallel processing)
+/// 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);
@@ -101,71 +97,3 @@ fn benchmark_load_from_file_contents(c: &mut Criterion) {
 
     group.finish();
 }
-
-/// Try to find optimal batch size
-fn benchmark_message_processing_batches(c: &mut Criterion) {
-    let messages = generate_messages(NUM_MESSAGES);
-    let (tx, rx) = mpsc::channel();
-
-    let mut group = c.benchmark_group("message_processing_batches");
-    group.throughput(criterion::Throughput::Elements(NUM_MESSAGES as u64));
-
-    // Sequential processing (baseline)
-    group.bench_function("sequential", |b| {
-        b.iter(|| {
-            let mut batch = messages.clone();
-            let loader = RrdLoader;
-            for msg in batch.drain(..) {
-                let data = re_data_loader::LoadedData::LogMsg(
-                    loader.name(),
-                    msg,
-                );
-                let _ = tx.send(data);
-            }
-            while rx.try_recv().is_ok() {}
-        });
-    });
-
-    // Parallel processing with different batch sizes
-    #[cfg(not(target_arch = "wasm32"))]
-    {
-        use rayon::prelude::*;
-
-        for batch_size in [10, 50, 100, 500, 1000] {
-            group.bench_function(format!("parallel_batch_{}", batch_size), |b| {
-                b.iter(|| {
-                    let mut messages = messages.clone();
-                    let mut processed_count = 0;
-
-                    while processed_count < messages.len() {
-                        let remaining = messages.len() - processed_count;
-                        let current_batch_size = batch_size.min(remaining);
-                        let batch: Vec<_> = messages
-                            .drain(processed_count..processed_count + current_batch_size)
-                            .collect();
-
-                        let processed: Vec<_> = batch
-                            .into_par_iter()
-                            .map(|msg| msg) // Simulate transform_message
-                            .collect();
-
-                        let loader = RrdLoader;
-                        for msg in processed {
-                            let data = re_data_loader::LoadedData::LogMsg(
-                                loader.name(),
-                                msg,
-                            );
-                            let _ = tx.send(data);
-                        }
-
-                        processed_count += current_batch_size;
-                    }
-
-                    while rx.try_recv().is_ok() {}
-                });
-            });
-        }
-    }
-
-    group.finish();
-}