turbo_persistence/
write_batch.rs

1use std::{
2    cell::SyncUnsafeCell,
3    io::Write,
4    mem::{replace, take},
5    path::PathBuf,
6    sync::atomic::{AtomicU32, AtomicU64, Ordering},
7};
8
9use anyhow::{Context, Result, bail};
10use byteorder::{BE, WriteBytesExt};
11use either::Either;
12use fs_err::File;
13use parking_lot::Mutex;
14use smallvec::SmallVec;
15use thread_local::ThreadLocal;
16
17use crate::{
18    FamilyConfig, FamilyKind, ValueBuffer,
19    collector::Collector,
20    collector_entry::CollectorEntry,
21    compression::{Compressor, checksum_block},
22    constants::{MAX_INLINE_VALUE_SIZE, MAX_MEDIUM_VALUE_SIZE, THREAD_LOCAL_SIZE_SHIFT},
23    db::WriteOperationGuard,
24    key::StoreKey,
25    meta_file::MetaEntryFlags,
26    meta_file_builder::MetaFileBuilder,
27    parallel_scheduler::ParallelScheduler,
28    shard::ShardBits,
29    static_sorted_file_builder::{StaticSortedFileBuilderMeta, write_static_stored_file},
30};
31
32/// A newly created database file (meta, SST, or blob), carrying its on-disk size so commit can sum
33/// written bytes without stat'ing the files afterwards.
34pub(crate) struct NewFile {
35    pub(crate) seq: u32,
36    pub(crate) file: File,
37    /// On-disk size in bytes.
38    pub(crate) size: u64,
39}
40
41/// The thread local state of a `WriteBatch`. `FAMILIES` should fit within a `u32`.
42//
43// NOTE: This type *must* use `usize`, even though the real type used in storage is `u32` because
44// there's no way to cast a `u32` to `usize` when declaring an array without the nightly
45// `min_generic_const_args` feature.
46struct ThreadLocalState<K: StoreKey + Send, const FAMILIES: usize> {
47    /// The collectors for each family.
48    collectors: [Option<Collector<K, THREAD_LOCAL_SIZE_SHIFT>>; FAMILIES],
49    /// The list of new blob files that have been created.
50    new_blob_files: Vec<NewFile>,
51}
52
53/// The result of a `WriteBatch::finish` operation.
54pub(crate) struct FinishResult {
55    pub(crate) sequence_number: u32,
56    pub(crate) new_meta_files: Vec<NewFile>,
57    pub(crate) new_sst_files: Vec<NewFile>,
58    pub(crate) new_blob_files: Vec<NewFile>,
59    /// Number of keys written in this batch.
60    pub(crate) keys_written: u64,
61}
62
63/// The most leading key hash bits that split a family's entries into collectors in memory as part
64/// of a commit.
65const MAX_COLLECTOR_SHARD_BITS: ShardBits = ShardBits::new(2);
66
67enum GlobalCollectorState<K: StoreKey + Send> {
68    /// Initial state. Single collector. Once the collector is full, we switch to sharded mode. When
69    /// written, its entries are split into one SST file per shard.
70    Unsharded(Collector<K>),
71    /// Sharded mode. One collector per key hash prefix of the given bits (see
72    /// [`WriteBatch::collector_bits`]).
73    Sharded(ShardBits, Box<[Collector<K>]>),
74}
75
76/// A write batch.
77pub struct WriteBatch<'db, K: StoreKey + Send, S: ParallelScheduler, const FAMILIES: usize> {
78    /// RAII guard that releases the write-operation slot (and rolls back on failure) on drop.
79    _write_guard: WriteOperationGuard<'db>,
80    /// Parallel scheduler
81    parallel_scheduler: S,
82    /// The database path
83    db_path: PathBuf,
84    /// Per-family storage configuration.
85    #[cfg_attr(not(feature = "verify_sst_content"), allow(dead_code))]
86    family_configs: [FamilyConfig; FAMILIES],
87    /// The shards of each family. SST files are split at shard boundaries.
88    shard_bits: [ShardBits; FAMILIES],
89    /// The current sequence number counter. Increased for every new SST file or blob file.
90    current_sequence_number: AtomicU32,
91    /// The thread local state.
92    thread_locals: ThreadLocal<SyncUnsafeCell<ThreadLocalState<K, FAMILIES>>>,
93    /// Collectors in use. The thread local collectors flush into these when they are full.
94    collectors: [Mutex<GlobalCollectorState<K>>; FAMILIES],
95    /// Meta file builders for each family.
96    meta_collectors: [Mutex<Vec<(u32, StaticSortedFileBuilderMeta<'static>)>>; FAMILIES],
97    /// The list of new SST files that have been created.
98    new_sst_files: Mutex<Vec<NewFile>>,
99}
100
101impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize>
102    WriteBatch<'db, K, S, FAMILIES>
103{
104    /// Creates a new write batch for a database with per-family configuration.
105    pub(crate) fn new(
106        write_guard: WriteOperationGuard<'db>,
107        path: PathBuf,
108        current: u32,
109        parallel_scheduler: S,
110        family_configs: [FamilyConfig; FAMILIES],
111        shard_bits: [ShardBits; FAMILIES],
112    ) -> Self {
113        const {
114            assert!(FAMILIES <= usize_from_u32(u32::MAX));
115        };
116        Self {
117            _write_guard: write_guard,
118            parallel_scheduler,
119            db_path: path,
120            family_configs,
121            shard_bits,
122            current_sequence_number: AtomicU32::new(current),
123            thread_locals: ThreadLocal::new(),
124            collectors: [(); FAMILIES]
125                .map(|_| Mutex::new(GlobalCollectorState::Unsharded(Collector::new()))),
126            meta_collectors: [(); FAMILIES].map(|_| Mutex::new(Vec::new())),
127            new_sst_files: Mutex::new(Vec::new()),
128        }
129    }
130
131    /// Marks the write operation as successfully completed.
132    ///
133    /// Must be called before dropping the `WriteBatch` to skip the rollback in the guard's `Drop`
134    /// impl. Typically called by `TurboPersistence::commit_write_batch` after a successful commit.
135    pub(crate) fn mark_succeeded(&mut self) {
136        self._write_guard.success();
137    }
138
139    /// Returns the thread local state for the current thread.
140    #[allow(clippy::mut_from_ref)]
141    fn thread_local_state(&self) -> &mut ThreadLocalState<K, FAMILIES> {
142        let cell = self.thread_locals.get_or(|| {
143            SyncUnsafeCell::new(ThreadLocalState {
144                collectors: [const { None }; FAMILIES],
145                new_blob_files: Vec::new(),
146            })
147        });
148        // Safety: We know that the cell is only accessed from the current thread.
149        unsafe { &mut *cell.get() }
150    }
151
152    /// Returns the collector for a family for the current thread.
153    fn thread_local_collector_mut<'l>(
154        &self,
155        state: &'l mut ThreadLocalState<K, FAMILIES>,
156        family: u32,
157    ) -> Result<&'l mut Collector<K, THREAD_LOCAL_SIZE_SHIFT>> {
158        debug_assert!(usize_from_u32(family) < FAMILIES);
159        let collector =
160            state.collectors[usize_from_u32(family)].get_or_insert_with(|| Collector::new());
161        if collector.is_full() {
162            self.flush_thread_local_collector(family, collector)?;
163        }
164        Ok(collector)
165    }
166
167    #[tracing::instrument(level = "trace", skip(self, collector), fields(family_name = self.family_configs[usize_from_u32(family)].name))]
168    fn flush_thread_local_collector(
169        &self,
170        family: u32,
171        collector: &mut Collector<K, THREAD_LOCAL_SIZE_SHIFT>,
172    ) -> Result<()> {
173        let mut full_collectors = SmallVec::<[_; 2]>::new();
174        {
175            let mut global_collector_state = self.collectors[usize_from_u32(family)].lock();
176            for entry in collector.drain() {
177                match &mut *global_collector_state {
178                    GlobalCollectorState::Unsharded(collector) => {
179                        collector.add_entry(entry);
180                        if collector.is_full() {
181                            let collector_bits = self.collector_bits(family);
182                            if collector_bits.count() == 1 {
183                                full_collectors.push(replace(collector, Collector::new()));
184                            } else {
185                                // When full, split the entries by key hash prefix.
186                                let mut shards = (0..collector_bits.count())
187                                    .map(|_| Collector::new())
188                                    .collect::<Box<[_]>>();
189                                for entry in collector.drain() {
190                                    let shard = collector_bits.shard_of(entry.key.hash) as usize;
191                                    shards[shard].add_entry(entry);
192                                }
193                                // There is a rare edge case where all entries are in the same
194                                // shard, and the collector is full
195                                // after the split.
196                                for collector in shards.iter_mut() {
197                                    if collector.is_full() {
198                                        full_collectors
199                                            .push(replace(&mut *collector, Collector::new()));
200                                    }
201                                }
202                                *global_collector_state =
203                                    GlobalCollectorState::Sharded(collector_bits, shards);
204                            }
205                        }
206                    }
207                    GlobalCollectorState::Sharded(collector_bits, shards) => {
208                        let shard = collector_bits.shard_of(entry.key.hash) as usize;
209                        let collector = &mut shards[shard];
210                        collector.add_entry(entry);
211                        if collector.is_full() {
212                            full_collectors.push(replace(&mut *collector, Collector::new()));
213                        }
214                    }
215                }
216            }
217        }
218        // After flushing write all the full global collectors to disk.
219        // TODO: This can distribute work unfairly
220        // * a thread could fill up multiple global collectors and then get stuck writing them all
221        //   out, if multiple threads could work on it we could take care of spare IO parallism
222        // * we can also have too much IO parallism with many threads concurrently writing files.
223        //
224        // Ideally we would limit the amount of data buffered in memory and control the amount of IO
225        // parallism.  Consider:
226        // * store full-buffers as a field on WireBatch (queued writes)
227        // * each thread will attempt to poll and flush a full buffer after flushing its local
228        //   buffer.
229        // This will distribute the writing work more fairly, but now we have the problem of to
230        // many concurrent writes contending for filesystem locks.  So we could also use a semaphore
231        // to restrict how many concurrent writes occur.  But then we would accumulate 'fullBuffers'
232        // leading to too much memory consumption.  So really we also need to slow down the threads
233        // submitting work data.  To do this we could simply use a tokio semaphore and make all
234        // these operations async, or we could integrate with the parallel::map operation that is
235        // driving the work to slow down task submission in this case.
236        for mut global_collector in full_collectors {
237            // When the global collector is full, we create new SST files, one per shard.
238            let ssts = self.create_sst_files(
239                family,
240                global_collector.sorted(self.family_configs[usize_from_u32(family)].kind),
241            )?;
242            self.new_sst_files.lock().extend(ssts);
243            drop(global_collector);
244        }
245        Ok(())
246    }
247
248    /// Puts a key-value pair into the write batch.
249    pub fn put(&self, family: u32, key: K, value: ValueBuffer<'_>) -> Result<()> {
250        let state = self.thread_local_state();
251        let collector = self.thread_local_collector_mut(state, family)?;
252        if value.len() <= MAX_MEDIUM_VALUE_SIZE {
253            collector.put(key, value);
254        } else {
255            let blob = self.create_blob(family, &value)?;
256            collector.put_blob(key, blob.seq);
257            state.new_blob_files.push(blob);
258        }
259        Ok(())
260    }
261
262    /// Puts a delete operation into the write batch. This deletes *all* values for `key`.
263    ///
264    /// Combining this with a [`WriteBatch::put`] of the same key in the same batch is **not
265    /// supported**: which one wins is undefined, and callers are expected to resolve the intent
266    /// themselves before writing.
267    pub fn delete(&self, family: u32, key: K) -> Result<()> {
268        let state = self.thread_local_state();
269        let collector = self.thread_local_collector_mut(state, family)?;
270        collector.delete(key);
271        Ok(())
272    }
273
274    /// Deletes a single key-value pair, leaving any other values for `key` intact.
275    ///
276    /// Only valid for [`FamilyKind::MultiValue`] families: in a `SingleValue` family a key has one
277    /// value and [`WriteBatch::delete`] already removes it exactly.
278    ///
279    /// Deleting a pair that is written in the same batch — by this or any other operation on the
280    /// key — is **not supported**, for the reason given on [`WriteBatch::delete`]: which one wins
281    /// is undefined, and it is the caller's job to resolve that before writing.
282    ///
283    /// Only values of at most [`MAX_INLINE_VALUE_SIZE`] bytes can be deleted this way.  This is a
284    /// simplifying limitation that could be relaxed if needed. Of course in general the storage
285    /// overhead of deleting large values by value makes it apriori inefficient.
286    pub fn delete_value(&self, family: u32, key: K, value: ValueBuffer<'_>) -> Result<()> {
287        let family_config = &self.family_configs[usize_from_u32(family)];
288        if family_config.kind != FamilyKind::MultiValue {
289            bail!(
290                "delete_value is only valid for MultiValue families, but family {} is SingleValue",
291                family_config.name
292            );
293        }
294        if value.len() > MAX_INLINE_VALUE_SIZE {
295            bail!(
296                "delete_value only supports values of at most {MAX_INLINE_VALUE_SIZE} bytes, got \
297                 {} bytes",
298                value.len()
299            );
300        }
301        let state = self.thread_local_state();
302        let collector = self.thread_local_collector_mut(state, family)?;
303        collector.delete_value(key, &value);
304        Ok(())
305    }
306
307    /// Flushes a family of the write batch, reducing the amount of buffered memory used.
308    /// Does not commit any data persistently.
309    ///
310    /// # Safety
311    ///
312    /// Caller must ensure that no concurrent put or delete operation is happening on the flushed
313    /// family.
314    #[tracing::instrument(level = "trace", skip(self), fields(family_name = self.family_configs[usize_from_u32(family)].name))]
315    pub unsafe fn flush(&self, family: u32) -> Result<()> {
316        // Flush the thread local collectors to the global collector.
317        let mut collectors = Vec::new();
318        for cell in self.thread_locals.iter() {
319            let state = unsafe { &mut *cell.get() };
320            if let Some(collector) = state.collectors[usize_from_u32(family)].take()
321                && !collector.is_empty()
322            {
323                collectors.push(collector);
324            }
325        }
326
327        self.parallel_scheduler
328            .try_parallel_for_each_owned(collectors, |mut collector| {
329                self.flush_thread_local_collector(family, &mut collector)?;
330                drop(collector);
331                anyhow::Ok(())
332            })?;
333
334        // Now we flush the global collector(s).
335        let family_usize = usize_from_u32(family);
336        let mut collector_state = self.collectors[family_usize].lock();
337        let family_config = self.family_configs[family_usize];
338        match &mut *collector_state {
339            GlobalCollectorState::Unsharded(collector) => {
340                if !collector.is_empty() {
341                    // A collector that never filled up was never split into shards, so its
342                    // entries can span all shards.
343                    let ssts =
344                        self.create_sst_files(family, collector.sorted(family_config.kind))?;
345                    collector.clear();
346                    self.new_sst_files.lock().extend(ssts);
347                }
348            }
349            GlobalCollectorState::Sharded(..) => {
350                let GlobalCollectorState::Sharded(_, mut shards) = replace(
351                    &mut *collector_state,
352                    GlobalCollectorState::Unsharded(Collector::new()),
353                ) else {
354                    unreachable!();
355                };
356                self.parallel_scheduler
357                    .try_parallel_for_each_mut(&mut shards, |collector| {
358                        if !collector.is_empty() {
359                            let ssts = self
360                                .create_sst_files(family, collector.sorted(family_config.kind))?;
361                            collector.clear_and_drop_capacity();
362                            self.new_sst_files.lock().extend(ssts);
363                        }
364                        anyhow::Ok(())
365                    })?;
366            }
367        }
368
369        Ok(())
370    }
371
372    /// Finishes the write batch by returning the new sequence number and the new SST files. This
373    /// writes all outstanding thread local data to disk.
374    #[tracing::instrument(level = "trace", skip_all)]
375    pub(crate) fn finish(
376        &mut self,
377        get_accessed_key_hashes: impl Fn(u32) -> qfilter::Filter + Send + Sync,
378    ) -> Result<FinishResult> {
379        let mut new_blob_files = Vec::new();
380
381        // First, we flush all thread local collectors to the global collectors.
382        {
383            let _span = tracing::trace_span!("flush thread local collectors").entered();
384            let mut collectors = [const { Vec::new() }; FAMILIES];
385            for cell in self.thread_locals.iter_mut() {
386                let state = cell.get_mut();
387                new_blob_files.append(&mut state.new_blob_files);
388                for (family, thread_local_collector) in state.collectors.iter_mut().enumerate() {
389                    if let Some(collector) = thread_local_collector.take()
390                        && !collector.is_empty()
391                    {
392                        collectors[family].push(collector);
393                    }
394                }
395            }
396            let to_flush = collectors
397                .into_iter()
398                .enumerate()
399                .flat_map(|(family, collector)| {
400                    collector
401                        .into_iter()
402                        .map(move |collector| (family as u32, collector))
403                })
404                .collect::<Vec<_>>();
405            self.parallel_scheduler.try_parallel_for_each_owned(
406                to_flush,
407                |(family, mut collector)| {
408                    self.flush_thread_local_collector(family, &mut collector)?;
409                    drop(collector);
410                    anyhow::Ok(())
411                },
412            )?;
413        }
414
415        let _span = tracing::trace_span!("flush collectors").entered();
416
417        // Now we reduce the global collectors in parallel
418        let mut new_sst_files = take(self.new_sst_files.get_mut());
419        let shared_new_sst_files = Mutex::new(&mut new_sst_files);
420
421        let new_collectors =
422            [(); FAMILIES].map(|_| Mutex::new(GlobalCollectorState::Unsharded(Collector::new())));
423        let collectors = replace(&mut self.collectors, new_collectors);
424        let collectors = collectors
425            .into_iter()
426            .enumerate()
427            .flat_map(|(family, state)| {
428                let collector = state.into_inner();
429                match collector {
430                    GlobalCollectorState::Unsharded(collector) => {
431                        Either::Left([(family, collector)].into_iter())
432                    }
433                    GlobalCollectorState::Sharded(_, shards) => {
434                        Either::Right(shards.into_iter().map(move |collector| (family, collector)))
435                    }
436                }
437            })
438            .collect::<Vec<_>>();
439        self.parallel_scheduler.try_parallel_for_each_owned(
440            collectors,
441            |(family, mut collector)| {
442                let family = family as u32;
443                if !collector.is_empty() {
444                    let ssts = self.create_sst_files(
445                        family,
446                        collector.sorted(self.family_configs[usize_from_u32(family)].kind),
447                    )?;
448                    collector.clear();
449                    drop(collector);
450                    shared_new_sst_files.lock().extend(ssts);
451                }
452                anyhow::Ok(())
453            },
454        )?;
455
456        // Now we need to write the new meta files.
457        let new_meta_collectors = [(); FAMILIES].map(|_| Mutex::new(Vec::new()));
458        let meta_collectors = replace(&mut self.meta_collectors, new_meta_collectors);
459        let keys_written = AtomicU64::new(0);
460        let file_to_write = meta_collectors
461            .into_iter()
462            .map(|mutex| mutex.into_inner())
463            .enumerate()
464            .filter(|(_, sst_files)| !sst_files.is_empty())
465            .collect::<Vec<_>>();
466        let new_meta_files = self
467            .parallel_scheduler
468            .parallel_map_collect_owned::<_, _, Result<Vec<_>>>(
469                file_to_write,
470                |(family, sst_files)| {
471                    let family = family as u32;
472                    let mut entries = 0;
473                    let mut builder = MetaFileBuilder::new(
474                        family,
475                        self.family_configs[usize_from_u32(family)].compression,
476                        self.shard_bits[usize_from_u32(family)],
477                    );
478                    for (seq, sst) in sst_files {
479                        entries += sst.entries;
480                        builder.add(seq, sst);
481                    }
482                    keys_written.fetch_add(entries, Ordering::Relaxed);
483                    let accessed_key_hashes = get_accessed_key_hashes(family);
484                    builder.set_used_key_hashes_amqf(accessed_key_hashes);
485                    let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
486                    let (file, size) = builder.write(&self.db_path, seq)?;
487                    Ok(NewFile { seq, file, size })
488                },
489            )?;
490
491        // Finally we return the new files and sequence number.
492        let seq = self.current_sequence_number.load(Ordering::SeqCst);
493        Ok(FinishResult {
494            sequence_number: seq,
495            new_meta_files,
496            new_sst_files,
497            new_blob_files,
498            keys_written: keys_written.into_inner(),
499        })
500    }
501
502    /// Creates a new blob file with the given value.
503    #[tracing::instrument(level = "trace", skip(self, value), fields(value_len = value.len()))]
504    fn create_blob(&self, family: u32, value: &[u8]) -> Result<NewFile> {
505        let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
506        let mut compressed = Vec::new();
507        let compression = self.family_configs[usize_from_u32(family)].compression;
508        Compressor::new(compression)?
509            .compress_into_buffer(value, &mut compressed)
510            .context("Compression of value for blob file failed")?;
511
512        let mut buffer = Vec::with_capacity(8 + compressed.len());
513        buffer.write_u32::<BE>(value.len() as u32)?;
514        buffer.write_u32::<BE>(checksum_block(&compressed))?;
515        buffer.extend_from_slice(&compressed);
516
517        let size = buffer.len() as u64;
518        let file = self.db_path.join(format!("{seq:08}.blob"));
519        let mut file = File::create(&file)?;
520        file.write_all(&buffer)?;
521        file.flush()?;
522        Ok(NewFile { seq, file, size })
523    }
524
525    /// The key hash bits that split the entries of `family` into collectors in memory: the shard
526    /// bits of the family, but at most [`MAX_COLLECTOR_SHARD_BITS`].
527    fn collector_bits(&self, family: u32) -> ShardBits {
528        self.shard_bits[usize_from_u32(family)].min(MAX_COLLECTOR_SHARD_BITS)
529    }
530
531    #[tracing::instrument(level = "trace", skip(self, entries), fields(family_name = self.family_configs[usize_from_u32(family)].name))]
532    /// Writes sorted entries that can span multiple shards into one SST file per shard.
533    fn create_sst_files(
534        &self,
535        family: u32,
536        entries: &[CollectorEntry<K>],
537    ) -> Result<SmallVec<[NewFile; 1]>> {
538        let shard_bits = self.shard_bits[usize_from_u32(family)];
539        let mut files = SmallVec::new();
540        let mut rest = entries;
541        while let Some(first) = rest.first() {
542            // Entries are sorted by key hash, so each shard is a contiguous range.
543            let shard = shard_bits.shard_of(first.key.hash);
544            let len = rest.partition_point(|e| shard_bits.shard_of(e.key.hash) == shard);
545            let (chunk, remaining) = rest.split_at(len);
546            files.push(self.create_sst_file(family, chunk)?);
547            rest = remaining;
548        }
549        Ok(files)
550    }
551
552    /// Creates a new SST file with the given collector data.
553    fn create_sst_file(&self, family: u32, entries: &[CollectorEntry<K>]) -> Result<NewFile> {
554        let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
555
556        let path = self.db_path.join(format!("{seq:08}.sst"));
557        let (meta, file) = self
558            .parallel_scheduler
559            .block_in_place(|| {
560                write_static_stored_file(
561                    entries,
562                    &path,
563                    MetaEntryFlags::FRESH,
564                    self.family_configs[usize_from_u32(family)].compression,
565                )
566            })
567            .with_context(|| format!("Unable to write SST file {seq:08}.sst"))?;
568
569        #[cfg(feature = "verify_sst_content")]
570        {
571            use core::panic;
572
573            use crate::{
574                collector_entry::CollectorEntryValue,
575                key::hash_key,
576                lookup_entry::LookupValue,
577                static_sorted_file::{
578                    BlockCache, SstLookupResult, StaticSortedFile, StaticSortedFileMetaData,
579                },
580                static_sorted_file_builder::Entry,
581            };
582
583            file.sync_all()?;
584            let sst = StaticSortedFile::open(
585                &self.db_path,
586                StaticSortedFileMetaData {
587                    sequence_number: seq,
588                    block_count: meta.block_count,
589                },
590                self.family_configs[usize_from_u32(family)].compression,
591                crate::mmap_access_mode(),
592            )?;
593            let cache2 = BlockCache::with(
594                10,
595                u64::MAX,
596                Default::default(),
597                Default::default(),
598                Default::default(),
599            );
600            let cache3 = BlockCache::with(
601                10,
602                u64::MAX,
603                Default::default(),
604                Default::default(),
605                Default::default(),
606            );
607            let mut key_buf = Vec::new();
608            let family_config = self.family_configs[usize_from_u32(family)].kind;
609            for entry in entries {
610                entry.write_key_to(&mut key_buf);
611                let result = sst
612                    .lookup::<_, true>(hash_key(&key_buf), &key_buf, &cache2, &cache3)
613                    .expect("key found");
614                key_buf.clear();
615                match result {
616                    SstLookupResult::Found(values) => {
617                        if values.len() > 1 {
618                            use crate::FamilyKind;
619
620                            assert!(
621                                values.len() == 1 || family_config == FamilyKind::MultiValue,
622                                "only multi-value tables can have more than one value, got {} \
623                                 values",
624                                values.len()
625                            )
626                        }
627                        match &entry.value {
628                            CollectorEntryValue::Large { blob } => {
629                                assert!(
630                                    values.contains(&LookupValue::Blob {
631                                        sequence_number: *blob
632                                    }),
633                                    "we wrote a blob but did not read it"
634                                );
635                            }
636                            // Key tombstones sort last within a key group, so a same-batch
637                            // `put(K, v); delete(K)` reads back as [v, KeyDeleted].
638                            CollectorEntryValue::KeyDeleted => assert!(
639                                values.last() == Some(&LookupValue::KeyDeleted),
640                                "we wrote a key tombstone but it was not last in results"
641                            ),
642                            CollectorEntryValue::KeyValueDeleted { value, len } => {
643                                let expected = &value[..*len as usize];
644                                assert!(
645                                    values.iter().any(|lv| matches!(
646                                        lv,
647                                        LookupValue::KeyValueDeleted { value } if &**value == expected
648                                    )),
649                                    "we wrote a key-value tombstone but did not read it back"
650                                )
651                            }
652                            v => {
653                                assert!(
654                                    values.into_iter().any(|lv| {
655                                        if let LookupValue::Slice { value } = lv {
656                                            &*value == v.as_bytes().unwrap()
657                                        } else {
658                                            false
659                                        }
660                                    }),
661                                    "we wrote a slice of bytes but did not read it"
662                                )
663                            }
664                        }
665                    }
666                    SstLookupResult::NotFound => panic!("All keys must exist"),
667                }
668            }
669        }
670
671        let size = meta.size;
672        self.meta_collectors[usize_from_u32(family)]
673            .lock()
674            .push((seq, meta));
675
676        Ok(NewFile { seq, file, size })
677    }
678}
679
680#[inline(always)]
681const fn usize_from_u32(value: u32) -> usize {
682    // This should always be true, as we assume at least a 32-bit width architecture for Turbopack.
683    // Since this is a const expression, we expect it to be compiled away.
684    const {
685        assert!(u32::BITS <= usize::BITS);
686    };
687    value as usize
688}
turbo_persistence/write_batch.rs

turbo_persistence/
write_batch.rs
