feldera · ryzhyk · Aug 12, 2026 · Aug 11, 2026 · Aug 11, 2026 · Aug 11, 2026
diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs
@@ -100,6 +100,7 @@ use super::dbsp_handle::Mode;
 /// Label name used to store operator's persistent id,
 /// i.e., id stable across circuit modifications.
 const LABEL_PERSISTENT_OPERATOR_ID: &str = "persistent_id";
+use rkyv::with::CopyOptimize;
 
 /// Value stored in the stream.
 struct StreamValue<D> {
@@ -7308,6 +7309,8 @@ where
 #[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)]
 #[archive_attr(derive(rkyv::CheckBytes))]
 struct CommittedClock {
+    // Copied in bulk, like the larger state blobs; see `CommittedZ1`.
+    #[with(CopyOptimize)]
     time: Vec<u8>,
 }
 

diff --git a/crates/dbsp/src/operator/dynamic/communication/shard.rs b/crates/dbsp/src/operator/dynamic/communication/shard.rs
@@ -6,7 +6,7 @@
 
 use feldera_storage::fbuf::{FBuf, FBufSerializer};
 use itertools::Itertools;
-use rkyv::{archived_root, ser::Serializer as _};
+use rkyv::archived_root;
 
 use crate::{
     Circuit, Runtime, Stream,
@@ -19,7 +19,8 @@ use crate::{
     operator::communication::{ExchangeActivity, Mailbox, new_exchange_operators},
     storage::file::SerializerInner,
     trace::{
-        Batch, BatchReader, Builder, IndexedWSetSerializer, deserialize_indexed_wset, merge_batches,
+        Batch, BatchReader, Builder, IndexedWSetSerializer, deserialize_indexed_wset,
+        merge_batches, serialize_offsets,
     },
 };
 
@@ -504,7 +505,7 @@ impl PairsSerializer {
     pub fn done(mut self, serializer: &mut SerializerInner) -> FBuf {
         serializer
             .with(FBufSerializer::new(&mut self.fbuf), |s| {
-                s.serialize_value(&self.offsets)
+                serialize_offsets(&self.offsets, s)
             })
             .unwrap();
         self.fbuf

diff --git a/crates/dbsp/src/operator/dynamic/time_series/window.rs b/crates/dbsp/src/operator/dynamic/time_series/window.rs
@@ -85,6 +85,11 @@ where
 }
 
 /// A window that is serialized to a file.
+//
+// The bounds stay on rkyv's per-element path, unlike the state blobs in
+// `CommittedZ1`: `CopyOptimize` applies to a `Vec` field, not through an
+// `Option` of a tuple, and a pair of serialized keys is too small to be worth
+// reshaping the struct for.
 #[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)]
 #[archive_attr(derive(rkyv::CheckBytes))]
 struct CommittedWindow {

diff --git a/crates/dbsp/src/operator/transaction_z1.rs b/crates/dbsp/src/operator/transaction_z1.rs
@@ -2,6 +2,7 @@ use std::{borrow::Cow, sync::Arc};
 
 use feldera_storage::{FileCommitter, StoragePath};
 use rkyv::bytecheck;
+use rkyv::with::CopyOptimize;
 use size_of::SizeOf;
 
 use crate::{
@@ -45,10 +46,15 @@ pub struct TransactionZ1<T> {
     new_value: T,
 }
 
+// The blob is a whole serialized state, so it is copied in bulk: `Vec<u8>`'s own
+// rkyv impl resolves it byte by byte, one write call each. `CopyOptimize` keeps
+// the archived layout, an `ArchivedVec<u8>`, so existing checkpoints still read.
 #[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)]
 #[archive_attr(derive(rkyv::CheckBytes))]
 pub struct CommittedTransactionZ1 {
+    #[with(CopyOptimize)]
     old_value: Vec<u8>,
+    #[with(CopyOptimize)]
     new_value: Vec<u8>,
 }
 

diff --git a/crates/dbsp/src/operator/z1.rs b/crates/dbsp/src/operator/z1.rs
@@ -28,6 +28,7 @@ use super::require_persistent_id;
 
 circuit_cache_key!(DelayedId<C, D>(StreamId => Stream<C, D>));
 circuit_cache_key!(NestedDelayedId<C, D>(StreamId => Stream<C, D>));
+use rkyv::with::CopyOptimize;
 
 /// Like [`FeedbackConnector`] but specialized for [`Z1`] feedback operator.
 ///
@@ -226,9 +227,13 @@ pub struct Z1<T> {
     values: T,
 }
 
+// The blob is a whole serialized state, so it is copied in bulk: `Vec<u8>`'s own
+// rkyv impl resolves it byte by byte, one write call each. `CopyOptimize` keeps
+// the archived layout, an `ArchivedVec<u8>`, so existing checkpoints still read.
 #[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)]
 #[archive_attr(derive(rkyv::CheckBytes))]
 pub struct CommittedZ1 {
+    #[with(CopyOptimize)]
     values: Vec<u8>,
 }
 
@@ -652,6 +657,34 @@ mod test {
         }
     }
 
+    /// The same struct without `#[with(CopyOptimize)]`, as a layout reference.
+    #[derive(rkyv::Serialize, rkyv::Archive)]
+    struct PerElementZ1 {
+        values: Vec<u8>,
+    }
+
+    /// Copying the state blob in bulk must leave the archived bytes exactly as
+    /// the per-element path wrote them, or a checkpoint taken by an earlier
+    /// build would no longer restore.
+    #[test]
+    fn bulk_copy_keeps_the_committed_layout() {
+        for length in [0, 1, 4096] {
+            let values: Vec<u8> = (0..length).map(|i| (i * 13 + 5) as u8).collect();
+            let bulk = crate::storage::file::to_bytes(&CommittedZ1 {
+                values: values.clone(),
+            })
+            .unwrap();
+            let per_element = crate::storage::file::to_bytes(&PerElementZ1 {
+                values: values.clone(),
+            })
+            .unwrap();
+            assert_eq!(bulk.as_slice(), per_element.as_slice(), "length {length}");
+
+            let restored: CommittedZ1 = crate::trace::aligned_deserialize(&bulk[..]);
+            assert_eq!(restored.values, values, "length {length}");
+        }
+    }
+
     #[tokio::test]
     async fn z1_test() {
         let mut z1 = Z1::new(0);

diff --git a/crates/dbsp/src/trace.rs b/crates/dbsp/src/trace.rs
@@ -44,7 +44,7 @@ use enum_map::Enum;
 use feldera_storage::fbuf::FBufSerializer;
 use feldera_storage::{FileCommitter, FileReader, StoragePath};
 use rand::{Rng, thread_rng};
-use rkyv::ser::Serializer as _;
+use rkyv::with::{CopyOptimize, With};
 use size_of::SizeOf;
 use std::any::TypeId;
 use std::future::Future;
@@ -1466,7 +1466,7 @@ where
             offsets.push(cursor.weight().serialize(s)?);
             cursor.step_key();
         }
-        s.serialize_value(&offsets)
+        serialize_offsets(&offsets, s)
     })
     .into_vec()
 }
@@ -1495,6 +1495,32 @@ where
 /// Separator that identifies the end of values for a key.
 const SEPARATOR: u64 = u64::MAX;
 
+/// A bulk copy of an offset table is only valid where a `usize` archives as
+/// itself. rkyv archives integers in native byte order, and the `size_64`
+/// feature makes the archived width 8, so this holds on every 64-bit target.
+const _: () = assert!(
+    size_of::<usize>() == size_of::<rkyv::Archived<usize>>(),
+    "serialize_offsets needs usize and its archived form to have one representation"
+);
+
+/// Serializes an offset table as the root value, copying it in one bulk write.
+///
+/// A `Vec<usize>` resolves element by element, one `write` call per offset,
+/// which measured 25x slower than the copy on a table of 3M offsets.
+/// `CopyOptimize` leaves the archived layout alone, so the readers still take
+/// `archived_root::<Vec<usize>>` and existing checkpoints stay readable.
+// The wrapper is implemented for `Vec<T>`, so a slice will not do here.
+#[allow(clippy::ptr_arg)]
+pub(crate) fn serialize_offsets<S>(
+    offsets: &Vec<usize>,
+    serializer: &mut S,
+) -> Result<usize, S::Error>
+where
+    S: rkyv::ser::Serializer + ?Sized,
+{
+    serializer.serialize_value(With::<_, CopyOptimize>::cast(offsets))
+}
+
 #[cfg(debug_assertions)]
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 enum State {
@@ -1594,7 +1620,7 @@ impl IndexedWSetSerializer {
             self.offsets[0] = self.n_keys;
             self.offsets[1] = self.n_values;
             serializer_inner.with(FBufSerializer::new(&mut self.fbuf), |s| {
-                s.serialize_value(&self.offsets).unwrap()
+                serialize_offsets(&self.offsets, s).unwrap()
             });
         }
         self.fbuf
@@ -1668,12 +1694,16 @@ where
 #[cfg(test)]
 mod serialize_test {
     use crate::{
-        DynZWeight, OrdIndexedZSet,
-        algebra::OrdIndexedZSet as DynOrdIndexedZSet,
+        DynZWeight, OrdIndexedZSet, OrdZSet,
+        algebra::{OrdIndexedZSet as DynOrdIndexedZSet, OrdZSet as DynOrdZSet},
         dynamic::DynData,
         indexed_zset,
         storage::file::SerializerInner,
-        trace::{BatchReader, deserialize_indexed_wset, serialize_indexed_wset},
+        trace::{
+            BatchReader, SEPARATOR, deserialize_indexed_wset, deserialize_wset,
+            serialize_indexed_wset, serialize_offsets, serialize_wset,
+        },
+        zset,
     };
 
     #[test]
@@ -1732,4 +1762,44 @@ mod serialize_test {
             assert_eq!(&*test, &deserialized);
         }
     }
+
+    #[test]
+    fn test_serialize_wset() {
+        let test1: OrdZSet<u64> = zset! {};
+        let test2 = zset! { 1u64 => 1 };
+        let test3 = zset! { 1u64 => 1, 2 => 2, 3 => -3 };
+
+        for test in [test1, test2, test3] {
+            let serialized = serialize_wset(&*test);
+            let deserialized = deserialize_wset::<DynOrdZSet<DynData>, DynData, DynZWeight>(
+                &test.factories(),
+                &serialized,
+            );
+
+            assert_eq!(&*test, &deserialized);
+        }
+    }
+
+    /// Every reader of a bulk-copied offset table takes it as
+    /// `archived_root::<Vec<usize>>`, so pin that the copy lands the offsets
+    /// where such a reader finds them, at the value each writer recorded.
+    #[test]
+    fn serialize_offsets_reads_back_as_a_vec() {
+        for table in [
+            vec![],
+            vec![0usize],
+            vec![0, 8, 24, SEPARATOR as usize, usize::MAX, 1 << 40],
+            (0..10_000).collect(),
+        ] {
+            let bytes = SerializerInner::to_fbuf_with_thread_local(|serializer| {
+                serialize_offsets(&table, serializer)
+            });
+            // SAFETY: the bytes come from serializing `table` just above.
+            let archived = unsafe { rkyv::archived_root::<Vec<usize>>(&bytes) };
+            assert_eq!(archived.len(), table.len());
+            for (index, offset) in table.iter().enumerate() {
+                assert_eq!(archived[index] as usize, *offset, "offset {index}");
+            }
+        }
+    }
 }
diff --git a/crates/sqllib/Cargo.toml b/crates/sqllib/Cargo.toml
@@ -59,3 +59,7 @@ derive_more = { version = "1.0", features = ["not"] }
 proptest = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 size-of = { workspace = true }
+
+[[bench]]
+name = "flat_variant_serialize"
+harness = false