diff --git a/README.md b/README.md index e79447d..09ff0ae 100644 --- a/README.md +++ b/README.md @@ -85,13 +85,15 @@ let seg = Blob::open(MmapBytes::open(path)?)?; seg.read_all(b"term", |v| { /* zero-copy borrow into the mapping */ })?; ``` -With the whole input in hand, `SegmentWriter::write_sorted` writes the same -bytes and sizes the segment's head reserve exactly, so a reader's first probe -covers the index without a second round trip and a small segment does not -carry a large one's worth of zeroes. `SegmentWrite` carries the per-file -settings -- compression, inline runs, sync spreading, and whether the reserve -holds a copy of the directory. `supdb::reserve` answers the sizing question on -its own, from lengths or from totals. +A segment's head reserve is what lets a reader's first probe cover the index +without a second round trip, and it has to be sized before the first key is +written. `supdb::reserve` computes it rather than guessing, so a small segment +does not carry a large one's worth of zeroes. With the whole input in hand, +`SegmentWriter::write_sorted` does it for you. Without: run the lengths through +`reserve::Planner`, which holds aggregates rather than records, then stream +through `SegmentWriter::create_with(path, opts, write, reserve)`, which takes +the per-file settings -- compression, inline runs, sync spreading -- and the +reserve together, since all of them must be set before the first key. The same segment in a browser, over ranged HTTP from a Web Worker: diff --git a/src/db.rs b/src/db.rs index a27db5d..3118d7d 100644 --- a/src/db.rs +++ b/src/db.rs @@ -904,6 +904,52 @@ fn superblock(fields: &[u64; 16]) -> [u8; crate::format::SUPER_BYTES] { } impl SegmentWriter { + /// Open `path` for a fresh segment with every per-file setting applied + /// and the head reserve set, so nothing is left to remember. + /// + /// The four things a segment writer can be configured to do -- inline + /// runs, compression, spread syncs, the head reserve -- all have to be set + /// before the first key, and each is silent when forgotten: a plain + /// segment where a compressed one was wanted, or a reserve of zero that + /// costs the sparse reader a round trip and raises nothing. Setting them + /// at construction is what makes forgetting one impossible. + /// + /// `reserve` is a number rather than a policy, because the caller may + /// know it exactly. `reserve::for_lengths` computes it for input in hand, + /// and its `Reserve` prices the hash-directory copy separately: + /// + /// ```no_run + /// # use supdb::{SegmentOptions, SegmentWrite, SegmentWriter, reserve}; + /// # let (path, opts, write) = (std::path::Path::new("s.sup"), SegmentOptions::default(), SegmentWrite::default()); + /// # let lengths: Vec<(usize, usize)> = Vec::new(); + /// let r = reserve::for_lengths(&lengths, opts.block_size, write.inline_max).unwrap(); + /// // `r.bytes()` for a lookup that plans from the probe; `without_directory` + /// // to save four bytes a key and let a lookup fetch the directory itself. + /// let w = SegmentWriter::create_with(path, &opts, &write, r.bytes())?; + /// # Ok::<(), std::io::Error>(()) + /// ``` + pub fn create_with( + path: &Path, + opts: &SegmentOptions, + write: &SegmentWrite, + reserve: usize, + ) -> Result { + let mut w = SegmentWriter::create(path, opts)?; + // Every field of `SegmentWrite`, and the reserve. If a setter is + // added to this writer without a field beside it here, this stops + // compiling, which is the point of the struct. + let SegmentWrite { + inline_max, + compress, + sync_every, + } = *write; + w.set_inline_max(inline_max); + w.set_compress(compress); + w.set_sync_every(sync_every); + w.set_head_reserve(reserve); + Ok(w) + } + /// Write a whole segment from input already in hand, sizing the head /// reserve exactly instead of guessing at it. /// @@ -914,9 +960,10 @@ impl SegmentWriter { /// reserve exactly, which is what `reserve::for_lengths` does and what /// this does for you. /// - /// `write` carries the per-file settings; every one of this writer's - /// setters has a field there, so nothing it can be configured to do is - /// unreachable from here. + /// The reserve it computes holds the hash-directory copy, so a lookup + /// plans its records from the probe. To trade that for four bytes a key, + /// take `reserve::for_lengths(..).without_directory()` and stream through + /// [`SegmentWriter::create_with`] instead. /// /// `items` must be sorted by key, as the streaming API requires. Returns /// the reserve it used, since a caller measuring segments wants to know. @@ -939,21 +986,11 @@ impl SegmentWriter { // block count -- and the table sized by it -- is the same either way. // What compression moves is where the key section lands, and the row // is already taken at its worst alignment. - let plan = crate::reserve::for_lengths(&lengths, opts.block_size, write.inline_max) - .ok_or_else(|| err("segment writer: this input cannot be a segment"))?; - let reserve = if write.directory_in_reserve { - plan.bytes() - } else { - plan.without_directory() - }; + let reserve = crate::reserve::for_lengths(&lengths, opts.block_size, write.inline_max) + .ok_or_else(|| err("segment writer: this input cannot be a segment"))? + .bytes(); - let mut w = SegmentWriter::create(path, opts)?; - // Every setter, in the order they must be called: all of these want - // to be set before the first key. - w.set_inline_max(write.inline_max); - w.set_compress(write.compress); - w.set_sync_every(write.sync_every); - w.set_head_reserve(reserve); + let mut w = SegmentWriter::create_with(path, opts, write, reserve)?; for (k, vals) in items { w.begin(k)?; for v in *vals { @@ -1571,8 +1608,16 @@ impl SegmentWriter { } } -/// The per-file settings a segment is written with: every one of -/// `SegmentWriter`'s setters, gathered so a batch write can apply them. +/// The per-file settings a segment is written with: one field for every one +/// of `SegmentWriter`'s setters, and nothing else. +/// +/// That invariant is the point. Every setter here must be called before the +/// first key, and forgetting one is silent -- so +/// [`SegmentWriter::create_with`] takes this struct and applies all of it, +/// and a new setter that does not appear here is a compile error there rather +/// than a quiet default. Nothing lives in this struct that `create_with` does +/// not apply, which is why the choice of what the head reserve holds is not +/// here: `create_with` is given the reserve as a number. /// /// These are separate from [`SegmentOptions`] on purpose, and the separation /// is the same one that struct's own note draws: `SegmentOptions` is the @@ -1580,14 +1625,7 @@ impl SegmentWriter { /// while these describe one file. A term index built to be downloaded wants /// compression and inline runs; the segments the seal writes and its own /// merge reads back want neither. -/// -/// It exists because the first `write_sorted` took `inline_max` as a bare -/// argument and had nowhere to put the rest, so compression silently did -/// nothing -- which is the failure `SegmentOptions` refuses a compression -/// field to avoid, reproduced one layer up. A struct with a field per setter -/// makes the next setter's absence a compile error in -/// `SegmentWriter::write_sorted` rather than a quiet default. -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Default)] pub struct SegmentWrite { /// Runs up to this many bytes go inline in the index record, and the /// segment is written records-first so they stream. Zero keeps every run @@ -1598,23 +1636,6 @@ pub struct SegmentWrite { /// fdatasync every this many bytes of blocks rather than once at the end. /// Zero for the single sync. pub sync_every: usize, - /// Put a copy of the hash directory in the head reserve. Four bytes a - /// key, and it is the difference between a lookup that plans its records - /// from the probe and one that fetches the directory first. On by - /// default: a segment written through this path is one whose whole input - /// was in hand, which is the shape that gets downloaded and read cold. - pub directory_in_reserve: bool, -} - -impl Default for SegmentWrite { - fn default() -> SegmentWrite { - SegmentWrite { - inline_max: 0, - compress: false, - sync_every: 0, - directory_in_reserve: true, - } - } } /// How a segment file is written. diff --git a/src/flatindex.rs b/src/flatindex.rs index 995fd94..86bb3fa 100644 --- a/src/flatindex.rs +++ b/src/flatindex.rs @@ -91,7 +91,7 @@ const FENCE_MIN_STRIDE: usize = 16; /// Stride for `n` keys: enough entries to narrow the search hard, few enough /// that the fence stays small enough to sit in cache. -fn fence_stride(n: usize) -> usize { +pub fn fence_stride(n: usize) -> usize { let want = n.div_ceil(fence_target()).max(FENCE_MIN_STRIDE); want.next_power_of_two() } @@ -490,7 +490,7 @@ fn record_len(klen: usize, next: usize) -> usize { /// `record_len` plus the record's tail: the bytes of its inline runs, padded /// so the next record stays 4-aligned. -fn record_len_tail(klen: usize, next: usize, tail: usize) -> usize { +pub fn record_len_tail(klen: usize, next: usize, tail: usize) -> usize { record_len(klen, next) + align_up(tail, REC_ALIGN) } @@ -653,12 +653,6 @@ pub fn plan_inline( insert_slack: usize, record_slack: bool, ) -> Option { - let mut cap = 1usize; - while cap < all.len() * 2 { - cap = cap.checked_mul(2)?; - } - cap = cap.max(16); - let mut rec_offs = Vec::with_capacity(all.len()); let mut at = 0usize; for (k, exts) in all { @@ -676,38 +670,95 @@ pub fn plan_inline( let tail = tails.get(rec_offs.len() - 1).map_or(0, |t| t.len()); at = at.checked_add(record_len_tail(k.len(), n, tail))?; } - if at > MAX_RECS { + // The fence samples every `stride`-th key. `fence_n + 1` offsets, so an + // entry's key is the span between its offset and the next. + let stride = fence_stride(all.len()); + let fence_n = all.len().div_ceil(stride); + let fence_blob_len: usize = (0..fence_n) + .map(|i| all[i * stride].0.len()) + .try_fold(0usize, |a, b| a.checked_add(b))?; + + let l = section_layout( + all.len(), + at, + fence_n, + fence_blob_len, + insert_slack, + record_slack, + )?; + Some(Plan { + hash_cap: l.hash_cap, + dir_cap: l.dir_cap, + recs_len: at, + recs_cap: l.recs_cap, + rec_offs, + fence_n, + fence_stride: stride, + fence_offs_off: l.fence_offs_off, + fence_blob_off: l.fence_blob_off, + fence_blob_len, + total: l.total, + written: l.recs_off + at, + }) +} + +/// Where a section's regions land, from the four aggregates that decide them: +/// the key count, the record bytes, and the fence's entry count and blob +/// length. Nothing here needs the keys themselves. +/// +/// Public and separate because the head reserve has to be sized before a key +/// is written, and a caller streaming its input cannot hand over slices of +/// everything. `plan_inline` calls this after counting; `reserve::Planner` +/// calls it after accumulating. Two copies of this arithmetic would be two +/// definitions of the section, and they would drift. +pub struct SectionLayout { + pub hash_cap: usize, + pub dir_cap: usize, + pub fence_offs_off: usize, + pub fence_blob_off: usize, + pub recs_off: usize, + pub recs_cap: usize, + pub total: usize, +} + +pub fn section_layout( + keys: usize, + rec_bytes: usize, + fence_n: usize, + fence_blob_len: usize, + insert_slack: usize, + record_slack: bool, +) -> Option { + if rec_bytes > MAX_RECS { return None; } + let mut cap = 1usize; + while cap < keys * 2 { + cap = cap.checked_mul(2)?; + } + cap = cap.max(16); + // Half again, so a store whose keys gain extents can publish updates // without rewriting anything -- unless the caller says the section is // never edited in place. let slack = if record_slack { - at * SLACK_NUM / SLACK_DEN + rec_bytes * SLACK_NUM / SLACK_DEN } else { 0 }; - let recs_cap = at.checked_add(slack)?; + let recs_cap = rec_bytes.checked_add(slack)?; if recs_cap > MAX_RECS { return None; } - // The fence samples every `stride`-th key. `fence_n + 1` offsets, so an - // entry's key is the span between its offset and the next. - let stride = fence_stride(all.len()); - let fence_n = all.len().div_ceil(stride); - let fence_blob_len: usize = (0..fence_n) - .map(|i| all[i * stride].0.len()) - .try_fold(0usize, |a, b| a.checked_add(b))?; - // One buffer when nothing asked for insert room, two when something did. let dir_cap = if insert_slack == 0 { 0 } else { - all.len().checked_add(insert_slack)? + keys.checked_add(insert_slack)? }; let dir_bytes = if dir_cap == 0 { - all.len() * 4 + keys * 4 } else { dir_cap.checked_mul(8)? }; @@ -717,20 +768,14 @@ pub fn plan_inline( // Records are 4-aligned within the section, and the blob is bytes, so the // record region is realigned after it. let recs_off = align_up(fence_blob_off + fence_blob_len, REC_ALIGN); - let total = recs_off + recs_cap; - Some(Plan { + Some(SectionLayout { hash_cap: cap, dir_cap, - recs_len: at, - recs_cap, - rec_offs, - fence_n, - fence_stride: stride, fence_offs_off, fence_blob_off, - fence_blob_len, - total, - written: recs_off + at, + recs_off, + recs_cap, + total: recs_off + recs_cap, }) } diff --git a/src/reserve.rs b/src/reserve.rs index eadc007..1ef93fc 100644 --- a/src/reserve.rs +++ b/src/reserve.rs @@ -12,24 +12,28 @@ //! as latency, never as a fault, which is the shape of defect this repository //! keeps a list of. //! -//! So the size is computed. [`for_lengths`] answers exactly, from the key and -//! run lengths a caller that gathered its input already has. [`from_totals`] -//! answers with an upper bound for a caller that knows only totals. Both -//! return the [`Reserve`] broken into its four pieces, because the last of -//! them is a decision: the directory copy costs four bytes a key and buys a -//! lookup that plans with no second wave, and only the caller knows whether -//! its readers are paying for round trips or for bytes. +//! So the size is computed. [`Planner`] accumulates the answer one key at a +//! time, holding aggregates rather than the input, for a caller that streams +//! its records and will not hold them. [`for_lengths`] is that planner over a +//! slice, for a caller that has one. [`from_totals`] answers with an upper +//! bound for a caller that knows only totals. All three return the +//! [`Reserve`] broken into its four pieces, because the last of them is a +//! decision: the directory copy costs four bytes a key and buys a lookup that +//! plans with no second wave, and only the caller knows whether its readers +//! are paying for round trips or for bytes. //! -//! **None of the layout arithmetic lives here.** `for_lengths` plans the key -//! section with [`crate::flatindex::plan_inline`], the same call the writer -//! makes, over placeholder keys of the caller's lengths; the block table's -//! size comes from [`crate::flatindex::block_table_len`], which -//! `encode_blocks` allocates by. A second copy of that arithmetic would be a -//! second definition of the format, and the two would drift the first time -//! one of them was edited. +//! **None of the layout arithmetic lives here.** Where the key section's +//! regions land comes from [`crate::flatindex::section_layout`], which +//! `plan_inline` calls after counting and the planner calls after +//! accumulating; the block table's size comes from +//! [`crate::flatindex::block_table_len`], which `encode_blocks` allocates by; +//! a record's bytes come from `flatindex::record_len_tail` and the fence's +//! stride from `flatindex::fence_stride`. What is left here is the cut into +//! blocks, which is four lines of the writer's own rule. A second copy of any +//! of that would be a second definition of the format, and the two would +//! drift the first time one of them was edited. use crate::flatindex; -use crate::index::{Ext, Extents}; /// The bytes a key's values encode to, which is what a block holds and what /// an inline run puts in the record. @@ -66,25 +70,206 @@ fn uvarint_len(mut v: u64) -> usize { /// block by itself and a key's values stay contiguous. Inline runs are not /// passed here -- they never reach a block. fn blocks_for(runs: impl Iterator, block_size: usize) -> usize { - let mut blocks = 0usize; let mut staged = 0usize; - for n in runs { - if staged != 0 && staged + n > block_size { - blocks += 1; - staged = 0; - } - staged += n; - if staged >= block_size { - blocks += 1; - staged = 0; - } - } + let mut blocks: usize = runs.map(|n| cut(&mut staged, n, block_size)).sum(); if staged != 0 { blocks += 1; } blocks } +/// One run through the cut, as the writer does it: a run that does not fit +/// beside what is staged closes a block first, and a builder at or over the +/// block size is flushed after the push. Returns how many blocks closed, and +/// leaves what is still staged in `staged` -- which the last block takes. +fn cut(staged: &mut usize, n: usize, block_size: usize) -> usize { + let mut closed = 0; + // `staged + n > block_size`, without the sum: what is staged is always + // below the block size, and the sum is what would overflow first where a + // usize is 32 bits. + if *staged != 0 && n > block_size - *staged { + closed += 1; + *staged = 0; + } + *staged = staged.saturating_add(n); + if *staged >= block_size { + closed += 1; + *staged = 0; + } + closed +} + +/// The longest run this planner will size, which is the longest one the +/// writer will store: an extent addresses its run with a `u32`, and a run +/// past that is refused rather than written. The headroom below that limit is +/// for the record framing's own 4-byte rounding, and it matters where a usize +/// is 32 bits and the limit is `usize::MAX` -- there the rounding is what +/// overflows, not the length. +const MAX_RUN: usize = (u32::MAX as usize) - 8; + +/// The reserve, accumulated one key at a time. +/// +/// This is the shape a caller wants who will not hold their records: the +/// answer depends on the key count, the record bytes, how the runs cut into +/// blocks, and the lengths of the keys the fence samples -- all of which are +/// aggregates. So a first pass over lengths alone, with no values retained, +/// is enough, and the second pass streams the records through +/// [`crate::SegmentWriter::create_with`] with the reserve already known. +/// +/// What it holds is one `u32` per sixteenth key, and nothing else that grows. +/// The fence samples every `stride`-th key and `stride` is not known until the +/// count is, but every stride the format can choose is a power of two at or +/// above the smallest one, so every sampled key is a multiple of that +/// smallest stride and keeping those is enough. At ten million keys that is +/// about 2.5 MB, against the 160 MB a slice of every key's lengths takes on a +/// 64-bit target and the gigabyte the records themselves would. +pub struct Planner { + block_size: usize, + inline_max: usize, + keys: usize, + rec_bytes: usize, + staged: usize, + blocks: usize, + /// Key lengths at every `sample_stride`-th key, for the fence. + sampled: Vec, + sample_stride: usize, + /// Cleared when the input cannot be a segment, so `finish` says so. + viable: bool, +} + +impl Planner { + /// `block_size` and `inline_max` are the writer's, and must be the ones + /// it will be given. + pub fn new(block_size: usize, inline_max: usize) -> Planner { + Planner { + block_size, + inline_max, + keys: 0, + rec_bytes: 0, + staged: 0, + blocks: 0, + sampled: Vec::new(), + // The smallest stride the fence can choose. Asking the format + // rather than restating it: every larger stride is a power of two + // multiple of this one, so a key the fence samples is always one + // of these. + sample_stride: flatindex::fence_stride(0), + viable: true, + } + } + + /// One key, by its key length and the bytes its values encode to. + /// [`run_len`] turns a key's value lengths into the second. Keys must + /// arrive in the order they will be written, which is key order. + pub fn push(&mut self, key_len: usize, run_len: usize) { + // A key the writer cannot frame with a u16 length, or a run it cannot + // address with a u32 extent, is refused here too: a planner that + // returned a number for a segment that cannot exist would be sizing a + // reserve for a file nobody can write. + if key_len > u16::MAX as usize || run_len > MAX_RUN { + self.viable = false; + return; + } + if self.keys.is_multiple_of(self.sample_stride) { + self.sampled.push(key_len as u32); + } + let inline = self.inline_max > 0 && run_len <= self.inline_max; + let tail = if inline { run_len } else { 0 }; + // One extent per key: that is what a segment writes. Checked, as + // `plan_inline` checks the same sum -- unchecked it would wrap where + // a usize is 32 bits and hand back a reserve for the wrapped total, + // which is a wrong number rather than a refusal. + match self + .rec_bytes + .checked_add(flatindex::record_len_tail(key_len, 1, tail)) + { + Some(n) => self.rec_bytes = n, + None => { + self.viable = false; + return; + } + } + if !inline { + self.blocks += cut(&mut self.staged, run_len, self.block_size); + } + self.keys += 1; + } + + /// How many keys have been pushed. + pub fn keys(&self) -> usize { + self.keys + } + + /// The bytes this planner holds that grow with the input: the sampled key + /// lengths, and nothing else. One `u32` per sixteenth key, so a caller + /// sizing a first pass can check the claim rather than trust it. + pub fn retained_bytes(&self) -> usize { + self.sampled.len() * std::mem::size_of::() + } + + /// The reserve for what has been pushed so far. + /// + /// Exact but for the checksum row, which can be twelve bytes over: the + /// row covers the key section in pieces cut on the *object's* pages, so + /// its length depends on where the section lands, which depends on this + /// answer. It is taken at its worst alignment, where the section starts + /// one byte before a page boundary and cuts one piece more than it + /// otherwise would. That is four bytes, and the 8-aligned boundary behind + /// the row can move by eight because of them. Nothing else rounds. + /// + /// `None` when what was pushed cannot be a segment: a key over 64 KiB, or + /// a key section past the flat index's limits. The writer would refuse it + /// too. + pub fn finish(&self) -> Option { + if !self.viable { + return None; + } + let stride = flatindex::fence_stride(self.keys); + let fence_n = self.keys.div_ceil(stride); + let mut fence_blob_len = 0usize; + for i in 0..fence_n { + // Every sampled key is a multiple of `sample_stride`, so it is in + // hand; if it ever is not, the format changed under this. + let at = (i * stride) / self.sample_stride; + fence_blob_len = fence_blob_len.checked_add(*self.sampled.get(at)? as usize)?; + } + let layout = flatindex::section_layout( + self.keys, + self.rec_bytes, + fence_n, + fence_blob_len, + // No insert room and no record slack: a segment is never edited + // in place, which is exactly how the writer plans it. + 0, + false, + )?; + + let mut blocks = self.blocks; + if self.staged != 0 { + blocks += 1; + } + let table = flatindex::block_table_len(blocks); + let row = flatindex::checksum_row_len( + layout.total, + flatindex::PIECE_SHIFT, + (1u64 << flatindex::PIECE_SHIFT) - 1, + ); + // The fence copy is the span the reader takes: from the offset array + // to the record region, which is what `fence_span` reports. + let fence = if fence_n == 0 { + 0 + } else { + layout.recs_off - layout.fence_offs_off + }; + Some(Reserve { + table, + row, + fence, + directory: self.keys * 4, + }) + } +} + /// What a segment's reserve holds, in bytes, piece by piece. /// /// The order is the writer's: the table first, then the checksum row, the @@ -123,91 +308,26 @@ impl Reserve { } } -/// The reserve a segment of these keys needs. +/// The reserve a segment of these keys needs, for a caller holding a slice. /// /// `keys` is one `(key length, run length)` per key, in key order; [`run_len`] /// turns a key's value lengths into the second. `inline_max` and `block_size` /// are the writer's, and must be the ones it will be given. /// -/// Exact but for the checksum row, which can be twelve bytes over. -/// -/// The row covers the key section in pieces cut on the *object's* pages, so -/// its length depends on where the section lands, which depends on this -/// answer, which is the one circularity in the layout. It is resolved the -/// only way it can be from here: the row is taken at its worst alignment, -/// where the section starts one byte before a page boundary and cuts one -/// piece more than it otherwise would. That is four bytes, and the 8-aligned -/// boundary behind the row can move by eight because of them. Nothing else -/// rounds. -/// -/// `None` when the input cannot be a segment at all: a key over 64 KiB, or a -/// key section past the flat index's limits. The writer would refuse it too. +/// This is [`Planner`] over a slice, and the exactness and the failure cases +/// are its. A caller who will not hold its records should use the planner +/// directly: a slice of lengths is `size_of::<(usize, usize)>()` a key, which +/// is sixteen bytes where a pointer is eight. pub fn for_lengths( keys: &[(usize, usize)], block_size: usize, inline_max: usize, ) -> Option { - let inline = |run: usize| inline_max > 0 && run <= inline_max; - - // Placeholder keys and one extent apiece: the planner reads their lengths - // and the extent count, never the bytes. A segment gives every key one - // extent, and its tail is the run when the run is inline. - let arena = vec![0u8; keys.iter().map(|&(k, _)| k).sum::()]; - let ext = Extents::One(Ext { - block: 0, - off: 0, - len: 0, - last: 0, - count: 0, - }); - let mut all: Vec<(&[u8], &Extents)> = Vec::with_capacity(keys.len()); - let mut at = 0usize; - for &(klen, _) in keys { - all.push((&arena[at..at + klen], &ext)); - at += klen; + let mut p = Planner::new(block_size, inline_max); + for &(key_len, run) in keys { + p.push(key_len, run); } - let tail_arena = vec![0u8; keys.iter().map(|&(_, r)| r).sum::()]; - let mut tails: Vec<&[u8]> = Vec::with_capacity(keys.len()); - let mut at = 0usize; - for &(_, run) in keys { - tails.push(if inline(run) { - &tail_arena[at..at + run] - } else { - &[] - }); - at += run; - } - // No insert room and no record slack: a segment is never edited in place, - // which is exactly how the writer plans it. - let plan = flatindex::plan_inline(&all, &tails, 0, false)?; - - let table = flatindex::block_table_len(blocks_for( - keys.iter().filter(|&&(_, r)| !inline(r)).map(|&(_, r)| r), - block_size, - )); - // The section's own length is the planner's total; the row is appended - // after it and covers everything before itself. - let row = flatindex::checksum_row_len( - plan.total, - flatindex::PIECE_SHIFT, - // The worst base: a section starting one byte before a page boundary - // cuts one more piece than one starting on it. - (1u64 << flatindex::PIECE_SHIFT) - 1, - ); - // The fence copy is the span the reader will take: from the offset array - // to the record region, which is what `fence_span` reports. - let recs_off = plan.total - plan.recs_cap; - let fence = if plan.fence_n == 0 { - 0 - } else { - recs_off - plan.fence_offs_off - }; - Some(Reserve { - table, - row, - fence, - directory: keys.len() * 4, - }) + p.finish() } /// An upper bound on the reserve, for a caller that knows only totals. @@ -237,8 +357,11 @@ pub fn from_totals( // long as the longest, runs as long as the longest, as many of both as // the totals allow. let per_key_run = run_bytes.div_ceil(keys).max(1).min(max_run_len); - let shaped: Vec<(usize, usize)> = (0..keys).map(|_| (max_key_len, per_key_run)).collect(); - let mut need = for_lengths(&shaped, block_size, inline_max)?; + let mut p = Planner::new(block_size, inline_max); + for _ in 0..keys { + p.push(max_key_len, per_key_run); + } + let mut need = p.finish()?; // `for_lengths` on an even shape cuts the blocks evenly, and an uneven one // cuts more. Every block but the last holds more than `block_size - @@ -249,7 +372,7 @@ pub fn from_totals( } else { run_bytes.div_ceil(block_size - max_run_len).min(keys) }; - let even_blocks = blocks_for(shaped.iter().map(|&(_, r)| r), block_size); + let even_blocks = blocks_for(std::iter::repeat_n(per_key_run, keys), block_size); if worst_blocks > even_blocks { need.table = flatindex::block_table_len(worst_blocks); } @@ -320,6 +443,90 @@ mod tests { } } + #[test] + fn the_planner_holds_aggregates_rather_than_the_input() { + // A million keys of sixteen bytes: the planner keeps one u32 per + // sixteenth key and nothing else that grows. + let mut p = Planner::new(64 << 10, 0); + for _ in 0..1_000_000 { + p.push(16, 100); + } + assert_eq!(p.keys(), 1_000_000); + assert_eq!(p.retained_bytes(), 1_000_000usize.div_ceil(16) * 4); + // Against a slice of the same lengths, which is a pair of usizes a + // key -- asked of the target rather than assumed to be sixteen, since + // this crate also builds for wasm32, where it is eight. + let sliced = 1_000_000 * std::mem::size_of::<(usize, usize)>(); + assert!(p.retained_bytes() * 30 < sliced); + assert!(p.finish().is_some()); + } + + #[test] + fn the_planner_and_the_slice_agree() { + // Uneven keys and runs, so the fence blob and the block cut both + // depend on the order they arrive in. + for n in [0usize, 1, 15, 16, 17, 4096, 40_000] { + let keys: Vec<(usize, usize)> = (0..n) + .map(|i| (8 + (i * 7) % 40, 1 + (i * 13) % 900)) + .collect(); + let sliced = for_lengths(&keys, 4096, 256); + let mut p = Planner::new(4096, 256); + for &(k, r) in &keys { + p.push(k, r); + } + assert_eq!(sliced, p.finish(), "{n} keys"); + } + } + + #[test] + fn a_key_too_long_to_frame_is_refused_rather_than_sized() { + let mut p = Planner::new(4096, 0); + p.push(16, 100); + p.push(u16::MAX as usize + 1, 100); + assert!(p.finish().is_none()); + } + + #[test] + fn a_run_past_what_an_extent_addresses_is_refused_rather_than_sized() { + let mut p = Planner::new(4096, 0); + p.push(16, 100); + p.push(16, MAX_RUN + 1); + assert!(p.finish().is_none()); + } + + #[test] + fn record_bytes_that_would_wrap_refuse_rather_than_return_the_wrapped_sum() { + // Runs at the largest a segment can hold, inline so every byte lands + // in the record region. The sum passes usize on a 32-bit target long + // before this many keys, and stays honest on a 64-bit one. + let mut p = Planner::new(4096, MAX_RUN); + for _ in 0..64 { + p.push(16, MAX_RUN); + } + // Either the sum overflowed and it refused, or it did not and the + // section is past what the index can address. Never a number. + assert!(p.finish().is_none()); + } + + #[test] + fn a_huge_run_does_not_wrap_the_block_cut() { + // `cut` used to add the run to what was staged and compare the sum, + // which is what overflows -- at `MAX_RUN` only where a usize is 32 + // bits, so the value here is one that overflows at any width and + // takes the same path. `from_totals` reaches this with whatever + // `max_run_len` its caller passes, so it is not a hypothetical. + // + // Wrapped, the sum comes out small: no block closes and the run is + // left staged. The counts below are what says which happened. + let huge = usize::MAX - 10; + let mut staged = 0usize; + assert_eq!(cut(&mut staged, huge, 4096), 1); + assert_eq!(staged, 0); + let mut staged = 100usize; + assert_eq!(cut(&mut staged, huge, 4096), 2, "the sum wrapped"); + assert_eq!(staged, 0); + } + #[test] fn no_keys_still_reserves_room_for_the_table() { let need = for_lengths(&[], 64 << 10, 0).expect("plannable"); diff --git a/tests/segwriter.rs b/tests/segwriter.rs index 1119627..4de9fe8 100644 --- a/tests/segwriter.rs +++ b/tests/segwriter.rs @@ -1252,13 +1252,6 @@ fn write_sorted_applies_every_per_file_setting() { ..Default::default() }, ), - ( - "no-directory", - supdb::SegmentWrite { - directory_in_reserve: false, - ..Default::default() - }, - ), ] { let batch_path = dir.join(format!("batch-{name}.sup")); let stream_path = dir.join(format!("stream-{name}.sup")); @@ -1353,3 +1346,224 @@ fn compression_does_not_move_the_reserve() { ); } } + +/// `create_with` applies every per-file setting and the reserve it is given, +/// so a streaming caller has nothing to remember. +/// +/// Each of those settings must be set before the first key and each is silent +/// when forgotten, which is why they belong on the constructor rather than in +/// four calls a caller can get wrong. The check is against the four setters +/// called by hand: same bytes, or the constructor is not doing what it says. +#[test] +fn create_with_applies_the_settings_and_the_reserve() { + let _g = serial(); + let dir = scratch("segwriter-create-with"); + let o = opts(); + let data = compressible(500, 0xC0DE); + + let write = supdb::SegmentWrite { + inline_max: INLINE, + compress: true, + sync_every: 8192, + }; + let lengths: Vec<(usize, usize)> = data + .iter() + .map(|(k, vals)| { + let lens: Vec = vals.iter().map(|v| v.len() as u32).collect(); + (k.len(), supdb::reserve::run_len(&lens)) + }) + .collect(); + let r = + supdb::reserve::for_lengths(&lengths, o.block_size, write.inline_max).expect("plannable"); + + let fill = |w: &mut SegmentWriter| { + for (k, vals) in &data { + w.begin(k).expect("begin"); + for v in vals { + w.value(v); + } + w.end().expect("end"); + } + }; + + for (name, reserve) in [ + ("whole", r.bytes()), + ("no-directory", r.without_directory()), + ] { + let via_ctor = dir.join(format!("ctor-{name}.sup")); + let via_setters = dir.join(format!("setters-{name}.sup")); + { + let mut w = + SegmentWriter::create_with(&via_ctor, &o, &write, reserve).expect("create_with"); + fill(&mut w); + w.finish(11).expect("finish"); + } + { + let mut w = SegmentWriter::create(&via_setters, &o).expect("create"); + w.set_inline_max(write.inline_max); + w.set_compress(write.compress); + w.set_sync_every(write.sync_every); + w.set_head_reserve(reserve); + fill(&mut w); + w.finish(11).expect("finish"); + } + assert_eq!( + without_the_clock(&std::fs::read(&via_ctor).unwrap()), + without_the_clock(&std::fs::read(&via_setters).unwrap()), + "{name}: create_with and the setters disagree" + ); + + // The reserve it was handed is the reserve the file has: the open + // plan fits the probe for it, and the smaller one is smaller. + let bytes = std::fs::read(&via_ctor).unwrap(); + let probe = 4096 + reserve as u64; + let head = bytes[..4096].to_vec(); + let plan = supdb::blob::open_sparse_ranges(&head, bytes.len() as u64).unwrap(); + for &(off, len) in &plan { + assert!( + off + len <= probe, + "{name}: the open plan reaches {off}+{len}, past a {probe}-byte probe" + ); + } + let blob = open(&via_ctor); + assert_eq!(blob.keys(), data.len(), "{name}: key count"); + } + assert!( + r.without_directory() < r.bytes(), + "dropping the directory copy saved nothing" + ); + + // And the batch writer is this constructor: same file, same reserve. + let borrowed: Vec<(&[u8], Vec<&[u8]>)> = data + .iter() + .map(|(k, vals)| (k.as_slice(), vals.iter().map(|v| v.as_slice()).collect())) + .collect(); + let items: Vec<(&[u8], &[&[u8]])> = borrowed + .iter() + .map(|(k, vals)| (*k, vals.as_slice())) + .collect(); + let batched = dir.join("batched.sup"); + let used = SegmentWriter::write_sorted(&batched, &o, &write, 11, &items).expect("write_sorted"); + assert_eq!( + used, + r.bytes(), + "write_sorted sized the reserve differently" + ); + assert_eq!( + without_the_clock(&std::fs::read(&batched).unwrap()), + without_the_clock(&std::fs::read(dir.join("ctor-whole.sup")).unwrap()), + "write_sorted and create_with disagree" + ); +} + +/// A segment written without ever holding its records. +/// +/// This is the shape the reserve estimator exists for. The caller streams its +/// input twice: once through `reserve::Planner`, which keeps aggregates and +/// one `u32` per sixteenth key, and once through `create_with`, which already +/// knows the reserve. Nothing in the test holds a key or a value beyond the +/// one it is looking at, which is the property under test -- a version that +/// collected the records into a Vec first would pass every assertion below +/// and prove nothing. +#[test] +fn a_segment_can_be_written_without_holding_its_records() { + let _g = serial(); + let dir = scratch("segwriter-streaming"); + let o = opts(); + let n = 20_000usize; + + // The source: `i` in, one key and its values out, nothing retained. + let key_of = |i: usize| format!("term={i:08}").into_bytes(); + let values_of = |i: usize| -> Vec> { + let count = 1 + (i * 7) % 6; + (0..count) + .map(|j| vec![b'a' + ((i + j) % 5) as u8; 40 + (i % 3) * 30]) + .collect() + }; + + // Pass one: lengths only. + let mut planner = supdb::reserve::Planner::new(o.block_size, INLINE); + for i in 0..n { + let lens: Vec = values_of(i).iter().map(|v| v.len() as u32).collect(); + planner.push(key_of(i).len(), supdb::reserve::run_len(&lens)); + } + let r = planner.finish().expect("plannable"); + let reserve = r.bytes(); + assert_eq!(planner.keys(), n); + // What it held, against what the records would have been. + let record_bytes: usize = (0..n) + .map(|i| key_of(i).len() + values_of(i).iter().map(|v| v.len()).sum::()) + .sum(); + assert!( + planner.retained_bytes() * 100 < record_bytes, + "the planner held {} bytes against {record_bytes} of records", + planner.retained_bytes() + ); + + // Pass two: the records, streamed, with the reserve already known. + let path = dir.join("streamed.sup"); + let write = supdb::SegmentWrite { + inline_max: INLINE, + ..Default::default() + }; + { + let mut w = SegmentWriter::create_with(&path, &o, &write, reserve).expect("create_with"); + for i in 0..n { + let k = key_of(i); + w.begin(&k).expect("begin"); + for v in values_of(i) { + w.value(&v); + } + w.end().expect("end"); + } + w.finish(5).expect("finish"); + } + + // The reserve computed without the records is the reserve the file needs: + // its open plan fits the probe, and the reader answers from it. + let bytes = std::fs::read(&path).unwrap(); + let probe = 4096 + reserve as u64; + assert!( + (bytes.len() as u64) > probe, + "the file fits the probe; proves nothing" + ); + let head = bytes[..4096].to_vec(); + let plan = supdb::blob::open_sparse_ranges(&head, bytes.len() as u64).unwrap(); + for &(off, len) in &plan { + assert!( + off + len <= probe, + "the open plan reaches {off}+{len}, past a {probe}-byte probe" + ); + } + + let blob = open(&path); + assert_eq!(blob.keys(), n); + for i in (0..n).step_by(997) { + let mut got: Vec> = Vec::new(); + blob.read_all(&key_of(i), |v| got.push(v.to_vec())) + .expect("read"); + assert_eq!(got, values_of(i), "key {i} read back differently"); + } + + // And it agrees with the batch writer, which holds everything. + let held: Vec<(Vec, Vec>)> = (0..n).map(|i| (key_of(i), values_of(i))).collect(); + let borrowed: Vec<(&[u8], Vec<&[u8]>)> = held + .iter() + .map(|(k, vals)| (k.as_slice(), vals.iter().map(|v| v.as_slice()).collect())) + .collect(); + let items: Vec<(&[u8], &[&[u8]])> = borrowed + .iter() + .map(|(k, vals)| (*k, vals.as_slice())) + .collect(); + let batched = dir.join("batched.sup"); + let used = SegmentWriter::write_sorted(&batched, &o, &write, 5, &items).expect("write_sorted"); + assert_eq!( + used, reserve, + "streaming and batch sized the reserve differently" + ); + assert_eq!( + without_the_clock(&std::fs::read(&path).unwrap()), + without_the_clock(&std::fs::read(&batched).unwrap()), + "streaming and batch produced different segments" + ); +}