risingwave_storage/hummock/sstable/
mod.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Hummock state store's SST builder, format and iterator
16
17// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
18mod block;
19
20use std::collections::HashSet;
21use std::fmt::{Debug, Formatter};
22use std::ops::{BitXor, Bound, Range};
23
24pub use block::*;
25mod block_iterator;
26pub use block_iterator::*;
27mod xor_filter;
28use serde::{Deserialize, Serialize};
29pub use xor_filter::{
30    BlockedXor8FilterBuilder, BlockedXor16FilterBuilder, Xor8FilterBuilder, Xor16FilterBuilder,
31    XorFilterReader,
32};
33pub mod builder;
34pub use builder::*;
35pub mod writer;
36use risingwave_common::catalog::TableId;
37pub use writer::*;
38mod forward_sstable_iterator;
39pub mod multi_builder;
40use bytes::{Buf, BufMut};
41pub use forward_sstable_iterator::*;
42use tracing::warn;
43mod backward_sstable_iterator;
44pub use backward_sstable_iterator::*;
45use risingwave_hummock_sdk::key::{FullKey, KeyPayloadType, UserKey, UserKeyRangeRef};
46use risingwave_hummock_sdk::{HummockEpoch, HummockSstableObjectId};
47
48mod filter;
49mod utils;
50
51pub use filter::{
52    DEFAULT_FILTER_HASH_PREALLOC_KEY_COUNT_CAP, FilterBuilder, FilterBuilderOptions,
53    NoneFilterBuilder,
54};
55pub use utils::{CompressionAlgorithm, xxhash64_checksum, xxhash64_verify};
56use utils::{get_length_prefixed_slice, put_length_prefixed_slice};
57use xxhash_rust::xxh64;
58
59use super::{HummockError, HummockResult};
60use crate::hummock::CachePolicy;
61use crate::store::ReadOptions;
62
63const MAGIC: u32 = 0x5785ab73;
64const OLD_VERSION: u32 = 1;
65const VERSION: u32 = 2;
66
67/// Assume that watermark1 is 5, watermark2 is 7, watermark3 is 11, delete ranges
68/// `{ [0, wmk1) in epoch1, [wmk1, wmk2) in epoch2, [wmk2, wmk3) in epoch3 }`
69/// can be transformed into events below:
70/// `{ <0, +epoch1> <wmk1, -epoch1> <wmk1, +epoch2> <wmk2, -epoch2> <wmk2, +epoch3> <wmk3,
71/// -epoch3> }`
72/// Then we can get monotonic events (they are in order by user key) as below:
73/// `{ <0, epoch1>, <wmk1, epoch2>, <wmk2, epoch3>, <wmk3, +inf> }`
74/// which means that delete range of [0, wmk1) is epoch1, delete range of [wmk1, wmk2) if epoch2,
75/// etc. In this example, at the event key wmk1 (5), delete range changes from epoch1 to epoch2,
76/// thus the `new epoch` is epoch2. epoch2 will be used from the event key wmk1 (5) and till the
77/// next event key wmk2 (7) (not inclusive).
78/// If there is no range deletes between current event key and next event key, `new_epoch` will be
79/// `HummockEpoch::MAX`.
80#[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize)]
81pub struct MonotonicDeleteEvent {
82    pub event_key:
83        risingwave_hummock_sdk::key::range_delete_backward_compatibility_serde_struct::PointRange,
84    pub new_epoch: HummockEpoch,
85}
86
87impl MonotonicDeleteEvent {
88    pub fn encode(&self, mut buf: impl BufMut) {
89        self.event_key
90            .left_user_key
91            .encode_length_prefixed(&mut buf);
92        buf.put_u8(if self.event_key.is_exclude_left_key {
93            1
94        } else {
95            0
96        });
97        buf.put_u64_le(self.new_epoch);
98    }
99
100    pub fn decode(buf: &mut &[u8]) -> Self {
101        use risingwave_hummock_sdk::key::range_delete_backward_compatibility_serde_struct::*;
102        let user_key = UserKey::decode_length_prefixed(buf);
103        let exclude_left_key_flag = buf.get_u8();
104        let is_exclude_left_key = match exclude_left_key_flag {
105            0 => false,
106            1 => true,
107            _ => panic!("exclusive flag should be either 0 or 1"),
108        };
109        let new_epoch = buf.get_u64_le();
110        Self {
111            event_key: PointRange {
112                left_user_key: user_key,
113                is_exclude_left_key,
114            },
115            new_epoch,
116        }
117    }
118}
119
120#[derive(Serialize, Deserialize)]
121struct SerdeSstable {
122    id: HummockSstableObjectId,
123    meta: SstableMeta,
124}
125
126impl From<SerdeSstable> for Sstable {
127    fn from(SerdeSstable { id, meta }: SerdeSstable) -> Self {
128        // Set skip_bloom_filter_in_serde to false because the behavior
129        // is determined by the serializer
130        Sstable::new(id, meta, false)
131    }
132}
133
134/// [`Sstable`] is a handle for accessing SST.
135#[derive(Clone, Deserialize)]
136#[serde(from = "SerdeSstable")]
137pub struct Sstable {
138    pub id: HummockSstableObjectId,
139    pub meta: SstableMeta,
140    #[serde(skip)]
141    pub filter_reader: XorFilterReader,
142    /// SST serde happens when an SST meta is written to meta disk cache.
143    /// Excluding the SST filter from serde can reduce the meta disk cache entry size
144    /// and reduce disk IO throughput at the cost of making the SST filter useless.
145    #[serde(skip)]
146    skip_bloom_filter_in_serde: bool,
147}
148
149impl Serialize for Sstable {
150    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
151    where
152        S: serde::Serializer,
153    {
154        let mut serde_sstable = SerdeSstable {
155            id: self.id,
156            meta: self.meta.clone(),
157        };
158        if !self.skip_bloom_filter_in_serde {
159            serde_sstable.meta.bloom_filter = self.filter_reader.encode_to_bytes();
160        }
161        serde_sstable.serialize(serializer)
162    }
163}
164
165impl Debug for Sstable {
166    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
167        f.debug_struct("Sstable")
168            .field("id", &self.id)
169            .field("meta", &self.meta)
170            .finish()
171    }
172}
173
174impl Sstable {
175    pub fn new(
176        id: HummockSstableObjectId,
177        mut meta: SstableMeta,
178        skip_bloom_filter_in_serde: bool,
179    ) -> Self {
180        let filter_data = std::mem::take(&mut meta.bloom_filter);
181        let filter_reader = XorFilterReader::new(&filter_data, &meta.block_metas);
182        Self {
183            id,
184            meta,
185            filter_reader,
186            skip_bloom_filter_in_serde,
187        }
188    }
189
190    /// Restore an SST handle directly from native metadata bytes.
191    pub fn from_meta_bytes(
192        id: HummockSstableObjectId,
193        bytes: &[u8],
194        skip_bloom_filter_in_serde: bool,
195    ) -> HummockResult<Self> {
196        let (meta, filter_reader) = SstableMeta::decode_with_filter(bytes, XorFilterReader::new)?;
197        Ok(Self {
198            id,
199            meta,
200            filter_reader,
201            skip_bloom_filter_in_serde,
202        })
203    }
204
205    #[inline(always)]
206    pub fn has_filter(&self) -> bool {
207        !self.filter_reader.is_empty()
208    }
209
210    pub fn calculate_block_info(&self, block_index: usize) -> (Range<usize>, usize) {
211        let block_meta = &self.meta.block_metas[block_index];
212        let range =
213            block_meta.offset as usize..block_meta.offset as usize + block_meta.len as usize;
214        let uncompressed_capacity = block_meta.uncompressed_size as usize;
215        (range, uncompressed_capacity)
216    }
217
218    #[inline(always)]
219    pub fn hash_for_filter(dist_key: &[u8], table_id: u32) -> u64 {
220        let dist_key_hash = xxh64::xxh64(dist_key, 0);
221        // congyi adds this because he aims to dedup keys in different tables
222        (table_id as u64).bitxor(dist_key_hash)
223    }
224
225    #[inline(always)]
226    pub fn may_match_hash(&self, user_key_range: &UserKeyRangeRef<'_>, hash: u64) -> bool {
227        self.filter_reader
228            .may_match(&self.meta.block_metas, user_key_range, hash)
229    }
230
231    #[inline(always)]
232    pub fn block_count(&self) -> usize {
233        self.meta.block_metas.len()
234    }
235
236    #[inline(always)]
237    pub fn estimated_meta_cache_memory_weight(&self) -> usize {
238        // This is for foyer's in-memory cache weighter. The disk tier uses foyer `Code`
239        // serialization and `estimated_size` instead.
240        std::mem::size_of::<Self>()
241            + self.meta.estimated_heap_size()
242            + self.filter_reader.estimated_heap_size()
243    }
244
245    /// Size of the meta disk-cache value, without cloning metadata or encoding its filter.
246    pub(crate) fn serde_encoded_size(&self) -> usize {
247        // Bincode's default format encodes a struct like a tuple and uses a fixed-width
248        // length prefix for byte vectors. Replacing the bloom payload changes only its
249        // byte count. Keep sizing the other fields through serde as the format evolves.
250        let size = bincode::serialized_size(&(self.id, &self.meta)).unwrap() as usize;
251        if self.skip_bloom_filter_in_serde {
252            size
253        } else {
254            size - self.meta.bloom_filter.len() + self.filter_reader.serialized_len()
255        }
256    }
257}
258
259#[derive(Clone, Default, Debug, Eq, PartialEq, Serialize, Deserialize)]
260pub struct BlockMeta {
261    #[serde(with = "serde_bytes")]
262    pub smallest_key: Vec<u8>,
263    pub offset: u32,
264    pub len: u32,
265    pub uncompressed_size: u32,
266    pub total_key_count: u32,
267    pub stale_key_count: u32,
268}
269
270impl BlockMeta {
271    /// Format:
272    ///
273    /// ```plain
274    /// | offset (4B) | len (4B) | uncompressed size (4B) | smallest key len (4B) | smallest key |
275    /// ```
276    pub fn encode(&self, mut buf: impl BufMut) {
277        buf.put_u32_le(self.offset);
278        buf.put_u32_le(self.len);
279        buf.put_u32_le(self.uncompressed_size);
280        buf.put_u32_le(self.total_key_count);
281        buf.put_u32_le(self.stale_key_count);
282        put_length_prefixed_slice(buf, &self.smallest_key);
283    }
284
285    pub fn decode(buf: &mut &[u8]) -> Self {
286        let offset = buf.get_u32_le();
287        let len = buf.get_u32_le();
288        let uncompressed_size = buf.get_u32_le();
289
290        let total_key_count = buf.get_u32_le();
291        let stale_key_count = buf.get_u32_le();
292        let smallest_key = get_length_prefixed_slice(buf);
293        Self {
294            smallest_key,
295            offset,
296            len,
297            uncompressed_size,
298            total_key_count,
299            stale_key_count,
300        }
301    }
302
303    pub fn decode_from_v1(buf: &mut &[u8]) -> Self {
304        let offset = buf.get_u32_le();
305        let len = buf.get_u32_le();
306        let uncompressed_size = buf.get_u32_le();
307        let total_key_count = 0;
308        let stale_key_count = 0;
309        let smallest_key = get_length_prefixed_slice(buf);
310        Self {
311            smallest_key,
312            offset,
313            len,
314            uncompressed_size,
315            total_key_count,
316            stale_key_count,
317        }
318    }
319
320    #[inline]
321    pub fn encoded_size(&self) -> usize {
322        24 /* offset + len + key len + uncompressed size + total key count + stale key count */ + self.smallest_key.len()
323    }
324
325    pub fn table_id(&self) -> TableId {
326        FullKey::decode(&self.smallest_key).user_key.table_id
327    }
328
329    fn estimated_heap_size(&self) -> usize {
330        self.smallest_key.capacity()
331    }
332}
333
334#[derive(Default, Clone, PartialEq, Eq, Debug, Serialize, Deserialize)]
335pub struct SstableMeta {
336    pub block_metas: Vec<BlockMeta>,
337    #[serde(with = "serde_bytes")]
338    pub bloom_filter: Vec<u8>,
339    pub estimated_size: u32,
340    pub key_count: u32,
341    #[serde(with = "serde_bytes")]
342    pub smallest_key: Vec<u8>,
343    #[serde(with = "serde_bytes")]
344    pub largest_key: Vec<u8>,
345    pub meta_offset: u64,
346    /// Assume that watermark1 is 5, watermark2 is 7, watermark3 is 11, delete ranges
347    /// `{ [0, wmk1) in epoch1, [wmk1, wmk2) in epoch2, [wmk2, wmk3) in epoch3 }`
348    /// can be transformed into events below:
349    /// `{ <0, +epoch1> <wmk1, -epoch1> <wmk1, +epoch2> <wmk2, -epoch2> <wmk2, +epoch3> <wmk3,
350    /// -epoch3> }`
351    /// Then we can get monotonic events (they are in order by user key) as below:
352    /// `{ <0, epoch1>, <wmk1, epoch2>, <wmk2, epoch3>, <wmk3, +inf> }`
353    /// which means that delete range of [0, wmk1) is epoch1, delete range of [wmk1, wmk2) if
354    /// epoch2, etc. In this example, at the event key wmk1 (5), delete range changes from
355    /// epoch1 to epoch2, thus the `new epoch` is epoch2. epoch2 will be used from the event
356    /// key wmk1 (5) and till the next event key wmk2 (7) (not inclusive).
357    /// If there is no range deletes between current event key and next event key, `new_epoch` will
358    /// be `HummockEpoch::MAX`.
359    #[deprecated]
360    pub monotonic_tombstone_events: Vec<MonotonicDeleteEvent>,
361    /// Format version, for further compatibility.
362    pub version: u32,
363}
364
365impl SstableMeta {
366    /// Format:
367    ///
368    /// ```plain
369    /// | N (4B) |
370    /// | block meta 0 | ... | block meta N-1 |
371    /// | SST filter len (4B) | SST filter |
372    /// | estimated size (4B) | key count (4B) |
373    /// | smallest key len (4B) | smallest key |
374    /// | largest key len (4B) | largest key |
375    /// | K (4B) |
376    /// | tombstone-event 0 | ... | tombstone-event K-1 |
377    /// | file offset of this meta block (8B) |
378    /// | checksum (8B) | version (4B) | magic (4B) |
379    /// ```
380    pub fn encode_to_bytes(&self) -> Vec<u8> {
381        let encoded_size = self.encoded_size();
382        let mut buf = Vec::with_capacity(encoded_size);
383        self.encode_to(&mut buf);
384        buf
385    }
386
387    pub fn encode_to(&self, mut buf: impl BufMut + AsRef<[u8]>) {
388        let start = buf.as_ref().len();
389
390        buf.put_u32_le(
391            utils::checked_into_u32(self.block_metas.len()).unwrap_or_else(|_| {
392                let tmp_full_key = FullKey::decode(&self.smallest_key);
393                panic!(
394                    "WARN overflow can't convert block_metas_len {} into u32 table {}",
395                    self.block_metas.len(),
396                    tmp_full_key.user_key.table_id,
397                )
398            }),
399        );
400        for block_meta in &self.block_metas {
401            block_meta.encode(&mut buf);
402        }
403        put_length_prefixed_slice(&mut buf, &self.bloom_filter);
404        buf.put_u32_le(self.estimated_size);
405        buf.put_u32_le(self.key_count);
406        put_length_prefixed_slice(&mut buf, &self.smallest_key);
407        put_length_prefixed_slice(&mut buf, &self.largest_key);
408        #[expect(deprecated)]
409        buf.put_u32_le(
410            utils::checked_into_u32(self.monotonic_tombstone_events.len()).unwrap_or_else(|_| {
411                let tmp_full_key = FullKey::decode(&self.smallest_key);
412                panic!(
413                    "WARN overflow can't convert monotonic_tombstone_events_len {} into u32 table {}",
414                    self.monotonic_tombstone_events.len(),
415                    tmp_full_key.user_key.table_id,
416                )
417            }),
418        );
419        #[expect(deprecated)]
420        for monotonic_tombstone_event in &self.monotonic_tombstone_events {
421            monotonic_tombstone_event.encode(&mut buf);
422        }
423        buf.put_u64_le(self.meta_offset);
424
425        let end = buf.as_ref().len();
426
427        let checksum = xxhash64_checksum(&buf.as_ref()[start..end]);
428        buf.put_u64_le(checksum);
429        buf.put_u32_le(VERSION);
430        buf.put_u32_le(MAGIC);
431    }
432
433    pub fn decode(buf: &[u8]) -> HummockResult<Self> {
434        let (mut meta, bloom_filter) = Self::decode_with_filter(buf, |data, _| data.to_vec())?;
435        meta.bloom_filter = bloom_filter;
436        Ok(meta)
437    }
438
439    fn decode_with_filter<F>(
440        buf: &[u8],
441        restore_filter: impl FnOnce(&[u8], &[BlockMeta]) -> F,
442    ) -> HummockResult<(Self, F)> {
443        let mut cursor = buf.len();
444
445        cursor -= 4;
446        let magic = (&buf[cursor..]).get_u32_le();
447        if magic != MAGIC {
448            return Err(HummockError::magic_mismatch(MAGIC, magic));
449        }
450
451        cursor -= 4;
452        let version = (&buf[cursor..cursor + 4]).get_u32_le();
453        if version != VERSION && version != OLD_VERSION {
454            return Err(HummockError::invalid_format_version(version));
455        }
456
457        cursor -= 8;
458        let checksum = (&buf[cursor..cursor + 8]).get_u64_le();
459        let buf = &mut &buf[..cursor];
460        xxhash64_verify(buf, checksum)?;
461
462        let block_meta_count = buf.get_u32_le() as usize;
463        let mut block_metas = Vec::with_capacity(block_meta_count);
464        if version == OLD_VERSION {
465            for _ in 0..block_meta_count {
466                block_metas.push(BlockMeta::decode_from_v1(buf));
467            }
468        } else {
469            for _ in 0..block_meta_count {
470                block_metas.push(BlockMeta::decode(buf));
471            }
472        }
473
474        let filter_len = buf.get_u32_le() as usize;
475        let filter_data = &buf[..filter_len];
476        buf.advance(filter_len);
477        let estimated_size = buf.get_u32_le();
478        let key_count = buf.get_u32_le();
479        let smallest_key = get_length_prefixed_slice(buf);
480        let largest_key = get_length_prefixed_slice(buf);
481        let tomb_event_count = buf.get_u32_le() as usize;
482        let mut monotonic_tombstone_events = Vec::with_capacity(tomb_event_count);
483        for _ in 0..tomb_event_count {
484            let monotonic_tombstone_event = MonotonicDeleteEvent::decode(buf);
485            monotonic_tombstone_events.push(monotonic_tombstone_event);
486        }
487        let meta_offset = buf.get_u64_le();
488
489        if !monotonic_tombstone_events.is_empty() {
490            warn!(
491                count = monotonic_tombstone_events.len(),
492                tables = ?monotonic_tombstone_events
493                    .iter()
494                    .map(|event| event.event_key.left_user_key.table_id)
495                    .collect::<HashSet<_>>(),
496                "read non-empty range tombstones");
497        }
498
499        let filter = restore_filter(filter_data, &block_metas);
500        #[expect(deprecated)]
501        let meta = Self {
502            block_metas,
503            bloom_filter: Vec::new(),
504            estimated_size,
505            key_count,
506            smallest_key,
507            largest_key,
508            meta_offset,
509            monotonic_tombstone_events,
510            version,
511        };
512        Ok((meta, filter))
513    }
514
515    #[inline]
516    pub fn encoded_size(&self) -> usize {
517        4 // block meta count
518            + self
519            .block_metas
520            .iter()
521            .map(|block_meta| block_meta.encoded_size())
522            .sum::<usize>()
523            + 4 // monotonic tombstone events len
524            + 4 // SST filter len
525            + self.bloom_filter.len()
526            + 4 // estimated size
527            + 4 // key count
528            + 4 // key len
529            + self.smallest_key.len()
530            + 4 // key len
531            + self.largest_key.len()
532            + 8 // footer
533            + 8 // checksum
534            + 4 // version
535            + 4 // magic
536    }
537
538    #[expect(
539        deprecated,
540        reason = "monotonic_tombstone_events is deprecated but still contributes to decoded meta heap size"
541    )]
542    fn estimated_heap_size(&self) -> usize {
543        self.block_metas.capacity() * std::mem::size_of::<BlockMeta>()
544            + self
545                .block_metas
546                .iter()
547                .map(BlockMeta::estimated_heap_size)
548                .sum::<usize>()
549            + self.bloom_filter.capacity()
550            + self.smallest_key.capacity()
551            + self.largest_key.capacity()
552            + self.monotonic_tombstone_events.capacity()
553                * std::mem::size_of::<MonotonicDeleteEvent>()
554    }
555}
556
557#[derive(Default)]
558pub struct SstableIteratorReadOptions {
559    pub cache_policy: CachePolicy,
560    /// The table being read by the state-store request. When present in the SST's table IDs,
561    /// SST iterators can avoid blocks for other tables colocated in the same physical SST.
562    /// Otherwise, they use the SST's full table ID range.
563    pub read_table_id: Option<TableId>,
564    /// The only hard upper bound of this scan. It limits the iterator's block window.
565    pub scan_end_user_key: Option<Bound<UserKey<KeyPayloadType>>>,
566    /// Whether to prefetch blocks within the already-bounded scan window.
567    pub prefetch: bool,
568}
569
570impl SstableIteratorReadOptions {
571    /// Builds options with prefetch disabled. Range scans enable it explicitly after setting the
572    /// scan bound, while point gets retain the default to avoid creating a prefetch stream.
573    pub fn from_read_options(read_options: &ReadOptions) -> Self {
574        Self {
575            cache_policy: read_options.cache_policy,
576            read_table_id: None,
577            scan_end_user_key: None,
578            prefetch: false,
579        }
580    }
581}
582
583#[cfg(test)]
584mod tests {
585    use super::*;
586    use crate::hummock::HummockValue;
587    use crate::hummock::iterator::test_utils::{
588        default_builder_opt_for_test, iterator_test_key_of,
589    };
590    use crate::hummock::test_utils::gen_test_sstable_data;
591    use crate::store::PrefetchOptions;
592
593    #[test]
594    fn test_iterator_read_options_do_not_enable_prefetch_for_point_get() {
595        let read_options = ReadOptions {
596            prefetch_options: PrefetchOptions::prefetch_for_large_range_scan(),
597            ..Default::default()
598        };
599
600        let iterator_options = SstableIteratorReadOptions::from_read_options(&read_options);
601        assert!(!iterator_options.prefetch);
602        assert!(iterator_options.scan_end_user_key.is_none());
603    }
604
605    #[test]
606    fn test_sstable_meta_enc_dec() {
607        #[expect(deprecated)]
608        let meta = SstableMeta {
609            block_metas: vec![
610                BlockMeta {
611                    smallest_key: b"0-smallest-key".to_vec(),
612                    len: 100,
613                    ..Default::default()
614                },
615                BlockMeta {
616                    smallest_key: b"5-some-key".to_vec(),
617                    offset: 100,
618                    len: 100,
619                    ..Default::default()
620                },
621            ],
622            bloom_filter: b"0123456789".to_vec(),
623            estimated_size: 123,
624            key_count: 123,
625            smallest_key: b"0-smallest-key".to_vec(),
626            largest_key: b"9-largest-key".to_vec(),
627            meta_offset: 123,
628            monotonic_tombstone_events: vec![],
629            version: VERSION,
630        };
631        let sz = meta.encoded_size();
632        let buf = meta.encode_to_bytes();
633        assert_eq!(sz, buf.len());
634        let decoded_meta = SstableMeta::decode(&buf[..]).unwrap();
635        assert_eq!(decoded_meta, meta);
636
637        // Bincode encodes structs like tuples. Plain Vec<u8> fields here reproduce the
638        // meta disk-cache format before serde_bytes, including nested block metadata.
639        let legacy_block_metas: Vec<_> = meta
640            .block_metas
641            .iter()
642            .map(|block| {
643                (
644                    &block.smallest_key,
645                    block.offset,
646                    block.len,
647                    block.uncompressed_size,
648                    block.total_key_count,
649                    block.stale_key_count,
650                )
651            })
652            .collect();
653        #[expect(deprecated)]
654        let legacy_meta = (
655            legacy_block_metas,
656            &meta.bloom_filter,
657            meta.estimated_size,
658            meta.key_count,
659            &meta.smallest_key,
660            &meta.largest_key,
661            meta.meta_offset,
662            &meta.monotonic_tombstone_events,
663            meta.version,
664        );
665        let legacy_bytes = bincode::serialize(&legacy_meta).unwrap();
666        assert_eq!(bincode::serialize(&meta).unwrap(), legacy_bytes);
667        let decoded_meta: SstableMeta = bincode::deserialize_from(&mut &legacy_bytes[..]).unwrap();
668        assert_eq!(decoded_meta, meta);
669    }
670
671    #[tokio::test]
672    async fn test_sstable_serde() {
673        use foyer::Code;
674
675        use crate::hummock::SstableMetaCacheValue;
676
677        let (_, meta) = gen_test_sstable_data(
678            default_builder_opt_for_test(),
679            (0..100).clone().map(|x| {
680                (
681                    iterator_test_key_of(x),
682                    HummockValue::put(format!("overlapped_new_{}", x).as_bytes().to_vec()),
683                )
684            }),
685        )
686        .await;
687
688        for skip_filter in [false, true] {
689            let sstable = Sstable::new(42.into(), meta.clone(), skip_filter);
690            let legacy_bytes = bincode::serialize(&sstable).unwrap();
691            let value = SstableMetaCacheValue::from(sstable);
692            let mut bytes = Vec::new();
693            value.encode(&mut bytes).unwrap();
694
695            assert_eq!(bytes, legacy_bytes);
696            assert_eq!(value.estimated_size(), bytes.len());
697
698            let decoded = SstableMetaCacheValue::decode(&mut legacy_bytes.as_slice()).unwrap();
699            assert_eq!(decoded.id, value.id);
700            assert_eq!(decoded.meta, value.meta);
701            assert!(value.has_filter());
702            if skip_filter {
703                assert!(!decoded.has_filter());
704            } else {
705                assert_eq!(
706                    decoded.filter_reader.encode_to_bytes(),
707                    value.filter_reader.encode_to_bytes()
708                );
709            }
710        }
711    }
712
713    #[tokio::test]
714    async fn test_sstable_from_meta_bytes() {
715        let (_, meta) = gen_test_sstable_data(
716            default_builder_opt_for_test(),
717            (0..100).map(|x| (iterator_test_key_of(x), HummockValue::put(vec![42]))),
718        )
719        .await;
720        let bytes = meta.encode_to_bytes();
721        for skip in [false, true] {
722            let expected = Sstable::new(42.into(), meta.clone(), skip);
723            let restored = {
724                let temporary = bytes.clone();
725                Sstable::from_meta_bytes(42.into(), &temporary, skip).unwrap()
726            };
727            assert_eq!(restored.id, expected.id);
728            assert_eq!(restored.meta, expected.meta);
729            assert_eq!(
730                restored.filter_reader.encode_to_bytes(),
731                expected.filter_reader.encode_to_bytes()
732            );
733            assert_eq!(
734                bincode::serialize(&restored).unwrap(),
735                bincode::serialize(&expected).unwrap()
736            );
737        }
738        let mut corrupt = bytes;
739        corrupt[0] ^= 1;
740        assert!(SstableMeta::decode(&corrupt).is_err());
741        assert!(Sstable::from_meta_bytes(42.into(), &corrupt, false).is_err());
742    }
743}
risingwave_storage/hummock/sstable/mod.rs

risingwave_storage/hummock/sstable/
mod.rs
