From 96885fde0024c3c6020bc4cc32725b957a1aa85b Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Tue, 11 Aug 2026 14:44:38 -0700 Subject: [PATCH 1/6] sqllib: benchmark FlatVariant serialization Generate 100,000 documents of exactly 16 KiB and serialize all of them with DbspSerializer, once through to_bytes, which allocates an FBuf per document, and once into a single recycled FBuf, the way the layer-file writer packs a data block. The gap between the two modes is buffer allocation, teardown, and first touch, so the shared-buffer rate is the ceiling the encoding can reach. cargo bench -p feldera-sqllib --bench flat_variant_serialize cargo bench -p feldera-sqllib --bench flat_variant_serialize -- \ --docs 10000 --size 4096 An unoptimized build runs 200 documents instead of 100,000, because cargo test --benches runs a harnessless bench binary with no arguments and its rates would mean nothing anyway. Signed-off-by: Leonid Ryzhyk --- crates/sqllib/Cargo.toml | 4 + .../sqllib/benches/flat_variant_serialize.rs | 815 ++++++++++++++++++ 2 files changed, 819 insertions(+) create mode 100644 crates/sqllib/benches/flat_variant_serialize.rs 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 +} From 7afc78da3342c6fbe58a65a16ccd40a7f65a3832 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Tue, 11 Aug 2026 14:44:47 -0700 Subject: [PATCH 2/6] sqllib: copy a FlatVariant document in bulk when serializing ArchivedVec::serialize_from_slice resolves a slice element by element unless rkyv's nightly-only copy feature is on, so serializing a document cost one write call per byte rather than the single bulk write the module documentation claims. serialize_copy_from_slice writes it once. A byte is trivially copyable, has no padding, and archives as itself, so the copy is exactly the encoding. Measured by benches/flat_variant_serialize.rs on 100,000 documents of 16 KiB, paired back to back in one machine state: mode before after fresh FBuf per document 3.689 s 424 MiB/s 0.056 s 27 GiB/s shared FBuf 3.574 s 438 MiB/s 0.033 s 47 GiB/s Signed-off-by: Leonid Ryzhyk --- crates/sqllib/src/flat_variant.rs | 22 +++++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) 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). From b65f9e2768a3af58f178c2446350a869dc2529bf Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Tue, 11 Aug 2026 15:07:59 -0700 Subject: [PATCH 3/6] sqllib: copy a ByteArray payload in bulk when serializing The derived rkyv::Serialize delegated to SmallVec's, which resolves the payload element by element, one write call per byte. Hand-writing the impl around serialize_copy_from_slice writes it once. A byte is trivially copyable, has no padding, and archives as itself, so the copy is exactly the payload, and the archived layout does not change: an ArchivedVec either way, so existing storage stays readable. Every BINARY and VARBINARY value pays this on each batch write to storage and each checkpoint, and so does Variant::Binary. Measured on 16 KiB values, against the untouched Vec path as a control in the same run: Vec (per element) 97.808 ms 0.31 GiB/s ByteArray (bulk copy) 1.118 ms 27.30 GiB/s Signed-off-by: Leonid Ryzhyk --- crates/sqllib/src/binary.rs | 47 ++++++++++++++++++++++++++++++++++++- 1 file changed, 46 insertions(+), 1 deletion(-) diff --git a/crates/sqllib/src/binary.rs b/crates/sqllib/src/binary.rs index 06846212ae9..28a9266f9aa 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,6 +57,19 @@ pub struct ByteArray { data: CompactVec, } +// rkyv::Serialize is 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); @@ -203,6 +217,37 @@ mod test_binary_deserializer { } } +#[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; From 0b17d2841d9617fbf750fa3102dcf875de04f204 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Tue, 11 Aug 2026 15:27:42 -0700 Subject: [PATCH 4/6] dbsp: copy batch offset tables in bulk when serializing The three batch serializers end by writing a Vec of offsets, one entry per key, value and weight. Vec's own rkyv impl resolves it element by element, one write call per offset. serialize_offsets wraps the table in rkyv's CopyOptimize so it lands in one write instead. The wrapper leaves the archived layout alone, an ArchivedVec of the same elements, so the readers still take archived_root::> and existing checkpoints stay readable. Copying is only valid where a usize archives as itself, which a static assertion pins: rkyv archives integers in native byte order, and the size_64 feature fixes the archived width at 8 bytes, so it holds on every 64-bit target. The table itself serializes 25x faster (3M offsets: 181.6 ms to 7.3 ms). End to end the gain is smaller, because the per-item serialization of the keys, values and weights dominates and is untouched. Checkpointing a 1M-row VecIndexedWSet, paired back to back: before 35.264 ms/pass 1.48 GiB/s after 30.874 ms/pass 1.69 GiB/s Signed-off-by: Leonid Ryzhyk --- .../operator/dynamic/communication/shard.rs | 7 +- crates/dbsp/src/trace.rs | 82 +++++++++++++++++-- crates/sqllib/src/binary.rs | 8 +- 3 files changed, 84 insertions(+), 13 deletions(-) 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/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/src/binary.rs b/crates/sqllib/src/binary.rs index 28a9266f9aa..93c26588d0c 100644 --- a/crates/sqllib/src/binary.rs +++ b/crates/sqllib/src/binary.rs @@ -57,10 +57,10 @@ pub struct ByteArray { data: CompactVec, } -// rkyv::Serialize is 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. +/// 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 From 9b0d3fa43c440bee539bf03f0f5689d350532e85 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Tue, 11 Aug 2026 15:50:44 -0700 Subject: [PATCH 5/6] dbsp: copy checkpointed state blobs in bulk Checkpoint::checkpoint returns state already serialized, and the committed structs that carry it hold it as Vec, whose rkyv impl copies it back out one byte per write call. rkyv's CopyOptimize wrapper writes it once. Annotated: CommittedZ1 and CommittedTransactionZ1, which carry a whole operator state, and CommittedClock, which is small but free to fix. CommittedWindow keeps the per-element path, because the wrapper applies to a Vec field rather than through an Option of a tuple and a pair of serialized keys is not worth reshaping the struct for; the reason is now recorded there. Serializing a CommittedZ1 with a 16 MiB blob, paired back to back: before 13.304 ms/pass 1.17 GiB/s after 1.419 ms/pass 11.01 GiB/s Restore gets faster too, since the wrapper deserializes with one copy_nonoverlapping instead of a per-element loop. The archived layout is unchanged, so checkpoints written by earlier builds still restore. bulk_copy_keeps_the_committed_layout pins exactly that: it compares the archived bytes against a reference struct that has no wrapper, and fails if the wrapper ever changes the layout, which a swap to AsBox confirms. Signed-off-by: Leonid Ryzhyk --- crates/dbsp/src/circuit/circuit_builder.rs | 3 ++ .../operator/dynamic/time_series/window.rs | 5 +++ crates/dbsp/src/operator/transaction_z1.rs | 6 ++++ crates/dbsp/src/operator/z1.rs | 33 +++++++++++++++++++ 4 files changed, 47 insertions(+) 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/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); From 0102a23aa91300d031646521f1b91f7cb6f2c494 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Tue, 11 Aug 2026 16:30:26 -0700 Subject: [PATCH 6/6] sqllib: count a spilled ByteArray payload in SizeOf ByteArray::size_of_children delegated to its SmallVec field, and there is no SizeOf impl for SmallVec, so the call resolved to the one for [u8]: it walked the bytes, reported none of the buffer holding them, and left a 16 KiB VARBINARY value claiming 48 bytes. Count the spilled allocation the way Vec counts its own. An inline payload needs nothing added, being part of size_of::() already. The undercount was total rather than approximate, and BINARY and VARBINARY are the only SQL types affected: SqlString wraps an ArcStr, which feldera-size-of does support, and FlatVariant reports its Arc buffer. What read the wrong number: the fallback batches size their contents with size_of().total_bytes() per key, value and weight (trace/ord/fallback/indexed_wset.rs:568-654) to decide when a batch moves to storage, so a batch of large binary values looked nearly free and stayed in memory. Expect such batches to spill sooner now, which is the point. Dropping the delegation also drops a per-byte loop that computed nothing. Found by the FlatVariant serialization benchmark, whose VARBINARY corpus reported 4.6 MiB of heap for 1.5 GiB of payloads. Signed-off-by: Leonid Ryzhyk --- crates/sqllib/src/binary.rs | 51 ++++++++++++++++++++++++++++++++++++- 1 file changed, 50 insertions(+), 1 deletion(-) diff --git a/crates/sqllib/src/binary.rs b/crates/sqllib/src/binary.rs index 93c26588d0c..2a1d0c70f42 100644 --- a/crates/sqllib/src/binary.rs +++ b/crates/sqllib/src/binary.rs @@ -72,7 +72,17 @@ impl rkyv::Serialize for ByteArray { 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(); + } } } @@ -217,6 +227,45 @@ 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};