diff --git a/data_store/benches/data_benchmark.rs b/data_store/benches/data_benchmark.rs
index 8ec451199cc1..fdbe4d523304 100644
--- a/data_store/benches/data_benchmark.rs
+++ b/data_store/benches/data_benchmark.rs
@@ -7,7 +7,8 @@ use criterion::{criterion_group, criterion_main, Criterion};
 
 use data_store::TypePathDataStore;
 use data_store::*;
-use log_types::{FieldName, IndexKey, LogId};
+use itertools::Itertools;
+use log_types::{FieldName, LogId};
 
 const NUM_FRAMES: i64 = 1_000; // this can have a big impact on performance
 const NUM_POINTS_PER_CAMERA: u64 = 1_000;
@@ -55,14 +56,31 @@ fn obj_path(camera: &str, index: u64) -> ObjPath {
     ]))
 }
 
+fn type_path() -> ObjTypePath {
+    ObjTypePath::new(vec![
+        TypePathComp::String("camera".into()),
+        TypePathComp::Index,
+        TypePathComp::String("point".into()),
+        TypePathComp::Index,
+    ])
+}
+
 #[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
 struct Time(i64);
 
-fn generate_date(individual_pos: bool, individual_radius: bool) -> TypePathDataStore<Time> {
+fn generate_data(individual_pos: bool, individual_radius: bool) -> TypePathDataStore<Time> {
     let mut data_store = TypePathDataStore::default();
 
+    let type_path = type_path();
+
+    let indices = (0..NUM_POINTS_PER_CAMERA)
+        .map(Index::Sequence)
+        .collect_vec();
+    let positions = vec![[1.0_f32; 3]; NUM_POINTS_PER_CAMERA as usize];
+    let radii = vec![1.0_f32; NUM_POINTS_PER_CAMERA as usize];
+
     for frame in 0..NUM_FRAMES {
-        let time_value = Time(frame);
+        let time_value = Time(frame as _);
         for camera in ["left", "right"] {
             if individual_pos {
                 for point in 0..NUM_POINTS_PER_CAMERA {
@@ -77,28 +95,15 @@ fn generate_date(individual_pos: bool, individual_radius: bool) -> TypePathDataS
                         .unwrap();
                 }
             } else {
-                let type_path = ObjTypePath::new(vec![
-                    TypePathComp::String("camera".into()),
-                    TypePathComp::Index,
-                    TypePathComp::String("point".into()),
-                    TypePathComp::Index,
-                ]);
                 let mut index_path_prefix = IndexPath::default();
                 index_path_prefix.push(Index::String(camera.into()));
                 index_path_prefix.push(Index::Placeholder);
 
-                let batch = Arc::new(
-                    (0..NUM_POINTS_PER_CAMERA)
-                        .map(|pi| {
-                            let pos: [f32; 3] = [1.0, 2.0, 3.0];
-                            (IndexKey::new(Index::Sequence(pi)), pos)
-                        })
-                        .collect(),
-                );
+                let batch = Arc::new(Batch::new(&indices, &positions));
 
                 data_store
                     .insert_batch(
-                        &ObjPath::new(type_path, index_path_prefix),
+                        &ObjPath::new(type_path.clone(), index_path_prefix),
                         FieldName::from("pos"),
                         time_value,
                         LogId::random(),
@@ -120,25 +125,15 @@ fn generate_date(individual_pos: bool, individual_radius: bool) -> TypePathDataS
                         .unwrap();
                 }
             } else {
-                let type_path = ObjTypePath::new(vec![
-                    TypePathComp::String("camera".into()),
-                    TypePathComp::Index,
-                    TypePathComp::String("point".into()),
-                    TypePathComp::Index,
-                ]);
                 let mut index_path_prefix = IndexPath::default();
                 index_path_prefix.push(Index::String(camera.into()));
                 index_path_prefix.push(Index::Placeholder);
 
-                let batch = Arc::new(
-                    (0..NUM_POINTS_PER_CAMERA)
-                        .map(|pi| (IndexKey::new(Index::Sequence(pi)), 1.0_f32))
-                        .collect(),
-                );
+                let batch = Arc::new(Batch::new(&indices, &radii));
 
                 data_store
                     .insert_batch(
-                        &ObjPath::new(type_path, index_path_prefix),
+                        &ObjPath::new(type_path.clone(), index_path_prefix),
                         FieldName::from("radius"),
                         time_value,
                         LogId::random(),
@@ -152,11 +147,103 @@ fn generate_date(individual_pos: bool, individual_radius: bool) -> TypePathDataS
     data_store
 }
 
-pub fn criterion_benchmark(c: &mut Criterion) {
+fn create_batch_thoughput(c: &mut Criterion) {
+    const NUM: usize = 100_000;
+    let indices = (0..NUM).map(|pi| Index::Sequence(pi as _)).collect_vec();
+    let positions = vec![[1.0_f32; 3]; NUM];
+
+    let mut group = c.benchmark_group("create-batch-throughput");
+    group.throughput(criterion::Throughput::Elements(NUM as _));
+
+    group.bench_function("Batch::new", |b| {
+        b.iter(|| Batch::new(&indices, &positions));
+    });
+
+    group.finish();
+}
+
+fn insert_batch_thoughput(c: &mut Criterion) {
+    const NUM_FRAMES: usize = 100;
+    const NUM_POINTS: usize = 10_000;
+    let indices = (0..NUM_POINTS)
+        .map(|pi| Index::Sequence(pi as _))
+        .collect_vec();
+    let positions = vec![[1.0_f32; 3]; NUM_POINTS];
+    let batch = std::sync::Arc::new(Batch::new(&indices, &positions));
+
+    let mut index_path_prefix = IndexPath::default();
+    index_path_prefix.push(Index::String("left".into()));
+    index_path_prefix.push(Index::Placeholder);
+
+    let mut group = c.benchmark_group("insert-batch-throughput");
+    group.throughput(criterion::Throughput::Elements(
+        (NUM_POINTS * NUM_FRAMES) as _,
+    ));
+
+    group.bench_function("insert_batch", |b| {
+        b.iter(|| {
+            let mut data_store = TypePathDataStore::default();
+            for frame in 0..NUM_FRAMES {
+                let time_value = Time(frame as _);
+                data_store
+                    .insert_batch(
+                        &ObjPath::new(type_path(), index_path_prefix.clone()),
+                        FieldName::from("pos"),
+                        time_value,
+                        LogId::random(),
+                        batch.clone(),
+                    )
+                    .unwrap();
+            }
+            data_store
+        });
+    });
+
+    group.finish();
+}
+
+fn insert_individual_thoughput(c: &mut Criterion) {
+    const NUM_FRAMES: usize = 100;
+    const NUM_POINTS: usize = 1000;
+
+    let mut index_path_prefix = IndexPath::default();
+    index_path_prefix.push(Index::String("left".into()));
+    index_path_prefix.push(Index::Placeholder);
+
+    let mut group = c.benchmark_group("insert-individual-throughput");
+    group.throughput(criterion::Throughput::Elements(
+        (NUM_POINTS * NUM_FRAMES) as _,
+    ));
+
+    group.bench_function("insert_individual", |b| {
+        b.iter(|| {
+            let mut data_store = TypePathDataStore::default();
+            for frame in 0..NUM_FRAMES {
+                let time_value = Time(frame as _);
+                for point in 0..NUM_POINTS {
+                    data_store
+                        .insert_individual::<[f32; 3]>(
+                            obj_path("left", point as _),
+                            FieldName::from("pos"),
+                            time_value,
+                            LogId::random(),
+                            [1.0, 2.0, 3.0],
+                        )
+                        .unwrap();
+                }
+            }
+            data_store
+        });
+    });
+
+    group.finish();
+}
+
+fn query_throughput(c: &mut Criterion) {
     let mut group = c.benchmark_group("query-points-throughput");
     group.throughput(criterion::Throughput::Elements(TOTAL_POINTS as _));
 
-    let data_store = generate_date(false, false);
+    let data_store = generate_data(false, false);
     group.bench_function("batched_pos_batched_radius", |b| {
         b.iter(|| {
             let points = points_from_store(&data_store, &TimeQuery::LatestAt(Time(NUM_FRAMES / 2)));
@@ -164,7 +251,7 @@ pub fn criterion_benchmark(c: &mut Criterion) {
         });
     });
 
-    let data_store = generate_date(true, true);
+    let data_store = generate_data(true, true);
     group.bench_function("individual_pos_individual_radius", |b| {
         b.iter(|| {
             let points = points_from_store(&data_store, &TimeQuery::LatestAt(Time(NUM_FRAMES / 2)));
@@ -172,7 +259,7 @@ pub fn criterion_benchmark(c: &mut Criterion) {
         });
     });
 
-    let data_store = generate_date(false, true);
+    let data_store = generate_data(false, true);
     group.bench_function("batched_pos_individual_radius", |b| {
         b.iter(|| {
             let points = points_from_store(&data_store, &TimeQuery::LatestAt(Time(NUM_FRAMES / 2)));
@@ -180,7 +267,7 @@ pub fn criterion_benchmark(c: &mut Criterion) {
         });
     });
 
-    let data_store = generate_date(true, false);
+    let data_store = generate_data(true, false);
     group.bench_function("individual_pos_batched_radius", |b| {
         b.iter(|| {
             let points = points_from_store(&data_store, &TimeQuery::LatestAt(Time(NUM_FRAMES / 2)));
@@ -191,5 +278,11 @@ pub fn criterion_benchmark(c: &mut Criterion) {
     group.finish();
 }
 
-criterion_group!(benches, criterion_benchmark);
+criterion_group!(
+    benches,
+    create_batch_thoughput,
+    insert_batch_thoughput,
+    insert_individual_thoughput,
+    query_throughput
+);
 criterion_main!(benches);
diff --git a/data_store/examples/memory_usage.rs b/data_store/examples/memory_usage.rs
index bb706eba37ee..c08e9dcc821f 100644
--- a/data_store/examples/memory_usage.rs
+++ b/data_store/examples/memory_usage.rs
@@ -45,7 +45,8 @@ unsafe impl std::alloc::GlobalAlloc for TrackingAllocator {
 }
 
 use data_store::*;
-use log_types::{IndexKey, LogId, TimeValue};
+use itertools::Itertools;
+use log_types::{LogId, TimeValue};
 
 impl TrackingAllocator {
     fn used_bytes(&self) -> usize {
@@ -142,19 +143,18 @@ fn big_clouds_batched() {
     const NUM_FRAMES: usize = 100;
     const NUM_POINTS_PER_CAMERA: usize = 1_000;
 
+    let indices = (0..NUM_POINTS_PER_CAMERA)
+        .map(|i| Index::Sequence(i as _))
+        .collect_vec();
+    let point: [f32; 3] = [1.0, 2.0, 3.0];
+    let positions = vec![point; NUM_POINTS_PER_CAMERA];
+
     let mut store = TypePathDataStore::default();
     let mut frame = 0;
     let mut num_points = 0;
     while frame < NUM_FRAMES {
         for camera in 0..NUM_CAMERAS {
-            let batch = std::sync::Arc::new(
-                (0..NUM_POINTS_PER_CAMERA)
-                    .map(|i| {
-                        let point: [f32; 3] = [1.0, 2.0, 3.0];
-                        (IndexKey::new(Index::Sequence(i as _)), point)
-                    })
-                    .collect(),
-            );
+            let batch = std::sync::Arc::new(Batch::new(&indices, &positions));
             let (obj_type_path, index_path) =
                 obj_path(camera as _, 0).into_type_path_and_index_path();
             let (index_path_prefix, _) = index_path.replace_last_with_placeholder();
diff --git a/data_store/src/log_store.rs b/data_store/src/log_store.rs
index 62843ec60214..bbbdbd37cadc 100644
--- a/data_store/src/log_store.rs
+++ b/data_store/src/log_store.rs
@@ -1,15 +1,18 @@
-use nohash_hasher::IntMap;
+use nohash_hasher::{IntMap, IntSet};
 
 use log_types::{
-    Data, DataMsg, DataPath, DataVec, IndexKey, ObjPath, ObjPathHash, TimeSource, TimeType,
+    Data, DataMsg, DataPath, DataVec, Index, IndexHash, LoggedData, ObjPath, ObjPathHash,
+    TimeSource, TimeType,
 };
 
-use crate::{Batch, TypePathDataStore};
+use crate::{ArcBatch, Batch, TypePathDataStore};
 
 #[derive(Default)]
 pub struct LogDataStore {
     store_from_time_source: IntMap<TimeSource, (TimeType, TypePathDataStore<i64>)>,
     obj_path_from_hash: IntMap<ObjPathHash, ObjPath>,
+    /// To avoid doing double-work filling in [`Self::obj_path_from_hash`].
+    regiestered_batch_paths: IntMap<ObjPathHash, IntSet<IndexHash>>,
 }
 
 impl LogDataStore {
@@ -43,12 +46,11 @@ impl LogDataStore {
     }
 
     pub fn insert(&mut self, data_msg: &DataMsg) -> crate::Result<()> {
-        self.register_obj_path(data_msg);
+        crate::profile_function!();
 
         let mut batcher = Batcher::default();
 
         for (time_source, time_value) in &data_msg.time_point.0 {
-            let store = self.entry(time_source, time_value.typ());
             let time = time_value.as_i64();
             let id = data_msg.id;
 
@@ -57,71 +59,63 @@ impl LogDataStore {
                 field_name: fname,
             } = data_msg.data_path.clone();
 
-            match data_msg.data.clone() {
-                Data::Batch { indices, data } => {
+            match &data_msg.data {
+                LoggedData::Single(data) => {
+                    self.obj_path_from_hash
+                        .entry(*op.hash())
+                        .or_insert_with(|| op.clone());
+
+                    let store = self.entry(time_source, time_value.typ());
+                    log_types::data_map!(data.clone(), |data| store
+                        .insert_individual(op, fname, time, id, data))
+                }
+                LoggedData::Batch { indices, data } => {
                     log_types::data_vec_map!(data, |vec| {
                         let batch = batcher.batch(indices, vec);
+                        self.register_batch_obj_paths(data_msg, batch.indices());
+                        let store = self.entry(time_source, time_value.typ());
                         store.insert_batch(&op, fname, time, id, batch)
                     })
                 }
-
-                Data::I32(data) => store.insert_individual(op, fname, time, id, data),
-                Data::F32(data) => store.insert_individual(op, fname, time, id, data),
-                Data::Color(data) => store.insert_individual(op, fname, time, id, data),
-                Data::String(data) => store.insert_individual(op, fname, time, id, data),
-
-                Data::Vec2(data) => store.insert_individual(op, fname, time, id, data),
-                Data::BBox2D(data) => store.insert_individual(op, fname, time, id, data),
-                Data::LineSegments2D(data) => store.insert_individual(op, fname, time, id, data),
-                Data::Image(data) => store.insert_individual(op, fname, time, id, data),
-
-                Data::Vec3(data) => store.insert_individual(op, fname, time, id, data),
-                Data::Box3(data) => store.insert_individual(op, fname, time, id, data),
-                Data::Path3D(data) => store.insert_individual(op, fname, time, id, data),
-                Data::LineSegments3D(data) => store.insert_individual(op, fname, time, id, data),
-                Data::Mesh3D(data) => store.insert_individual(op, fname, time, id, data),
-                Data::Camera(data) => store.insert_individual(op, fname, time, id, data),
-
-                Data::Vecf32(data) => store.insert_individual(op, fname, time, id, data),
-
-                Data::Space(data) => store.insert_individual(op, fname, time, id, data),
+                LoggedData::BatchSplat(data) => {
+                    let store = self.entry(time_source, time_value.typ());
+                    log_types::data_map!(data.clone(), |data| store
+                        .insert_batch_splat(op, fname, time, id, data))
+                }
             }?;
         }
 
         Ok(())
     }
 
-    fn register_obj_path(&mut self, data_msg: &DataMsg) {
+    #[inline(never)]
+    fn register_batch_obj_paths(
+        &mut self,
+        data_msg: &DataMsg,
+        indices: std::slice::Iter<'_, (IndexHash, Index)>,
+    ) {
+        crate::profile_function!();
         let obj_path = data_msg.data_path.obj_path();
 
-        if let Data::Batch { indices, .. } = &data_msg.data {
-            for index_path_suffix in indices {
-                crate::profile_scope!("Register batch obj paths");
+        let registered_suffixes = self
+            .regiestered_batch_paths
+            .entry(*data_msg.data_path.obj_path.hash())
+            .or_default();
+
+        for (index_hash, index) in indices {
+            if registered_suffixes.insert(*index_hash) {
+                // TODO: speed this up. A lot. Please.
                 let (obj_type_path, index_path_prefix) =
                     obj_path.clone().into_type_path_and_index_path();
-                // TODO: speed this up. A lot. Please.
                 let mut index_path = index_path_prefix.clone();
-                index_path.replace_last_placeholder_with(index_path_suffix.clone().into());
+                index_path.replace_last_placeholder_with(index.clone());
                 let obj_path = ObjPath::new(obj_type_path.clone(), index_path);
                 self.obj_path_from_hash.insert(*obj_path.hash(), obj_path);
             }
-        } else {
-            self.obj_path_from_hash
-                .entry(*obj_path.hash())
-                .or_insert_with(|| obj_path.clone());
         }
     }
 }
 
-fn batch<T>(indices: Vec<log_types::Index>, data: Vec<T>) -> Batch<T> {
-    assert_eq!(indices.len(), data.len()); // TODO: return Result instead
-    std::sync::Arc::new(
-        itertools::izip!(indices, data)
-            .map(|(index, value)| (IndexKey::new(index), value))
-            .collect(),
-    )
-}
-
 /// Converts data to a batch ONCE, then reuses that batch for other time sources
 #[derive(Default)]
 struct Batcher {
@@ -129,11 +123,15 @@ struct Batcher {
 }
 
 impl Batcher {
-    pub fn batch<T: 'static>(&mut self, indices: Vec<log_types::Index>, data: Vec<T>) -> Batch<T> {
+    pub fn batch<T: 'static + Clone>(
+        &mut self,
+        indices: &[log_types::Index],
+        data: &[T],
+    ) -> ArcBatch<T> {
         if let Some(batch) = &self.batch {
-            batch.downcast_ref::<Batch<T>>().unwrap().clone()
+            batch.downcast_ref::<ArcBatch<T>>().unwrap().clone()
         } else {
-            let batch = batch(indices, data);
+            let batch = std::sync::Arc::new(Batch::new(indices, data));
             self.batch = Some(Box::new(batch.clone()));
             batch
         }
diff --git a/data_store/src/obj_path_query.rs b/data_store/src/obj_path_query.rs
index e9a29de51a00..f4c6ac16f304 100644
--- a/data_store/src/obj_path_query.rs
+++ b/data_store/src/obj_path_query.rs
@@ -1,6 +1,6 @@
 //! Queries of the type "read these fields, from the object at this [`ObjPath`], over this time interval"
 
-use log_types::{DataTrait, FieldName, IndexPath, LogId, ObjPath};
+use log_types::{DataTrait, FieldName, IndexPath, IndexPathHash, LogId, ObjPath};
 
 use crate::{storage::*, TimeQuery};
 
@@ -19,7 +19,8 @@ pub fn visit_obj_data<'s, Time: 'static + Copy + Ord, T: DataTrait>(
     if let Some(primary_data) = obj_store.get::<T>(field_name) {
         match primary_data {
             DataStore::Individual(primary) => {
-                if let Some(primary) = primary.values.get(index_path) {
+                let index_path = IndexPathHash::from_path(index_path);
+                if let Some(primary) = primary.values.get(&index_path) {
                     query(
                         &primary.history,
                         time_query,
@@ -32,6 +33,9 @@ pub fn visit_obj_data<'s, Time: 'static + Copy + Ord, T: DataTrait>(
             DataStore::Batched(primary) => {
                 let (index_path_prefix, index_path_suffix) =
                     index_path.clone().replace_last_with_placeholder();
+                let index_path_prefix = IndexPathHash::from_path(&index_path_prefix);
+                let index_path_suffix = index_path_suffix.hash();
+
                 if let Some(primary) = primary.batches_over_time.get(&index_path_prefix) {
                     query(
                         &primary.history,
@@ -48,6 +52,9 @@ pub fn visit_obj_data<'s, Time: 'static + Copy + Ord, T: DataTrait>(
                     );
                 }
             }
+            DataStore::BatchSplat(_) => {
+                tracing::error!("Used BatchSplat for a primary field {field_name:?}");
+            }
         }
     }
 
@@ -70,7 +77,10 @@ pub fn visit_obj_data_1<'s, Time: 'static + Copy + Ord, T: DataTrait, S1: DataTr
 
         match primary_data {
             DataStore::Individual(primary) => {
-                if let Some(primary) = primary.values.get(index_path) {
+                let index_path = IndexPathHash::from_path(index_path);
+
+                if let Some(primary) = primary.values.get(&index_path) {
+                    let index_path_split = &primary.index_path_split;
                     let child1_reader = IndividualDataReader::<Time, S1>::new(obj_store, &child1);
 
                     query(
@@ -81,7 +91,7 @@ pub fn visit_obj_data_1<'s, Time: 'static + Copy + Ord, T: DataTrait, S1: DataTr
                                 &primary.obj_path,
                                 log_id,
                                 primary_value,
-                                child1_reader.latest_at(index_path, time),
+                                child1_reader.latest_at(&index_path, index_path_split, time),
                             );
                         },
                     );
@@ -90,6 +100,9 @@ pub fn visit_obj_data_1<'s, Time: 'static + Copy + Ord, T: DataTrait, S1: DataTr
             DataStore::Batched(primary) => {
                 let (index_path_prefix, index_path_suffix) =
                     index_path.clone().replace_last_with_placeholder();
+                let index_path_prefix = IndexPathHash::from_path(&index_path_prefix);
+                let index_path_suffix = index_path_suffix.hash();
+
                 if let Some(primary) = primary.batches_over_time.get(&index_path_prefix) {
                     let child1_store = obj_store.get::<S1>(&child1);
 
@@ -112,6 +125,9 @@ pub fn visit_obj_data_1<'s, Time: 'static + Copy + Ord, T: DataTrait, S1: DataTr
                     );
                 }
             }
+            DataStore::BatchSplat(_) => {
+                tracing::error!("Used BatchSplat for a primary field {field_name:?}");
+            }
         }
     }
 
@@ -141,7 +157,10 @@ pub fn visit_obj_data_2<
 
         match primary_data {
             DataStore::Individual(primary) => {
-                if let Some(primary) = primary.values.get(index_path) {
+                let index_path = IndexPathHash::from_path(index_path);
+
+                if let Some(primary) = primary.values.get(&index_path) {
+                    let index_path_split = &primary.index_path_split;
                     let child1_reader = IndividualDataReader::<Time, S1>::new(obj_store, &child1);
                     let child2_reader = IndividualDataReader::<Time, S2>::new(obj_store, &child2);
 
@@ -153,8 +172,8 @@ pub fn visit_obj_data_2<
                                 &primary.obj_path,
                                 log_id,
                                 primary_value,
-                                child1_reader.latest_at(index_path, time),
-                                child2_reader.latest_at(index_path, time),
+                                child1_reader.latest_at(&index_path, index_path_split, time),
+                                child2_reader.latest_at(&index_path, index_path_split, time),
                             );
                         },
                     );
@@ -163,6 +182,9 @@ pub fn visit_obj_data_2<
             DataStore::Batched(primary) => {
                 let (index_path_prefix, index_path_suffix) =
                     index_path.clone().replace_last_with_placeholder();
+                let index_path_prefix = IndexPathHash::from_path(&index_path_prefix);
+                let index_path_suffix = index_path_suffix.hash();
+
                 if let Some(primary) = primary.batches_over_time.get(&index_path_prefix) {
                     let child1_store = obj_store.get::<S1>(&child1);
                     let child2_store = obj_store.get::<S2>(&child2);
@@ -189,6 +211,9 @@ pub fn visit_obj_data_2<
                     );
                 }
             }
+            DataStore::BatchSplat(_) => {
+                tracing::error!("Used BatchSplat for a primary field {field_name:?}");
+            }
         }
     }
 
@@ -220,7 +245,10 @@ pub fn visit_obj_data_3<
 
         match primary_data {
             DataStore::Individual(primary) => {
-                if let Some(primary) = primary.values.get(index_path) {
+                let index_path = IndexPathHash::from_path(index_path);
+
+                if let Some(primary) = primary.values.get(&index_path) {
+                    let index_path_split = &primary.index_path_split;
                     let child1_reader = IndividualDataReader::<Time, S1>::new(obj_store, &child1);
                     let child2_reader = IndividualDataReader::<Time, S2>::new(obj_store, &child2);
                     let child3_reader = IndividualDataReader::<Time, S3>::new(obj_store, &child3);
@@ -233,9 +261,9 @@ pub fn visit_obj_data_3<
                                 &primary.obj_path,
                                 log_id,
                                 primary_value,
-                                child1_reader.latest_at(index_path, time),
-                                child2_reader.latest_at(index_path, time),
-                                child3_reader.latest_at(index_path, time),
+                                child1_reader.latest_at(&index_path, index_path_split, time),
+                                child2_reader.latest_at(&index_path, index_path_split, time),
+                                child3_reader.latest_at(&index_path, index_path_split, time),
                             );
                         },
                     );
@@ -244,6 +272,9 @@ pub fn visit_obj_data_3<
             DataStore::Batched(primary) => {
                 let (index_path_prefix, index_path_suffix) =
                     index_path.clone().replace_last_with_placeholder();
+                let index_path_prefix = IndexPathHash::from_path(&index_path_prefix);
+                let index_path_suffix = index_path_suffix.hash();
+
                 if let Some(primary) = primary.batches_over_time.get(&index_path_prefix) {
                     let child1_store = obj_store.get::<S1>(&child1);
                     let child2_store = obj_store.get::<S2>(&child2);
@@ -274,6 +305,9 @@ pub fn visit_obj_data_3<
                     );
                 }
             }
+            DataStore::BatchSplat(_) => {
+                tracing::error!("Used BatchSplat for a primary field {field_name:?}");
+            }
         }
     }
 
diff --git a/data_store/src/storage.rs b/data_store/src/storage.rs
index dd919745ae4c..64669a494b20 100644
--- a/data_store/src/storage.rs
+++ b/data_store/src/storage.rs
@@ -1,22 +1,20 @@
-use std::collections::BTreeMap;
+use std::collections::{hash_map::Entry, BTreeMap};
 use std::sync::Arc;
 
 use ahash::AHashMap;
 use nohash_hasher::IntMap;
 
 use log_types::{
-    data_types, DataTrait, DataType, DataVec, FieldName, IndexKey, IndexPath, LogId, ObjPath,
+    data_types, DataTrait, DataType, DataVec, FieldName, Index, IndexHash, IndexPath,
+    IndexPathHash, LogId, ObjPath,
 };
 
 use crate::{ObjTypePath, TimeQuery};
 
 #[derive(Clone, Copy, Debug)]
 pub enum Error {
-    /// First stored as a batch, then individually. Not supported.
-    BatchFollowedByIndividual,
-
-    /// First stored individually, then followed by a batch. Not supported.
-    IndividualFollowedByBatch,
+    /// The same type batch must be either individual, batch, or a batch splat.
+    MixingIndividualBatchesOrSplats,
 
     /// One type was first logged, then another.
     WrongType,
@@ -26,16 +24,63 @@ pub type Result<T> = std::result::Result<T, Error>;
 
 // ----------------------------------------------------------------------------
 
-/// Can be shared between different timelines.
+/// Can be shared between different timelines with [`ArcBatch`].
 ///
-/// The [`IndexKey`] is the last path of the [`IndexPath`].
-pub type Batch<T> = Arc<IntMap<IndexKey, T>>;
+/// Each [`Index`] is the last path of the [`IndexPath`].
+pub struct Batch<T> {
+    map: IntMap<IndexHash, T>,
+    hashed_indices: Vec<(IndexHash, Index)>,
+}
+
+impl<T: Clone> Batch<T> {
+    #[inline(never)]
+    pub fn new(indices: &[log_types::Index], data: &[T]) -> Self {
+        crate::profile_function!(std::any::type_name::<T>());
+        assert_eq!(indices.len(), data.len()); // TODO: return Result instead
+        let mut hashed_indices = Vec::with_capacity(indices.len());
+        let map = itertools::izip!(indices, data)
+            .map(|(index, value)| {
+                let index_hash = IndexHash::hash(index);
+                hashed_indices.push((index_hash, index.clone()));
+                (index_hash, value.clone())
+            })
+            .collect();
+        Self {
+            map,
+            hashed_indices,
+        }
+    }
+}
+
+impl<T> Batch<T> {
+    #[inline]
+    pub fn get(&self, index: &IndexHash) -> Option<&T> {
+        self.map.get(index)
+    }
+
+    #[inline]
+    pub fn indices(&self) -> std::slice::Iter<'_, (IndexHash, Index)> {
+        self.hashed_indices.iter()
+    }
+
+    #[inline]
+    pub fn values(&self) -> impl ExactSizeIterator<Item = &T> {
+        self.map.values()
+    }
+
+    #[inline]
+    pub fn iter(&self) -> impl ExactSizeIterator<Item = (&IndexHash, &T)> {
+        self.map.iter()
+    }
+}
+
+pub type ArcBatch<T> = Arc<Batch<T>>;
 
 // ----------------------------------------------------------------------------
 
 /// We have one of these per each time source.
 pub struct TypePathDataStore<Time> {
-    objects: AHashMap<ObjTypePath, ObjStore<Time>>,
+    objects: IntMap<ObjTypePath, ObjStore<Time>>,
 }
 
 impl<Time> Default for TypePathDataStore<Time> {
@@ -47,6 +92,7 @@ impl<Time> Default for TypePathDataStore<Time> {
 }
 
 impl<Time: 'static + Copy + Ord> TypePathDataStore<Time> {
+    #[inline(never)]
     pub fn insert_individual<T: DataTrait>(
         &mut self,
         obj_path: ObjPath,
@@ -63,28 +109,47 @@ impl<Time: 'static + Copy + Ord> TypePathDataStore<Time> {
             .insert_individual(index_path, field_name, time, obj_path, log_id, value)
     }
 
-    /// `index_path_prefix` should have `Index::Placeholder` in the last position.
+    /// `parent_obj_path.index_path` should have `Index::Placeholder` in the last position.
+    #[inline(never)]
     pub fn insert_batch<T: DataTrait>(
         &mut self,
         parent_obj_path: &ObjPath,
         field_name: FieldName,
         time: Time,
         log_id: LogId,
-        batch: Batch<T>,
+        batch: ArcBatch<T>,
     ) -> Result<()> {
-        crate::profile_function!();
+        crate::profile_function!(std::any::type_name::<T>());
+        assert!(parent_obj_path.index_path().has_placeholder_last());
+        let index_path = IndexPathHash::from_path(parent_obj_path.index_path());
 
         self.objects
             .entry(parent_obj_path.obj_type_path().clone())
             .or_default()
-            .insert_batch(
-                parent_obj_path.index_path().clone(),
-                field_name,
-                time,
-                parent_obj_path,
-                log_id,
-                batch,
-            )
+            .insert_batch(index_path, field_name, time, parent_obj_path, log_id, batch)
+    }
+
+    /// Use the same value for all indices of a batch.
+    ///
+    /// `parent_obj_path.index_path` should have `Index::Placeholder` in the last position.
+    #[inline(never)]
+    pub fn insert_batch_splat<T: DataTrait>(
+        &mut self,
+        parent_obj_path: ObjPath,
+        field_name: FieldName,
+        time: Time,
+        log_id: LogId,
+        value: T,
+    ) -> Result<()> {
+        crate::profile_function!();
+        let (obj_type_path, index_path) = parent_obj_path.into_type_path_and_index_path();
+        assert!(index_path.has_placeholder_last());
+        let index_path = IndexPathHash::from_path(&index_path);
+
+        self.objects
+            .entry(obj_type_path)
+            .or_default()
+            .insert_batch_splat(index_path, field_name, time, log_id, value)
     }
 
     #[inline]
@@ -117,7 +182,7 @@ pub struct ObjStore<Time> {
     fields: AHashMap<FieldName, DataStoreTypeErased<Time>>,
 
     /// For each index suffix we know, what is the full object path?
-    obj_paths_from_batch_suffix: nohash_hasher::IntMap<IndexKey, ObjPath>,
+    obj_paths_from_batch_suffix: nohash_hasher::IntMap<IndexHash, ObjPath>,
 }
 
 impl<Time> Default for ObjStore<Time> {
@@ -153,25 +218,14 @@ impl<Time: 'static + Copy + Ord> ObjStore<Time> {
 
     fn insert_batch<T: DataTrait>(
         &mut self,
-        index_path_prefix: IndexPath,
+        index_path_prefix: IndexPathHash,
         field_name: FieldName,
         time: Time,
         parent_obj_path: &ObjPath,
         log_id: LogId,
-        batch: Batch<T>,
+        batch: ArcBatch<T>,
     ) -> Result<()> {
-        for index_path_suffix in batch.keys() {
-            if !self
-                .obj_paths_from_batch_suffix
-                .contains_key(index_path_suffix)
-            {
-                let obj_path = parent_obj_path
-                    .clone()
-                    .replace_last_placeholder_with(index_path_suffix.clone());
-                self.obj_paths_from_batch_suffix
-                    .insert(index_path_suffix.clone(), obj_path);
-            }
-        }
+        self.store_batch_obj_paths(&batch, parent_obj_path);
 
         if let Some(store) = self
             .fields
@@ -185,6 +239,39 @@ impl<Time: 'static + Copy + Ord> ObjStore<Time> {
         }
     }
 
+    fn insert_batch_splat<T: DataTrait>(
+        &mut self,
+        index_path_prefix: IndexPathHash,
+        field_name: FieldName,
+        time: Time,
+        log_id: LogId,
+        value: T,
+    ) -> Result<()> {
+        if let Some(store) = self
+            .fields
+            .entry(field_name)
+            .or_insert_with(|| DataStoreTypeErased::new_batch_splat::<T>())
+            .write::<T>()
+        {
+            store.insert_batch_splat(index_path_prefix, time, log_id, value)
+        } else {
+            Err(Error::WrongType)
+        }
+    }
+
+    #[inline(never)]
+    fn store_batch_obj_paths<T: DataTrait>(&mut self, batch: &Batch<T>, parent_obj_path: &ObjPath) {
+        crate::profile_function!();
+        for (index_hash, index) in batch.indices() {
+            if let Entry::Vacant(entry) = self.obj_paths_from_batch_suffix.entry(*index_hash) {
+                let obj_path = parent_obj_path
+                    .clone()
+                    .replace_last_placeholder_with(index.clone());
+                entry.insert(obj_path);
+            }
+        }
+    }
+
     pub fn get_field(&self, field_name: &FieldName) -> Option<&DataStoreTypeErased<Time>> {
         self.fields.get(field_name)
     }
@@ -199,7 +286,8 @@ impl<Time: 'static + Copy + Ord> ObjStore<Time> {
         self.fields.iter()
     }
 
-    pub(crate) fn obj_path_or_die(&self, index_path_suffix: &IndexKey) -> &ObjPath {
+    #[inline]
+    pub(crate) fn obj_path_or_die(&self, index_path_suffix: &IndexHash) -> &ObjPath {
         self.obj_paths_from_batch_suffix
             .get(index_path_suffix)
             .unwrap()
@@ -232,6 +320,14 @@ impl<Time: 'static + Copy + Ord> DataStoreTypeErased<Time> {
         }
     }
 
+    fn new_batch_splat<T: DataTrait>() -> Self {
+        Self {
+            data_store: Box::new(DataStore::<Time, T>::new_batch_splat()),
+            data_type: T::data_typ(),
+            _phantom: Default::default(),
+        }
+    }
+
     pub fn is<T: DataTrait>(&self) -> bool {
         self.data_store.is::<DataStore<Time, T>>()
     }
@@ -305,6 +401,8 @@ pub enum DataStore<Time, T> {
     Individual(IndividualDataHistory<Time, T>),
 
     Batched(BatchedDataHistory<Time, T>),
+
+    BatchSplat(BatchSplatDataHistory<Time, T>),
 }
 
 impl<Time: Copy + Ord, T: DataTrait> DataStore<Time, T> {
@@ -316,6 +414,10 @@ impl<Time: Copy + Ord, T: DataTrait> DataStore<Time, T> {
         Self::Batched(Default::default())
     }
 
+    fn new_batch_splat() -> Self {
+        Self::BatchSplat(Default::default())
+    }
+
     fn insert_individual(
         &mut self,
         index_path: IndexPath,
@@ -329,23 +431,39 @@ impl<Time: Copy + Ord, T: DataTrait> DataStore<Time, T> {
                 individual.insert(index_path, time, obj_path, log_id, value);
                 Ok(())
             }
-            Self::Batched(_) => Err(Error::BatchFollowedByIndividual),
+            _ => Err(Error::MixingIndividualBatchesOrSplats),
         }
     }
 
     fn insert_batch(
         &mut self,
-        index_path_prefix: IndexPath,
+        index_path_prefix: IndexPathHash,
         time: Time,
         log_id: LogId,
-        batch: Batch<T>,
+        batch: ArcBatch<T>,
     ) -> Result<()> {
         match self {
-            Self::Individual(_) => Err(Error::IndividualFollowedByBatch),
             Self::Batched(batched) => {
                 batched.insert(index_path_prefix, time, log_id, batch);
                 Ok(())
             }
+            _ => Err(Error::MixingIndividualBatchesOrSplats),
+        }
+    }
+
+    fn insert_batch_splat(
+        &mut self,
+        index_path_prefix: IndexPathHash,
+        time: Time,
+        log_id: LogId,
+        value: T,
+    ) -> Result<()> {
+        match self {
+            Self::BatchSplat(batched) => {
+                batched.insert(index_path_prefix, time, log_id, value);
+                Ok(())
+            }
+            _ => Err(Error::MixingIndividualBatchesOrSplats),
         }
     }
 
@@ -356,8 +474,11 @@ impl<Time: Copy + Ord, T: DataTrait> DataStore<Time, T> {
         time_query: &TimeQuery<Time>,
     ) -> (Vec<Time>, Vec<LogId>, Vec<T>) {
         match self {
-            Self::Individual(data_history) => data_history.query_object(&index_path, time_query),
+            Self::Individual(data_history) => {
+                data_history.query_object(&IndexPathHash::from_path(&index_path), time_query)
+            }
             Self::Batched(data_history) => data_history.query_object(index_path, time_query),
+            Self::BatchSplat(data_history) => data_history.query_object(index_path, time_query),
         }
     }
 }
@@ -367,7 +488,7 @@ impl<Time: Copy + Ord, T: DataTrait> DataStore<Time, T> {
 /// For a specific [`ObjTypePath`].
 pub struct IndividualDataHistory<Time, T> {
     /// fast to find latest value at a certain time.
-    pub(crate) values: IntMap<IndexPath, IndividualHistory<Time, T>>,
+    pub(crate) values: IntMap<IndexPathHash, IndividualHistory<Time, T>>,
 }
 
 pub struct IndividualHistory<Time, T> {
@@ -375,6 +496,9 @@ pub struct IndividualHistory<Time, T> {
     ///
     /// This is so that we can quickly check for object visibility when rendering.
     pub(crate) obj_path: ObjPath,
+    /// The index path prefix (full path minus the last component) and the last component.
+    /// This is used to find the batch of a sibling field if necessary.
+    pub(crate) index_path_split: (IndexPathHash, IndexHash),
     pub(crate) history: BTreeMap<Time, (LogId, T)>,
 }
 
@@ -396,16 +520,19 @@ impl<Time: Copy + Ord, T> IndividualDataHistory<Time, T> {
         value: T,
     ) {
         self.values
-            .entry(index_path)
+            .entry(IndexPathHash::from_path(&index_path))
             .or_insert_with(|| IndividualHistory {
                 obj_path,
+                index_path_split: index_path_split(index_path),
                 history: Default::default(),
             })
             .history
             .insert(time, (log_id, value));
     }
 
-    pub fn iter(&self) -> impl ExactSizeIterator<Item = (&IndexPath, &IndividualHistory<Time, T>)> {
+    pub fn iter(
+        &self,
+    ) -> impl ExactSizeIterator<Item = (&IndexPathHash, &IndividualHistory<Time, T>)> {
         self.values.iter()
     }
 }
@@ -414,7 +541,7 @@ impl<Time: Copy + Ord, T: Clone> IndividualDataHistory<Time, T> {
     /// Returns vectors of equal lengths.
     pub fn query_object(
         &self,
-        index_path: &IndexPath,
+        index_path: &IndexPathHash,
         time_query: &TimeQuery<Time>,
     ) -> (Vec<Time>, Vec<LogId>, Vec<T>) {
         crate::profile_function!();
@@ -438,11 +565,11 @@ impl<Time: Copy + Ord, T: Clone> IndividualDataHistory<Time, T> {
 /// For a specific [`ObjTypePath`].
 pub struct BatchedDataHistory<Time, T> {
     /// The index is the path prefix (everything but the last value).
-    pub(crate) batches_over_time: IntMap<IndexPath, BatchHistory<Time, T>>,
+    pub(crate) batches_over_time: IntMap<IndexPathHash, BatchHistory<Time, T>>,
 }
 
 pub struct BatchHistory<Time, T> {
-    pub(crate) history: BTreeMap<Time, (LogId, Batch<T>)>,
+    pub(crate) history: BTreeMap<Time, (LogId, ArcBatch<T>)>,
 }
 
 impl<Time: Copy + Ord, T> Default for BatchedDataHistory<Time, T> {
@@ -462,12 +589,18 @@ impl<Time: Copy + Ord, T> Default for BatchHistory<Time, T> {
 }
 
 impl<Time: Copy + Ord, T> BatchedDataHistory<Time, T> {
-    fn insert(&mut self, index_path_prefix: IndexPath, time: Time, log_id: LogId, batch: Batch<T>) {
+    fn insert(
+        &mut self,
+        index_path_prefix: IndexPathHash,
+        time: Time,
+        log_id: LogId,
+        batch: ArcBatch<T>,
+    ) {
         let batch_history = self.batches_over_time.entry(index_path_prefix).or_default();
         batch_history.history.insert(time, (log_id, batch));
     }
 
-    pub fn iter(&self) -> impl ExactSizeIterator<Item = (&IndexPath, &BatchHistory<Time, T>)> {
+    pub fn iter(&self) -> impl ExactSizeIterator<Item = (&IndexPathHash, &BatchHistory<Time, T>)> {
         self.batches_over_time.iter()
     }
 }
@@ -487,6 +620,7 @@ impl<Time: Copy + Ord, T: Clone> BatchedDataHistory<Time, T> {
 
         if index_path.has_placeholder_last() {
             // get all matches
+            let index_path = IndexPathHash::from_path(&index_path);
             if let Some(batch_history) = self.batches_over_time.get(&index_path) {
                 query(
                     &batch_history.history,
@@ -502,6 +636,9 @@ impl<Time: Copy + Ord, T: Clone> BatchedDataHistory<Time, T> {
             }
         } else {
             let (index_path_prefix, index_path_suffix) = index_path.replace_last_with_placeholder();
+            let index_path_prefix = IndexPathHash::from_path(&index_path_prefix);
+            let index_path_suffix = index_path_suffix.hash();
+
             if let Some(batch_history) = self.batches_over_time.get(&index_path_prefix) {
                 query(
                     &batch_history.history,
@@ -523,6 +660,69 @@ impl<Time: Copy + Ord, T: Clone> BatchedDataHistory<Time, T> {
 
 // ----------------------------------------------------------------------------
 
+/// For a specific [`ObjTypePath`].
+pub struct BatchSplatDataHistory<Time, T> {
+    pub(crate) values: IntMap<IndexPathHash, BTreeMap<Time, (LogId, T)>>,
+}
+
+impl<Time: Copy + Ord, T> Default for BatchSplatDataHistory<Time, T> {
+    fn default() -> Self {
+        Self {
+            values: Default::default(),
+        }
+    }
+}
+
+impl<Time: Copy + Ord, T> BatchSplatDataHistory<Time, T> {
+    fn insert(&mut self, index_path_prefix: IndexPathHash, time: Time, log_id: LogId, value: T) {
+        self.values
+            .entry(index_path_prefix)
+            .or_default()
+            .insert(time, (log_id, value));
+    }
+}
+
+impl<Time: Copy + Ord, T: Clone> BatchSplatDataHistory<Time, T> {
+    /// Returns vectors of equal lengths.
+    pub fn query_object(
+        &self,
+        index_path: IndexPath,
+        time_query: &TimeQuery<Time>,
+    ) -> (Vec<Time>, Vec<LogId>, Vec<T>) {
+        crate::profile_function!();
+
+        let mut times = vec![];
+        let mut ids = vec![];
+        let mut values = vec![];
+
+        if index_path.has_placeholder_last() {
+            let index_path = IndexPathHash::from_path(&index_path);
+            if let Some(history) = self.values.get(&index_path) {
+                query(history, time_query, |time, (log_id, value)| {
+                    times.push(*time);
+                    ids.push(*log_id);
+                    values.push(value.clone()); // TODO: return references instead
+                });
+            }
+        } else {
+            let (index_path_prefix, _index_path_suffix) =
+                index_path.replace_last_with_placeholder();
+            let index_path_prefix = IndexPathHash::from_path(&index_path_prefix);
+            if let Some(history) = self.values.get(&index_path_prefix) {
+                query(history, time_query, |time, (log_id, value)| {
+                    times.push(*time);
+                    ids.push(*log_id);
+                    values.push(value.clone()); // TODO: return references instead
+                });
+            }
+        }
+
+        (times, ids, values)
+    }
+}
+
+// ----------------------------------------------------------------------------
+
 pub(crate) fn latest_at<'data, Time: Copy + Ord, T>(
     data_over_time: &'data BTreeMap<Time, T>,
     query_time: &'_ Time,
@@ -564,6 +764,7 @@ pub(crate) enum IndividualDataReader<'store, Time, T> {
     None,
     Individual(&'store IndividualDataHistory<Time, T>),
     Batched(&'store BatchedDataHistory<Time, T>),
+    BatchSplat(&'store BatchSplatDataHistory<Time, T>),
 }
 
 impl<'store, Time: 'static + Copy + Ord, T: DataTrait> IndividualDataReader<'store, Time, T> {
@@ -572,24 +773,35 @@ impl<'store, Time: 'static + Copy + Ord, T: DataTrait> IndividualDataReader<'sto
             match data {
                 DataStore::Individual(individual) => Self::Individual(individual),
                 DataStore::Batched(batched) => Self::Batched(batched),
+                DataStore::BatchSplat(batched) => Self::BatchSplat(batched),
             }
         } else {
             Self::None
         }
     }
 
-    pub fn latest_at(&self, index_path: &IndexPath, query_time: &Time) -> Option<&'store T> {
+    pub fn latest_at(
+        &self,
+        full_index_path: &IndexPathHash,
+        index_path_split: &(IndexPathHash, IndexHash),
+        query_time: &Time,
+    ) -> Option<&'store T> {
         match self {
             Self::None => None,
             Self::Individual(history) => {
-                latest_at(&history.values.get(index_path)?.history, query_time)
+                latest_at(&history.values.get(full_index_path)?.history, query_time)
                     .map(|(_time, (_log_id, value))| value)
             }
             Self::Batched(data) => {
-                let (prefix, suffix) = index_path.clone().replace_last_with_placeholder();
+                let (prefix, suffix) = index_path_split;
                 let (_time, (_log_id, batch)) =
-                    latest_at(&data.batches_over_time.get(&prefix)?.history, query_time)?;
-                batch.get(&suffix)
+                    latest_at(&data.batches_over_time.get(prefix)?.history, query_time)?;
+                batch.get(suffix)
+            }
+            Self::BatchSplat(data) => {
+                let (prefix, _suffix) = index_path_split;
+                let (_time, (_log_id, value)) = latest_at(data.values.get(prefix)?, query_time)?;
+                Some(value)
             }
         }
     }
@@ -599,14 +811,15 @@ impl<'store, Time: 'static + Copy + Ord, T: DataTrait> IndividualDataReader<'sto
 
 pub(crate) enum BatchedDataReader<'store, Time, T> {
     None,
-    Individual(IndexPath, Time, &'store IndividualDataHistory<Time, T>),
-    Batched(&'store IntMap<IndexKey, T>),
+    Individual(IndexPathHash, Time, &'store IndividualDataHistory<Time, T>),
+    Batched(&'store Batch<T>),
+    BatchSplat(&'store T),
 }
 
 impl<'store, Time: Copy + Ord, T: DataTrait> BatchedDataReader<'store, Time, T> {
     pub fn new(
         data: Option<&'store DataStore<Time, T>>,
-        index_path_prefix: &IndexPath,
+        index_path_prefix: &IndexPathHash,
         query_time: &Time,
     ) -> Self {
         data.and_then(|data| Self::new_opt(data, index_path_prefix, query_time))
@@ -615,12 +828,12 @@ impl<'store, Time: Copy + Ord, T: DataTrait> BatchedDataReader<'store, Time, T>
 
     fn new_opt(
         data: &'store DataStore<Time, T>,
-        index_path_prefix: &IndexPath,
+        index_path_prefix: &IndexPathHash,
         query_time: &Time,
     ) -> Option<Self> {
         match data {
             DataStore::Individual(individual) => Some(Self::Individual(
-                index_path_prefix.clone(),
+                *index_path_prefix,
                 *query_time,
                 individual,
             )),
@@ -630,19 +843,37 @@ impl<'store, Time: Copy + Ord, T: DataTrait> BatchedDataReader<'store, Time, T>
                 let (_time, (_log_id, map)) = latest_at(everything_per_time, query_time)?;
                 Some(Self::Batched(map))
             }
+            DataStore::BatchSplat(batch_splat) => {
+                let splat_per_time = &batch_splat.values.get(index_path_prefix)?;
+                let (_time, (_log_id, value)) = latest_at(splat_per_time, query_time)?;
+                Some(Self::BatchSplat(value))
+            }
         }
     }
 
-    pub fn latest_at(&self, index_path_suffix: &IndexKey) -> Option<&'store T> {
+    pub fn latest_at(&self, index_path_suffix: &IndexHash) -> Option<&'store T> {
         match self {
             Self::None => None,
             Self::Individual(index_path_prefix, query_time, history) => {
-                let mut index_path = index_path_prefix.clone();
-                index_path.replace_last_placeholder_with(index_path_suffix.clone());
+                let mut index_path = *index_path_prefix;
+                index_path.replace_last_placeholder_with(index_path_suffix);
                 latest_at(&history.values.get(&index_path)?.history, query_time)
                     .map(|(_time, (_log_id, value))| value)
             }
             Self::Batched(data) => data.get(index_path_suffix),
+            Self::BatchSplat(value) => Some(value),
         }
     }
 }
+
+fn index_path_split(index_path: IndexPath) -> (IndexPathHash, IndexHash) {
+    if index_path.is_empty() {
+        (IndexPathHash::default(), IndexHash::PLACEHOLDER) // whatever: it won't be used
+    } else {
+        let (index_path_prefix, index_path_suffix) = index_path.replace_last_with_placeholder();
+        (
+            IndexPathHash::from_path(&index_path_prefix),
+            index_path_suffix.hash(),
+        )
+    }
+}
diff --git a/data_store/src/type_path_query.rs b/data_store/src/type_path_query.rs
index d8e5f5c6aadd..a6d786e744d8 100644
--- a/data_store/src/type_path_query.rs
+++ b/data_store/src/type_path_query.rs
@@ -1,4 +1,4 @@
-//! Queries of the type "read these fields, from all objects of this [`ObjTypePath`], over this time interval"
+//! Queries of the type "read these fields, from all objects of this [`log_types::ObjTypePath`], over this time interval"
 
 use log_types::{DataTrait, FieldName, LogId, ObjPath};
 
@@ -45,6 +45,9 @@ pub fn visit_type_data<'s, Time: 'static + Copy + Ord, T: DataTrait>(
                     );
                 }
             }
+            DataStore::BatchSplat(_) => {
+                tracing::error!("Used BatchSplat for a primary field {field_name:?}");
+            }
         }
     }
 
@@ -68,6 +71,7 @@ pub fn visit_type_data_1<'s, Time: 'static + Copy + Ord, T: DataTrait, S1: DataT
                 let child1_reader = IndividualDataReader::<Time, S1>::new(obj_store, &child1);
 
                 for (index_path, primary) in primary.iter() {
+                    let index_path_split = &primary.index_path_split;
                     query(
                         &primary.history,
                         time_query,
@@ -76,7 +80,7 @@ pub fn visit_type_data_1<'s, Time: 'static + Copy + Ord, T: DataTrait, S1: DataT
                                 &primary.obj_path,
                                 log_id,
                                 primary_value,
-                                child1_reader.latest_at(index_path, time),
+                                child1_reader.latest_at(index_path, index_path_split, time),
                             );
                         },
                     );
@@ -105,6 +109,9 @@ pub fn visit_type_data_1<'s, Time: 'static + Copy + Ord, T: DataTrait, S1: DataT
                     );
                 }
             }
+            DataStore::BatchSplat(_) => {
+                tracing::error!("Used BatchSplat for a primary field {field_name:?}");
+            }
         }
     }
 
@@ -136,6 +143,7 @@ pub fn visit_type_data_2<
                 let child2_reader = IndividualDataReader::<Time, S2>::new(obj_store, &child2);
 
                 for (index_path, primary) in primary.iter() {
+                    let index_path_split = &primary.index_path_split;
                     query(
                         &primary.history,
                         time_query,
@@ -144,8 +152,8 @@ pub fn visit_type_data_2<
                                 &primary.obj_path,
                                 log_id,
                                 primary_value,
-                                child1_reader.latest_at(index_path, time),
-                                child2_reader.latest_at(index_path, time),
+                                child1_reader.latest_at(index_path, index_path_split, time),
+                                child2_reader.latest_at(index_path, index_path_split, time),
                             );
                         },
                     );
@@ -178,6 +186,9 @@ pub fn visit_type_data_2<
                     );
                 }
             }
+            DataStore::BatchSplat(_) => {
+                tracing::error!("Used BatchSplat for a primary field {field_name:?}");
+            }
         }
     }
 
@@ -212,6 +223,7 @@ pub fn visit_type_data_3<
                 let child3_reader = IndividualDataReader::<Time, S3>::new(obj_store, &child3);
 
                 for (index_path, primary) in primary.iter() {
+                    let index_path_split = &primary.index_path_split;
                     query(
                         &primary.history,
                         time_query,
@@ -220,9 +232,9 @@ pub fn visit_type_data_3<
                                 &primary.obj_path,
                                 log_id,
                                 primary_value,
-                                child1_reader.latest_at(index_path, time),
-                                child2_reader.latest_at(index_path, time),
-                                child3_reader.latest_at(index_path, time),
+                                child1_reader.latest_at(index_path, index_path_split, time),
+                                child2_reader.latest_at(index_path, index_path_split, time),
+                                child3_reader.latest_at(index_path, index_path_split, time),
                             );
                         },
                     );
@@ -259,6 +271,9 @@ pub fn visit_type_data_3<
                     );
                 }
             }
+            DataStore::BatchSplat(_) => {
+                tracing::error!("Used BatchSplat for a primary field {field_name:?}");
+            }
         }
     }
 
diff --git a/data_store/tests/test_data_store.rs b/data_store/tests/test_data_store.rs
index 0ed2a28326cb..8abe2b7a7f62 100644
--- a/data_store/tests/test_data_store.rs
+++ b/data_store/tests/test_data_store.rs
@@ -1,5 +1,6 @@
 use data_store::*;
-use log_types::{FieldName, IndexKey, LogId};
+use itertools::Itertools;
+use log_types::{FieldName, LogId};
 
 #[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
 struct Time(i64);
@@ -30,9 +31,10 @@ pub fn points_from_store<'store, Time: 'static + Copy + Ord>(
     points
 }
 
-fn batch<T, const N: usize>(batch: [(IndexKey, T); N]) -> Batch<T> {
-    let batch: nohash_hasher::IntMap<IndexKey, T> = batch.into_iter().collect();
-    std::sync::Arc::new(batch)
+fn batch<T: Clone, const N: usize>(batch: &[(Index, T); N]) -> ArcBatch<T> {
+    let indices = batch.iter().map(|(index, _)| index.clone()).collect_vec();
+    let values = batch.iter().map(|(_, value)| value.clone()).collect_vec();
+    std::sync::Arc::new(Batch::new(&indices, &values))
 }
 
 fn id() -> LogId {
@@ -186,8 +188,8 @@ fn test_batches() -> data_store::Result<()> {
         values
     }
 
-    fn index(seq: u64) -> IndexKey {
-        IndexKey::new(Index::Sequence(seq))
+    fn index(seq: u64) -> Index {
+        Index::Sequence(seq)
     }
 
     let mut store = TypePathDataStore::default();
@@ -197,7 +199,7 @@ fn test_batches() -> data_store::Result<()> {
         "pos".into(),
         Time(1),
         id(),
-        batch([
+        batch(&[
             (index(0), 0_i32),
             (index(1), 1_i32),
             (index(2), 2_i32),
@@ -209,7 +211,7 @@ fn test_batches() -> data_store::Result<()> {
         "pos".into(),
         Time(2),
         id(),
-        batch([
+        batch(&[
             (index(0), 1_000_i32),
             (index(1), 1_001_i32),
             (index(2), 1_002_i32),
@@ -221,7 +223,7 @@ fn test_batches() -> data_store::Result<()> {
         "pos".into(),
         Time(3),
         id(),
-        batch([
+        batch(&[
             // 0, 1 omitted = dropped
             (index(2), 22_i32),
             (index(3), 33_i32),
@@ -232,14 +234,14 @@ fn test_batches() -> data_store::Result<()> {
         "label".into(),
         Time(4),
         id(),
-        batch([(index(1), s("one")), (index(2), s("two"))]),
+        batch(&[(index(1), s("one")), (index(2), s("two"))]),
     )?;
     store.insert_batch(
         &obj_path_batch("right"),
         "label".into(),
         Time(5),
         id(),
-        batch([
+        batch(&[
             (index(0), s("r0")),
             (index(1), s("r1")),
             (index(2), s("r2")),
@@ -252,7 +254,7 @@ fn test_batches() -> data_store::Result<()> {
         "pos".into(),
         Time(6),
         id(),
-        batch([
+        batch(&[
             (index(3), 1_003_i32),
             (index(4), 1_004_i32),
             (index(5), 1_005_i32),
@@ -263,7 +265,7 @@ fn test_batches() -> data_store::Result<()> {
         "label".into(),
         Time(7),
         id(),
-        batch([
+        batch(&[
             (index(3), s("r3_new")),
             // omitted = replaced
         ]),
@@ -390,8 +392,8 @@ fn test_batched_and_individual() -> data_store::Result<()> {
         values
     }
 
-    fn index(seq: u64) -> IndexKey {
-        IndexKey::new(Index::Sequence(seq))
+    fn index(seq: u64) -> Index {
+        Index::Sequence(seq)
     }
 
     let mut store = TypePathDataStore::default();
@@ -401,7 +403,7 @@ fn test_batched_and_individual() -> data_store::Result<()> {
         "pos".into(),
         Time(1),
         id(),
-        batch([
+        batch(&[
             (index(0), 0_i32),
             (index(1), 1_i32),
             (index(2), 2_i32),
@@ -413,7 +415,7 @@ fn test_batched_and_individual() -> data_store::Result<()> {
         "pos".into(),
         Time(2),
         id(),
-        batch([
+        batch(&[
             (index(0), 1_000_i32),
             (index(1), 1_001_i32),
             (index(2), 1_002_i32),
@@ -425,7 +427,7 @@ fn test_batched_and_individual() -> data_store::Result<()> {
         "pos".into(),
         Time(3),
         id(),
-        batch([
+        batch(&[
             // 0, 1 omitted = dropped
             (index(2), 22_i32),
             (index(3), 33_i32),
@@ -453,7 +455,7 @@ fn test_batched_and_individual() -> data_store::Result<()> {
         "pos".into(),
         Time(6),
         id(),
-        batch([
+        batch(&[
             (index(3), 1_003_i32),
             (index(4), 1_004_i32),
             (index(5), 1_005_i32),
@@ -582,8 +584,8 @@ fn test_individual_and_batched() -> data_store::Result<()> {
         values
     }
 
-    fn index(seq: u64) -> IndexKey {
-        IndexKey::new(Index::Sequence(seq))
+    fn index(seq: u64) -> Index {
+        Index::Sequence(seq)
     }
 
     let mut store = TypePathDataStore::default();
@@ -595,7 +597,7 @@ fn test_individual_and_batched() -> data_store::Result<()> {
         "label".into(),
         Time(3),
         id(),
-        batch([(index(1), s("one")), (index(2), s("two"))]),
+        batch(&[(index(1), s("one")), (index(2), s("two"))]),
     )?;
     store.insert_individual(obj_path("left", 2), "pos".into(), Time(4), id(), 2_i32)?;
     store.insert_individual(obj_path("left", 3), "pos".into(), Time(4), id(), 3_i32)?;
@@ -604,7 +606,7 @@ fn test_individual_and_batched() -> data_store::Result<()> {
         "label".into(),
         Time(5),
         id(),
-        batch([(index(2), s("two")), (index(3), s("three"))]),
+        batch(&[(index(2), s("two")), (index(3), s("three"))]),
     )?;
 
     assert_eq!(values(&store, 0), vec![]);
diff --git a/log_types/src/data.rs b/log_types/src/data.rs
index 0c17e93960d1..88487474b9f2 100644
--- a/log_types/src/data.rs
+++ b/log_types/src/data.rs
@@ -1,4 +1,4 @@
-use crate::{impl_into_enum, Index, ObjPath};
+use crate::{impl_into_enum, ObjPath};
 
 // ----------------------------------------------------------------------------
 
@@ -160,13 +160,6 @@ pub mod data_types {
 #[derive(Clone, Debug, PartialEq)]
 #[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
 pub enum Data {
-    Batch {
-        indices: Vec<Index>,
-        data: DataVec,
-    },
-
-    // ----------------------------
-
     // 1D:
     I32(i32),
     F32(f32),
@@ -201,10 +194,8 @@ pub enum Data {
 
 impl Data {
     #[inline]
-    pub fn typ(&self) -> DataType {
+    pub fn data_type(&self) -> DataType {
         match self {
-            Self::Batch { data, .. } => data.data_type(),
-
             Self::I32(_) => DataType::I32,
             Self::F32(_) => DataType::F32,
             Self::Color(_) => DataType::Color,
@@ -267,6 +258,37 @@ pub enum DataVec {
     Space(Vec<ObjPath>),
 }
 
+/// Do the same thing with all members of a [`Data`].
+///
+/// ```
+/// # use log_types::{Data, data_map};
+/// # let data: Data = Data::F32(0.0);
+/// data_map!(data, |data| { dbg!(data); });
+/// ```
+#[macro_export]
+macro_rules! data_map(
+    ($data: expr, |$value: pat_param| $action: expr) => ({
+        match $data {
+            Data::I32($value) => $action,
+            Data::F32($value) => $action,
+            Data::Color($value) => $action,
+            Data::String($value) => $action,
+            Data::Vec2($value) => $action,
+            Data::BBox2D($value) => $action,
+            Data::LineSegments2D($value) => $action,
+            Data::Image($value) => $action,
+            Data::Vec3($value) => $action,
+            Data::Box3($value) => $action,
+            Data::Path3D($value) => $action,
+            Data::LineSegments3D($value) => $action,
+            Data::Mesh3D($value) => $action,
+            Data::Camera($value) => $action,
+            Data::Vecf32($value) => $action,
+            Data::Space($value) => $action,
+        }
+    });
+);
+
 /// Do the same thing with all members of a [`DataVec`].
 ///
 /// ```
diff --git a/log_types/src/hash.rs b/log_types/src/hash.rs
index ede9f2d3ef0c..a288da68c7d7 100644
--- a/log_types/src/hash.rs
+++ b/log_types/src/hash.rs
@@ -49,6 +49,8 @@ impl std::fmt::Debug for Hash64 {
 pub struct Hash128([u64; 2]);
 
 impl Hash128 {
+    pub const ZERO: Hash128 = Hash128([0; 2]);
+
     pub fn hash(value: impl std::hash::Hash + Copy) -> Self {
         Self(double_hash(value))
     }
diff --git a/log_types/src/index.rs b/log_types/src/index.rs
new file mode 100644
index 000000000000..8c7d8860d8d2
--- /dev/null
+++ b/log_types/src/index.rs
@@ -0,0 +1,119 @@
+use crate::hash::Hash128;
+
+// ----------------------------------------------------------------------------
+
+/// The key of a table.
+#[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord)]
+#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
+pub enum Index {
+    /// For arrays, assumed to be dense (0, 1, 2, …).
+    Sequence(u64),
+
+    /// X,Y pixel coordinates, from top left.
+    Pixel([u64; 2]),
+
+    /// Any integer, e.g. a hash or an arbitrary identifier.
+    Integer(i128),
+
+    /// UUID/GUID
+    Uuid(uuid::Uuid),
+
+    /// Anything goes.
+    String(String),
+
+    /// Used as the last index when logging a batch of data.
+    Placeholder,
+}
+
+impl Index {
+    #[inline]
+    pub fn is_placeholder(&self) -> bool {
+        matches!(self, Self::Placeholder)
+    }
+
+    #[inline]
+    pub fn hash(&self) -> IndexHash {
+        IndexHash::hash(self)
+    }
+}
+
+impl std::fmt::Display for Index {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Sequence(seq) => format!("#{seq}").fmt(f),
+            Self::Pixel([x, y]) => format!("[{x}, {y}]").fmt(f),
+            Self::Integer(value) => value.fmt(f),
+            Self::Uuid(value) => value.fmt(f),
+            Self::String(value) => format!("{value:?}").fmt(f), // put it in quotes
+            Self::Placeholder => '_'.fmt(f),                    // put it in quotes
+        }
+    }
+}
+
+crate::impl_into_enum!(String, Index, String);
+
+// ----------------------------------------------------------------------------
+
+/// A 128 bit hash of [`Index`] with negligible chance of collision.
+#[derive(Copy, Clone, Eq)]
+pub struct IndexHash(Hash128);
+
+impl IndexHash {
+    pub const PLACEHOLDER: IndexHash = Self(Hash128::ZERO);
+
+    #[inline]
+    pub fn hash(index: &Index) -> Self {
+        if index.is_placeholder() {
+            Self(Hash128::ZERO)
+        } else {
+            Self(Hash128::hash(index))
+        }
+    }
+
+    /// Is this equal to `IndexHash::hash(&Index::Placeholder)` ?
+    #[inline]
+    pub fn is_placeholder(&self) -> bool {
+        self.0 == Hash128::ZERO
+    }
+
+    #[inline]
+    pub fn hash64(&self) -> u64 {
+        self.0.hash64()
+    }
+
+    #[inline]
+    pub fn first64(&self) -> u64 {
+        self.0.first64()
+    }
+
+    #[inline]
+    pub fn second64(&self) -> u64 {
+        self.0.second64()
+    }
+}
+
+impl std::hash::Hash for IndexHash {
+    #[inline]
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        state.write_u64(self.0.hash64());
+    }
+}
+
+impl std::cmp::PartialEq for IndexHash {
+    #[inline]
+    fn eq(&self, other: &Self) -> bool {
+        self.0 == other.0
+    }
+}
+
+impl nohash_hasher::IsEnabled for IndexHash {}
+
+impl std::fmt::Debug for IndexHash {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&format!(
+            "IndexHash({:016X}{:016X})",
+            self.0.first64(),
+            self.0.second64()
+        ))
+    }
+}
diff --git a/log_types/src/lib.rs b/log_types/src/lib.rs
index 69ec22184b71..03f4c8c97920 100644
--- a/log_types/src/lib.rs
+++ b/log_types/src/lib.rs
@@ -7,11 +7,13 @@ pub mod encoding;
 
 mod data;
 pub mod hash;
+mod index;
 pub mod objects;
 mod path;
 mod time;
 
 pub use data::*;
+pub use index::*;
 pub use objects::ObjectType;
 pub use path::*;
 pub use time::{Duration, Time};
@@ -122,7 +124,7 @@ pub struct DataMsg {
     pub data_path: DataPath,
 
     /// The value of this.
-    pub data: Data,
+    pub data: LoggedData,
 }
 
 #[inline]
@@ -130,7 +132,7 @@ pub fn data_msg(
     time_point: &TimePoint,
     obj_path: impl Into<ObjPath>,
     field_name: impl Into<FieldName>,
-    data: impl Into<Data>,
+    data: impl Into<LoggedData>,
 ) -> DataMsg {
     DataMsg {
         time_point: time_point.clone(),
@@ -142,6 +144,67 @@ pub fn data_msg(
 
 // ----------------------------------------------------------------------------
 
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
+pub enum LoggedData {
+    /// A single data value
+    Single(Data),
+
+    /// Log multiple values at once.
+    ///
+    /// The index replaces the last index in [`DataMsg.data_path`], which should be [`Index::Placeholder]`.
+    Batch { indices: Vec<Index>, data: DataVec },
+
+    /// Log the same value for all objects sharing the same index prefix (i.e. ignoring the last index).
+    ///
+    /// The last index in [`DataMsg.data_path`] should be [`Index::Placeholder]`.
+    ///
+    /// You can only use this for optional fields such as `color`, `space` etc.
+    /// You can NOT use it for primary fields such as `pos`.
+    BatchSplat(Data),
+}
+
+impl LoggedData {
+    #[inline]
+    pub fn data_type(&self) -> DataType {
+        match self {
+            Self::Single(data) | Self::BatchSplat(data) => data.data_type(),
+            Self::Batch { data, .. } => data.data_type(),
+        }
+    }
+}
+
+impl From<Data> for LoggedData {
+    #[inline]
+    fn from(data: Data) -> Self {
+        Self::Single(data)
+    }
+}
+
+#[macro_export]
+macro_rules! impl_into_logged_data {
+    ($from_ty: ty, $data_enum_variant: ident) => {
+        impl From<$from_ty> for LoggedData {
+            #[inline]
+            fn from(value: $from_ty) -> Self {
+                Self::Single(Data::$data_enum_variant(value))
+            }
+        }
+    };
+}
+
+impl_into_logged_data!(i32, I32);
+impl_into_logged_data!(f32, F32);
+impl_into_logged_data!(BBox2D, BBox2D);
+impl_into_logged_data!(Image, Image);
+impl_into_logged_data!(Box3, Box3);
+impl_into_logged_data!(Mesh3D, Mesh3D);
+impl_into_logged_data!(Camera, Camera);
+impl_into_logged_data!(Vec<f32>, Vecf32);
+impl_into_logged_data!(ObjPath, Space);
+
+// ----------------------------------------------------------------------------
+
 rr_string_interner::declare_new_type!(
     /// The name of a time source. Often something like `"time"` or `"frame"`.
     pub struct TimeSource;
diff --git a/log_types/src/path.rs b/log_types/src/path.rs
index d3f631f77d8f..ba29383cf8b8 100644
--- a/log_types/src/path.rs
+++ b/log_types/src/path.rs
@@ -6,13 +6,15 @@ pub(crate) mod obj_path_impl;
 mod obj_type_path;
 
 pub use data_path::DataPath;
-pub use index_path::{IndexKey, IndexPath};
+pub use index_path::{IndexPath, IndexPathHash};
 pub use obj_path::{ObjPath, ObjPathHash};
 pub use obj_path_builder::ObjPathBuilder;
 pub use obj_type_path::ObjTypePath;
 
 use rr_string_interner::InternedString;
 
+use crate::Index;
+
 rr_string_interner::declare_new_type!(
     /// The name of a object field, e.g. "pos" or "color".
     pub struct FieldName;
@@ -57,46 +59,6 @@ impl From<&str> for ObjPathComp {
 
 // ----------------------------------------------------------------------------
 
-/// The key of a table.
-#[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord)]
-#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
-pub enum Index {
-    /// For arrays, assumed to be dense (0, 1, 2, …).
-    Sequence(u64),
-
-    /// X,Y pixel coordinates, from top left.
-    Pixel([u64; 2]),
-
-    /// Any integer, e.g. a hash or an arbitrary identifier.
-    Integer(i128),
-
-    /// UUID/GUID
-    Uuid(uuid::Uuid),
-
-    /// Anything goes.
-    String(String),
-
-    /// Used as the last index when logging a batch of data.
-    Placeholder,
-}
-
-impl std::fmt::Display for Index {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Self::Sequence(seq) => format!("#{seq}").fmt(f),
-            Self::Pixel([x, y]) => format!("[{x}, {y}]").fmt(f),
-            Self::Integer(value) => value.fmt(f),
-            Self::Uuid(value) => value.fmt(f),
-            Self::String(value) => format!("{value:?}").fmt(f), // put it in quotes
-            Self::Placeholder => '_'.fmt(f),                    // put it in quotes
-        }
-    }
-}
-
-crate::impl_into_enum!(String, Index, String);
-
-// ----------------------------------------------------------------------------
-
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
 #[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
 pub enum TypePathComp {
diff --git a/log_types/src/path/index_path.rs b/log_types/src/path/index_path.rs
index ab975d3d802c..162928744cc4 100644
--- a/log_types/src/path/index_path.rs
+++ b/log_types/src/path/index_path.rs
@@ -1,89 +1,69 @@
-use crate::{hash::Hash128, path::Index};
+use crate::{Index, IndexHash};
 
 // ----------------------------------------------------------------------------
 
-/// Like `Index` but also includes a precomputed hash.
-#[derive(Clone, Eq)]
-pub struct IndexKey {
-    hash: Hash128,
-    // boxed to keep size of self small.
-    index: Box<Index>,
-}
+/// A 128 bit hash of a [`IndexPath`] with negligible risk of collision.
+#[derive(Clone, Copy, Default, Eq)]
+pub struct IndexPathHash([u64; 2]);
 
-impl IndexKey {
-    #[inline]
-    pub fn new(index: Index) -> Self {
-        Self {
-            hash: Hash128::hash(&index),
-            index: Box::new(index),
+impl IndexPathHash {
+    pub fn from_path(path: &IndexPath) -> Self {
+        let mut hash = Self::default();
+        for index in &path.components {
+            hash.push(&index.hash());
         }
+        hash
     }
 
-    #[inline]
-    pub fn index(&self) -> &Index {
-        &self.index
-    }
-}
-
-// ----------------------------------------------------------------------------
-
-impl std::cmp::PartialOrd for IndexKey {
-    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
-        self.index.partial_cmp(&other.index)
+    pub fn push(&mut self, index_hash: &IndexHash) {
+        self.0[0] = self.0[0].rotate_left(5);
+        self.0[1] = self.0[1].rotate_left(5);
+        self.0[0] ^= index_hash.first64();
+        self.0[1] ^= index_hash.second64();
     }
-}
 
-impl std::cmp::Ord for IndexKey {
-    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
-        self.index.cmp(&other.index)
+    pub fn replace_last_placeholder_with(&mut self, index_hash: &IndexHash) {
+        // `Index::Placeholder` has zero as hash, so we can easily replace it:
+        assert!(!index_hash.is_placeholder());
+        self.0[0] ^= index_hash.first64();
+        self.0[1] ^= index_hash.second64();
     }
 }
 
-impl std::cmp::PartialEq for IndexKey {
-    #[inline]
-    fn eq(&self, other: &Self) -> bool {
-        self.hash == other.hash // much faster, and low chance of collision
-    }
-}
-
-impl std::hash::Hash for IndexKey {
+impl std::hash::Hash for IndexPathHash {
     #[inline]
     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
-        self.hash.hash(state);
+        state.write_u64(self.0[0]);
     }
 }
 
-impl nohash_hasher::IsEnabled for IndexKey {}
-
-impl From<Index> for IndexKey {
+impl PartialEq for IndexPathHash {
     #[inline]
-    fn from(index: Index) -> Self {
-        IndexKey::new(index)
+    fn eq(&self, other: &Self) -> bool {
+        self.0 == other.0
     }
 }
 
-impl std::fmt::Debug for IndexKey {
+impl nohash_hasher::IsEnabled for IndexPathHash {}
+
+impl std::fmt::Debug for IndexPathHash {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        self.index.fmt(f)
+        f.write_str(&format!("Hash128({:016X}{:016X})", self.0[0], self.0[1]))
     }
 }
 
 // ----------------------------------------------------------------------------
 
-#[derive(Clone, Default, Eq)]
+#[derive(Clone, Default, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
 pub struct IndexPath {
     components: Vec<Index>,
-    hashes: [u64; 2], // 128 bit to avoid collisions
 }
 
 impl IndexPath {
     #[inline]
     pub fn new(components: Vec<Index>) -> Self {
-        let mut slf = Self::default();
-        for index in components {
-            slf.push(index);
-        }
-        slf
+        Self { components }
     }
 
     #[inline]
@@ -106,37 +86,26 @@ impl IndexPath {
         self.components.iter()
     }
 
-    pub fn push(&mut self, index: impl Into<IndexKey>) {
-        let index = index.into();
-
-        self.components.push(*index.index);
-        self.hashes[0] = self.hashes[0].rotate_left(5);
-        self.hashes[1] = self.hashes[1].rotate_left(5);
-        self.hashes[0] ^= index.hash.first64();
-        self.hashes[1] ^= index.hash.second64();
+    pub fn push(&mut self, index: Index) {
+        self.components.push(index);
     }
 
-    pub fn pop(&mut self) -> Option<IndexKey> {
-        let index = IndexKey::new(self.components.pop()?);
-        self.hashes[0] ^= index.hash.first64();
-        self.hashes[1] ^= index.hash.second64();
-        self.hashes[0] = self.hashes[0].rotate_right(5);
-        self.hashes[1] = self.hashes[1].rotate_right(5);
-        Some(index)
+    pub fn pop(&mut self) -> Option<Index> {
+        self.components.pop()
     }
 
     /// Replace last component with [`Index::Placeholder`], and return what was there.
-    pub fn replace_last_with_placeholder(mut self) -> (IndexPath, IndexKey) {
+    pub fn replace_last_with_placeholder(mut self) -> (IndexPath, Index) {
         let index = self.pop().unwrap();
-        assert_ne!(index, IndexKey::new(Index::Placeholder));
+        assert_ne!(index, Index::Placeholder);
         self.push(Index::Placeholder);
         (self, index)
     }
 
     /// Replace last [`Index::Placeholder`] with the given key.
-    pub fn replace_last_placeholder_with(&mut self, key: IndexKey) {
+    pub fn replace_last_placeholder_with(&mut self, key: Index) {
         let index = self.pop().unwrap();
-        assert_eq!(index, IndexKey::new(Index::Placeholder));
+        assert_eq!(index, Index::Placeholder);
         self.push(key);
     }
 
@@ -148,38 +117,6 @@ impl IndexPath {
 
 pub type Iter<'a> = std::slice::Iter<'a, Index>;
 
-#[cfg(feature = "serde")]
-impl serde::Serialize for IndexPath {
-    #[inline]
-    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
-        self.as_slice().serialize(serializer)
-    }
-}
-
-#[cfg(feature = "serde")]
-impl<'de> serde::Deserialize<'de> for IndexPath {
-    #[inline]
-    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
-        <Vec<Index>>::deserialize(deserializer).map(IndexPath::new)
-    }
-}
-
-impl std::cmp::PartialEq for IndexPath {
-    #[inline]
-    fn eq(&self, other: &Self) -> bool {
-        self.hashes == other.hashes // much faster, and low chance of collision
-    }
-}
-
-impl std::hash::Hash for IndexPath {
-    #[inline]
-    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
-        state.write_u64(self.hashes[0]);
-    }
-}
-
-impl nohash_hasher::IsEnabled for IndexPath {}
-
 impl std::fmt::Debug for IndexPath {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         self.as_slice().fmt(f)
@@ -204,9 +141,9 @@ fn test_index_path_key() {
 
     let (key1_prefix, seq0) = key1.replace_last_with_placeholder();
     assert_eq!(key1_prefix.components.len(), 1);
-    assert_eq!(seq0, IndexKey::new(Index::Sequence(0)));
+    assert_eq!(seq0, Index::Sequence(0));
 
     let (key2_prefix, seq1) = key2.replace_last_with_placeholder();
     assert_eq!(key2_prefix.components.len(), 2);
-    assert_eq!(seq1, IndexKey::new(Index::Sequence(1)));
+    assert_eq!(seq1, Index::Sequence(1));
 }
diff --git a/log_types/src/path/obj_path.rs b/log_types/src/path/obj_path.rs
index fa1586b7139b..aa76b5dddf9f 100644
--- a/log_types/src/path/obj_path.rs
+++ b/log_types/src/path/obj_path.rs
@@ -1,11 +1,12 @@
 use crate::{
     hash::Hash128,
     path::{obj_path_impl::ObjPathImpl, IndexPath, ObjPathBuilder, ObjTypePath},
+    Index,
 };
 
 // ----------------------------------------------------------------------------
 
-/// A 128 bit hash with negligible chance of collision.
+/// A 128 bit hash of [`ObjPath`] with negligible chance of collision.
 #[derive(Copy, Clone, Eq)]
 pub struct ObjPathHash(Hash128);
 
@@ -110,7 +111,7 @@ impl ObjPath {
 
     /// Replace last [`Index::Placeholder`] with the given key.
     #[must_use]
-    pub fn replace_last_placeholder_with(self, key: crate::IndexKey) -> Self {
+    pub fn replace_last_placeholder_with(self, key: Index) -> Self {
         Self::from(self.path.replace_last_placeholder_with(key))
     }
 }
diff --git a/log_types/src/path/obj_path_builder.rs b/log_types/src/path/obj_path_builder.rs
index 981ddde824c8..1b09c213aac7 100644
--- a/log_types/src/path/obj_path_builder.rs
+++ b/log_types/src/path/obj_path_builder.rs
@@ -1,7 +1,4 @@
-use crate::{
-    path::{Index, ObjPathComp},
-    ObjPath,
-};
+use crate::{path::ObjPathComp, Index, ObjPath};
 
 use super::obj_path::ObjPathComponentRef;
 
diff --git a/log_types/src/path/obj_path_impl.rs b/log_types/src/path/obj_path_impl.rs
index 8432057e6510..5fd2bf2afb35 100644
--- a/log_types/src/path/obj_path_impl.rs
+++ b/log_types/src/path/obj_path_impl.rs
@@ -1,8 +1,11 @@
-use crate::path::{Index, IndexPath, ObjPathBuilder, ObjPathComp, ObjTypePath, TypePathComp};
+use crate::{
+    path::{IndexPath, ObjPathBuilder, ObjPathComp, ObjTypePath, TypePathComp},
+    Index,
+};
 
 /// `camera / "left" / points / #42`
 ///
-/// Wrapped by [`ObjPath`] together with a hash.
+/// Wrapped by [`crate::ObjPath`] together with a hash.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 #[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
 pub(crate) struct ObjPathImpl {
@@ -76,7 +79,7 @@ impl ObjPathImpl {
 
     /// Replace last [`Index::Placeholder`] with the given key.
     #[must_use]
-    pub fn replace_last_placeholder_with(self, key: crate::IndexKey) -> Self {
+    pub fn replace_last_placeholder_with(self, key: Index) -> Self {
         let (type_path, mut index_path) = self.into_type_path_and_index_path();
         index_path.replace_last_placeholder_with(key);
         Self::new(type_path, index_path)
diff --git a/nyud/src/lib.rs b/nyud/src/lib.rs
index 500422f5a370..72daac290a33 100644
--- a/nyud/src/lib.rs
+++ b/nyud/src/lib.rs
@@ -57,8 +57,32 @@ fn log_dataset_zip(path: &Path, logger: &Logger<'_>) -> anyhow::Result<()> {
 
     let mut file_contents = vec![];
 
-    let mut num_depth_images = 0;
-    const MAX_DEPTH_IMAGES: usize = 8; // They are so slow
+    // logging depth images is slow (atm), so we don't log every frame
+    let mut depth_images_counter = 0;
+    const DEPTH_IMAGE_INTERVAL: usize = 16;
+
+    let obj_path = ObjPathBuilder::from("points") / Index::Placeholder;
+
+    {
+        // TODO: better way to do "forever and always"
+        let time_point =
+            time_point([("time", TimeValue::Time(Time::from_seconds_since_epoch(0.0)))]);
+
+        logger.log(data_msg(
+            &time_point,
+            &obj_path,
+            "space",
+            LoggedData::BatchSplat(Data::Space(ObjPath::from("world"))),
+        ));
+
+        // TODO: base color on depth?
+        logger.log(data_msg(
+            &time_point,
+            &obj_path,
+            "color",
+            LoggedData::BatchSplat(Data::Color([255_u8; 4])),
+        ));
+    }
 
     for i in 0..archive.len() {
         let file = archive.by_index_raw(i).unwrap();
@@ -77,7 +101,6 @@ fn log_dataset_zip(path: &Path, logger: &Logger<'_>) -> anyhow::Result<()> {
             let file_name_parts = file_name.split('-').collect_vec();
             let time = file_name_parts[file_name_parts.len() - 2];
             let time = Time::from_seconds_since_epoch(time.parse().unwrap());
-
             let time_point = time_point([("time", TimeValue::Time(time))]);
 
             if file_name.ends_with(".ppm") {
@@ -99,9 +122,9 @@ fn log_dataset_zip(path: &Path, logger: &Logger<'_>) -> anyhow::Result<()> {
                 ));
             }
 
-            if file_name.ends_with(".pgm") && num_depth_images < MAX_DEPTH_IMAGES {
-                num_depth_images += 1;
-
+            let is_depth_image = file_name.ends_with(".pgm");
+            depth_images_counter += is_depth_image as usize;
+            if depth_images_counter % DEPTH_IMAGE_INTERVAL == 1 {
                 let depth_image = image::load_from_memory(&file_contents)
                     .unwrap()
                     .into_luma16();
@@ -131,8 +154,8 @@ fn log_dataset_zip(path: &Path, logger: &Logger<'_>) -> anyhow::Result<()> {
                 ]);
                 let world_from_pixel = intrinsics.inverse();
 
-                let mut indices = vec![];
-                let mut positions = vec![];
+                let mut indices = Vec::with_capacity((w * h) as usize);
+                let mut positions = Vec::with_capacity((w * h) as usize);
 
                 for y in 0..h {
                     for x in 0..w {
@@ -149,40 +172,17 @@ fn log_dataset_zip(path: &Path, logger: &Logger<'_>) -> anyhow::Result<()> {
                     }
                 }
 
-                let obj_path = ObjPathBuilder::from("points") / Index::Placeholder;
-
                 logger.log(data_msg(
                     &time_point,
                     &obj_path,
                     "pos",
-                    Data::Batch {
+                    LoggedData::Batch {
                         indices: indices.clone(),
                         data: DataVec::Vec3(positions),
                     },
                 ));
 
-                let spaces = vec![ObjPath::from("world"); indices.len()];
-                logger.log(data_msg(
-                    &time_point,
-                    &obj_path,
-                    "space",
-                    Data::Batch {
-                        indices: indices.clone(),
-                        data: DataVec::Space(spaces),
-                    },
-                ));
-
-                // TODO: base color on depth?
-                let colors = vec![[255_u8; 4]; indices.len()];
-                logger.log(data_msg(
-                    &time_point,
-                    &obj_path,
-                    "color",
-                    Data::Batch {
-                        indices,
-                        data: DataVec::Color(colors),
-                    },
-                ));
+                tracing::info!("{} points in point cloud", indices.len());
             }
         }
     }
diff --git a/nyud/src/main.rs b/nyud/src/main.rs
index c3c806b5a876..82a1fd0f024d 100644
--- a/nyud/src/main.rs
+++ b/nyud/src/main.rs
@@ -17,6 +17,10 @@ struct Args {
     #[cfg(feature = "web")]
     #[clap(long)]
     open: bool,
+
+    /// Start with the puffin profiler running.
+    #[clap(long)]
+    profile: bool,
 }
 
 #[cfg(not(feature = "web"))]
@@ -31,6 +35,11 @@ fn main() {
         "Requires at least one file directory"
     );
 
+    let mut profiler = viewer::Profiler::default();
+    if args.profile {
+        profiler.start();
+    }
+
     let (rerun_tx, rerun_rx) = std::sync::mpsc::channel();
 
     let handle = std::thread::spawn(move || {
diff --git a/viewer/src/app.rs b/viewer/src/app.rs
index 13099806ca9a..8b5d97d7e625 100644
--- a/viewer/src/app.rs
+++ b/viewer/src/app.rs
@@ -67,8 +67,9 @@ impl eframe::App for App {
 
     fn update(&mut self, egui_ctx: &egui::Context, frame: &mut eframe::Frame) {
         if let Some(rx) = &mut self.rx {
-            while let Ok(data_msg) = rx.try_recv() {
-                self.log_db.add(data_msg);
+            crate::profile_scope!("receive_messages");
+            while let Ok(msg) = rx.try_recv() {
+                self.log_db.add(msg);
             }
         }
 
diff --git a/viewer/src/misc/log_db.rs b/viewer/src/misc/log_db.rs
index 4bf8830a0fe0..bf70147bdc04 100644
--- a/viewer/src/misc/log_db.rs
+++ b/viewer/src/misc/log_db.rs
@@ -1,6 +1,7 @@
 use std::collections::{BTreeMap, BTreeSet, HashMap};
 
 use itertools::Itertools as _;
+use nohash_hasher::IntMap;
 
 use log_types::*;
 use rr_string_interner::InternedString;
@@ -11,8 +12,8 @@ use crate::misc::TimePoints;
 pub(crate) struct LogDb {
     /// Messages in the order they arrived
     chronological_message_ids: Vec<LogId>,
-    log_messages: nohash_hasher::IntMap<LogId, LogMsg>,
-    pub object_types: ahash::AHashMap<ObjTypePath, ObjectType>,
+    log_messages: IntMap<LogId, LogMsg>,
+    pub object_types: IntMap<ObjTypePath, ObjectType>,
     pub time_points: TimePoints,
     pub data_tree: ObjectTree,
     pub data_store: data_store::LogDataStore,
@@ -189,7 +190,7 @@ impl DataColumns {
             .insert(msg.id);
 
         self.per_type
-            .entry(msg.data.typ())
+            .entry(msg.data.data_type())
             .or_default()
             .insert(msg.id);
     }
diff --git a/viewer/src/ui/context_panel.rs b/viewer/src/ui/context_panel.rs
index 6d17e97c6a55..46e55ab19a2f 100644
--- a/viewer/src/ui/context_panel.rs
+++ b/viewer/src/ui/context_panel.rs
@@ -1,6 +1,7 @@
-use data_store::ObjPath;
 use itertools::Itertools;
-use log_types::{Data, DataMsg, DataPath, LogId};
+
+use data_store::ObjPath;
+use log_types::{Data, DataMsg, DataPath, LogId, LoggedData};
 
 use crate::{LogDb, Preview, Selection, ViewerContext};
 
@@ -166,7 +167,7 @@ pub(crate) fn show_detailed_data_msg(
         data,
     } = msg;
 
-    let is_image = matches!(msg.data, Data::Image(_));
+    let is_image = matches!(msg.data, LoggedData::Single(Data::Image(_)));
 
     egui::Grid::new("fields")
         .striped(true)
@@ -185,12 +186,12 @@ pub(crate) fn show_detailed_data_msg(
 
             if !is_image {
                 ui.monospace("data:");
-                crate::space_view::ui_data(context, ui, id, data, Preview::Medium);
+                crate::space_view::ui_logged_data(context, ui, id, data, Preview::Medium);
                 ui.end_row();
             }
         });
 
-    if let Data::Image(image) = &msg.data {
+    if let LoggedData::Single(Data::Image(image)) = &msg.data {
         show_image(context, ui, id, image);
     }
 }
diff --git a/viewer/src/ui/log_table_view.rs b/viewer/src/ui/log_table_view.rs
index f195fe259a30..e70e7e922073 100644
--- a/viewer/src/ui/log_table_view.rs
+++ b/viewer/src/ui/log_table_view.rs
@@ -82,7 +82,7 @@ pub(crate) fn message_table(
 }
 
 fn row_height(msg: &DataMsg) -> f32 {
-    if matches!(&msg.data, log_types::Data::Image(_)) {
+    if matches!(msg.data.data_type(), DataType::Image) {
         48.0
     } else {
         18.0
@@ -114,6 +114,6 @@ fn table_row(
         context.data_path_button(ui, data_path);
     });
     row.col(|ui| {
-        crate::space_view::ui_data(context, ui, id, data, Preview::Specific(row_height));
+        crate::space_view::ui_logged_data(context, ui, id, data, Preview::Specific(row_height));
     });
 }
diff --git a/viewer/src/ui/space_view.rs b/viewer/src/ui/space_view.rs
index 1a7a289ea706..3ca03a5e6882 100644
--- a/viewer/src/ui/space_view.rs
+++ b/viewer/src/ui/space_view.rs
@@ -330,7 +330,7 @@ pub(crate) fn show_log_msg(
             ui.end_row();
 
             ui.monospace("data:");
-            ui_data(context, ui, id, data, preview);
+            ui_logged_data(context, ui, id, data, preview);
             ui.end_row();
         });
 }
@@ -351,6 +351,26 @@ pub(crate) fn ui_time_point(
     });
 }
 
+pub(crate) fn ui_logged_data(
+    context: &mut ViewerContext,
+    ui: &mut egui::Ui,
+    id: &LogId,
+    data: &LoggedData,
+    preview: Preview,
+) -> egui::Response {
+    match data {
+        LoggedData::Batch { data, .. } => ui.label(format!("batch: {:?}", data)),
+        LoggedData::Single(data) => ui_data(context, ui, id, data, preview),
+        LoggedData::BatchSplat(data) => {
+            ui.horizontal(|ui| {
+                ui.label("Batch Splat:");
+                ui_data(context, ui, id, data, preview)
+            })
+            .response
+        }
+    }
+}
+
 pub(crate) fn ui_data(
     context: &mut ViewerContext,
     ui: &mut egui::Ui,
@@ -359,8 +379,6 @@ pub(crate) fn ui_data(
     preview: Preview,
 ) -> egui::Response {
     match data {
-        Data::Batch { data, .. } => ui.label(format!("batch: {:?}", data)),
-
         Data::I32(value) => ui.label(value.to_string()),
         Data::F32(value) => ui.label(value.to_string()),
         Data::Color([r, g, b, a]) => {
diff --git a/viewer/src/ui/view3d/rendering.rs b/viewer/src/ui/view3d/rendering.rs
index a007e58038bf..c3a9c47c6544 100644
--- a/viewer/src/ui/view3d/rendering.rs
+++ b/viewer/src/ui/view3d/rendering.rs
@@ -348,6 +348,5 @@ pub fn paint_with_three_d(
 }
 
 fn color_to_three_d([r, g, b, a]: [u8; 4]) -> three_d::Color {
-    // TODO: figure out why three_d colors are messed up. Are they in linear space and in bytes!?!?!
-    three_d::Color::new(r, g, b, a)
+    three_d::Color { r, g, b, a }
 }