support per-component row-id ranges and fix static writes accordingly

Author: teh-cmc

crates/re_chunk/src/chunk.rs

@@ -217,6 +217,64 @@ impl Chunk {
             })
             .collect()
     }
+
+    /// Computes the `RowId` range covered by each individual component column on each timeline.
+    ///
+    /// This is different from the `RowId` range covered by the [`Chunk`] as a whole because component
+    /// columns are potentially sparse.
+    ///
+    /// This is crucial for indexing and queries to work properly.
+    //
+    // TODO(cmc): This needs to be stored in chunk metadata and transported across IPC.
+    pub fn row_id_range_per_component(&self) -> BTreeMap<ComponentName, (RowId, RowId)> {
+        re_tracing::profile_function!();
+
+        let row_ids = self.row_ids().collect_vec();
+
+        if self.is_sorted() {
+            self.components
+                .iter()
+                .filter_map(|(component_name, list_array)| {
+                    let mut row_id_min = None;
+                    let mut row_id_max = None;
+
+                    for (i, &row_id) in row_ids.iter().enumerate() {
+                        if list_array.is_valid(i) {
+                            row_id_min = Some(row_id);
+                        }
+                    }
+                    for (i, &row_id) in row_ids.iter().enumerate().rev() {
+                        if list_array.is_valid(i) {
+                            row_id_max = Some(row_id);
+                        }
+                    }
+
+                    Some((*component_name, (row_id_min?, row_id_max?)))
+                })
+                .collect()
+        } else {
+            self.components
+                .iter()
+                .filter_map(|(component_name, list_array)| {
+                    let mut row_id_min = Some(RowId::MAX);
+                    let mut row_id_max = Some(RowId::ZERO);
+
+                    for (i, &row_id) in row_ids.iter().enumerate() {
+                        if list_array.is_valid(i) && Some(row_id) > row_id_min {
+                            row_id_min = Some(row_id);
+                        }
+                    }
+                    for (i, &row_id) in row_ids.iter().enumerate().rev() {
+                        if list_array.is_valid(i) && Some(row_id) < row_id_max {
+                            row_id_max = Some(row_id);
+                        }
+                    }
+
+                    Some((*component_name, (row_id_min?, row_id_max?)))
+                })
+                .collect()
+        }
+    }
 }
 
 // ---

crates/re_chunk_store/src/writes.rs

@@ -49,18 +49,17 @@ impl ChunkStore {
 
         self.insert_id += 1;
 
-        let row_id_min = row_id_range.0;
-        let row_id_max = row_id_range.1;
-
         self.chunks_per_chunk_id.insert(chunk.id(), chunk.clone());
         self.chunk_ids_per_min_row_id
-            .entry(row_id_min)
+            .entry(row_id_range.0)
             .or_default()
             .push(chunk.id());
 
         if chunk.is_static() {
             // Static data: make sure to keep the most recent chunk available for each component column.
 
+            let row_id_range_per_component = chunk.row_id_range_per_component();
+
             for (&component_name, list_array) in chunk.components() {
                 let is_empty = list_array
                     .validity()
@@ -69,6 +68,12 @@ impl ChunkStore {
                     continue;
                 }
 
+                let Some((_row_id_min, row_id_max)) =
+                    row_id_range_per_component.get(&component_name)
+                else {
+                    continue;
+                };
+
                 self.static_chunk_ids_per_entity
                     .entry(chunk.entity_path().clone())
                     .or_default()
@@ -81,11 +86,12 @@ impl ChunkStore {
                             RowId::ZERO,
                             |chunk| {
                                 chunk
-                                    .row_id_range()
-                                    .map_or(RowId::ZERO, |(_, row_id_max)| row_id_max)
+                                    .row_id_range_per_component()
+                                    .get(&component_name)
+                                    .map_or(RowId::ZERO, |(_, row_id_max)| *row_id_max)
                             },
                         );
-                        if row_id_max > cur_row_id_max {
+                        if *row_id_max > cur_row_id_max {
                             *cur_chunk_id = chunk.id();
                         }
                     })