1use crate::wire::manifest::MetadataRow;
23use serde::{Deserialize, Serialize};
24use std::io::Read;
25use std::num::NonZeroUsize;
26use thiserror::Error;
27use xxhash_rust::xxh64::xxh64;
28
29pub const DEFAULT_TARGET_BLOCK_BYTES: usize = 64 * 1024;
31pub const DEFAULT_MAX_DELTA_RUNS: usize = 8;
33pub const DEFAULT_MAX_ROWS_PER_SEGMENT: usize = 65_536;
35pub const DEFAULT_TARGET_SEGMENT_BYTES: usize = 8 * 1024 * 1024;
37pub const DEFAULT_MAX_REORGANIZATION_INPUT_RUNS: usize = 8;
39pub const DEFAULT_MAX_REORGANIZATION_INPUT_ROWS: usize = 131_072;
41pub const DEFAULT_MAX_REORGANIZATION_INPUT_BYTES: usize = 64 * 1024 * 1024;
43pub const DEFAULT_INLINE_FILTER_MAX_BYTES: u32 = 1024;
45pub const RESTART_INTERVAL: usize = 16;
47pub const FILTER_BITS_PER_KEY: usize = 10;
49pub const FILTER_HASH_COUNT: u32 = 7;
51
52const FILTER_HASH_SEED_ONE: u64 = 0;
53const FILTER_HASH_SEED_TWO: u64 = 0x9e37_79b9_7f4a_7c15;
54pub(crate) const ZSTD_LEVEL: i32 = 3;
56
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
60#[serde(deny_unknown_fields)]
61pub struct BlockHandle {
62 pub offset: u64,
64 pub stored_len: u32,
66 pub decoded_len: u32,
68 pub crc32c: u32,
70}
71
72#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
74#[serde(deny_unknown_fields)]
75pub struct SegmentIndexEntry {
76 pub last_row_key: String,
78 pub block: BlockHandle,
80}
81
82#[derive(Debug, Clone, PartialEq, Eq)]
85pub struct BuiltSegmentBlocks {
86 pub bytes: Vec<u8>,
88 pub index: BlockHandle,
90 pub filter: BlockHandle,
92 pub row_count: u64,
94 pub min_row_key: String,
96 pub max_row_key: String,
98}
99
100impl BuiltSegmentBlocks {
101 pub fn inline_filter_hex(&self) -> Option<String> {
103 (self.filter.stored_len <= DEFAULT_INLINE_FILTER_MAX_BYTES).then(|| {
104 let start = self.filter.offset as usize;
105 crate::wire::hex::hex_encode_bytes(
106 &self.bytes[start..start + self.filter.stored_len as usize],
107 )
108 })
109 }
110}
111
112pub fn key_range_may_intersect(
114 min_row_key: &str,
115 max_row_key: &str,
116 row_count: u64,
117 lower_bound: &str,
118 upper_bound: Option<&str>,
119) -> bool {
120 row_count != 0
121 && max_row_key >= lower_bound
122 && upper_bound.is_none_or(|upper_bound| min_row_key < upper_bound)
123}
124
125#[derive(Debug, Clone, PartialEq, Eq)]
129pub struct DecodedDataBlock<R = MetadataRow> {
130 pub row_keys: Vec<String>,
132 pub rows: Vec<R>,
134}
135
136#[derive(Debug, Clone, PartialEq, Eq)]
138pub struct SegmentFilter {
139 n_hashes: u32,
140 bit_len: u64,
141 bits: Vec<u8>,
142}
143
144#[derive(Debug, Clone, PartialEq, Eq, Error)]
148#[non_exhaustive]
149pub enum SstBlockCodecError {
150 #[error("segment must contain at least one row")]
152 EmptySegment,
153 #[error("row key `{offered}` is not in ascending order after `{previous}`")]
155 RowKeysOutOfOrder {
156 previous: String,
158 offered: String,
160 },
161 #[error("stored bytes length {actual} does not match handle length {expected}")]
163 StoredLengthMismatch {
164 expected: u32,
166 actual: usize,
168 },
169 #[error("block checksum mismatch: expected {expected:#010x}, actual {actual:#010x}")]
171 ChecksumMismatch {
172 expected: u32,
174 actual: u32,
176 },
177 #[error("decoded length {actual} does not match handle length {expected}")]
179 DecodedLengthMismatch {
180 expected: u32,
182 actual: usize,
184 },
185 #[error("malformed block: {0}")]
187 Malformed(String),
188 #[error("block codec error: {0}")]
190 Codec(String),
191}
192
193#[derive(Debug)]
195#[must_use]
196pub struct SegmentBlocksBuilder {
197 target_block_bytes: usize,
198 entries: Vec<u8>,
199 restarts: Vec<u32>,
200 entry_count: usize,
201 previous_key: String,
206 bytes: Vec<u8>,
207 index: Vec<SegmentIndexEntry>,
208 decoded_data_bytes: usize,
209 filter_hashes: Vec<(u64, u64)>,
210 row_count: u64,
211 min_row_key: String,
212}
213
214impl Default for SegmentBlocksBuilder {
215 fn default() -> Self {
216 Self::new(const { NonZeroUsize::new(DEFAULT_TARGET_BLOCK_BYTES).unwrap() })
217 }
218}
219
220impl SegmentBlocksBuilder {
221 pub fn new(target_block_bytes: NonZeroUsize) -> Self {
223 Self {
224 target_block_bytes: target_block_bytes.get(),
225 entries: Vec::new(),
226 restarts: Vec::new(),
227 entry_count: 0,
228 previous_key: String::new(),
229 bytes: Vec::new(),
230 index: Vec::new(),
231 decoded_data_bytes: 0,
232 filter_hashes: Vec::new(),
233 row_count: 0,
234 min_row_key: String::new(),
235 }
236 }
237
238 pub fn push<R: Serialize>(
243 &mut self,
244 row_key: &str,
245 filter_key: &str,
246 row: &R,
247 ) -> Result<(), SstBlockCodecError> {
248 self.push_with_encoded_row(row_key, filter_key, row)
249 .map(drop)
250 }
251
252 pub fn push_with_encoded_row<R: Serialize>(
254 &mut self,
255 row_key: &str,
256 filter_key: &str,
257 row: &R,
258 ) -> Result<Vec<u8>, SstBlockCodecError> {
259 if self.row_count > 0 && row_key < self.previous_key.as_str() {
260 return Err(SstBlockCodecError::RowKeysOutOfOrder {
261 previous: self.previous_key.clone(),
262 offered: row_key.to_owned(),
263 });
264 }
265 if self.row_count == 0 {
266 self.min_row_key = row_key.to_owned();
267 }
268 self.filter_hashes.push(filter_key_hashes(filter_key));
269
270 let restart = self.entry_count % RESTART_INTERVAL == 0;
271 if restart {
272 self.restarts.push(self.entries.len() as u32);
273 }
274 let shared_len = if restart {
275 0
276 } else {
277 shared_prefix_len(&self.previous_key, row_key)
278 };
279 let suffix = &row_key.as_bytes()[shared_len..];
280 let mut row_bytes = Vec::new();
281 ciborium::ser::into_writer(row, &mut row_bytes)
282 .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
283 let previous_len = self.entries.len();
284 write_varint(&mut self.entries, shared_len as u64);
285 write_varint(&mut self.entries, suffix.len() as u64);
286 self.entries.extend_from_slice(suffix);
287 write_varint(&mut self.entries, row_bytes.len() as u64);
288 self.entries.extend_from_slice(&row_bytes);
289
290 self.decoded_data_bytes += self.entries.len() - previous_len;
291 self.entry_count += 1;
292 self.row_count += 1;
293 self.previous_key.clear();
294 self.previous_key.push_str(row_key);
295 if self.entries.len() >= self.target_block_bytes {
296 self.finish_data_block()?;
297 }
298 Ok(row_bytes)
299 }
300
301 fn finish_data_block(&mut self) -> Result<(), SstBlockCodecError> {
302 if self.entries.is_empty() {
303 return Ok(());
304 }
305 let mut payload = std::mem::take(&mut self.entries);
306 for restart in &self.restarts {
307 payload.extend_from_slice(&restart.to_le_bytes());
308 }
309 payload.extend_from_slice(&(self.restarts.len() as u32).to_le_bytes());
310 self.decoded_data_bytes += (self.restarts.len() + 1) * 4;
311 self.restarts.clear();
312 self.entry_count = 0;
313 let block = append_section(&mut self.bytes, &payload, true)?;
314 self.index.push(SegmentIndexEntry {
315 last_row_key: self.previous_key.clone(),
316 block,
317 });
318 Ok(())
319 }
320
321 pub fn row_count(&self) -> u64 {
323 self.row_count
324 }
325
326 pub fn decoded_data_bytes(&self) -> usize {
329 self.decoded_data_bytes
330 + if self.entries.is_empty() {
331 0
332 } else {
333 (self.restarts.len() + 1) * 4
334 }
335 }
336
337 pub fn finish(mut self) -> Result<BuiltSegmentBlocks, SstBlockCodecError> {
339 if self.row_count == 0 {
340 return Err(SstBlockCodecError::EmptySegment);
341 }
342 self.finish_data_block()?;
343 let mut bytes = self.bytes;
344 let index = self.index;
345
346 let filter_payload = build_filter_payload(&self.filter_hashes);
347 let filter = append_section(&mut bytes, &filter_payload, false)?;
348
349 let mut index_payload = Vec::new();
350 ciborium::ser::into_writer(&index, &mut index_payload)
351 .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
352 let index = append_section(&mut bytes, &index_payload, true)?;
353
354 Ok(BuiltSegmentBlocks {
355 bytes,
356 index,
357 filter,
358 row_count: self.row_count,
359 min_row_key: self.min_row_key,
360 max_row_key: self.previous_key,
361 })
362 }
363}
364
365pub fn decode_index_block(
367 stored: &[u8],
368 handle: &BlockHandle,
369) -> Result<Vec<SegmentIndexEntry>, SstBlockCodecError> {
370 let payload = decode_section(stored, handle, true)?;
371 let entries: Vec<SegmentIndexEntry> = ciborium::de::from_reader(payload.as_slice())
372 .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
373 let mut previous: Option<(&String, u64)> = None;
379 for entry in &entries {
380 let end = entry
381 .block
382 .offset
383 .checked_add(u64::from(entry.block.stored_len))
384 .ok_or_else(|| {
385 SstBlockCodecError::Malformed(format!(
386 "index block `{}` byte range overflows",
387 entry.last_row_key
388 ))
389 })?;
390 if let Some((previous_key, previous_end)) = previous {
391 if previous_key > &entry.last_row_key {
392 return Err(SstBlockCodecError::Malformed(format!(
393 "index blocks out of key order: `{}` follows `{previous_key}`",
394 entry.last_row_key
395 )));
396 }
397 if entry.block.offset != previous_end {
398 return Err(SstBlockCodecError::Malformed(format!(
399 "index block `{}` does not start where `{previous_key}` ends",
400 entry.last_row_key
401 )));
402 }
403 }
404 previous = Some((&entry.last_row_key, end));
405 }
406 Ok(entries)
407}
408
409pub fn decode_data_block(
411 stored: &[u8],
412 handle: &BlockHandle,
413) -> Result<DecodedDataBlock, SstBlockCodecError> {
414 decode_data_block_rows::<MetadataRow>(stored, handle)
415}
416
417pub fn decode_data_block_rows<R: serde::de::DeserializeOwned>(
420 stored: &[u8],
421 handle: &BlockHandle,
422) -> Result<DecodedDataBlock<R>, SstBlockCodecError> {
423 let payload = decode_section(stored, handle, true)?;
424 if payload.len() < 4 {
425 return Err(SstBlockCodecError::Malformed(
426 "data block shorter than its restart count".to_owned(),
427 ));
428 }
429 let (body, restart_count_bytes) = payload.split_at(payload.len() - 4);
430 let restart_count = u32::from_le_bytes(
431 restart_count_bytes
432 .try_into()
433 .expect("split_at should leave exactly four bytes"),
434 ) as usize;
435 let restarts_len = restart_count
436 .checked_mul(4)
437 .filter(|len| *len <= body.len())
438 .ok_or_else(|| SstBlockCodecError::Malformed("restart array exceeds block".to_owned()))?;
439 let (entries, restarts) = body.split_at(body.len() - restarts_len);
440 let mut restarts = restarts.chunks_exact(4);
441
442 let mut row_keys = Vec::new();
443 let mut rows = Vec::new();
444 let mut cursor = 0usize;
445 let mut previous_key = String::new();
446 while cursor < entries.len() {
447 let restart = rows.len() % RESTART_INTERVAL == 0;
448 if restart {
449 let offset = restarts.next().map(|bytes| {
450 u32::from_le_bytes(
451 bytes
452 .try_into()
453 .expect("restart offsets should have four bytes"),
454 )
455 });
456 if offset.map(u64::from) != Some(cursor as u64) {
457 return Err(SstBlockCodecError::Malformed(
458 "restart offset disagrees with its entry".to_owned(),
459 ));
460 }
461 }
462 let shared_len = read_entry_length(entries, &mut cursor)?;
463 let suffix_len = read_entry_length(entries, &mut cursor)?;
464 if restart && shared_len != 0 {
465 return Err(SstBlockCodecError::Malformed(
466 "restart entry has a shared prefix".to_owned(),
467 ));
468 }
469 if shared_len > previous_key.len() || !previous_key.is_char_boundary(shared_len) {
470 return Err(SstBlockCodecError::Malformed(
471 "shared prefix exceeds previous key".to_owned(),
472 ));
473 }
474 let suffix = take_slice(entries, &mut cursor, suffix_len)?;
475 let suffix = std::str::from_utf8(suffix)
476 .map_err(|_| SstBlockCodecError::Malformed("row key is not utf-8".to_owned()))?;
477 let key_len = shared_len.checked_add(suffix.len()).ok_or_else(|| {
478 SstBlockCodecError::Malformed("row key length exceeds address space".to_owned())
479 })?;
480 let mut key = String::with_capacity(key_len);
481 key.push_str(&previous_key[..shared_len]);
482 key.push_str(suffix);
483 let row_len = read_entry_length(entries, &mut cursor)?;
484 let row_bytes = take_slice(entries, &mut cursor, row_len)?;
485 let row: R = ciborium::de::from_reader(row_bytes)
486 .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
487 if key.as_str() < previous_key.as_str() {
490 return Err(SstBlockCodecError::Malformed(format!(
491 "rows out of row-key order: `{key}` follows `{previous_key}`"
492 )));
493 }
494 previous_key.clear();
495 previous_key.push_str(&key);
496 row_keys.push(key);
497 rows.push(row);
498 }
499 if restarts.next().is_some() {
500 return Err(SstBlockCodecError::Malformed(
501 "restart array contains unused offsets".to_owned(),
502 ));
503 }
504 Ok(DecodedDataBlock { row_keys, rows })
505}
506
507pub fn decode_filter_block(
509 stored: &[u8],
510 handle: &BlockHandle,
511) -> Result<SegmentFilter, SstBlockCodecError> {
512 let payload = decode_section(stored, handle, false)?;
513 if payload.len() < 12 {
514 return Err(SstBlockCodecError::Malformed(
515 "filter block shorter than its header".to_owned(),
516 ));
517 }
518 let n_hashes = u32::from_le_bytes(
519 payload[0..4]
520 .try_into()
521 .expect("header length should be checked above"),
522 );
523 let bit_len = u64::from_le_bytes(
524 payload[4..12]
525 .try_into()
526 .expect("header length should be checked above"),
527 );
528 if n_hashes != FILTER_HASH_COUNT {
529 return Err(SstBlockCodecError::Malformed(
530 "filter hash count must be seven".to_owned(),
531 ));
532 }
533 if bit_len < 64 {
534 return Err(SstBlockCodecError::Malformed(
535 "filter bit length must be at least 64".to_owned(),
536 ));
537 }
538 let bits = &payload[12..];
539 if bit_len.div_ceil(8) != bits.len() as u64 {
540 return Err(SstBlockCodecError::Malformed(
541 "filter bit length disagrees with its bytes".to_owned(),
542 ));
543 }
544 Ok(SegmentFilter {
545 n_hashes,
546 bit_len,
547 bits: bits.to_vec(),
548 })
549}
550
551impl SegmentFilter {
552 pub fn may_contain(&self, filter_key: &str) -> bool {
555 let (h1, h2) = filter_key_hashes(filter_key);
556 for probe in 0..u64::from(self.n_hashes) {
557 let bit = h1.wrapping_add(probe.wrapping_mul(h2)) % self.bit_len;
558 let byte = self.bits[(bit / 8) as usize];
559 if byte & (1 << (bit % 8)) == 0 {
560 return false;
561 }
562 }
563 true
564 }
565}
566
567pub fn string_prefix_upper_bound(prefix: &str) -> Option<String> {
573 let mut bytes = prefix.as_bytes().to_vec();
574 for index in (0..bytes.len()).rev() {
575 if bytes[index] != u8::MAX {
576 bytes[index] += 1;
577 bytes.truncate(index + 1);
578 return String::from_utf8(bytes).ok();
579 }
580 }
581 None
582}
583
584pub fn index_blocks_for_key_range(
587 index: &[SegmentIndexEntry],
588 lower_bound: &str,
589 upper_bound: Option<&str>,
590) -> std::ops::Range<usize> {
591 let start = index.partition_point(|entry| entry.last_row_key.as_str() < lower_bound);
592 let end = upper_bound.map_or(index.len(), |upper_bound| {
593 index
596 .partition_point(|entry| entry.last_row_key.as_str() < upper_bound)
597 .saturating_add(1)
598 .min(index.len())
599 });
600 start..end.max(start)
601}
602
603fn append_section(
604 bytes: &mut Vec<u8>,
605 payload: &[u8],
606 compress: bool,
607) -> Result<BlockHandle, SstBlockCodecError> {
608 let stored = if compress {
609 zstd::bulk::compress(payload, ZSTD_LEVEL)
610 .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?
611 } else {
612 payload.to_vec()
613 };
614 let handle = BlockHandle {
615 offset: bytes.len() as u64,
616 stored_len: stored.len() as u32,
617 decoded_len: payload.len() as u32,
618 crc32c: crc32c::crc32c(&stored),
619 };
620 bytes.extend_from_slice(&stored);
621 Ok(handle)
622}
623
624fn decode_section(
625 stored: &[u8],
626 handle: &BlockHandle,
627 compressed: bool,
628) -> Result<Vec<u8>, SstBlockCodecError> {
629 if stored.len() != handle.stored_len as usize {
630 return Err(SstBlockCodecError::StoredLengthMismatch {
631 expected: handle.stored_len,
632 actual: stored.len(),
633 });
634 }
635 let actual = crc32c::crc32c(stored);
636 if actual != handle.crc32c {
637 return Err(SstBlockCodecError::ChecksumMismatch {
638 expected: handle.crc32c,
639 actual,
640 });
641 }
642 let payload = if compressed {
643 let mut payload =
644 Vec::with_capacity((handle.decoded_len as usize).min(DEFAULT_TARGET_BLOCK_BYTES));
645 zstd::Decoder::new(stored)
646 .and_then(|decoder| {
647 decoder
648 .take(u64::from(handle.decoded_len) + 1)
649 .read_to_end(&mut payload)
650 })
651 .map_err(|error| SstBlockCodecError::Codec(error.to_string()))?;
652 payload
653 } else {
654 if stored.len() != handle.decoded_len as usize {
655 return Err(SstBlockCodecError::DecodedLengthMismatch {
656 expected: handle.decoded_len,
657 actual: stored.len(),
658 });
659 }
660 stored.to_vec()
661 };
662 if payload.len() != handle.decoded_len as usize {
663 return Err(SstBlockCodecError::DecodedLengthMismatch {
664 expected: handle.decoded_len,
665 actual: payload.len(),
666 });
667 }
668 Ok(payload)
669}
670
671fn build_filter_payload(hashes: &[(u64, u64)]) -> Vec<u8> {
672 let bit_len = (hashes.len() * FILTER_BITS_PER_KEY).max(64) as u64;
673 let mut bits = vec![0u8; bit_len.div_ceil(8) as usize];
674 for (h1, h2) in hashes {
675 for probe in 0..u64::from(FILTER_HASH_COUNT) {
676 let bit = h1.wrapping_add(probe.wrapping_mul(*h2)) % bit_len;
677 bits[(bit / 8) as usize] |= 1 << (bit % 8);
678 }
679 }
680 let mut payload = Vec::with_capacity(12 + bits.len());
681 payload.extend_from_slice(&FILTER_HASH_COUNT.to_le_bytes());
682 payload.extend_from_slice(&bit_len.to_le_bytes());
683 payload.extend_from_slice(&bits);
684 payload
685}
686
687fn filter_key_hashes(filter_key: &str) -> (u64, u64) {
688 (
689 xxh64(filter_key.as_bytes(), FILTER_HASH_SEED_ONE),
690 xxh64(filter_key.as_bytes(), FILTER_HASH_SEED_TWO),
691 )
692}
693
694fn shared_prefix_len(previous: &str, current: &str) -> usize {
695 let mut len = previous
696 .as_bytes()
697 .iter()
698 .zip(current.as_bytes())
699 .take_while(|(a, b)| a == b)
700 .count();
701 while !current.is_char_boundary(len) {
704 len -= 1;
705 }
706 len
707}
708
709pub fn write_varint(bytes: &mut Vec<u8>, mut value: u64) {
711 loop {
712 let byte = (value & 0x7f) as u8;
713 value >>= 7;
714 if value == 0 {
715 bytes.push(byte);
716 return;
717 }
718 bytes.push(byte | 0x80);
719 }
720}
721
722pub fn read_varint(bytes: &[u8], cursor: &mut usize) -> Result<u64, SstBlockCodecError> {
724 let mut value = 0u64;
725 let mut shift = 0u32;
726 loop {
727 let byte = *bytes.get(*cursor).ok_or_else(|| {
728 SstBlockCodecError::Malformed("varint runs past the block".to_owned())
729 })?;
730 *cursor += 1;
731 if shift == 63 && byte > 1 {
732 return Err(SstBlockCodecError::Malformed(
733 "varint exceeds 64 bits".to_owned(),
734 ));
735 }
736 value |= u64::from(byte & 0x7f) << shift;
737 if byte & 0x80 == 0 {
738 return Ok(value);
739 }
740 shift += 7;
741 }
742}
743
744fn read_entry_length(bytes: &[u8], cursor: &mut usize) -> Result<usize, SstBlockCodecError> {
745 usize::try_from(read_varint(bytes, cursor)?)
746 .map_err(|_| SstBlockCodecError::Malformed("entry length exceeds address space".to_owned()))
747}
748
749fn take_slice<'a>(
750 bytes: &'a [u8],
751 cursor: &mut usize,
752 len: usize,
753) -> Result<&'a [u8], SstBlockCodecError> {
754 let end = cursor.checked_add(len).filter(|end| *end <= bytes.len());
755 match end {
756 Some(end) => {
757 let slice = &bytes[*cursor..end];
758 *cursor = end;
759 Ok(slice)
760 }
761 None => Err(SstBlockCodecError::Malformed(
762 "entry runs past the block".to_owned(),
763 )),
764 }
765}
766
767#[cfg(test)]
768mod tests {
769 use super::*;
770 use crate::{ChangeSeq, InodeId, InodeKind};
771
772 #[test]
773 fn decoded_byte_accounting_includes_closed_and_open_blocks() {
774 let mut builder =
775 SegmentBlocksBuilder::new(NonZeroUsize::new(512).expect("nonzero block target"));
776 for index in 0..17 {
777 let (key, filter, row) = inode_row(index);
778 builder.push(&key, &filter, &row).expect("encode row");
779 }
780 let decoded_bytes = builder.decoded_data_bytes();
781 assert_eq!(builder.row_count(), 17);
782 assert!(
783 !builder.index.is_empty(),
784 "completed blocks are already compressed"
785 );
786 let built = builder.finish().expect("finish segment");
787 let start = built.index.offset as usize;
788 let end = start + built.index.stored_len as usize;
789 let index = decode_index_block(&built.bytes[start..end], &built.index).expect("index");
790 assert_eq!(
791 decoded_bytes,
792 index
793 .iter()
794 .map(|entry| entry.block.decoded_len as usize)
795 .sum::<usize>(),
796 );
797 }
798
799 fn inode_row(inode_id: u64) -> (String, String, MetadataRow) {
800 let row = MetadataRow::Inode(crate::manifest::InodeRecord {
801 inode_id: InodeId(inode_id),
802 inode_kind: InodeKind::File,
803 created_seq: ChangeSeq(inode_id),
804 commit_id: crate::CommitId::parse(format!("c_row_{inode_id}"))
805 .expect("valid commit id"),
806 created_by: crate::ActorId::loonfs(),
807 created_at_ms: inode_id,
808 });
809 let key = row.row_key();
810 (key.clone(), key, row)
811 }
812
813 fn build_segment(rows: usize) -> BuiltSegmentBlocks {
814 let mut builder = SegmentBlocksBuilder::default();
815 for index in 0..rows {
816 let (key, filter_key, row) = inode_row(index as u64);
817 builder.push(&key, &filter_key, &row).expect("push row");
818 }
819 builder.finish().expect("finish segment")
820 }
821
822 const SHARED_SEGMENT_ROWS: usize = 5_000;
823
824 fn shared_segment() -> &'static BuiltSegmentBlocks {
825 static SEGMENT: std::sync::OnceLock<BuiltSegmentBlocks> = std::sync::OnceLock::new();
826 SEGMENT.get_or_init(|| build_segment(SHARED_SEGMENT_ROWS))
827 }
828
829 fn section<'a>(bytes: &'a [u8], handle: &BlockHandle) -> &'a [u8] {
830 &bytes[handle.offset as usize..handle.offset as usize + handle.stored_len as usize]
831 }
832
833 fn encode_index(entries: &[SegmentIndexEntry]) -> (Vec<u8>, BlockHandle) {
834 let mut payload = Vec::new();
835 ciborium::ser::into_writer(entries, &mut payload).expect("encode index");
836 let mut bytes = Vec::new();
837 let handle = append_section(&mut bytes, &payload, true).expect("append section");
838 (bytes, handle)
839 }
840
841 fn index_entry(last_row_key: &str, offset: u64, stored_len: u32) -> SegmentIndexEntry {
842 SegmentIndexEntry {
843 last_row_key: last_row_key.to_owned(),
844 block: BlockHandle {
845 offset,
846 stored_len,
847 decoded_len: stored_len,
848 crc32c: 0,
849 },
850 }
851 }
852
853 #[test]
854 fn segment_round_trips_every_row_through_index_and_blocks() {
855 let rows = SHARED_SEGMENT_ROWS;
856 let built = shared_segment();
857 let index =
858 decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
859 assert!(index.len() > 1, "5k inode rows should span several blocks");
860
861 let mut recovered = Vec::new();
862 for entry in &index {
863 let block = decode_data_block(section(&built.bytes, &entry.block), &entry.block)
864 .expect("data block");
865 assert_eq!(block.row_keys.len(), block.rows.len());
866 assert_eq!(
867 block.row_keys.last().expect("blocks are never empty"),
868 &entry.last_row_key
869 );
870 recovered.extend(block.row_keys.iter().cloned());
871 }
872 let expected: Vec<String> = (0..rows).map(|i| inode_row(i as u64).0).collect();
873 assert_eq!(recovered, expected);
874 assert_eq!(built.row_count, rows as u64);
875 assert_eq!(built.min_row_key, expected[0]);
876 assert_eq!(&built.max_row_key, expected.last().expect("rows"));
877 }
878
879 #[test]
880 fn index_narrows_point_lookups_to_one_block() {
881 let built = shared_segment();
882 let index =
883 decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
884 let (key, _, row) = inode_row(3_217);
885 let upper = format!("{key}\0");
886 let range = index_blocks_for_key_range(&index, &key, Some(&upper));
887 assert_eq!(range.len(), 1, "a point lookup should touch one block");
888 let entry = &index[range.start];
889 let block =
890 decode_data_block(section(&built.bytes, &entry.block), &entry.block).expect("block");
891 let position = block
892 .row_keys
893 .binary_search_by(|candidate| candidate.as_str().cmp(key.as_str()))
894 .expect("row should be present");
895 assert_eq!(block.rows[position], row);
896 }
897
898 #[test]
899 fn key_range_scan_covers_exactly_the_matching_blocks() {
900 let built = shared_segment();
901 let index =
902 decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
903 let lower = inode_row(1_000).0;
904 let upper = inode_row(1_500).0;
905 let range = index_blocks_for_key_range(&index, &lower, Some(&upper));
906 let mut keys = Vec::new();
907 for entry in &index[range] {
908 let block = decode_data_block(section(&built.bytes, &entry.block), &entry.block)
909 .expect("block");
910 keys.extend(block.row_keys);
911 }
912 let keys: Vec<&String> = keys
913 .iter()
914 .filter(|key| key.as_str() >= lower.as_str() && key.as_str() < upper.as_str())
915 .collect();
916 assert_eq!(keys.len(), 500);
917 }
918
919 #[test]
920 fn out_of_order_and_empty_segments_are_rejected() {
921 let mut builder = SegmentBlocksBuilder::default();
922 let (key_b, filter_b, row_b) = inode_row(2);
923 let (key_a, filter_a, row_a) = inode_row(1);
924 builder.push(&key_b, &filter_b, &row_b).expect("first row");
925 let error = builder
926 .push(&key_a, &filter_a, &row_a)
927 .expect_err("descending key should be rejected");
928 assert!(matches!(
929 error,
930 SstBlockCodecError::RowKeysOutOfOrder { .. }
931 ));
932
933 let error = SegmentBlocksBuilder::default()
934 .finish()
935 .expect_err("empty segment should be rejected");
936 assert!(matches!(error, SstBlockCodecError::EmptySegment));
937 }
938
939 #[test]
940 fn max_key_survives_a_last_row_that_closes_its_block() {
941 let rows = 100usize;
946 let single_block = build_segment(rows);
947 let calibration = decode_index_block(
948 section(&single_block.bytes, &single_block.index),
949 &single_block.index,
950 )
951 .expect("index");
952 assert_eq!(calibration.len(), 1, "the calibration segment is one block");
953 let restarts = rows.div_ceil(RESTART_INTERVAL);
954 let entry_bytes = calibration[0].block.decoded_len as usize - 4 * restarts - 4;
955
956 let mut builder =
957 SegmentBlocksBuilder::new(NonZeroUsize::new(entry_bytes).expect("positive target"));
958 for index in 0..rows {
959 let (key, filter_key, row) = inode_row(index as u64);
960 builder.push(&key, &filter_key, &row).expect("push row");
961 }
962 let built = builder.finish().expect("finish segment");
963
964 let expected_max = inode_row((rows - 1) as u64).0;
965 assert_eq!(built.min_row_key, inode_row(0).0);
966 assert_eq!(built.max_row_key, expected_max);
967 assert_eq!(built.row_count, rows as u64);
968 assert_eq!(built.bytes, single_block.bytes);
971 let index =
972 decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
973 assert_eq!(index.len(), 1);
974 assert_eq!(index[0].last_row_key, expected_max);
975 }
976
977 #[test]
978 fn block_geometry_does_not_change_the_segment_key_range() {
979 let rows = 400usize;
980 let expected: Vec<String> = (0..rows).map(|index| inode_row(index as u64).0).collect();
981 for target in [1usize, 64, 257, 1_024, 4_096, 65_536] {
982 let mut builder =
983 SegmentBlocksBuilder::new(NonZeroUsize::new(target).expect("positive target"));
984 for index in 0..rows {
985 let (key, filter_key, row) = inode_row(index as u64);
986 builder.push(&key, &filter_key, &row).expect("push row");
987 }
988 let built = builder.finish().expect("finish segment");
989 assert_eq!(built.min_row_key, expected[0], "target {target}");
990 assert_eq!(
991 &built.max_row_key,
992 expected.last().expect("rows"),
993 "target {target}"
994 );
995 assert_eq!(built.row_count, rows as u64, "target {target}");
996
997 let index = decode_index_block(section(&built.bytes, &built.index), &built.index)
998 .expect("index");
999 let mut recovered = Vec::new();
1000 for entry in &index {
1001 let block = decode_data_block(section(&built.bytes, &entry.block), &entry.block)
1002 .expect("data block");
1003 assert_eq!(
1006 block.row_keys.last().expect("blocks are never empty"),
1007 &entry.last_row_key,
1008 "target {target}"
1009 );
1010 recovered.extend(block.row_keys);
1011 }
1012 assert_eq!(recovered, expected, "target {target}");
1013 assert_eq!(
1014 index.last().expect("blocks").last_row_key,
1015 built.max_row_key,
1016 "target {target}"
1017 );
1018 }
1019 }
1020
1021 #[test]
1022 fn a_descending_row_after_a_block_boundary_is_rejected() {
1023 let mut builder = SegmentBlocksBuilder::new(NonZeroUsize::MIN);
1024 let (key_high, filter_high, row_high) = inode_row(9);
1025 builder
1026 .push(&key_high, &filter_high, &row_high)
1027 .expect("first row");
1028 let (key_low, filter_low, row_low) = inode_row(3);
1029 let error = builder
1030 .push(&key_low, &filter_low, &row_low)
1031 .expect_err("a descending key across a block boundary should be rejected");
1032 assert!(
1033 matches!(
1034 &error,
1035 SstBlockCodecError::RowKeysOutOfOrder { previous, offered }
1036 if previous == &key_high && offered == &key_low
1037 ),
1038 "unexpected error: {error}"
1039 );
1040 }
1041
1042 #[test]
1043 fn adjacent_equal_keys_are_permitted() {
1044 let mut builder = SegmentBlocksBuilder::default();
1045 let (key, filter_key, row) = inode_row(7);
1046 builder.push(&key, &filter_key, &row).expect("first copy");
1047 builder.push(&key, &filter_key, &row).expect("second copy");
1048 let built = builder.finish().expect("finish");
1049 assert_eq!(built.row_count, 2);
1050 }
1051
1052 #[test]
1053 fn corrupted_sections_fail_their_checksums() {
1054 let built = build_segment(200);
1055 let index =
1056 decode_index_block(section(&built.bytes, &built.index), &built.index).expect("index");
1057
1058 let mut corrupted = built.bytes.clone();
1059 let target = index[0].block.offset as usize + 3;
1060 corrupted[target] ^= 0xff;
1061 let error = decode_data_block(section(&corrupted, &index[0].block), &index[0].block)
1062 .expect_err("corrupted data block should fail");
1063 assert!(matches!(error, SstBlockCodecError::ChecksumMismatch { .. }));
1064
1065 let mut corrupted = built.bytes.clone();
1066 let target = built.index.offset as usize + 3;
1067 corrupted[target] ^= 0xff;
1068 let error = decode_index_block(section(&corrupted, &built.index), &built.index)
1069 .expect_err("corrupted index should fail");
1070 assert!(matches!(error, SstBlockCodecError::ChecksumMismatch { .. }));
1071
1072 let mut corrupted = built.bytes.clone();
1073 let target = built.filter.offset as usize + 12;
1074 corrupted[target] ^= 0xff;
1075 let error = decode_filter_block(section(&corrupted, &built.filter), &built.filter)
1076 .expect_err("corrupted filter should fail");
1077 assert!(matches!(error, SstBlockCodecError::ChecksumMismatch { .. }));
1078 }
1079
1080 #[test]
1081 fn block_expansion_stops_after_the_first_byte_beyond_the_declared_length() {
1082 let mut bytes = Vec::new();
1083 let mut handle = append_section(&mut bytes, &[0; 1024], true).expect("compress block");
1084 handle.decoded_len = 8;
1085 assert_eq!(
1086 decode_data_block(&bytes, &handle).expect_err("expansion exceeds the handle"),
1087 SstBlockCodecError::DecodedLengthMismatch {
1088 expected: 8,
1089 actual: 9,
1090 }
1091 );
1092 }
1093
1094 #[test]
1095 fn a_large_declared_length_does_not_require_a_large_initial_allocation() {
1096 let mut bytes = Vec::new();
1097 let mut handle = append_section(&mut bytes, &[0; 4], true).expect("compress block");
1098 handle.decoded_len = u32::MAX;
1099 assert_eq!(
1100 decode_index_block(&bytes, &handle).expect_err("declared length exceeds expansion"),
1101 SstBlockCodecError::DecodedLengthMismatch {
1102 expected: u32::MAX,
1103 actual: 4,
1104 }
1105 );
1106 }
1107
1108 #[test]
1109 fn truncated_index_and_filter_sections_are_rejected_before_decoding() {
1110 let built = shared_segment();
1111 let index = section(&built.bytes, &built.index);
1112 let filter = section(&built.bytes, &built.filter);
1113 for (handle, error) in [
1114 (
1115 built.index,
1116 decode_index_block(&index[..index.len() - 1], &built.index)
1117 .expect_err("truncated index"),
1118 ),
1119 (
1120 built.filter,
1121 decode_filter_block(&filter[..filter.len() - 1], &built.filter)
1122 .expect_err("truncated filter"),
1123 ),
1124 ] {
1125 assert_eq!(
1126 error,
1127 SstBlockCodecError::StoredLengthMismatch {
1128 expected: handle.stored_len,
1129 actual: handle.stored_len as usize - 1,
1130 }
1131 );
1132 }
1133 }
1134
1135 #[test]
1136 fn filter_headers_reject_zero_bits_and_invalid_hash_counts() {
1137 for (n_hashes, bit_len) in [(FILTER_HASH_COUNT, 0u64), (u32::MAX, 64), (0, 64)] {
1138 let mut payload = Vec::new();
1139 payload.extend_from_slice(&n_hashes.to_le_bytes());
1140 payload.extend_from_slice(&bit_len.to_le_bytes());
1141 payload.resize(12 + bit_len.div_ceil(8) as usize, 0xff);
1142 let mut bytes = Vec::new();
1143 let handle = append_section(&mut bytes, &payload, false).expect("filter section");
1144 assert!(matches!(
1145 decode_filter_block(&bytes, &handle),
1146 Err(SstBlockCodecError::Malformed(_))
1147 ));
1148 }
1149 }
1150
1151 #[test]
1152 fn restart_entries_bound_shared_prefix_reconstruction() {
1153 let mut payload = Vec::new();
1154 let mut restart_offsets = Vec::new();
1155 for index in 0..=RESTART_INTERVAL {
1156 if index % RESTART_INTERVAL == 0 {
1157 restart_offsets.push(payload.len() as u32);
1158 }
1159 write_varint(&mut payload, u64::from(index != 0));
1160 write_varint(&mut payload, u64::from(index == 0));
1161 if index == 0 {
1162 payload.push(b'a');
1163 }
1164 write_varint(&mut payload, 1);
1165 payload.push(0);
1166 }
1167 for offset in restart_offsets {
1168 payload.extend_from_slice(&offset.to_le_bytes());
1169 }
1170 payload.extend_from_slice(&2u32.to_le_bytes());
1171 let mut bytes = Vec::new();
1172 let handle = append_section(&mut bytes, &payload, true).expect("data section");
1173 assert_eq!(
1174 decode_data_block_rows::<u8>(&bytes, &handle).expect_err("shared restart prefix"),
1175 SstBlockCodecError::Malformed("restart entry has a shared prefix".to_owned())
1176 );
1177 }
1178
1179 #[test]
1180 fn restart_offsets_must_name_their_entries() {
1181 let payload = [0, 1, b'a', 1, 0, 1, 0, 0, 0, 1, 0, 0, 0];
1182 let mut bytes = Vec::new();
1183 let handle = append_section(&mut bytes, &payload, true).expect("data section");
1184 assert_eq!(
1185 decode_data_block_rows::<u8>(&bytes, &handle).expect_err("wrong restart offset"),
1186 SstBlockCodecError::Malformed("restart offset disagrees with its entry".to_owned())
1187 );
1188 }
1189
1190 #[test]
1191 fn varints_reject_overflow_in_the_tenth_byte() {
1192 let mut bytes = [0xff; 10];
1193 for last in [2, 0x80] {
1194 bytes[9] = last;
1195 assert_eq!(
1196 read_varint(&bytes, &mut 0).expect_err("overflowing varint"),
1197 SstBlockCodecError::Malformed("varint exceeds 64 bits".to_owned())
1198 );
1199 }
1200 bytes[9] = 1;
1201 assert_eq!(
1202 read_varint(&bytes, &mut 0).expect("largest varint"),
1203 u64::MAX
1204 );
1205 }
1206
1207 #[test]
1208 fn filter_has_no_false_negatives_and_few_false_positives() {
1209 let rows = 2_000;
1210 let built = build_segment(rows);
1211 let filter = decode_filter_block(section(&built.bytes, &built.filter), &built.filter)
1212 .expect("filter");
1213 for index in 0..rows {
1214 let (key, _, _) = inode_row(index as u64);
1215 assert!(filter.may_contain(&key), "inserted key must stay positive");
1216 }
1217 let mut false_positives = 0usize;
1218 let probes = 10_000usize;
1219 for index in 0..probes {
1220 let (absent, _, _) = inode_row((rows + 10_000 + index) as u64);
1221 if filter.may_contain(&absent) {
1222 false_positives += 1;
1223 }
1224 }
1225 let rate = false_positives as f64 / probes as f64;
1226 assert!(rate < 0.02, "false positive rate {rate} exceeds 2%");
1227 }
1228
1229 #[test]
1230 fn decoding_rejects_out_of_order_rows_in_a_block() {
1231 let mut entries = Vec::new();
1234 for inode in [9u64, 3u64] {
1235 let (key, _, row) = inode_row(inode);
1236 let mut row_bytes = Vec::new();
1237 ciborium::ser::into_writer(&row, &mut row_bytes).expect("encode row");
1238 write_varint(&mut entries, 0);
1239 write_varint(&mut entries, key.len() as u64);
1240 entries.extend_from_slice(key.as_bytes());
1241 write_varint(&mut entries, row_bytes.len() as u64);
1242 entries.extend_from_slice(&row_bytes);
1243 }
1244 let mut payload = entries;
1245 payload.extend_from_slice(&0u32.to_le_bytes());
1246 payload.extend_from_slice(&1u32.to_le_bytes());
1247 let mut bytes = Vec::new();
1248 let handle = append_section(&mut bytes, &payload, true).expect("append section");
1249
1250 let error =
1251 decode_data_block(&bytes, &handle).expect_err("descending rows should be rejected");
1252 assert!(
1253 matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("row-key order")),
1254 "unexpected error: {error}"
1255 );
1256 }
1257
1258 #[test]
1259 fn decoding_rejects_a_shared_prefix_inside_a_utf8_code_point() {
1260 let (_, _, row) = inode_row(1);
1261 let mut row_bytes = Vec::new();
1262 ciborium::ser::into_writer(&row, &mut row_bytes).expect("encode row");
1263 let mut payload = Vec::new();
1264 write_varint(&mut payload, 0);
1265 write_varint(&mut payload, "é".len() as u64);
1266 payload.extend_from_slice("é".as_bytes());
1267 write_varint(&mut payload, row_bytes.len() as u64);
1268 payload.extend_from_slice(&row_bytes);
1269 write_varint(&mut payload, 1);
1270 write_varint(&mut payload, 0);
1271 write_varint(&mut payload, row_bytes.len() as u64);
1272 payload.extend_from_slice(&row_bytes);
1273 payload.extend_from_slice(&0u32.to_le_bytes());
1274 payload.extend_from_slice(&1u32.to_le_bytes());
1275
1276 let mut bytes = Vec::new();
1277 let handle = append_section(&mut bytes, &payload, true).expect("append section");
1278 let error = decode_data_block(&bytes, &handle)
1279 .expect_err("a partial utf-8 prefix should be rejected");
1280 assert!(matches!(
1281 &error,
1282 SstBlockCodecError::Malformed(message)
1283 if message == "shared prefix exceeds previous key"
1284 ));
1285 }
1286
1287 #[test]
1288 fn decoding_rejects_out_of_order_index_entries() {
1289 let entries = vec![
1290 index_entry("inode-00000000000000000009", 0, 1),
1291 index_entry("inode-00000000000000000003", 0, 1),
1292 ];
1293 let (bytes, handle) = encode_index(&entries);
1294
1295 let error =
1296 decode_index_block(&bytes, &handle).expect_err("descending index should be rejected");
1297 assert!(
1298 matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("key order")),
1299 "unexpected error: {error}"
1300 );
1301 }
1302
1303 #[test]
1304 fn decoding_rejects_out_of_order_index_offsets() {
1305 let entries = [index_entry("a", 10, 1), index_entry("b", 5, 1)];
1306 let (bytes, handle) = encode_index(&entries);
1307
1308 let error = decode_index_block(&bytes, &handle)
1309 .expect_err("descending block offsets should be rejected");
1310 assert!(
1311 matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("does not start where")),
1312 "unexpected error: {error}"
1313 );
1314 }
1315
1316 #[test]
1317 fn decoding_rejects_overlapping_index_ranges() {
1318 let entries = [index_entry("a", 10, 5), index_entry("b", 14, 1)];
1319 let (bytes, handle) = encode_index(&entries);
1320
1321 let error = decode_index_block(&bytes, &handle)
1322 .expect_err("overlapping block ranges should be rejected");
1323 assert!(
1324 matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("does not start where")),
1325 "unexpected error: {error}"
1326 );
1327 }
1328
1329 #[test]
1330 fn decoding_rejects_a_gap_between_index_blocks() {
1331 let entries = [index_entry("a", 0, 10), index_entry("b", 20, 1)];
1332 let (bytes, handle) = encode_index(&entries);
1333
1334 let error = decode_index_block(&bytes, &handle)
1335 .expect_err("a gap between block ranges should be rejected");
1336 assert!(
1337 matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("does not start where")),
1338 "unexpected error: {error}"
1339 );
1340 }
1341
1342 #[test]
1343 fn decoding_rejects_a_single_block_range_past_the_integer_edge() {
1344 let entries = [index_entry("a", u64::MAX - 10, 100)];
1345 let (bytes, handle) = encode_index(&entries);
1346
1347 let error = decode_index_block(&bytes, &handle)
1348 .expect_err("an overflowing single range should be rejected");
1349 assert!(
1350 matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("byte range overflows")),
1351 "unexpected error: {error}"
1352 );
1353 }
1354
1355 #[test]
1356 fn decoding_rejects_a_final_block_range_past_the_integer_edge() {
1357 let entries = [
1358 index_entry("a", u64::MAX - 110, 100),
1359 index_entry("b", u64::MAX - 10, 100),
1360 ];
1361 let (bytes, handle) = encode_index(&entries);
1362
1363 let error = decode_index_block(&bytes, &handle)
1364 .expect_err("a trailing overflowing range should be rejected");
1365 assert!(
1366 matches!(&error, SstBlockCodecError::Malformed(message) if message.contains("byte range overflows")),
1367 "unexpected error: {error}"
1368 );
1369 }
1370}