Add ChunkStoreDiff::SchemaAddition and use it for heuristics

### Related

* Closes RR-4028.

### What

Our heuristics and recommondations are purely additive, i.e. once
something has been recommended once it will only be recommended. An
additional assumption is that we never look at the actual values of
chunks to make decisions, but rather infer everything from the
per-column information. Before this PR, we were recomputed heuristics
for each chunk that would arrive. This was particularly expensive for
chunks with _any scalar_ components.

This PR introduces a new `ChunkStoreDiff::SchemaColumnAddition` event to
chunk stores, that gets emitted iff a new column is added to the chunk
store's schema. This drastically cuts down on how often we need to
recompute heuristics and significantly speeds up ingestion.

Source-Ref: e89fd4af59885940dd14edc7fca55e1b305ac1c3

Author: grtlr

crates/store/re_chunk_store/src/events.rs

@@ -14,7 +14,7 @@ use crate::{ChunkId, ChunkStore, ChunkStoreSubscriber, RowId};
 /// Per-component information for chunks.
 ///
 /// Created from either a physical chunk or virtual manifest metadata.
-#[derive(Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ChunkComponentMeta {
     pub descriptor: re_sdk_types::ComponentDescriptor,
 
@@ -28,15 +28,18 @@ pub struct ChunkComponentMeta {
     /// For virtual this means `row_count > 0`.
     pub has_data: bool,
 
-    /// Whether this component only has static data.
-    pub is_static_only: bool,
+    /// Whether this component has ever been written as static data.
+    ///
+    /// Once a component is static, it stays static. This flag is monotonic
+    /// and never transitions back to `false`.
+    pub is_static: bool,
 }
 
 /// Chunk meta originating from either a virtual or physical chunk.
 ///
 /// Useful for chunk store subscribers that do the same logic
 /// for physical and virtual additions.
-#[derive(Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ChunkMeta {
     pub entity_path: re_chunk::EntityPath,
     pub components: Vec<ChunkComponentMeta>,
@@ -52,7 +55,7 @@ impl ChunkMeta {
                 descriptor: column.descriptor.clone(),
                 inner_arrow_datatype: Some(column.list_array.value_type()),
                 has_data: !column.list_array.values().is_empty(),
-                is_static_only: chunk.is_static(),
+                is_static: chunk.is_static(),
             })
             .collect();
 
@@ -138,6 +141,22 @@ pub enum ChunkStoreDiff {
 
     /// When a physical chunk has been evicted.
     Deletion(ChunkStoreDiffDeletion),
+
+    /// Newly discovered entity/component columns in the schema.
+    ///
+    /// Also emitted when a component's `is_static` flag transitions from `false` to `true`.
+    /// Note: `has_data` does not influence the emission of `SchemaAddition` events.
+    SchemaAddition(ChunkStoreDiffSchemaAddition),
+}
+
+/// Describes newly added columns to the store schema.
+///
+/// This event is emitted when previously unseen entity/component pairs are
+/// discovered, either from a physical chunk addition or from an RRD manifest.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ChunkStoreDiffSchemaAddition {
+    /// Newly discovered entity/component pairs, grouped by entity.
+    pub new_columns: Vec<ChunkMeta>,
 }
 
 impl From<ChunkStoreDiffAddition> for ChunkStoreDiff {
@@ -158,6 +177,12 @@ impl From<ChunkStoreDiffDeletion> for ChunkStoreDiff {
     }
 }
 
+impl From<ChunkStoreDiffSchemaAddition> for ChunkStoreDiff {
+    fn from(value: ChunkStoreDiffSchemaAddition) -> Self {
+        Self::SchemaAddition(value)
+    }
+}
+
 impl ChunkStoreDiff {
     pub fn addition(
         chunk_before_processing: Arc<Chunk>,
@@ -191,6 +216,10 @@ impl ChunkStoreDiff {
         matches!(self, Self::Deletion(_))
     }
 
+    pub fn is_schema_addition(&self) -> bool {
+        matches!(self, Self::SchemaAddition(_))
+    }
+
     pub fn into_addition(self) -> Option<ChunkStoreDiffAddition> {
         match self {
             Self::Addition(addition) => Some(addition),
@@ -238,7 +267,7 @@ impl ChunkStoreDiff {
     pub fn delta(&self) -> i64 {
         match self {
             Self::Addition(_) => 1,
-            Self::VirtualAddition(_) => 0,
+            Self::VirtualAddition(_) | Self::SchemaAddition(_) => 0,
             Self::Deletion(_) => -1,
         }
     }
@@ -256,7 +285,7 @@ impl ChunkStoreDiff {
     pub fn delta_chunk(&self) -> Option<&Arc<Chunk>> {
         match self {
             Self::Addition(addition) => Some(addition.delta_chunk()),
-            Self::VirtualAddition(_) => None,
+            Self::VirtualAddition(_) | Self::SchemaAddition(_) => None,
             Self::Deletion(deletion) => Some(&deletion.chunk),
         }
     }
@@ -376,7 +405,7 @@ impl ChunkStoreDiffAddition {
                 descriptor: column.descriptor.clone(),
                 inner_arrow_datatype: Some(column.list_array.value_type()),
                 has_data: !column.list_array.values().is_empty(),
-                is_static_only: delta_chunk.is_static(),
+                is_static: delta_chunk.is_static(),
             })
             .collect();
 
@@ -435,16 +464,16 @@ impl ChunkStoreDiffVirtualAddition {
                         inner_arrow_datatype: Some(inner_arrow_datatype),
                         // These fields are filled in later in this function
                         has_data: false,
-                        is_static_only: false,
+                        is_static: false,
                     },
                 )
             })
             .collect();
 
-        /// Helper to track what's know about a component from the manifest's static/temporal maps.
+        /// Helper to track what's known about a component from the manifest's static/temporal maps.
         #[derive(Default)]
         struct VirtualComponentInfo {
-            has_temporal: bool,
+            is_static: bool,
 
             has_rows: bool,
         }
@@ -462,7 +491,7 @@ impl ChunkStoreDiffVirtualAddition {
                 entry.insert(
                     component,
                     VirtualComponentInfo {
-                        has_temporal: false,
+                        is_static: true,
                         has_rows: true,
                     },
                 );
@@ -480,7 +509,6 @@ impl ChunkStoreDiffVirtualAddition {
                     let has_rows = per_chunk.values().any(|e| e.num_rows > 0);
 
                     let existing = entry.entry(component).or_default();
-                    existing.has_temporal = true;
                     existing.has_rows |= has_rows;
                 }
             }
@@ -494,17 +522,17 @@ impl ChunkStoreDiffVirtualAddition {
                     .into_iter()
                     .map(|(component, info)| {
                         let has_data = info.has_rows;
-                        let is_static_only = !info.has_temporal;
+                        let is_static = info.is_static;
                         if let Some(meta) = component_schema_info.get(&component) {
                             ChunkComponentMeta {
                                 has_data,
-                                is_static_only,
+                                is_static,
                                 ..meta.clone()
                             }
                         } else {
                             ChunkComponentMeta {
                                 has_data,
-                                is_static_only,
+                                is_static,
                                 descriptor: re_sdk_types::ComponentDescriptor::partial(component),
                                 inner_arrow_datatype: None,
                             }
@@ -562,7 +590,7 @@ impl ChunkStoreDiffDeletion {
 
 #[cfg(test)]
 mod tests {
-    use std::collections::BTreeMap;
+    use std::collections::{BTreeMap, BTreeSet};
 
     use re_chunk::{RowId, TimelineName};
     use re_log_types::example_components::{MyColor, MyIndex, MyPoint, MyPoints};
@@ -612,7 +640,9 @@ mod tests {
 
             for event in events {
                 let delta = event.delta();
-                let delta_chunk = event.delta_chunk().unwrap();
+                let Some(delta_chunk) = event.delta_chunk() else {
+                    continue;
+                };
                 let delta_rows = delta * delta_chunk.num_rows() as i64;
 
                 for row_id in delta_chunk.row_ids() {
@@ -645,6 +675,18 @@ mod tests {
         }
     }
 
+    /// Helper to extract the set of new component descriptors from a [`ChunkStoreDiff::SchemaAddition`].
+    fn schema_addition_descriptors(event: &ChunkStoreEvent) -> BTreeSet<ComponentDescriptor> {
+        match &event.diff {
+            ChunkStoreDiff::SchemaAddition(sa) => sa
+                .new_columns
+                .iter()
+                .flat_map(|m| m.components.iter().map(|c| c.descriptor.clone()))
+                .collect(),
+            other => panic!("expected SchemaAddition, got {other:?}"),
+        }
+    }
+
     #[test]
     fn store_events() -> anyhow::Result<()> {
         let mut store = ChunkStore::new(
@@ -673,7 +715,18 @@ mod tests {
             )
             .build()?;
 
-        view.on_events(&store.insert_chunk(&Arc::new(chunk1))?);
+        let events = store.insert_chunk(&Arc::new(chunk1))?;
+
+        // chunk1 introduces entity_a with MyIndex — expect Addition + SchemaAddition.
+        assert_eq!(events.len(), 2);
+        assert!(events[0].is_addition());
+        assert!(events[1].is_schema_addition());
+        assert_eq!(
+            schema_addition_descriptors(&events[1]),
+            BTreeSet::from([MyIndex::partial_descriptor()]),
+        );
+
+        view.on_events(&events);
 
         similar_asserts::assert_eq!(
             GlobalCounts::new(
@@ -725,7 +778,18 @@ mod tests {
                 .build()?
         };
 
-        view.on_events(&store.insert_chunk(&Arc::new(chunk2))?);
+        let events = store.insert_chunk(&Arc::new(chunk2))?;
+
+        // chunk2 introduces entity_b with Points + Colors — expect Addition + SchemaAddition.
+        assert_eq!(events.len(), 2);
+        assert!(events[0].is_addition());
+        assert!(events[1].is_schema_addition());
+        assert_eq!(
+            schema_addition_descriptors(&events[1]),
+            BTreeSet::from([MyPoints::descriptor_points(), MyPoints::descriptor_colors(),]),
+        );
+
+        view.on_events(&events);
 
         similar_asserts::assert_eq!(
             GlobalCounts::new(
@@ -777,7 +841,21 @@ mod tests {
                 .build()?
         };
 
-        view.on_events(&store.insert_chunk(&Arc::new(chunk3))?);
+        let events = store.insert_chunk(&Arc::new(chunk3))?;
+
+        // chunk3 adds MyIndex to entity_b (new!) and re-uses Colors (not new, but transitions to static).
+        // Colors already existed on entity_b, but this is a static chunk so Colors gets an
+        // is_static transition (false → true). MyIndex is new on entity_b.
+        assert_eq!(events.len(), 2);
+        assert!(events[0].is_addition());
+        assert!(events[1].is_schema_addition());
+        assert_eq!(
+            schema_addition_descriptors(&events[1]),
+            BTreeSet::from([MyIndex::partial_descriptor(), MyPoints::descriptor_colors(),]),
+            "MyIndex is new on entity_b; Colors gets is_static transition"
+        );
+
+        view.on_events(&events);
 
         similar_asserts::assert_eq!(
             GlobalCounts::new(
@@ -811,6 +889,16 @@ mod tests {
         );
 
         let events = store.gc(&GarbageCollectionOptions::gc_everything()).0;
+
+        // GC should only produce Deletion events, never SchemaAddition.
+        for event in &events {
+            assert!(
+                event.is_deletion(),
+                "GC should only produce deletions, got: {:?}",
+                event.diff
+            );
+        }
+
         view.on_events(&events);
 
         similar_asserts::assert_eq!(

crates/store/re_chunk_store/src/lib.rs

@@ -45,7 +45,7 @@ pub use self::dataframe::{
 };
 pub use self::events::{
     ChunkComponentMeta, ChunkMeta, ChunkStoreDiff, ChunkStoreDiffAddition, ChunkStoreDiffDeletion,
-    ChunkStoreDiffVirtualAddition, ChunkStoreEvent,
+    ChunkStoreDiffSchemaAddition, ChunkStoreDiffVirtualAddition, ChunkStoreEvent,
 };
 pub use self::gc::{GarbageCollectionOptions, GarbageCollectionTarget};
 pub use self::lineage::{ChunkDirectLineage, ChunkDirectLineageReport};