diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs index 7f930670e5a..81570ec1ddf 100644 --- 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 { @@ -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, } diff --git a/crates/dbsp/src/operator/dynamic/communication/shard.rs b/crates/dbsp/src/operator/dynamic/communication/shard.rs index f1e212d6f71..fe263982c22 100644 --- 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 index fd507700ecf..982bc9dcc3b 100644 --- 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 index c81a91ae3d3..63f97b81a82 100644 --- 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 { new_value: T, } +// The blob is a whole serialized state, so it is copied in bulk: `Vec`'s own +// rkyv impl resolves it byte by byte, one write call each. `CopyOptimize` keeps +// the archived layout, an `ArchivedVec`, 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, + #[with(CopyOptimize)] new_value: Vec, } diff --git a/crates/dbsp/src/operator/z1.rs b/crates/dbsp/src/operator/z1.rs index ff9b436d70e..94e6fa9e5a6 100644 --- 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(StreamId => Stream)); circuit_cache_key!(NestedDelayedId(StreamId => Stream)); +use rkyv::with::CopyOptimize; /// Like [`FeedbackConnector`] but specialized for [`Z1`] feedback operator. /// @@ -226,9 +227,13 @@ pub struct Z1 { values: T, } +// The blob is a whole serialized state, so it is copied in bulk: `Vec`'s own +// rkyv impl resolves it byte by byte, one write call each. `CopyOptimize` keeps +// the archived layout, an `ArchivedVec`, so existing checkpoints still read. #[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)] #[archive_attr(derive(rkyv::CheckBytes))] pub struct CommittedZ1 { + #[with(CopyOptimize)] values: Vec, } @@ -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, + } + + /// 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 = (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 index 53540573224..478daa07922 100644 --- 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::() == size_of::>(), + "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` 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::>` and existing checkpoints stay readable. +// The wrapper is implemented for `Vec`, so a slice will not do here. +#[allow(clippy::ptr_arg)] +pub(crate) fn serialize_offsets( + offsets: &Vec, + serializer: &mut S, +) -> Result +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 = 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::, DynData, DynZWeight>( + &test.factories(), + &serialized, + ); + + assert_eq!(&*test, &deserialized); + } + } + + /// Every reader of a bulk-copied offset table takes it as + /// `archived_root::>`, 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::>(&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 index 47dbf3de83f..cc2dcb6ef17 100644 --- 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 diff --git a/crates/sqllib/benches/flat_variant_serialize.rs b/crates/sqllib/benches/flat_variant_serialize.rs new file mode 100644 index 00000000000..46b700b2ba4 --- /dev/null +++ b/crates/sqllib/benches/flat_variant_serialize.rs @@ -0,0 +1,815 @@ +//! Serialization throughput for [`FlatVariant`], the flat-buffer SQL VARIANT. +//! +//! A `FlatVariant` archives as its own encoding, so serializing one document is +//! a bulk copy of the document buffer plus a 16-byte `ArchivedVec` header. This +//! benchmark measures that copy at scale: it builds `--docs` documents of +//! `--size` encoded bytes each, then serializes all of them with +//! `DbspSerializer` through each of the two paths the runtime uses. +//! +//! | mode | models | +//! |---------------------------|----------------------------------------------------| +//! | fresh `FBuf` per document | `to_bytes`: checkpoints, worker-to-worker payloads | +//! | shared `FBuf` | the layer-file writer filling a data block | +//! +//! The gap between the two is buffer allocation, teardown, and first touch, so +//! the shared-buffer rate is the ceiling the encoding can reach. +//! +//! The default workload is 100,000 documents of 16 KiB, which holds 1.6 GiB of +//! documents live, so expect a little over 2 GiB of RSS. +//! +//! ```text +//! cargo bench -p feldera-sqllib --bench flat_variant_serialize +//! cargo bench -p feldera-sqllib --bench flat_variant_serialize -- --docs 10000 --size 4096 +//! ``` + +use std::env::args; +use std::hint::black_box; +use std::process::exit; +use std::time::{Duration, Instant}; + +use dbsp::storage::buffer_cache::{FBuf, FBufSerializer}; +use dbsp::storage::file::{SerializerInner, to_bytes}; +use dbsp::trace::aligned_deserialize; +use feldera_sqllib::FlatVariant; +use rkyv::ser::Serializer as _; +use size_of::SizeOf; + +/// Bytes rkyv writes for one document beyond the document itself: the +/// `ArchivedVec` that points at the copied encoding. +const HEADER_BYTES: usize = size_of::>(); + +/// rkyv aligns each archived value, so a document that is a whole number of +/// these needs no padding between its bytes and its header. +const ALIGNMENT: usize = 8; + +fn main() { + let config = Config::from_args(); + let workload = Workload::calibrate(&config); + report_configuration(&config, &workload); + + let corpus = generate(&config, &workload); + report_generation(&corpus); + + let modes = Mode::ALL.map(|mode| (mode, mode.measure(&corpus, &config))); + report_serialization(&config, &corpus, &modes); + + if config.check { + check_round_trip(&corpus); + } +} + +// Configuration + +struct Config { + documents: usize, + /// Encoded bytes per document; a multiple of [`ALIGNMENT`]. + document_bytes: usize, + passes: usize, + seed: u64, + /// Bytes to accumulate in the shared buffer before recycling it. + flush_bytes: usize, + check: bool, +} + +impl Default for Config { + fn default() -> Self { + Self { + documents: if SMOKE { 200 } else { 100_000 }, + document_bytes: 16 * 1024, + passes: if SMOKE { 1 } else { 3 }, + seed: 0x5eed_1234_5678_9abc, + flush_bytes: 1 << 20, + check: true, + } + } +} + +/// An unoptimized build cannot produce a meaningful rate, and `cargo test +/// --benches` runs this binary with no arguments, so a build that is not the +/// `bench` or `release` profile does a smoke run by default. `--docs` and +/// `--passes` still override it. +const SMOKE: bool = cfg!(debug_assertions); + +impl Config { + fn from_args() -> Self { + let mut config = Config::default(); + let mut args = args().skip(1); + while let Some(arg) = args.next() { + match arg.as_str() { + "--docs" => config.documents = number(&mut args, &arg), + "--size" => config.document_bytes = number(&mut args, &arg), + "--passes" => config.passes = number(&mut args, &arg), + "--seed" => config.seed = number(&mut args, &arg) as u64, + "--flush" => config.flush_bytes = number(&mut args, &arg), + "--no-check" => config.check = false, + // `cargo bench` passes `--bench`, and a test runner may pass + // `--test` to ask for a smoke run rather than a measurement. + "--bench" => (), + "--test" => { + config.documents = 200; + config.passes = 1; + } + "--help" | "-h" => { + usage(); + exit(0); + } + other => fail(&format!("unknown argument {other:?}; try --help")), + } + } + if config.documents < 2 { + fail("--docs must be at least 2"); + } + if config.passes == 0 { + fail("--passes must be at least 1"); + } + if config.flush_bytes < config.document_bytes { + fail("--flush must be at least one document"); + } + // Round up rather than reject: an unaligned size is unreachable, and the + // report prints the size actually used. + config.document_bytes = config.document_bytes.next_multiple_of(ALIGNMENT); + config + } + + fn serialized_bytes(&self) -> usize { + self.document_bytes + HEADER_BYTES + } +} + +fn number(args: &mut impl Iterator, flag: &str) -> usize { + let Some(value) = args.next() else { + fail(&format!("{flag} requires a value")); + }; + let (digits, radix) = match value.strip_prefix("0x") { + Some(hex) => (hex, 16), + None => (value.as_str(), 10), + }; + match usize::from_str_radix(&digits.replace('_', ""), radix) { + Ok(number) => number, + Err(error) => fail(&format!("{flag}: {value:?} is not a number: {error}")), + } +} + +fn usage() { + println!( + "FlatVariant serialization benchmark + + --docs N documents to serialize (default 100000) + --size N encoded bytes per document, rounded up to a multiple of 8 + (default 16384) + --passes N times to repeat each serialization mode (default 3) + --seed N seed for document contents (default 0x5eed123456789abc) + --flush N bytes to accumulate in the shared buffer before recycling it + (default 1048576) + --no-check skip the round-trip check + --test smoke run: 200 documents, one pass" + ); +} + +fn fail(message: &str) -> ! { + eprintln!("error: {message}"); + exit(2); +} + +// Document generation +// +// A generated document imitates a customer telemetry record: a root map with a +// nested session map, a `props` map of sub-maps holding most of the fields, an +// array of event maps, and an array of tags. Field names come from a small +// vocabulary, so keys repeat across documents the way real ones do. +// +// Every string in a document has a fixed length and every number encodes to a +// fixed-width payload, so all documents encode to exactly the same size whatever +// contents the generator picks. `Workload::calibrate` relies on that to hit +// `--size` exactly, and generation verifies it for every document. + +/// Field-name vocabulary, so keys read like telemetry rather than `f0001`. +const WORDS: [&str; 12] = [ + "clicks", + "dwell_ms", + "last_seen", + "referrer", + "variant", + "platform", + "app_version", + "locale", + "utm_source", + "churn_risk", + "impressions", + "bucket", +]; + +/// Kind of placeholder a [`Patch`] refills, which fixes the alphabet it draws +/// from and so keeps the template valid JSON. +#[derive(Clone, Copy)] +enum Placeholder { + /// Characters inside a string literal. + Text, + /// Digits of an integer literal, whose first digit must not be zero. + Integer, + /// Digits after the `0.` of a fraction, where a leading zero is fine. + Fraction, +} + +/// A region of the JSON template refilled for each document. The region's length +/// never changes, so neither does the encoded document size. +struct Patch { + at: usize, + len: usize, + placeholder: Placeholder, +} + +/// The document shape: how many of each repeated element the template holds. +#[derive(Clone, Copy)] +struct Shape { + /// Sub-maps of `props`, the knob calibration turns to reach `--size`. + sections: usize, + fields_per_section: usize, + events: usize, + tags: usize, + /// Length of the `pad` string, which closes the last bytes to `--size`. + pad: usize, +} + +impl Shape { + /// Root fields: id, ts, active, score, region, session, props, events, tags, + /// zzz_pad. + const ROOT_KEYS: usize = 10; + /// Fields of the nested `session` map. + const SESSION_KEYS: usize = 6; + /// Fields of one `events` entry. + const EVENT_KEYS: usize = 5; + + fn with_sections(sections: usize, pad: usize) -> Self { + Self { + sections, + fields_per_section: 24, + events: 4, + tags: 8, + pad, + } + } + + fn keys(&self) -> usize { + Self::ROOT_KEYS + + Self::SESSION_KEYS + + self.sections * (1 + self.fields_per_section) + + self.events * Self::EVENT_KEYS + } +} + +/// A JSON document with its variable regions marked, reused for every document +/// so that generating one costs a refill and a parse rather than a text build +/// and a parse. +struct Template { + json: Vec, + patches: Vec, +} + +impl Template { + fn build(shape: &Shape) -> Self { + let mut template = Template { + json: Vec::with_capacity(4096), + patches: Vec::new(), + }; + template.raw("{"); + template.key("id"); + template.text(32); + template.raw(","); + template.key("ts"); + template.integer(13); + template.raw(","); + template.key("active"); + template.raw("true,"); + template.key("score"); + template.fraction(6); + template.raw(","); + template.key("region"); + template.text(12); + template.raw(","); + + template.key("session"); + template.raw("{"); + template.key("start"); + template.integer(13); + template.raw(","); + template.key("agent"); + template.text(24); + template.raw(","); + template.key("ip"); + template.text(15); + template.raw(","); + template.key("referrer"); + template.text(28); + template.raw(","); + template.key("depth"); + template.integer(3); + template.raw(","); + template.key("bounced"); + template.raw("false},"); + + template.key("props"); + template.raw("{"); + for section in 0..shape.sections { + if section > 0 { + template.raw(","); + } + let name = format!("section_{section:02}"); + template.key(&name); + template.raw("{"); + for field in 0..shape.fields_per_section { + if field > 0 { + template.raw(","); + } + let name = format!("{}_{field:02}", WORDS[(section + field) % WORDS.len()]); + template.key(&name); + // The mix of value kinds a telemetry record carries, so the + // encoding holds variable-width and fixed-width payloads alike. + match field % 6 { + 0 => template.text(12), + 1 => template.integer(10), + 2 => template.text(24), + 3 => template.fraction(6), + 4 => template.raw("true"), + _ => template.raw("null"), + } + } + template.raw("}"); + } + template.raw("},"); + + template.key("events"); + template.raw("["); + for event in 0..shape.events { + if event > 0 { + template.raw(","); + } + template.raw("{"); + template.key("kind"); + template.text(10); + template.raw(","); + template.key("at"); + template.integer(13); + template.raw(","); + template.key("ok"); + template.raw("true,"); + template.key("weight"); + template.fraction(6); + template.raw(","); + template.key("note"); + template.text(20); + template.raw("}"); + } + template.raw("],"); + + template.key("tags"); + template.raw("["); + for tag in 0..shape.tags { + if tag > 0 { + template.raw(","); + } + template.text(10); + } + template.raw("],"); + + // Map keys are stored sorted, and this one sorts after every other key, + // so a document ends in randomized text. A serializer that loses trailing + // bytes then fails the round-trip check instead of landing on a payload + // byte that happens to be zero. + template.key("zzz_pad"); + template.text(shape.pad); + template.raw("}"); + template + } + + fn raw(&mut self, text: &str) { + self.json.extend_from_slice(text.as_bytes()); + } + + fn key(&mut self, name: &str) { + self.raw("\""); + self.raw(name); + self.raw("\":"); + } + + fn text(&mut self, len: usize) { + self.raw("\""); + self.placeholder(Placeholder::Text, len); + self.raw("\""); + } + + fn integer(&mut self, digits: usize) { + self.placeholder(Placeholder::Integer, digits); + } + + fn fraction(&mut self, digits: usize) { + self.raw("0."); + self.placeholder(Placeholder::Fraction, digits); + } + + /// Reserves a variable region, filled with a valid initial value so that the + /// template parses even before the first refill. + fn placeholder(&mut self, placeholder: Placeholder, len: usize) { + if len == 0 { + return; + } + let at = self.json.len(); + match placeholder { + Placeholder::Text => self.json.resize(at + len, b'a'), + Placeholder::Integer => { + self.json.push(b'1'); + self.json.resize(at + len, b'0'); + } + Placeholder::Fraction => self.json.resize(at + len, b'0'), + } + self.patches.push(Patch { + at, + len, + placeholder, + }); + } + + /// Refills every variable region and parses the result. Structure and sizes + /// are the template's; only contents change. + fn document(&mut self, rng: &mut Rng) -> FlatVariant { + for patch in &self.patches { + let region = &mut self.json[patch.at..patch.at + patch.len]; + match patch.placeholder { + Placeholder::Text => fill(region, rng, TEXT_ALPHABET), + Placeholder::Fraction => fill(region, rng, DIGIT_ALPHABET), + Placeholder::Integer => { + fill(region, rng, DIGIT_ALPHABET); + // JSON forbids a leading zero in an integer literal. + region[0] = b'1' + (region[0] - b'0') % 9; + } + } + } + serde_json::from_slice(&self.json).expect("template is valid JSON") + } +} + +/// 32 characters, so one character costs 5 bits of randomness and no division. +const TEXT_ALPHABET: &[u8; 32] = b"abcdefghijklmnopqrstuvwxyz-_.012"; +/// Eight digits rather than ten, for the same reason. +const DIGIT_ALPHABET: &[u8; 32] = b"01234567012345670123456701234567"; + +fn fill(region: &mut [u8], rng: &mut Rng, alphabet: &[u8; 32]) { + // One draw feeds twelve characters, five bits each. + for chunk in region.chunks_mut(12) { + let mut bits = rng.next(); + for byte in chunk { + *byte = alphabet[(bits & 31) as usize]; + bits >>= 5; + } + } +} + +/// SplitMix64, hand-rolled so that the workload is identical on every machine +/// and across dependency upgrades; the generator needs only uniform bits. +struct Rng(u64); + +impl Rng { + fn next(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9e37_79b9_7f4a_7c15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + z ^ (z >> 31) + } +} + +// Calibration + +/// A document shape whose encoded size is exactly `Config::document_bytes`. +struct Workload { + shape: Shape, + probes: usize, +} + +impl Workload { + /// Sizes the document shape to `config.document_bytes`. + /// + /// Encoded size grows linearly in [`Shape::sections`] and by one byte per + /// [`Shape::pad`] character, so two probes fix the slope, a division picks + /// the section count, and the pad closes the remainder. The last walk pushes + /// the pad to the top of the range that still serializes to the target, + /// which is where the document needs no alignment padding and its size is + /// therefore the target exactly. + fn calibrate(config: &Config) -> Self { + let target = config.serialized_bytes(); + let mut probes = 0; + let mut probe = |sections: usize, pad: usize| { + probes += 1; + let mut template = Template::build(&Shape::with_sections(sections, pad)); + let document = template.document(&mut Rng(config.seed)); + to_bytes(&document).expect("serialize document").len() + }; + + let one = probe(1, 0); + let two = probe(2, 0); + let per_section = two - one; + let sectionless = one - per_section; + if sectionless > target { + fail(&format!( + "--size {} is too small for this document shape; use at least {}", + config.document_bytes, + sectionless - HEADER_BYTES + )); + } + + let mut sections = (target - sectionless) / per_section; + while sections > 0 && probe(sections, 0) > target { + sections -= 1; + } + let mut pad = target - probe(sections, 0); + while probe(sections, pad + 1) == target { + pad += 1; + } + + let serialized = probe(sections, pad); + assert_eq!( + serialized, target, + "calibration missed the target document size" + ); + Workload { + shape: Shape::with_sections(sections, pad), + probes, + } + } +} + +// Generation + +struct Corpus { + documents: Vec, + /// Bytes one pass over the corpus writes. + serialized_bytes: usize, + wall: Duration, +} + +/// Builds the documents and checks that each one serializes to the calibrated +/// size. That check doubles as a warm-up: it faults in every document buffer, so +/// the first measured pass is not the one paying for first touch. +fn generate(config: &Config, workload: &Workload) -> Corpus { + let expected = config.serialized_bytes(); + let mut template = Template::build(&workload.shape); + let mut rng = Rng(config.seed); + let mut documents = Vec::with_capacity(config.documents); + + let start = Instant::now(); + for index in 0..config.documents { + let document = template.document(&mut rng); + let serialized = to_bytes(&document).expect("serialize document").len(); + assert_eq!( + serialized, expected, + "document {index} serialized to {serialized} bytes, expected {expected}" + ); + documents.push(document); + } + let wall = start.elapsed(); + + Corpus { + serialized_bytes: documents.len() * expected, + documents, + wall, + } +} + +// Measurement + +#[derive(Clone, Copy)] +enum Mode { + /// One freshly allocated buffer per document, as `to_bytes` does. + FreshBuffer, + /// Many documents into one recycled buffer, as the layer-file writer does. + SharedBuffer, +} + +impl Mode { + const ALL: [Mode; 2] = [Mode::FreshBuffer, Mode::SharedBuffer]; + + fn label(&self) -> &'static str { + match self { + Mode::FreshBuffer => "fresh FBuf per document", + Mode::SharedBuffer => "shared FBuf", + } + } + + /// Serializes the whole corpus `config.passes` times. + fn measure(&self, corpus: &Corpus, config: &Config) -> Vec { + (0..config.passes) + .map(|_| { + let pass = match self { + Mode::FreshBuffer => fresh_buffer(&corpus.documents), + Mode::SharedBuffer => shared_buffer(&corpus.documents, config.flush_bytes), + }; + assert_eq!( + pass.bytes, + corpus.serialized_bytes, + "{} wrote {} bytes, expected {}", + self.label(), + pass.bytes, + corpus.serialized_bytes + ); + pass.wall + }) + .collect() + } +} + +struct Pass { + wall: Duration, + bytes: usize, +} + +fn fresh_buffer(documents: &[FlatVariant]) -> Pass { + let mut bytes = 0; + let start = Instant::now(); + for document in documents { + let buffer = to_bytes(document).expect("serialize document"); + bytes += buffer.len(); + black_box(buffer.as_slice()); + } + Pass { + wall: start.elapsed(), + bytes, + } +} + +fn shared_buffer(documents: &[FlatVariant], flush_bytes: usize) -> Pass { + let mut inner = SerializerInner::new(); + let mut buffer = FBuf::with_capacity(flush_bytes); + let mut bytes = 0; + + let start = Instant::now(); + for document in documents { + let before = buffer.len(); + inner + .with(FBufSerializer::new(&mut buffer), |serializer| { + serializer.serialize_value(document) + }) + .expect("serialize document"); + bytes += buffer.len() - before; + if buffer.len() >= flush_bytes { + black_box(buffer.as_slice()); + buffer.clear(); + } + } + black_box(buffer.as_slice()); + Pass { + wall: start.elapsed(), + bytes, + } +} + +// Checks + +/// Round-trips a sample of documents through the archived form, and confirms +/// that the comparison the round trip rests on can tell two documents apart. +fn check_round_trip(corpus: &Corpus) { + let documents = &corpus.documents; + let sampled = 16.min(documents.len()); + let stride = documents.len() / sampled; + for index in (0..sampled).map(|sample| sample * stride) { + let document = &documents[index]; + let bytes = to_bytes(document).expect("serialize document"); + let restored: FlatVariant = aligned_deserialize(&bytes[..]); + assert_eq!(&restored, document, "document {index} changed in transit"); + assert_ne!( + &restored, + &documents[(index + 1) % documents.len()], + "documents are indistinguishable, so the round trip proves nothing" + ); + } + println!("round trip: {sampled} documents restored and compared equal"); +} + +// Reporting + +fn report_configuration(config: &Config, workload: &Workload) { + let shape = &workload.shape; + println!("FlatVariant serialization through DbspSerializer\n"); + if SMOKE { + println!("built without optimizations: smoke run, the rates below mean nothing\n"); + } + println!("configuration"); + row("documents", &count(config.documents), ""); + row( + "document bytes", + &count(config.document_bytes), + &bytes(config.document_bytes as f64), + ); + row( + "serialized bytes", + &count(config.serialized_bytes()), + &format!("document + {HEADER_BYTES} B ArchivedVec header"), + ); + row( + "keys per document", + &count(shape.keys()), + &format!( + "{} sections x {} fields, {} events, {} tags, {} B pad", + shape.sections, shape.fields_per_section, shape.events, shape.tags, shape.pad + ), + ); + row( + "calibration probes", + &count(workload.probes), + "documents built to size the shape", + ); + row("seed", &format!("{:#x}", config.seed), ""); + row("passes", &count(config.passes), "per mode"); + row( + "shared buffer flush", + &count(config.flush_bytes), + "bytes before the buffer is recycled", + ); + println!(); +} + +fn report_generation(corpus: &Corpus) { + let seconds = corpus.wall.as_secs_f64(); + let documents = corpus.documents.len(); + let heap = corpus.documents.size_of().total_bytes(); + println!("generation (JSON parse plus the size check, not measured)"); + row("wall", &format!("{seconds:.3} s"), ""); + row( + "rate", + &count((documents as f64 / seconds) as usize), + &format!( + "documents/s, {}/s", + bytes(corpus.serialized_bytes as f64 / seconds) + ), + ); + row( + "document heap", + &bytes(heap as f64), + &format!("size_of, {} B/document", count(heap / documents)), + ); + println!(); +} + +fn report_serialization(config: &Config, corpus: &Corpus, modes: &[(Mode, Vec)]) { + let mut header = format!(" {:<24}", "mode"); + for pass in 1..=config.passes { + let label = format!("pass {pass}"); + header.push_str(&format!("{label:>10}")); + } + header.push_str(&format!( + "{:>10}{:>14}{:>14}", + "median", "documents/s", "bytes/s" + )); + println!("serialization"); + println!("{header}"); + + for (mode, passes) in modes { + let mut line = format!(" {:<24}", mode.label()); + for wall in passes { + line.push_str(&format!("{:>10.3}", wall.as_secs_f64())); + } + let median = median_seconds(passes); + let rate = count((corpus.documents.len() as f64 / median) as usize); + let throughput = format!("{}/s", bytes(corpus.serialized_bytes as f64 / median)); + line.push_str(&format!("{median:>10.3}{rate:>14}{throughput:>14}")); + println!("{line}"); + } + println!(); +} + +fn median_seconds(passes: &[Duration]) -> f64 { + let mut seconds: Vec = passes.iter().map(Duration::as_secs_f64).collect(); + seconds.sort_by(f64::total_cmp); + seconds[seconds.len() / 2] +} + +fn row(label: &str, value: &str, note: &str) { + if note.is_empty() { + println!(" {label:<22}{value:>13}"); + } else { + println!(" {label:<22}{value:>13} {note}"); + } +} + +/// Formats a byte count in the largest unit that keeps it above one. +fn bytes(mut count: f64) -> String { + const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"]; + let mut unit = 0; + while count >= 1024.0 && unit + 1 < UNITS.len() { + count /= 1024.0; + unit += 1; + } + format!("{count:.1} {}", UNITS[unit]) +} + +/// Groups digits so that six- and ten-digit counts stay readable. +fn count(number: usize) -> String { + let digits = number.to_string(); + let mut grouped = String::with_capacity(digits.len() + digits.len() / 3); + for (index, digit) in digits.chars().enumerate() { + if index > 0 && (digits.len() - index).is_multiple_of(3) { + grouped.push(','); + } + grouped.push(digit); + } + grouped +} diff --git a/crates/sqllib/src/binary.rs b/crates/sqllib/src/binary.rs index 06846212ae9..2a1d0c70f42 100644 --- a/crates/sqllib/src/binary.rs +++ b/crates/sqllib/src/binary.rs @@ -14,6 +14,8 @@ use feldera_types::serde_with_context::{ use flate2::read::GzDecoder; use hex::ToHex; use md5::{Digest, Md5}; +use rkyv::ser::Serializer as RkyvSerializer; +use rkyv::vec::ArchivedVec; use serde::{ Deserialize, Deserializer, Serialize, Serializer, de::{Error as _, Visitor}, @@ -46,7 +48,6 @@ type CompactVec = SmallVec<[u8; THRESHOLD]>; Serialize, Deserialize, rkyv::Archive, - rkyv::Serialize, rkyv::Deserialize, IsNone, )] @@ -56,9 +57,32 @@ pub struct ByteArray { data: CompactVec, } +/// Hand written rather than derived so that the payload is copied in one bulk +/// write. The derived impl delegates to `SmallVec`'s, which resolves the bytes +/// element by element, one `write` call per byte, and measured 47x slower on a +/// 16 KiB value. +impl rkyv::Serialize for ByteArray { + fn serialize(&self, serializer: &mut S) -> Result { + // SAFETY: a byte is trivially copyable, has no padding, and archives as + // itself, so the copy is exactly the payload. + let data = unsafe { ArchivedVec::serialize_copy_from_slice(&self.data, serializer)? }; + Ok(ByteArrayResolver { data }) + } +} + impl SizeOf for ByteArray { fn size_of_children(&self, context: &mut size_of::Context) { - self.data.size_of_children(context); + // `SmallVec` has no `SizeOf` impl, so delegating to `self.data` resolved + // to the one for `[u8]`, which walks the bytes and reports none of the + // buffer holding them: a spilled payload counted as zero, whatever its + // size. An inline payload needs nothing added, being part of + // `size_of::()` already; a spilled one is one allocation, + // counted here the way `Vec` counts its own. + if self.data.spilled() { + context + .add_vectorlike(self.data.len(), self.data.capacity(), size_of::()) + .add_distinct_allocation(); + } } } @@ -203,6 +227,76 @@ mod test_binary_deserializer { } } +#[cfg(test)] +mod test_binary_size_of { + use super::{ByteArray, THRESHOLD}; + use size_of::SizeOf; + use std::mem::size_of; + + /// A spilled payload is nearly the whole footprint of a large VARBINARY + /// value, and the batch spill accounting reads footprints through `SizeOf`, + /// so it has to be counted. + #[test] + fn size_of_counts_a_spilled_payload() { + let inline = ByteArray::new(&[7u8; THRESHOLD]); + assert!(!inline.data.spilled()); + assert_eq!(inline.size_of().total_bytes(), size_of::()); + assert_eq!(inline.size_of().distinct_allocations(), 0); + + // The payload is counted, and it scales with the payload rather than + // landing on some fixed overhead. + for length in [THRESHOLD + 1, 4096, 1 << 20] { + let spilled = ByteArray::new(&vec![7u8; length]); + assert!(spilled.data.spilled(), "length {length}"); + assert_eq!( + spilled.size_of().total_bytes(), + size_of::() + spilled.data.capacity(), + "length {length}" + ); + assert!( + spilled.size_of().total_bytes() >= size_of::() + length, + "length {length}" + ); + assert_eq!( + spilled.size_of().distinct_allocations(), + 1, + "length {length}" + ); + } + } +} + +#[cfg(test)] +mod test_binary_rkyv { + use super::{ByteArray, THRESHOLD}; + + /// The archived form holds the payload verbatim. `Serialize` copies it in + /// bulk through an unsafe rkyv entry point, so pin the bytes themselves + /// rather than only the value a round trip recovers: a copy that drops or + /// shifts bytes can still deserialize equal when the bytes it mangles + /// happen to match rkyv's alignment padding. + #[test] + fn rkyv_archives_the_payload_verbatim() { + // Inline, exactly at the spill point, and heap allocated. + for length in [0, 1, THRESHOLD, THRESHOLD + 1, 4096] { + let payload: Vec = (0..length).map(|i| (i * 7 + 1) as u8).collect(); + let array = ByteArray::new(&payload); + + let serialized = dbsp::storage::file::to_bytes(&array).unwrap(); + // SAFETY: the bytes come from serializing `array` just above. + let archived = unsafe { rkyv::archived_root::(&serialized) }; + assert_eq!( + archived.data.as_slice(), + payload.as_slice(), + "length {length}" + ); + + let restored: ByteArray = dbsp::trace::aligned_deserialize(&serialized[..]); + assert_eq!(restored, array, "length {length}"); + } + } +} + #[doc(hidden)] impl NumEntries for &ByteArray { const CONST_NUM_ENTRIES: Option = None; diff --git a/crates/sqllib/src/flat_variant.rs b/crates/sqllib/src/flat_variant.rs index c0495edae11..c9503faba5a 100644 --- a/crates/sqllib/src/flat_variant.rs +++ b/crates/sqllib/src/flat_variant.rs @@ -1111,7 +1111,12 @@ impl rkyv::Archive for FlatVariant { impl rkyv::Serialize for FlatVariant { fn serialize(&self, serializer: &mut S) -> Result { - ArchivedVec::serialize_from_slice(self.as_bytes(), serializer) + // SAFETY: a byte is trivially copyable, has no padding, and archives as + // itself, so the copy is exactly the encoding. The safe + // `serialize_from_slice` resolves a slice element by element, one + // `write` call per byte, which measured 60x slower on a 16 KiB document + // (`benches/flat_variant_serialize.rs`). + unsafe { ArchivedVec::serialize_copy_from_slice(self.as_bytes(), serializer) } } } @@ -1596,6 +1601,21 @@ mod tests { } } + /// The archived form is the encoding verbatim. `Serialize` copies the + /// document in bulk through an unsafe rkyv entry point, so pin the bytes + /// themselves rather than only the value a round trip recovers: a copy that + /// drops or shifts bytes can still deserialize equal when the bytes it + /// mangles happen to match rkyv's alignment padding. + #[test] + fn rkyv_archives_the_encoding_verbatim() { + let document: FlatVariant = + serde_json::from_str(r#"{"a":{"b":[1,2.5,"three",null,true]},"z":"tail"}"#).unwrap(); + let bytes = dbsp::storage::file::to_bytes(&document).unwrap(); + // SAFETY: the bytes come from serializing `document` just above. + let archived = unsafe { rkyv::archived_root::(&bytes) }; + assert_eq!(archived.as_bytes(), document.as_bytes()); + } + /// The MAP cast must treat a JSON null member exactly like the enum /// path, where `Option::try_from` always wraps in `Some` /// (MapTests#mapValuesVariant regression).