re_datastore: component chunks & streamlining batches

crates/re_arrow_store/Cargo.toml

@@ -39,6 +39,12 @@ thiserror.workspace = true
 criterion = "0.4"
 itertools = "0.10"
 mimalloc = "0.1"
+polars-core = { workspace = true, features = [
+  "dtype-date",
+  "dtype-time",
+  "dtype-struct",
+  "fmt",
+] }
 tracing-subscriber = "0.3"
 
 [lib]

crates/re_arrow_store/benches/data_store.rs

@@ -1,7 +1,7 @@
 #[global_allocator]
 static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
-use arrow2::array::{Array, ListArray, StructArray};
+use arrow2::array::{Array, StructArray};
 use criterion::{criterion_group, criterion_main, Criterion};
 
 use re_arrow_store::{DataStore, TimeQuery, TimelineQuery};
@@ -89,9 +89,7 @@ fn query_messages(store: &mut DataStore) -> Box<dyn Array> {
     let mut results = store.get(&[component], &row_indices);
 
     let row = std::mem::take(&mut results[0]).unwrap();
-    let list = row.as_any().downcast_ref::<ListArray<i32>>().unwrap();
-    let rects = list.value(0);
-    let rects = rects.as_any().downcast_ref::<StructArray>().unwrap();
+    let rects = row.as_any().downcast_ref::<StructArray>().unwrap();
     assert_eq!(NUM_RECTS as usize, rects.len());
 
     row

crates/re_arrow_store/src/store.rs

@@ -836,14 +836,50 @@ pub struct ComponentBucket {
     /// The offset of this bucket in the global table.
     pub(crate) row_offset: RowIndex,
 
+    /// Has this bucket been retired yet?
+    ///
+    /// At any given moment, all buckets except the currently active one have to be retired.
+    pub(crate) retired: bool,
+
     /// The time ranges (plural!) covered by this bucket.
     /// Buckets are never sorted over time, so these time ranges can grow arbitrarily large.
     ///
     /// These are only used for garbage collection.
     pub(crate) time_ranges: HashMap<Timeline, TimeRange>,
 
-    /// All the data for this bucket. This is a single column!
-    pub(crate) data: Box<dyn Array>,
+    /// All the data for this bucket: many rows of a single column.
+    ///
+    /// Each chunk is a list of list of components, i.e. `ListArray<ListArray<StructArray>>`:
+    /// - the first list layer corresponds to the different rows,
+    /// - the second list layer corresponds to the different instances within a single row,
+    /// - and finally the struct layer is the component itself.
+    /// E.g.:
+    /// ```ignore
+    /// [
+    ///   [{x: 8.687487, y: 1.9590926}, {x: 2.0559108, y: 0.1494348}, {x: 7.09219, y: 0.9616637}],
+    ///   [{x: 7.158843, y: 0.68897724}, {x: 8.934421, y: 2.8420508}],
+    /// ]
+    /// ```
+    ///
+    /// During the active lifespan of the bucket, this can contain any number of chunks,
+    /// depending on how the data was inserted (e.g. single insertions vs. batches).
+    /// All of these chunks get compacted into one contiguous array when the bucket is retired,
+    /// i.e. when the bucket is full and a new one is created.
+    ///
+    /// Note that, as of today, we do not actually support batched insertion nor do we support
+    /// chunks of non-unit length (batches are inserted on a per-row basis internally).
+    /// As a result, chunks always contain one and only one row's worth of data, at least until
+    /// the bucket is retired and compacted.
+    /// See also #589.
+    pub(crate) chunks: Vec<Box<dyn Array>>,
+
+    /// The total number of rows present in this bucket, across all chunks.
+    pub(crate) total_rows: u64,
+    /// The size of this bucket in bytes, across all chunks.
+    ///
+    /// Accurately computing the size of arrow arrays is surprisingly costly, which is why we
+    /// cache this.
+    pub(crate) total_size_bytes: u64,
 }
 
 impl std::fmt::Display for ComponentBucket {
@@ -862,14 +898,16 @@ impl std::fmt::Display for ComponentBucket {
             // - all buckets that follow are lazily instantiated when data get inserted
             //
             // TODO(#439): is that still true with deletion?
+            // TODO(#589): support for non-unit-length chunks
             self.row_offset.as_u64()
                 + self
-                    .data
+                    .chunks
                     .len()
                     .checked_sub(1)
                     .expect("buckets are never empty") as u64,
         ))?;
 
+        f.write_fmt(format_args!("retired: {}\n", self.retired))?;
         f.write_str("time ranges:\n")?;
         for (timeline, time_range) in &self.time_ranges {
             f.write_fmt(format_args!(
@@ -878,7 +916,12 @@ impl std::fmt::Display for ComponentBucket {
             ))?;
         }
 
-        let chunk = Chunk::new(vec![self.data()]);
+        let rows = {
+            use arrow2::compute::concatenate::concatenate;
+            let chunks = self.chunks.iter().map(|chunk| &**chunk).collect::<Vec<_>>();
+            vec![concatenate(&chunks).unwrap()]
+        };
+        let chunk = Chunk::new(rows);
         f.write_str(&arrow2::io::print::write(&[chunk], &[self.name.as_str()]))?;
 
         Ok(())
@@ -888,12 +931,12 @@ impl std::fmt::Display for ComponentBucket {
 impl ComponentBucket {
     /// Returns the number of rows stored across this bucket.
     pub fn total_rows(&self) -> u64 {
-        self.data.len() as u64
+        self.total_rows
     }
 
     /// Returns the size of the data stored across this bucket, in bytes.
     pub fn total_size_bytes(&self) -> u64 {
-        arrow2::compute::aggregate::estimated_bytes_size(&*self.data) as u64
+        self.total_size_bytes
     }
 }
 

crates/re_arrow_store/src/store_read.rs

@@ -1,7 +1,7 @@
 use std::sync::atomic::Ordering;
 
 use arrow2::{
-    array::{Array, Int64Array, MutableArray, UInt64Array, UInt64Vec},
+    array::{Array, Int64Array, ListArray, MutableArray, UInt64Array, UInt64Vec},
     datatypes::{DataType, TimeUnit},
 };
 
@@ -110,8 +110,7 @@ impl DataStore {
     ///             .map(|(&component, col)| Series::try_from((component, col)).unwrap())
     ///             .collect();
     ///
-    ///         let df = DataFrame::new(series).unwrap();
-    ///         df.explode(df.get_column_names()).unwrap()
+    ///         DataFrame::new(series).unwrap()
     ///     };
     ///
     ///     df
@@ -176,8 +175,8 @@ impl DataStore {
 
     /// Retrieves the data associated with a list of `components` at the specified `indices`.
     ///
-    /// If the associated data is found, it will be written to returned array at the appropriate
-    /// index, or `None` otherwise.
+    /// If the associated data is found, it will be written into the returned array at the
+    /// appropriate index, or `None` otherwise.
     ///
     /// `row_indices` takes a list of options so that one can easily re-use the results obtained
     /// from [`Self::query`].
@@ -575,25 +574,38 @@ impl ComponentTable {
 }
 
 impl ComponentBucket {
-    /// Get this `ComponentBucket`s debug name
+    /// Returns the name of the component stored in this bucket.
     #[allow(dead_code)]
     pub fn name(&self) -> &str {
         &self.name
     }
 
-    // Panics on out-of-bounds
+    /// Returns a shallow clone of the row data present at the given `row_idx`.
     pub fn get(&self, row_idx: RowIndex) -> Box<dyn Array> {
         let row_idx = row_idx.as_u64() - self.row_offset.as_u64();
-        self.data.slice(row_idx as usize, 1)
+        // This has to be safe to unwrap, otherwise it would never have made it past insertion.
+        if self.retired {
+            self.chunks[0]
+                .as_any()
+                .downcast_ref::<ListArray<i32>>()
+                .unwrap()
+                .value(row_idx as _)
+        } else {
+            self.chunks[row_idx as usize]
+                .as_any()
+                .downcast_ref::<ListArray<i32>>()
+                .unwrap()
+                .value(0)
+        }
     }
 
-    /// Returns the entire data Array in this component
-    pub fn data(&self) -> Box<dyn Array> {
-        // shallow copy
-        self.data.clone()
+    /// Returns a shallow clone of all the chunks in this bucket.
+    #[allow(dead_code)]
+    pub fn data(&self) -> Vec<Box<dyn Array>> {
+        self.chunks.clone() // shallow
     }
 
-    /// Return an iterator over the time ranges in this bucket
+    /// Return an iterator over the time ranges in this bucket.
     #[allow(dead_code)]
     pub fn iter_time_ranges(&self) -> impl Iterator<Item = (&Timeline, &TimeRange)> {
         self.time_ranges.iter()