Add migration tool for legacy .rrd files

Cargo.lock

@@ -7205,7 +7205,6 @@ dependencies = [
  "re_tuid",
  "re_types_core",
  "serde",
- "serde_bytes",
  "similar-asserts",
  "static_assertions",
  "thiserror 1.0.65",
@@ -8415,10 +8414,13 @@ version = "0.24.0-alpha.1+dev"
 dependencies = [
  "anyhow",
  "arrow",
+ "camino",
  "clap",
  "crossbeam",
  "document-features",
  "env_filter",
+ "indexmap 2.8.0",
+ "indicatif",
  "itertools 0.14.0",
  "log",
  "puffin",
@@ -9068,15 +9070,6 @@ dependencies = [
  "wasm-bindgen",
 ]
 
-[[package]]
-name = "serde_bytes"
-version = "0.11.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "387cc504cb06bb40a96c8e04e951fe01854cf6bc921053c954e4a606d9675c6a"
-dependencies = [
- "serde",
-]
-
 [[package]]
 name = "serde_derive"
 version = "1.0.219"

Cargo.toml

@@ -287,7 +287,6 @@ rustdoc-json = "0.9.4"
 rustdoc-types = "0.35.0"
 seq-macro = "0.3"
 serde = { version = "1", features = ["derive"] }
-serde_bytes = "0.11"
 serde_json = { version = "1", default-features = false, features = ["std"] }
 serde-wasm-bindgen = "0.6.5"
 serde_yaml = { version = "0.9.21", default-features = false }

crates/store/re_chunk/src/chunk.rs

@@ -32,7 +32,7 @@ pub enum ChunkError {
     #[error("Detected malformed Chunk: {reason}")]
     Malformed { reason: String },
 
-    #[error(transparent)]
+    #[error("Arrow: {0}")]
     Arrow(#[from] arrow::error::ArrowError),
 
     #[error("{kind} index out of bounds: {index} (len={len})")]

crates/store/re_log_types/Cargo.toml

@@ -31,7 +31,6 @@ default = []
 ## Enable (de)serialization using serde.
 serde = [
   "dep:serde",
-  "dep:serde_bytes",
   "fixed/serde",
   "re_build_info/serde",
   "re_string_interner/serde",
@@ -79,7 +78,6 @@ web-time.workspace = true
 
 # Optional dependencies:
 serde = { workspace = true, optional = true, features = ["derive", "rc"] }
-serde_bytes = { workspace = true, optional = true }
 
 [dev-dependencies]
 criterion.workspace = true

crates/store/re_log_types/src/arrow_msg.rs

@@ -91,99 +91,3 @@ impl Drop for ArrowMsg {
         }
     }
 }
-
-#[cfg(feature = "serde")]
-impl serde::Serialize for ArrowMsg {
-    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: serde::Serializer,
-    {
-        re_tracing::profile_scope!("ArrowMsg::serialize");
-        use serde::ser::SerializeTuple as _;
-
-        let mut ipc_bytes = Vec::<u8>::new();
-
-        let mut writer =
-            arrow::ipc::writer::StreamWriter::try_new(&mut ipc_bytes, self.batch.schema_ref())
-                .map_err(|err| serde::ser::Error::custom(err.to_string()))?;
-        writer
-            .write(&self.batch)
-            .map_err(|err| serde::ser::Error::custom(err.to_string()))?;
-        writer
-            .finish()
-            .map_err(|err| serde::ser::Error::custom(err.to_string()))?;
-
-        let mut inner = serializer.serialize_tuple(3)?;
-        inner.serialize_element(&self.chunk_id)?;
-        inner.serialize_element(&self.timepoint_max)?;
-        inner.serialize_element(&serde_bytes::ByteBuf::from(ipc_bytes))?;
-        inner.end()
-    }
-}
-
-#[cfg(feature = "serde")]
-impl<'de> serde::Deserialize<'de> for ArrowMsg {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        struct FieldVisitor;
-
-        impl<'de> serde::de::Visitor<'de> for FieldVisitor {
-            type Value = ArrowMsg;
-
-            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-                formatter.write_str("(table_id, timepoint, buf)")
-            }
-
-            fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
-            where
-                A: serde::de::SeqAccess<'de>,
-            {
-                re_tracing::profile_scope!("ArrowMsg::deserialize");
-
-                let table_id: Option<re_tuid::Tuid> = seq.next_element()?;
-                let timepoint_max: Option<TimePoint> = seq.next_element()?;
-                let ipc_bytes: Option<serde_bytes::ByteBuf> = seq.next_element()?;
-
-                if let (Some(chunk_id), Some(timepoint_max), Some(buf)) =
-                    (table_id, timepoint_max, ipc_bytes)
-                {
-                    use arrow::ipc::reader::StreamReader;
-
-                    let stream = StreamReader::try_new(std::io::Cursor::new(buf), None)
-                        .map_err(|err| serde::de::Error::custom(format!("Arrow error: {err}")))?;
-                    let batches: Result<Vec<_>, _> = stream.collect();
-
-                    let batches = batches
-                        .map_err(|err| serde::de::Error::custom(format!("Arrow error: {err}")))?;
-
-                    if batches.is_empty() {
-                        return Err(serde::de::Error::custom("No RecordBatch in stream"));
-                    }
-                    if batches.len() > 1 {
-                        return Err(serde::de::Error::custom(format!(
-                            "Found {} batches in stream - expected just one.",
-                            batches.len()
-                        )));
-                    }
-                    #[allow(clippy::unwrap_used)] // is_empty check above
-                    let batch = batches.into_iter().next().unwrap();
-
-                    Ok(ArrowMsg {
-                        chunk_id,
-                        timepoint_max,
-                        batch,
-                        on_release: None,
-                    })
-                } else {
-                    Err(serde::de::Error::custom(
-                        "Expected (table_id, timepoint, buf)",
-                    ))
-                }
-            }
-        }
-
-        deserializer.deserialize_tuple(3, FieldVisitor)
-    }
-}

crates/store/re_sorbet/src/index_column_descriptor.rs

@@ -137,7 +137,7 @@ impl TryFrom<&ArrowField> for IndexColumnDescriptor {
         let name = if let Some(name) = field.metadata().get("rerun.index_name") {
             name.to_owned()
         } else {
-            re_log::warn_once!(
+            re_log::debug_once!(
                 "Timeline '{}' is missing 'rerun.index_name' metadata. Falling back on field/column name",
                 field.name()
             );

crates/store/re_sorbet/src/lib.rs

@@ -46,7 +46,6 @@ pub use self::{
         ArrowBatchMetadata, ArrowFieldMetadata, MetadataExt, MissingFieldMetadata,
         MissingMetadataKey,
     },
-    migration::migrate_record_batch,
     row_id_column_descriptor::{RowIdColumnDescriptor, WrongDatatypeError},
     selectors::{
         ColumnSelector, ColumnSelectorParseError, ComponentColumnSelector, TimeColumnSelector,

crates/store/re_sorbet/src/migration.rs

@@ -1,12 +1,95 @@
 //! Handles migrating old `re_types` to new ones.
 //!
 //!
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, sync::Arc};
 
 use arrow::{
-    array::{ArrayRef as ArrowArrayRef, RecordBatch as ArrowRecordBatch, RecordBatchOptions},
+    array::{
+        ArrayRef as ArrowArrayRef, AsArray as _, RecordBatch as ArrowRecordBatch,
+        RecordBatchOptions,
+    },
     datatypes::{Field as ArrowField, FieldRef as ArrowFieldRef, Schema as ArrowSchema},
 };
+use itertools::Itertools as _;
+use re_log::ResultExt as _;
+use re_tuid::Tuid;
+use re_types_core::{arrow_helpers::as_array_ref, Loggable as _};
+
+use crate::ColumnKind;
+
+/// Migrate TUID:s with the pre-0.23 encoding.
+pub fn migrate_tuids(batch: &ArrowRecordBatch) -> ArrowRecordBatch {
+    re_tracing::profile_function!();
+
+    let num_columns = batch.num_columns();
+    let mut fields: Vec<ArrowFieldRef> = Vec::with_capacity(num_columns);
+    let mut columns: Vec<ArrowArrayRef> = Vec::with_capacity(num_columns);
+
+    for (field, array) in itertools::izip!(batch.schema().fields(), batch.columns()) {
+        let (mut field, mut array) = (field.clone(), array.clone());
+
+        let is_tuid = field.extension_type_name() == Some("rerun.datatypes.TUID")
+            || field.name() == "rerun.controls.RowId";
+        if is_tuid {
+            (field, array) = migrate_tuid_column(field, array);
+        }
+
+        fields.push(field);
+        columns.push(array);
+    }
+
+    let schema = Arc::new(ArrowSchema::new_with_metadata(
+        fields,
+        batch.schema().metadata.clone(),
+    ));
+
+    ArrowRecordBatch::try_new_with_options(
+        schema.clone(),
+        columns,
+        &RecordBatchOptions::default().with_row_count(Some(batch.num_rows())),
+    )
+    .ok_or_log_error()
+    .unwrap_or_else(|| ArrowRecordBatch::new_empty(schema))
+}
+
+/// Migrate TUID:s with the pre-0.23 encoding.
+fn migrate_tuid_column(
+    field: ArrowFieldRef,
+    array: ArrowArrayRef,
+) -> (ArrowFieldRef, ArrowArrayRef) {
+    re_tracing::profile_function!();
+
+    if let Some(struct_array) = array.as_struct_opt() {
+        // Maybe legacy struct (from Rerun 0.22 or earlier):
+        let [nanos, counters] = struct_array.columns() else {
+            return (field, array);
+        };
+
+        let Some(nanos) = nanos.as_primitive_opt::<arrow::datatypes::UInt64Type>() else {
+            return (field, array);
+        };
+
+        let Some(counters) = counters.as_primitive_opt::<arrow::datatypes::UInt64Type>() else {
+            return (field, array);
+        };
+
+        re_tracing::profile_function!();
+
+        let tuids: Vec<Tuid> = itertools::izip!(nanos.values(), counters.values())
+            .map(|(&nanos, &inc)| Tuid::from_nanos_and_inc(nanos, inc))
+            .collect();
+
+        let new_field = ArrowField::new(field.name(), Tuid::arrow_datatype(), false)
+            .with_metadata(field.metadata().clone());
+        let new_array = re_types_core::tuids_to_arrow(&tuids);
+
+        re_log::debug_once!("Migrated legacy TUID encoding of column {}", field.name());
+
+        (new_field.into(), as_array_ref(new_array))
+    } else {
+        (field, array)
+    }
+}
 
 /// Migrate old renamed types to new types.
 pub fn migrate_record_batch(batch: &ArrowRecordBatch) -> ArrowRecordBatch {
@@ -82,12 +165,65 @@ pub fn migrate_record_batch(batch: &ArrowRecordBatch) -> ArrowRecordBatch {
         columns.push(array.clone());
     }
 
-    let schema = ArrowSchema::new_with_metadata(fields, batch.schema().metadata.clone());
+    let schema = Arc::new(ArrowSchema::new_with_metadata(
+        fields,
+        batch.schema().metadata.clone(),
+    ));
 
     ArrowRecordBatch::try_new_with_options(
-        schema.into(),
+        schema.clone(),
         columns,
         &RecordBatchOptions::default().with_row_count(Some(batch.num_rows())),
     )
-    .expect("Can't fail")
+    .ok_or_log_error()
+    .unwrap_or_else(|| ArrowRecordBatch::new_empty(schema))
+}
+
+/// Put row-id first, then time columns, and last data columns.
+pub fn reorder_columns(batch: &ArrowRecordBatch) -> ArrowRecordBatch {
+    re_tracing::profile_function!();
+
+    let mut row_ids = vec![];
+    let mut indices = vec![];
+    let mut components = vec![];
+
+    for (field, array) in itertools::izip!(batch.schema().fields(), batch.columns()) {
+        let field = field.clone();
+        let array = array.clone();
+        let column_kind = ColumnKind::try_from(field.as_ref()).unwrap_or(ColumnKind::Component);
+        match column_kind {
+            ColumnKind::RowId => row_ids.push((field, array)),
+            ColumnKind::Index => indices.push((field, array)),
+            ColumnKind::Component => components.push((field, array)),
+        }
+    }
+
+    let (fields, arrays): (Vec<ArrowFieldRef>, Vec<ArrowArrayRef>) =
+        itertools::chain!(row_ids, indices, components).unzip();
+
+    let schema = Arc::new(ArrowSchema::new_with_metadata(
+        fields,
+        batch.schema().metadata.clone(),
+    ));
+
+    if schema.fields() != batch.schema().fields() {
+        re_log::debug!(
+            "Reordered columns. Before: {:?}, after: {:?}",
+            batch
+                .schema()
+                .fields()
+                .iter()
+                .map(|f| f.name())
+                .collect_vec(),
+            schema.fields().iter().map(|f| f.name()).collect_vec()
+        );
+    }
+
+    ArrowRecordBatch::try_new_with_options(
+        schema.clone(),
+        arrays,
+        &RecordBatchOptions::default().with_row_count(Some(batch.num_rows())),
+    )
+    .ok_or_log_error()
+    .unwrap_or_else(|| ArrowRecordBatch::new_empty(schema))
 }