Store keys in key order in SST blocks that omit hashes

@@ -122,12 +122,14 @@ The hashes are sorted.  A Key block contains n keys, which specify n key value pairs. -The block type determines whether the key hash is stored per entry:+The block type determines whether the key hash is stored per entry, and with it the order the+entries are stored in: -- Block type 1 (with hash): Full 8-byte hash stored per entry-- Block type 2 (no hash): No hash stored (for keys ≤ 32 bytes)+- Block type 1 (with hash): Full 8-byte hash stored per entry. Entries are sorted by+  `(key hash, key)`.+- Block type 2 (no hash): No hash stored (for keys ≤ 32 bytes). Entries are sorted by **key**. -During lookup, if block type is 2, the full hash is recomputed from the key data.+See [Entry ordering](#entry-ordering) for why the two differ.  Depending on the `type` field entry has a different format: @@ -167,7 +169,13 @@ Depending on the `type` field entry has a different format: Both ranged kinds are open-ended, so a decoder must test the key-value tombstone range **before** the inline range. -The entries are sorted by key hash and key.+##### Entry ordering++Logically keys are ordered by hash (this is how we chose file and block assignments). However, within a single key block, however, the order is chosen per block type:++- **With hash (types 1 and 3):** sorted by `(key hash, key)`.+- **No hash (types 2 and 4):** sorted by **key** alone.+  ##### Key-value tombstones @@ -236,10 +244,11 @@ Reading start from the current sequence number and goes downwards.   - Check AMQF from SST file for key existence -> if not continue   - let block = 0   - loop-    - Index Block: find key range that contains the key by binary search+    - Index Block: find key range that contains the key by binary search using the **hash** of the key       - found -> set block, continue       - not found -> break-    - Key Block: find key by binary search+    - Key Block: find key by binary search, comparing `(hash, key)` in blocks that store a hash and+      the key alone in blocks that do not (see [Entry ordering](#entry-ordering))       - found -> lookup value from value block, return           - read value as inline, or by using the block index in the key to find the value elsewhere in the file.       - not found -> break

@@ -1161,8 +1161,8 @@ impl Entry for BenchEntry {         8     } -    fn write_key_to(&self, buf: &mut Vec<u8>) {-        buf.extend_from_slice(&self.key);+    fn key_bytes(&self) -> &[u8] {+        &self.key     }      fn value(&self) -> EntryValue<'_> {

@@ -145,8 +145,8 @@ impl<K: StoreKey> Entry for CollectorEntry<K> {         self.key.data.len()     } -    fn write_key_to(&self, buf: &mut Vec<u8>) {-        self.key.data.write_to(buf);+    fn key_bytes(&self) -> &[u8] {+        self.key.data.as_slice()     }      fn value(&self) -> EntryValue<'_> {

@@ -118,105 +118,118 @@ impl<T: KeyBase> KeyBase for &'_ T { /// comparison with a byte slice (total order). pub trait QueryKey: KeyBase {     fn cmp(&self, key: &[u8]) -> std::cmp::Ordering;+    fn eq(&self, key: &[u8]) -> bool; }  impl QueryKey for &'_ [u8] {     fn cmp(&self, key: &[u8]) -> std::cmp::Ordering {         Ord::cmp(self, &key)     }++    fn eq(&self, key: &[u8]) -> bool {+        PartialEq::eq(*self, key)+    } }  impl<const N: usize> QueryKey for [u8; N] {     fn cmp(&self, key: &[u8]) -> std::cmp::Ordering {         Ord::cmp(&self[..], key)     }+    fn eq(&self, key: &[u8]) -> bool {+        PartialEq::eq(self, key)+    } }  impl QueryKey for Vec<u8> {     fn cmp(&self, key: &[u8]) -> std::cmp::Ordering {         Ord::cmp(&**self, key)     }+    fn eq(&self, key: &[u8]) -> bool {+        PartialEq::eq(self.as_slice(), key)+    } }  impl QueryKey for Box<[u8]> {     fn cmp(&self, key: &[u8]) -> std::cmp::Ordering {         Ord::cmp(&**self, key)     }+    fn eq(&self, key: &[u8]) -> bool {+        PartialEq::eq(&**self, key)+    } }  impl QueryKey for u8 {     fn cmp(&self, key: &[u8]) -> std::cmp::Ordering {         Ord::cmp(&[*self][..], key)     }+    fn eq(&self, key: &[u8]) -> bool {+        PartialEq::eq(&[*self][..], key)+    } }  impl<A: QueryKey, B: QueryKey> QueryKey for (A, B) {-    fn cmp(&self, mut key: &[u8]) -> std::cmp::Ordering {+    fn cmp(&self, key: &[u8]) -> std::cmp::Ordering {         let (a, b) = self;         let len = a.len();         let key_len = key.len();-        let key_part = &key[..min(key_len, len)];-        match a.cmp(key_part) {-            std::cmp::Ordering::Equal => {-                key = &key[len..];-                b.cmp(key)-            }-            ord => ord,-        }+        let (key_part, value_part) = key.split_at(min(key_len, len));+        a.cmp(key_part).then_with(|| b.cmp(value_part))+    }+    fn eq(&self, key: &[u8]) -> bool {+        let (a, b) = self;+        let len = a.len();+        let key_len = key.len();+        let (key_part, value_part) = &key.split_at(min(key_len, len));+        a.eq(key_part) && b.eq(value_part)     } }  impl<T: QueryKey> QueryKey for &'_ T {     fn cmp(&self, key: &[u8]) -> std::cmp::Ordering {         (*self).cmp(key)     }+    fn eq(&self, key: &[u8]) -> bool {+        (*self).eq(key)+    } }  /// A trait for keys that can be stored in the database. They need to allow hashing and comparison. pub trait StoreKey: KeyBase + Ord {-    fn write_to(&self, buf: &mut Vec<u8>);+    /// The key's bytes.+    fn as_slice(&self) -> &[u8];++    fn write_to(&self, buf: &mut Vec<u8>) {+        buf.extend_from_slice(self.as_slice());+    } }  impl<const N: usize> StoreKey for [u8; N] {-    fn write_to(&self, buf: &mut Vec<u8>) {-        buf.extend_from_slice(&self[..]);+    fn as_slice(&self) -> &[u8] {+        &self[..]     } }  impl StoreKey for Vec<u8> {-    fn write_to(&self, buf: &mut Vec<u8>) {-        buf.extend_from_slice(self);+    fn as_slice(&self) -> &[u8] {+        self     } }  impl StoreKey for Box<[u8]> {-    fn write_to(&self, buf: &mut Vec<u8>) {-        buf.extend_from_slice(self);+    fn as_slice(&self) -> &[u8] {+        self     } }  impl StoreKey for &'_ [u8] {-    fn write_to(&self, buf: &mut Vec<u8>) {-        buf.extend_from_slice(self);-    }-}--impl StoreKey for u8 {-    fn write_to(&self, buf: &mut Vec<u8>) {-        buf.push(*self);-    }-}--impl<A: StoreKey, B: StoreKey> StoreKey for (A, B) {-    fn write_to(&self, buf: &mut Vec<u8>) {-        self.0.write_to(buf);-        self.1.write_to(buf);+    fn as_slice(&self) -> &[u8] {+        self     } }  impl<T: StoreKey> StoreKey for &'_ T {-    fn write_to(&self, buf: &mut Vec<u8>) {-        (*self).write_to(buf);+    fn as_slice(&self) -> &[u8] {+        (*self).as_slice()     } } 

@@ -82,8 +82,8 @@ pub use key::{KeyBase, QueryKey, StoreKey, hash_key}; pub use meta_file::MetaEntryFlags; pub use parallel_scheduler::{ParallelScheduler, SerialScheduler}; pub use static_sorted_file::{-    BlockCache, BlockCacheLifecycle, BlockWeighter, SstLookupResult, StaticSortedFile,-    StaticSortedFileMetaData,+    BlockCache, BlockCacheLifecycle, BlockWeighter, KeyBlockLayout, SstLookupResult,+    StaticSortedFile, StaticSortedFileMetaData, }; pub use static_sorted_file_builder::{     BLOCK_HEADER_SIZE, Entry, EntryValue, StreamingSstWriter, write_static_stored_file,

@@ -70,8 +70,8 @@ impl Entry for LookupEntry {         self.key.len()     } -    fn write_key_to(&self, buf: &mut Vec<u8>) {-        buf.extend_from_slice(&self.key);+    fn key_bytes(&self) -> &[u8] {+        &self.key     }      fn value(&self) -> EntryValue<'_> {

@@ -33,13 +33,57 @@ use crate::{ pub const BLOCK_TYPE_INDEX: u8 = 0; /// The block header for a key block with 8-byte hash per entry. pub const BLOCK_TYPE_KEY_WITH_HASH: u8 = 1;-/// The block header for a key block without hash.+/// The block header for a key block without hash. Entries are ordered by key. pub const BLOCK_TYPE_KEY_NO_HASH: u8 = 2; /// The block header for a fixed-size key block with 8-byte hash per entry. pub const BLOCK_TYPE_FIXED_KEY_WITH_HASH: u8 = 3;-/// The block header for a fixed-size key block without hash.+/// The block header for a fixed-size key block without hash. Entries are ordered by key. pub const BLOCK_TYPE_FIXED_KEY_NO_HASH: u8 = 4; +/// Whether a key block stores a hash per entry, and therefore what order its entries are in.+#[derive(Clone, Copy, PartialEq, Eq, Debug)]+pub enum KeyBlockLayout {+    /// 8-byte hash stored ahead of each key; entries sorted by `(hash, key)`.+    HashThenKey,+    /// No hash stored; entries sorted by key.+    KeyOnly,+}++impl KeyBlockLayout {+    /// Bytes each entry spends on its stored hash: 8, or 0 when the hash is omitted.+    #[inline]+    pub fn hash_len(self) -> u8 {+        match self {+            KeyBlockLayout::HashThenKey => size_of::<u64>() as u8,+            KeyBlockLayout::KeyOnly => 0,+        }+    }++    /// The on-disk block type byte for this layout, for `fixed`-size or variable-size entries.+    #[inline]+    pub fn block_type(self, fixed: bool) -> u8 {+        match (self, fixed) {+            (KeyBlockLayout::HashThenKey, false) => BLOCK_TYPE_KEY_WITH_HASH,+            (KeyBlockLayout::KeyOnly, false) => BLOCK_TYPE_KEY_NO_HASH,+            (KeyBlockLayout::HashThenKey, true) => BLOCK_TYPE_FIXED_KEY_WITH_HASH,+            (KeyBlockLayout::KeyOnly, true) => BLOCK_TYPE_FIXED_KEY_NO_HASH,+        }+    }++    /// Decodes a key block's type byte into its layout, plus whether entries are fixed-size.+    /// Returns `None` for a byte that is not a key block type.+    #[inline]+    pub fn from_block_type(block_type: u8) -> Option<(Self, bool)> {+        match block_type {+            BLOCK_TYPE_KEY_WITH_HASH => Some((KeyBlockLayout::HashThenKey, false)),+            BLOCK_TYPE_KEY_NO_HASH => Some((KeyBlockLayout::KeyOnly, false)),+            BLOCK_TYPE_FIXED_KEY_WITH_HASH => Some((KeyBlockLayout::HashThenKey, true)),+            BLOCK_TYPE_FIXED_KEY_NO_HASH => Some((KeyBlockLayout::KeyOnly, true)),+            _ => None,+        }+    }+}+ /// Written in a fixed-size key block header's value type field when entries share a value size but /// not a value type. Each entry then carries its own type byte ahead of its value. pub const FIXED_KEY_BLOCK_MIXED_VALUE_TYPE: u8 = 4;@@ -296,23 +340,18 @@ impl StaticSortedFile {             verified_blocks: &self.verified_blocks,         };         let block_type = be::read_u8(&key_block_arc);-        match block_type {-            BLOCK_TYPE_KEY_WITH_HASH | BLOCK_TYPE_KEY_NO_HASH => {-                let has_hash = block_type == BLOCK_TYPE_KEY_WITH_HASH;-                self.lookup_key_block::<K, FIND_ALL>(key_block_arc, key_hash, key, has_hash, reader)+        match KeyBlockLayout::from_block_type(block_type) {+            Some((layout, false)) => {+                self.lookup_key_block::<K, FIND_ALL>(key_block_arc, key_hash, key, layout, reader)             }--            BLOCK_TYPE_FIXED_KEY_WITH_HASH | BLOCK_TYPE_FIXED_KEY_NO_HASH => {-                let has_hash = block_type == BLOCK_TYPE_FIXED_KEY_WITH_HASH;-                self.lookup_fixed_key_block::<K, FIND_ALL>(-                    key_block_arc,-                    key_hash,-                    key,-                    has_hash,-                    reader,-                )-            }-            _ => {+            Some((layout, true)) => self.lookup_fixed_key_block::<K, FIND_ALL>(+                key_block_arc,+                key_hash,+                key,+                layout,+                reader,+            ),+            None => {                 bail!("Invalid block type");             }         }@@ -349,10 +388,10 @@ impl StaticSortedFile {         block: ArcBytes,         key_hash: u64,         key: &K,-        has_hash: bool,+        layout: KeyBlockLayout,         reader: ArcBlockCacheReader<'_>,     ) -> Result<SstLookupResult> {-        let hash_len: u8 = if has_hash { 8 } else { 0 };+        let hash_len = layout.hash_len();         ensure!(block.len() >= 4, "key block too short");         let entry_count = be::read_u24(&block[1..]) as usize;         let data = &block[4..];@@ -363,9 +402,15 @@ impl StaticSortedFile {         let offsets = &data[..entry_count * 4];         let entries = &data[entry_count * 4..]; -        self.lookup_block_inner::<K, FIND_ALL>(&block, entry_count, key_hash, key, reader, |i| {-            get_key_entry(offsets, entries, entry_count, i, hash_len)-        })+        self.lookup_block_inner::<K, FIND_ALL>(+            &block,+            entry_count,+            key_hash,+            key,+            layout,+            reader,+            |i| get_key_entry(offsets, entries, entry_count, i, hash_len),+        )     }      /// Looks up a key in a fixed-size key block.@@ -377,10 +422,10 @@ impl StaticSortedFile {         block: ArcBytes,         key_hash: u64,         key: &K,-        has_hash: bool,+        layout: KeyBlockLayout,         reader: ArcBlockCacheReader<'_>,     ) -> Result<SstLookupResult> {-        let hash_len: u8 = if has_hash { 8 } else { 0 };+        let hash_len = layout.hash_len();         ensure!(block.len() >= 6, "fixed key block too short");         let entry_count = be::read_u24(&block[1..]) as usize;         let key_size = be::read_u8(&block[4..]) as usize;@@ -397,9 +442,15 @@ impl StaticSortedFile {             "fixed key block for {entry_count} entries must is the wrong size"         ); -        self.lookup_block_inner::<K, FIND_ALL>(&block, entry_count, key_hash, key, reader, |i| {-            get_fixed_key_entry(entries, i, hash_len, key_size, value_type, stride)-        })+        self.lookup_block_inner::<K, FIND_ALL>(+            &block,+            entry_count,+            key_hash,+            key,+            layout,+            reader,+            |i| get_fixed_key_entry(entries, i, hash_len, key_size, value_type, stride),+        )     }      /// Shared binary search + collection logic for both key block variants.@@ -412,6 +463,7 @@ impl StaticSortedFile {         entry_count: usize,         key_hash: u64,         key: &K,+        layout: KeyBlockLayout,         reader: ArcBlockCacheReader<'_>,         get_entry: impl Fn(usize) -> Result<GetKeyEntryResult<'a>>,     ) -> Result<SstLookupResult> {@@ -427,7 +479,7 @@ impl StaticSortedFile {                 val,             } = get_entry(m)?; -            let comparison = compare_hash_key(mid_hash, mid_key, key_hash, key);+            let comparison = compare_hash_key(layout, mid_hash, mid_key, key_hash, key);              match comparison {                 Ordering::Less => r = m,@@ -450,7 +502,7 @@ impl StaticSortedFile {                             ty,                             val,                         } = get_entry(i)?;-                        if !entry_matches_key(hash, entry_key, key_hash, key) {+                        if !entry_matches_key(layout, hash, entry_key, key_hash, key) {                             break;                         }                         results.push(self.handle_key_match(ty, val, block, reader)?);@@ -469,7 +521,7 @@ impl StaticSortedFile {                             ty,                             val,                         } = get_entry(i)?;-                        if !entry_matches_key(hash, entry_key, key_hash, key) {+                        if !entry_matches_key(layout, hash, entry_key, key_hash, key) {                             break;                         }                         results.push(self.handle_key_match(ty, val, block, reader)?);@@ -767,24 +819,58 @@ pub struct StaticSortedFileIter {  enum CurrentKeyBlockKind {     /// Variable-size entries with an offset table for random access.-    Variable { offsets: RcBytes, hash_len: u8 },+    Variable { offsets: RcBytes },     /// Fixed-size entries with uniform key size and value size (no offset table).     Fixed {-        hash_len: u8,         key_size: usize,         /// The type shared by every entry, or `None` if each entry carries its own type byte.         value_type: Option<u8>,         stride: usize,     }, } +impl CurrentKeyBlockKind {+    /// Decodes entry `index`, dispatching on the block's entry layout.+    fn entry<'l>(+        &self,+        entries: &'l [u8],+        entry_count: u32,+        index: usize,+        hash_len: u8,+    ) -> Result<GetKeyEntryResult<'l>> {+        match self {+            CurrentKeyBlockKind::Variable { offsets } => {+                get_key_entry(offsets, entries, entry_count as usize, index, hash_len)+            }+            CurrentKeyBlockKind::Fixed {+                key_size,+                value_type,+                stride,+            } => get_fixed_key_entry(entries, index, hash_len, *key_size, *value_type, *stride),+        }+    }+}++/// One entry of a [`CurrentKeyBlock::hash_order`] plan: the key's hash and the index of the entry+/// it was computed from.+struct HashOrderEntry {+    hash: u64,+    entry_index: u32,+}+ struct CurrentKeyBlock {     kind: CurrentKeyBlockKind,+    /// Whether entries carry a hash, and so what order they are stored in.+    layout: KeyBlockLayout,     entries: RcBytes,     /// Number of entries in this key block (max ~819 per 16 KiB block).     entry_count: u32,-    /// Current position within the key block.+    /// Current iteration position. Indexes `hash_order` when that is present, and the block's+    /// entries directly otherwise.     index: u32,+    /// Iteration plan for a [`KeyBlockLayout::KeyOnly`] block, in `(hash, key)` order.+    /// `None` for [`KeyBlockLayout::HashThenKey`], whose entries are already stored in that order.+    hash_order: Option<Vec<HashOrderEntry>>, }  impl Iterator for StaticSortedFileIter {@@ -861,87 +947,77 @@ impl StaticSortedFileIter {         ensure!(data.len() >= 4, "key block too short");         let block_type = data[0];         let entry_count = be::read_u24(&data[1..]);-        match block_type {-            BLOCK_TYPE_KEY_WITH_HASH | BLOCK_TYPE_KEY_NO_HASH => {-                let hash_len = if block_type == BLOCK_TYPE_KEY_WITH_HASH {-                    8-                } else {-                    0-                };-                let n = entry_count as usize;-                let offsets_range = 4..4 + n * 4;-                let entries_range = 4 + n * 4..block.len();-                let offsets = block.clone().slice(offsets_range);-                let entries = block.slice(entries_range);-                Ok(CurrentKeyBlock {-                    kind: CurrentKeyBlockKind::Variable { offsets, hash_len },-                    entries,-                    entry_count,-                    index: 0,-                })-            }-            BLOCK_TYPE_FIXED_KEY_WITH_HASH | BLOCK_TYPE_FIXED_KEY_NO_HASH => {-                let hash_len = if block_type == BLOCK_TYPE_FIXED_KEY_WITH_HASH {-                    8-                } else {-                    0-                };-                let key_size = data[4] as usize;-                let FixedValueLayout {+        let block_len = block.len();+        let Some((layout, fixed)) = KeyBlockLayout::from_block_type(block_type) else {+            bail!("Invalid key block type: {block_type}");+        };+        let hash_len = layout.hash_len();++        let (kind, entries) = if fixed {+            ensure!(data.len() >= 6, "fixed key block too short");+            // In fixed blocks the size of the keys (<=32) is stored immediately after the block len+            // (retrieved above)+            let key_size = data[4] as usize;+            let FixedValueLayout {+                value_type,+                val_size,+                header_size,+            } = fixed_value_layout(data, data[5])?;+            let stride = hash_len as usize + key_size + val_size;+            let entries = block.slice(header_size..block_len);+            (+                CurrentKeyBlockKind::Fixed {+                    key_size,                     value_type,-                    val_size,-                    header_size,-                } = fixed_value_layout(data, data[5])?;-                let stride = hash_len as usize + key_size + val_size;-                let entries_range = header_size..block.len();-                let entries = block.slice(entries_range);-                Ok(CurrentKeyBlock {-                    kind: CurrentKeyBlockKind::Fixed {-                        hash_len,-                        key_size,-                        value_type,-                        stride,-                    },-                    entries,-                    entry_count,-                    index: 0,-                })-            }-            _ => {-                bail!("Invalid key block type: {block_type}");-            }-        }+                    stride,+                },+                entries,+            )+        } else {+            let offset_table_begin = 4usize;+            let offset_table_end = offset_table_begin + (entry_count as usize) * 4;+            // In variable blocks the offsets table starts immediately after the entry count+            let offsets = block.clone().slice(offset_table_begin..offset_table_end);+            let entries = block.slice(offset_table_end..block_len);+            (CurrentKeyBlockKind::Variable { offsets }, entries)+        };++        // Compute the hash order if needed+        let hash_order = match layout {+            KeyBlockLayout::HashThenKey => None,+            KeyBlockLayout::KeyOnly => Some(hash_order_for_block(entry_count, |i| {+                kind.entry(&entries, entry_count, i, hash_len)+            })?),+        };++        Ok(CurrentKeyBlock {+            kind,+            layout,+            entries,+            entry_count,+            index: 0,+            hash_order,+        })     }      /// Gets the next entry in the file and moves the cursor.     fn next_internal(&mut self) -> Result<Option<LookupEntry>> {         loop {             let kb = &mut self.current_key_block;             if kb.index < kb.entry_count {-                let index = kb.index as usize;-                let entry_count = kb.entry_count as usize;-                let GetKeyEntryResult { hash, key, ty, val } = match &kb.kind {-                    CurrentKeyBlockKind::Variable { offsets, hash_len } => {-                        get_key_entry(offsets, &kb.entries, entry_count, index, *hash_len)?+                let (precomputed_hash, index) = match &kb.hash_order {+                    None => (None, kb.index as usize),+                    Some(hash_order) => {+                        let HashOrderEntry { hash, entry_index } = hash_order[kb.index as usize];+                        (Some(hash), entry_index as usize)                     }-                    CurrentKeyBlockKind::Fixed {-                        hash_len,-                        key_size,-                        value_type,-                        stride,-                    } => get_fixed_key_entry(-                        &kb.entries,-                        index,-                        *hash_len,-                        *key_size,-                        *value_type,-                        *stride,-                    )?,                 };-                let full_hash = if hash.is_empty() {-                    crate::key::hash_key(&key)-                } else {-                    be::read_u64(hash)+                let GetKeyEntryResult { hash, key, ty, val } =+                    kb.kind+                        .entry(&kb.entries, kb.entry_count, index, kb.layout.hash_len())?;+                let full_hash = match precomputed_hash {+                    Some(hash) => hash,+                    None => be::read_u64(hash),                 };                 let value = if ty == KEY_BLOCK_ENTRY_TYPE_MEDIUM {                     let block = be::read_u16(val);@@ -991,47 +1067,63 @@ struct GetKeyEntryResult<'l> {     val: &'l [u8], } -/// Compares a query (full_hash + query_key) against an entry (entry_hash + entry_key).-/// Returns the ordering of query relative to entry.-/// When entry_hash is empty, computes full hash from entry_key.+/// Computes `(key hash, entry index)` for every entry of a no-hash key block, in `(hash, key)`+/// order.+fn hash_order_for_block<'l>(+    entry_count: u32,+    get_entry: impl Fn(usize) -> Result<GetKeyEntryResult<'l>>,+) -> Result<Vec<HashOrderEntry>> {+    let mut order = Vec::with_capacity(entry_count as usize);+    for entry_index in 0..entry_count {+        let key = get_entry(entry_index as usize)?.key;+        order.push(HashOrderEntry {+            hash: crate::key::hash_key(&key),+            entry_index,+        });+    }+    // Stable sort by hash, stability is important to preserve the original hash order+    // This keeps tombstones in their correct relative positions.+    order.sort_by_key(|entry| entry.hash);+    Ok(order)+}++/// Compares a query against an entry, returning the ordering of the query relative to the entry in+/// the block's own sort order. fn compare_hash_key<K: QueryKey>(+    layout: KeyBlockLayout,     entry_hash: &[u8],     entry_key: &[u8],     full_hash: u64,     query_key: &K, ) -> Ordering {-    if entry_hash.is_empty() {-        // No hash stored - compute full hash from entry key-        let entry_full_hash = crate::key::hash_key(&entry_key);-        match full_hash.cmp(&entry_full_hash) {-            Ordering::Equal => query_key.cmp(entry_key),-            ord => ord,+    match layout {+        KeyBlockLayout::KeyOnly => {+            debug_assert!(entry_hash.is_empty(), "KeyOnly entries carry no hash");+            query_key.cmp(entry_key)         }-    } else {-        // Full 8-byte hash stored - compare hashes first-        let full_hash_bytes = full_hash.to_be_bytes();-        match full_hash_bytes[..].cmp(entry_hash) {+        KeyBlockLayout::HashThenKey => match full_hash.to_be_bytes()[..].cmp(entry_hash) {             Ordering::Equal => query_key.cmp(entry_key),             ord => ord,-        }+        },     } } -/// Checks if a query key equals an entry key, optionally comparing stored hashes first.-/// When a hash is stored (8 bytes), compares hashes before keys for speed.-/// When no hash is stored, compares keys directly (avoiding hash recomputation).+/// Checks whether a query key names the same entry, used to walk a key group outward from a hit. fn entry_matches_key<K: QueryKey>(+    layout: KeyBlockLayout,     entry_hash: &[u8],     entry_key: &[u8],     full_hash: u64,     query_key: &K, ) -> bool {-    if entry_hash.is_empty() {-        // No hash stored - compare keys directly instead of recomputing hash-        query_key.cmp(entry_key) == Ordering::Equal-    } else {-        // Hash stored - cheap 8-byte comparison first, then key comparison-        full_hash.to_be_bytes()[..] == *entry_hash && query_key.cmp(entry_key) == Ordering::Equal+    match layout {+        KeyBlockLayout::KeyOnly => {+            debug_assert!(entry_hash.is_empty(), "KeyOnly entries carry no hash");+            query_key.eq(entry_key)+        }+        KeyBlockLayout::HashThenKey => {+            full_hash.to_be_bytes()[..] == *entry_hash && query_key.eq(entry_key)+        }     } } @@ -1150,3 +1242,51 @@ fn get_fixed_key_entry<'l>(         val: &entries[val_start..(index + 1) * stride],     }) }++#[cfg(test)]+mod tests {+    use super::*;++    /// `block_type` and `from_block_type` must be inverses over every layout and both entry+    /// sizings. This is what lets the writer and the readers agree: the writer picks a variant and+    /// encodes it, and each reader decodes the same variant back.+    #[test]+    fn block_type_round_trips() {+        for layout in [KeyBlockLayout::HashThenKey, KeyBlockLayout::KeyOnly] {+            for fixed in [false, true] {+                let byte = layout.block_type(fixed);+                assert_eq!(+                    KeyBlockLayout::from_block_type(byte),+                    Some((layout, fixed)),+                    "{layout:?} (fixed={fixed}) encoded as {byte} did not round-trip"+                );+            }+        }+    }++    /// The four key block types must be distinct, and must not collide with the index block type —+    /// a collision would silently route a key block into the index decoder or vice versa.+    #[test]+    fn block_types_are_distinct() {+        let mut seen = vec![BLOCK_TYPE_INDEX];+        for layout in [KeyBlockLayout::HashThenKey, KeyBlockLayout::KeyOnly] {+            for fixed in [false, true] {+                let byte = layout.block_type(fixed);+                assert!(!seen.contains(&byte), "block type {byte} is used twice");+                seen.push(byte);+            }+        }+        assert!(KeyBlockLayout::from_block_type(BLOCK_TYPE_INDEX).is_none());+    }++    /// Only `HashThenKey` stores hash bytes, and it stores exactly a `u64` of them. `get_key_entry`+    /// and `get_fixed_key_entry` both slice the entry using this length.+    #[test]+    fn hash_len_matches_layout() {+        assert_eq!(+            KeyBlockLayout::HashThenKey.hash_len() as usize,+            size_of::<u64>()+        );+        assert_eq!(KeyBlockLayout::KeyOnly.hash_len(), 0);+    }+}

@@ -14,13 +14,11 @@ use crate::{     constants::{MAX_INLINE_VALUE_SIZE, MAX_SMALL_VALUE_SIZE, MIN_SMALL_VALUE_BLOCK_SIZE},     meta_file::MetaEntryFlags,     static_sorted_file::{-        BLOB_VALUE_REF_SIZE, BLOCK_TYPE_FIXED_KEY_NO_HASH, BLOCK_TYPE_FIXED_KEY_WITH_HASH,-        BLOCK_TYPE_INDEX, BLOCK_TYPE_KEY_NO_HASH, BLOCK_TYPE_KEY_WITH_HASH,-        FIXED_KEY_BLOCK_MIXED_VALUE_TYPE, KEY_BLOCK_ENTRY_TYPE_BLOB,-        KEY_BLOCK_ENTRY_TYPE_INLINE_MIN, KEY_BLOCK_ENTRY_TYPE_KEY_DELETED,-        KEY_BLOCK_ENTRY_TYPE_KEY_VALUE_DELETED_MIN, KEY_BLOCK_ENTRY_TYPE_MEDIUM,-        KEY_BLOCK_ENTRY_TYPE_SMALL, KEY_DELETED_REF_SIZE, MEDIUM_VALUE_REF_SIZE,-        SMALL_VALUE_REF_SIZE,+        BLOB_VALUE_REF_SIZE, BLOCK_TYPE_INDEX, FIXED_KEY_BLOCK_MIXED_VALUE_TYPE,+        KEY_BLOCK_ENTRY_TYPE_BLOB, KEY_BLOCK_ENTRY_TYPE_INLINE_MIN,+        KEY_BLOCK_ENTRY_TYPE_KEY_DELETED, KEY_BLOCK_ENTRY_TYPE_KEY_VALUE_DELETED_MIN,+        KEY_BLOCK_ENTRY_TYPE_MEDIUM, KEY_BLOCK_ENTRY_TYPE_SMALL, KEY_DELETED_REF_SIZE,+        KeyBlockLayout, MEDIUM_VALUE_REF_SIZE, SMALL_VALUE_REF_SIZE,     }, }; @@ -56,9 +54,8 @@ const BLOCK_INDEX_CAPACITY_BUFFER: usize = 16;  /// Minimum key size (in bytes) for attempting LZ4 compression on key blocks. ///-/// Keys are sorted by hash, so we should not expect correlation in the data between nearby keys in-/// a block. For small keys (below this threshold), compression is unlikely to be able to exploit-/// patterns and only wastes CPU time. We skip the compression attempt entirely in this case.+/// For small keys (below this threshold), compression is unlikely to find enough to work with and+/// only wastes CPU time, so we skip the attempt entirely. const MIN_KEY_SIZE_FOR_COMPRESSION: usize = 16;  /// Maximum key length that can use fixed-size key block layout.@@ -146,9 +143,17 @@ impl KeyBlockFormat { #[derive(Clone, Copy)] struct KeyBlockFlushInfo {     max_key_len: usize,+    min_key_len: usize,     format: KeyBlockFormat, } +impl KeyBlockFlushInfo {+    /// The shared key length when every entry in the block has the same one, else `None`.+    fn uniform_key_len(&self) -> Option<usize> {+        (self.min_key_len == self.max_key_len).then_some(self.max_key_len)+    }+}+ /// Tracks the accumulated state of the current incomplete key block. /// /// During streaming, this sits on [`StreamingSstWriter`] and tracks the tail of the resolved@@ -161,6 +166,7 @@ struct KeyBlockAccumulator {     entry_count: usize,     /// Maximum key length among accumulated entries (determines whether hashes are stored).     max_key_len: usize,+    min_key_len: usize,     /// Hash of the most recently added entry (used to avoid splitting entries with equal hashes     /// across blocks).     last_hash: u64,@@ -174,6 +180,7 @@ impl KeyBlockAccumulator {             size: 0,             entry_count: 0,             max_key_len: 0,+            min_key_len: usize::MAX,             last_hash: 0,             format: KeyBlockFormat::Unknown,         }@@ -183,6 +190,7 @@ impl KeyBlockAccumulator {     fn add(&mut self, key_len: usize, key_hash: u64, value_type: EntryType) {         self.size += key_len + KEY_BLOCK_ENTRY_META_OVERHEAD;         self.max_key_len = self.max_key_len.max(key_len);+        self.min_key_len = self.min_key_len.min(key_len);         self.entry_count += 1;         self.last_hash = key_hash;         self.format.update(key_len, value_type);@@ -192,6 +200,7 @@ impl KeyBlockAccumulator {     fn flush_info(&self) -> KeyBlockFlushInfo {         KeyBlockFlushInfo {             max_key_len: self.max_key_len,+            min_key_len: self.min_key_len,             format: self.format,         }     }@@ -215,14 +224,31 @@ impl KeyBlockAccumulator {         self.size = 0;         self.entry_count = 0;         self.max_key_len = 0;+        self.min_key_len = usize::MAX;         self.format = KeyBlockFormat::Unknown;         // last_hash is intentionally not reset -- it is overwritten on the next add() call.     } } -/// Determines whether to store the hash per entry based on max key length.-fn use_hash(max_key_len: usize) -> bool {-    max_key_len > 32+/// Chooses a key block's layout from the longest key it holds.+fn choose_layout(max_key_len: usize) -> KeyBlockLayout {+    // Short keys are cheap enough to compare directly that storing an 8-byte hash per entry costs+    // more space than the comparison saves, so those blocks omit it and reorder entries by key.+    if max_key_len > 32 {+        KeyBlockLayout::HashThenKey+    } else {+        KeyBlockLayout::KeyOnly+    }+}++#[inline]+fn be_key_u32(key: &[u8]) -> u32 {+    u32::from_be_bytes(key.try_into().expect("4-byte key"))+}++#[inline]+fn be_key_u64(key: &[u8]) -> u64 {+    u64::from_be_bytes(key.try_into().expect("8-byte key")) }  /// Trait for entries from that SST files can be created@@ -231,8 +257,12 @@ pub trait Entry {     fn key_hash(&self) -> u64;     /// Returns the length of the key     fn key_len(&self) -> usize;+    /// Returns the key's bytes.+    fn key_bytes(&self) -> &[u8];     /// Writes the key to a buffer-    fn write_key_to(&self, buf: &mut Vec<u8>);+    fn write_key_to(&self, buf: &mut Vec<u8>) {+        buf.extend_from_slice(self.key_bytes());+    }      /// Returns the value     fn value(&self) -> EntryValue<'_>;@@ -245,8 +275,8 @@ impl<E: Entry> Entry for &E {     fn key_len(&self) -> usize {         (*self).key_len()     }-    fn write_key_to(&self, buf: &mut Vec<u8>) {-        (*self).write_key_to(buf)+    fn key_bytes(&self) -> &[u8] {+        (*self).key_bytes()     }     fn value(&self) -> EntryValue<'_> {         (*self).value()@@ -884,12 +914,35 @@ impl<E: Entry> StreamingSstWriter<E> {     }      /// Flushes a single key block from `pending_keys[start..end]`.+    ///+    /// Potentially reorders the keys into key order if we are not storing hashes.     fn flush_key_block(&mut self, start: usize, end: usize, info: KeyBlockFlushInfo) -> Result<()> {         let entry_count = end - start;-        let has_hash = use_hash(info.max_key_len);+        let layout = choose_layout(info.max_key_len);         let try_compress = info.max_key_len >= MIN_KEY_SIZE_FOR_COMPRESSION; -        self.key_buffer.clear();+        // Read the boundary hash before reordering, which would move a different entry to `start`.+        // The index block must keep routing by the block's lowest hash.+        let first_hash = self.pending_keys[start].entry.key_hash();+        // Split the borrow of `self` so the block builders can hold `&mut key_buffer` while the+        // loops read `pending_keys`.+        let Self {+            key_buffer,+            pending_keys,+            ..+        } = self;+        key_buffer.clear();+        let build_key_order = |start: usize, end: usize| -> Vec<&PendingEntry<E>> {+            let mut key_order: Vec<&PendingEntry<E>> = pending_keys.range(start..end).collect();++            // Stable sort is important to preserve relative order of tombstones+            match info.uniform_key_len() {+                Some(4) => key_order.sort_by_key(|&e| be_key_u32(e.entry.key_bytes())),+                Some(8) => key_order.sort_by_key(|&e| be_key_u64(e.entry.key_bytes())),+                _ => key_order.sort_by_key(|&e| e.entry.key_bytes()),+            }+            key_order+        };          if let KeyBlockFormat::Fixed {             key_len: key_size,@@ -898,32 +951,38 @@ impl<E: Entry> StreamingSstWriter<E> {         } = info.format         {             let mut builder = FixedKeyBlockBuilder::new(-                &mut self.key_buffer,+                key_buffer,                 entry_count as u32,-                has_hash,+                layout,                 key_size,                 val_size,                 value_type,             );-            for i in start..end {-                let pending = &self.pending_keys[i];-                builder.put(&pending.entry, &pending.value_ref, has_hash);+            if layout == KeyBlockLayout::KeyOnly {+                for pending in build_key_order(start, end) {+                    builder.put(&pending.entry, &pending.value_ref);+                }+            } else {+                for pending in pending_keys.range(start..end) {+                    builder.put_with_hash(&pending.entry, &pending.value_ref);+                }             }             builder.finish();         } else {-            let mut builder =-                KeyBlockBuilder::new(&mut self.key_buffer, entry_count as u32, has_hash);--            for i in start..end {-                let pending = &self.pending_keys[i];-                builder.put(&pending.entry, &pending.value_ref, has_hash);+            let mut builder = KeyBlockBuilder::new(key_buffer, entry_count as u32, layout);+            if layout == KeyBlockLayout::KeyOnly {+                for pending in build_key_order(start, end) {+                    builder.put(&pending.entry, &pending.value_ref);+                }+            } else {+                for pending in pending_keys.range(start..end) {+                    builder.put_with_hash(&pending.entry, &pending.value_ref);+                }             }              builder.finish();         } -        // Record boundary-        let first_hash = self.pending_keys[start].entry.key_hash();         let block_index = write_block_to_file(             self.file.as_mut().unwrap(),             &mut self.compress_buffer,@@ -1122,16 +1181,12 @@ const KEY_BLOCK_HEADER_SIZE: usize = 4;  impl<'l> KeyBlockBuilder<'l> {     /// Creates a new key block builder for the number of entries.-    fn new(buffer: &'l mut Vec<u8>, entry_count: u32, has_hash: bool) -> Self {+    fn new(buffer: &'l mut Vec<u8>, entry_count: u32, layout: KeyBlockLayout) -> Self {         debug_assert!(entry_count < (1 << 24));          const ESTIMATED_KEY_SIZE: usize = 16;         buffer.reserve(entry_count as usize * ESTIMATED_KEY_SIZE);-        let block_type = if has_hash {-            BLOCK_TYPE_KEY_WITH_HASH-        } else {-            BLOCK_TYPE_KEY_NO_HASH-        };+        let block_type = layout.block_type(false);         buffer.write_u8(block_type).unwrap();         buffer.write_u24::<BE>(entry_count).unwrap();         for _ in 0..entry_count {@@ -1152,18 +1207,22 @@ impl<'l> KeyBlockBuilder<'l> {         BE::write_u32(&mut self.buffer[header_offset..header_offset + 4], header);     } +    /// Writes a single entry (header +  key + value data) to the block.+    fn put<E: Entry>(&mut self, entry: &E, value_ref: &ValueRef) {+        self.write_entry_header(value_ref.entry_type());+        entry.write_key_to(self.buffer);+        value_ref.write_value_to(self.buffer);+        self.current_entry += 1;+    }     /// Writes a single entry (header + hash + key + value data) to the block.-    fn put<E: Entry>(&mut self, entry: &E, value_ref: &ValueRef, has_hash: bool) {+    fn put_with_hash<E: Entry>(&mut self, entry: &E, value_ref: &ValueRef) {         self.write_entry_header(value_ref.entry_type());-        if has_hash {-            self.buffer-                .extend_from_slice(&entry.key_hash().to_be_bytes());-        }+        self.buffer+            .extend_from_slice(&entry.key_hash().to_be_bytes());         entry.write_key_to(self.buffer);         value_ref.write_value_to(self.buffer);         self.current_entry += 1;     }-     /// Returns the key block buffer.     fn finish(self) -> &'l mut Vec<u8> {         self.buffer@@ -1193,21 +1252,17 @@ impl<'l> FixedKeyBlockBuilder<'l> {     fn new(         buffer: &'l mut Vec<u8>,         entry_count: u32,-        has_hash: bool,+        layout: KeyBlockLayout,         key_size: u8,         val_size: u8,         value_type: Option<EntryType>,     ) -> Self {-        let hash_len: usize = if has_hash { 8 } else { 0 };+        let hash_len = layout.hash_len() as usize;         let per_entry_type = value_type.is_none();         let stride = hash_len + key_size as usize + val_size as usize + usize::from(per_entry_type);         buffer.reserve(FIXED_KEY_BLOCK_HEADER_SIZE + entry_count as usize * stride); -        let block_type = if has_hash {-            BLOCK_TYPE_FIXED_KEY_WITH_HASH-        } else {-            BLOCK_TYPE_FIXED_KEY_NO_HASH-        };+        let block_type = layout.block_type(true);         buffer.extend_from_slice(&[             block_type,             (entry_count >> 16) as u8,@@ -1228,19 +1283,25 @@ impl<'l> FixedKeyBlockBuilder<'l> {         }     } -    /// Writes a single entry (hash + key + optional type byte + value data) to the block.-    fn put<E: Entry>(&mut self, entry: &E, value_ref: &ValueRef, has_hash: bool) {-        if has_hash {-            self.buffer-                .extend_from_slice(&entry.key_hash().to_be_bytes());-        }+    /// Writes a single entry (key + optional type byte + value data) to the block.+    fn put<E: Entry>(&mut self, entry: &E, value_ref: &ValueRef) {         entry.write_key_to(self.buffer);         if self.per_entry_type {             self.buffer.push(value_ref.entry_type().0);         }         value_ref.write_value_to(self.buffer);     } +    /// Writes a single entry (hash + key + optional type byte + value data) to the block.+    fn put_with_hash<E: Entry>(&mut self, entry: &E, value_ref: &ValueRef) {+        self.buffer+            .extend_from_slice(&entry.key_hash().to_be_bytes());+        entry.write_key_to(self.buffer);+        if self.per_entry_type {+            self.buffer.push(value_ref.entry_type().0);+        }+        value_ref.write_value_to(self.buffer);+    }     fn finish(self) -> &'l mut Vec<u8> {         self.buffer     }@@ -1393,8 +1454,8 @@ mod tests {             self.key.len()         } -        fn write_key_to(&self, buf: &mut Vec<u8>) {-            buf.extend_from_slice(&self.key);+        fn key_bytes(&self) -> &[u8] {+            &self.key         }          fn value(&self) -> EntryValue<'_> {@@ -1415,9 +1476,9 @@ mod tests {         }     } -    /// Sort entries by hash (required by SST writer).+    /// Sort entries by (hash, key) (required by SST writer).     fn sort_entries(entries: &mut [TestEntry]) {-        entries.sort_by_key(|e| e.hash);+        entries.sort_by(|a, b| a.hash.cmp(&b.hash).then_with(|| a.key.cmp(&b.key)));     }      /// Open an SST file for lookup given a path and metadata.@@ -1913,8 +1974,9 @@ mod tests {         let meta = write_sst(dir.path(), 1, &entries, MetaEntryFlags::default())?;         let block = read_first_block(dir.path(), 1, meta.block_count)?; -        assert!(-            block[0] == BLOCK_TYPE_FIXED_KEY_WITH_HASH || block[0] == BLOCK_TYPE_FIXED_KEY_NO_HASH,+        assert_eq!(+            KeyBlockLayout::from_block_type(block[0]).map(|(_, fixed)| fixed),+            Some(true),             "mixed value types of equal size should stay in fixed layout, got block type {}",             block[0]         );@@ -1951,8 +2013,9 @@ mod tests {          let meta = write_sst(dir.path(), 1, &entries, MetaEntryFlags::default())?;         let block = read_first_block(dir.path(), 1, meta.block_count)?;-        assert!(-            block[0] == BLOCK_TYPE_KEY_WITH_HASH || block[0] == BLOCK_TYPE_KEY_NO_HASH,+        assert_eq!(+            KeyBlockLayout::from_block_type(block[0]).map(|(_, fixed)| fixed),+            Some(false),             "differing value sizes should use variable layout, got block type {}",             block[0]         );@@ -2048,4 +2111,70 @@ mod tests {         corrupt_sst_byte(dir.path(), 1, BLOCK_HEADER_SIZE as u64 + 1);         assert_corruption_detected(dir.path(), 1, &meta, &entries);     }++    #[test]+    fn be_key_order_matches_byte_order() {+        let keys4: Vec<[u8; 4]> = vec![+            [0, 0, 0, 0],+            [0, 0, 0, 1],+            [0, 0, 1, 0],+            [0x7f, 0xff, 0xff, 0xff],+            [0x80, 0, 0, 0],+            [0xff, 0xfe, 0, 0],+            [0xff, 0xff, 0xff, 0xff],+        ];+        for a in &keys4 {+            for b in &keys4 {+                assert_eq!(+                    be_key_u32(a).cmp(&be_key_u32(b)),+                    a[..].cmp(&b[..]),+                    "u32 order disagrees with byte order for {a:?} vs {b:?}"+                );+            }+        }+        let keys8: Vec<[u8; 8]> = vec![+            [0; 8],+            [0, 0, 0, 0, 0, 0, 0, 1],+            [0x7f, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff],+            [0x80, 0, 0, 0, 0, 0, 0, 0],+            [0xff; 8],+        ];+        for a in &keys8 {+            for b in &keys8 {+                assert_eq!(+                    be_key_u64(a).cmp(&be_key_u64(b)),+                    a[..].cmp(&b[..]),+                    "u64 order disagrees with byte order for {a:?} vs {b:?}"+                );+            }+        }+    }++    /// `uniform_key_len` must only report a length when the block's keys really are all that long,+    /// since the specialized sorts are unsound otherwise.+    #[test]+    fn uniform_key_len_requires_equal_lengths() {+        let mut acc = KeyBlockAccumulator::new();+        assert_eq!(acc.flush_info().uniform_key_len(), None, "empty block");++        let ty = EntryType(KEY_BLOCK_ENTRY_TYPE_INLINE_MIN);+        acc.add(8, 1, ty);+        acc.add(8, 2, ty);+        assert_eq!(acc.flush_info().uniform_key_len(), Some(8));++        acc.add(4, 3, ty);+        assert_eq!(+            acc.flush_info().uniform_key_len(),+            None,+            "mixed lengths must not report a uniform length"+        );++        acc.reset();+        acc.add(4, 4, ty);+        assert_eq!(+            acc.flush_info().uniform_key_len(),+            Some(4),+            "reset clears min"+        );+    } }

@@ -106,6 +106,13 @@ impl ParallelScheduler for RayonParallelScheduler {     } } +fn tuple_key(prefix: u8, suffix: [u8; 4]) -> Box<[u8]> {+    let mut key = Vec::with_capacity(1 + suffix.len());+    key.push(prefix);+    key.extend_from_slice(&suffix);+    key.into_boxed_slice()+}+ #[test] fn full_cycle() -> Result<()> {     let mut test_cases = Vec::new();@@ -514,13 +521,9 @@ fn persist_changes() -> Result<()> {     let path = tempdir.path();      const READ_COUNT: u32 = 2_000; // we'll read every 10th value, so writes are 10x this value-    fn put(-        b: &WriteBatch<(u8, [u8; 4]), RayonParallelScheduler, 1>,-        key: u8,-        value: u8,-    ) -> Result<()> {+    fn put(b: &WriteBatch<Box<[u8]>, RayonParallelScheduler, 1>, key: u8, value: u8) -> Result<()> {         for i in 0..(READ_COUNT * 10) {-            b.put(0, (key, i.to_be_bytes()), vec![value].into())?;+            b.put(0, tuple_key(key, i.to_be_bytes()), vec![value].into())?;         }         Ok(())     }@@ -646,13 +649,9 @@ fn partial_compaction() -> Result<()> {     let path = tempdir.path();      const READ_COUNT: u32 = 2_000; // we'll read every 10th value, so writes are 10x this value-    fn put(-        b: &WriteBatch<(u8, [u8; 4]), RayonParallelScheduler, 1>,-        key: u8,-        value: u8,-    ) -> Result<()> {+    fn put(b: &WriteBatch<Box<[u8]>, RayonParallelScheduler, 1>, key: u8, value: u8) -> Result<()> {         for i in 0..(READ_COUNT * 10) {-            b.put(0, (key, i.to_be_bytes()), vec![value].into())?;+            b.put(0, tuple_key(key, i.to_be_bytes()), vec![value].into())?;         }         Ok(())     }@@ -747,14 +746,14 @@ fn merge_file_removal() -> Result<()> {      const READ_COUNT: u32 = 2_000; // we'll read every 10th value, so writes are 10x this value     fn put(-        b: &WriteBatch<(u8, [u8; 4]), RayonParallelScheduler, 1>,+        b: &WriteBatch<Box<[u8]>, RayonParallelScheduler, 1>,         key: u8,         value: u32,     ) -> Result<()> {         for i in 0..(READ_COUNT * 10) {             b.put(                 0,-                (key, i.to_be_bytes()),+                tuple_key(key, i.to_be_bytes()),                 value.to_be_bytes().to_vec().into(),             )?;         }

@@ -272,8 +272,8 @@ impl KeyBase for WriteBuffer<'_> { }  impl StoreKey for WriteBuffer<'_> {-    fn write_to(&self, buf: &mut Vec<u8>) {-        buf.extend_from_slice(self);+    fn as_slice(&self) -> &[u8] {+        self     } }