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    #[inline(always)]
191    pub fn has_filter(&self) -> bool {
192        !self.filter_reader.is_empty()
193    }
194
195    pub fn calculate_block_info(&self, block_index: usize) -> (Range<usize>, usize) {
196        let block_meta = &self.meta.block_metas[block_index];
197        let range =
198            block_meta.offset as usize..block_meta.offset as usize + block_meta.len as usize;
199        let uncompressed_capacity = block_meta.uncompressed_size as usize;
200        (range, uncompressed_capacity)
201    }
202
203    #[inline(always)]
204    pub fn hash_for_filter(dist_key: &[u8], table_id: u32) -> u64 {
205        let dist_key_hash = xxh64::xxh64(dist_key, 0);
206        // congyi adds this because he aims to dedup keys in different tables
207        (table_id as u64).bitxor(dist_key_hash)
208    }
209
210    #[inline(always)]
211    pub fn may_match_hash(&self, user_key_range: &UserKeyRangeRef<'_>, hash: u64) -> bool {
212        self.filter_reader
213            .may_match(&self.meta.block_metas, user_key_range, hash)
214    }
215
216    #[inline(always)]
217    pub fn block_count(&self) -> usize {
218        self.meta.block_metas.len()
219    }
220
221    #[inline(always)]
222    pub fn estimated_meta_cache_memory_weight(&self) -> usize {
223        // This is for foyer's in-memory cache weighter. The disk tier uses foyer `Code`
224        // serialization and `estimated_size` instead.
225        std::mem::size_of::<Self>()
226            + self.meta.estimated_heap_size()
227            + self.filter_reader.estimated_heap_size()
228    }
229}
230
231#[derive(Clone, Default, Debug, Eq, PartialEq, Serialize, Deserialize)]
232pub struct BlockMeta {
233    pub smallest_key: Vec<u8>,
234    pub offset: u32,
235    pub len: u32,
236    pub uncompressed_size: u32,
237    pub total_key_count: u32,
238    pub stale_key_count: u32,
239}
240
241impl BlockMeta {
242    /// Format:
243    ///
244    /// ```plain
245    /// | offset (4B) | len (4B) | uncompressed size (4B) | smallest key len (4B) | smallest key |
246    /// ```
247    pub fn encode(&self, mut buf: impl BufMut) {
248        buf.put_u32_le(self.offset);
249        buf.put_u32_le(self.len);
250        buf.put_u32_le(self.uncompressed_size);
251        buf.put_u32_le(self.total_key_count);
252        buf.put_u32_le(self.stale_key_count);
253        put_length_prefixed_slice(buf, &self.smallest_key);
254    }
255
256    pub fn decode(buf: &mut &[u8]) -> Self {
257        let offset = buf.get_u32_le();
258        let len = buf.get_u32_le();
259        let uncompressed_size = buf.get_u32_le();
260
261        let total_key_count = buf.get_u32_le();
262        let stale_key_count = buf.get_u32_le();
263        let smallest_key = get_length_prefixed_slice(buf);
264        Self {
265            smallest_key,
266            offset,
267            len,
268            uncompressed_size,
269            total_key_count,
270            stale_key_count,
271        }
272    }
273
274    pub fn decode_from_v1(buf: &mut &[u8]) -> Self {
275        let offset = buf.get_u32_le();
276        let len = buf.get_u32_le();
277        let uncompressed_size = buf.get_u32_le();
278        let total_key_count = 0;
279        let stale_key_count = 0;
280        let smallest_key = get_length_prefixed_slice(buf);
281        Self {
282            smallest_key,
283            offset,
284            len,
285            uncompressed_size,
286            total_key_count,
287            stale_key_count,
288        }
289    }
290
291    #[inline]
292    pub fn encoded_size(&self) -> usize {
293        24 /* offset + len + key len + uncompressed size + total key count + stale key count */ + self.smallest_key.len()
294    }
295
296    pub fn table_id(&self) -> TableId {
297        FullKey::decode(&self.smallest_key).user_key.table_id
298    }
299
300    fn estimated_heap_size(&self) -> usize {
301        self.smallest_key.capacity()
302    }
303}
304
305#[derive(Default, Clone, PartialEq, Eq, Debug, Serialize, Deserialize)]
306pub struct SstableMeta {
307    pub block_metas: Vec<BlockMeta>,
308    pub bloom_filter: Vec<u8>,
309    pub estimated_size: u32,
310    pub key_count: u32,
311    pub smallest_key: Vec<u8>,
312    pub largest_key: Vec<u8>,
313    pub meta_offset: u64,
314    /// Assume that watermark1 is 5, watermark2 is 7, watermark3 is 11, delete ranges
315    /// `{ [0, wmk1) in epoch1, [wmk1, wmk2) in epoch2, [wmk2, wmk3) in epoch3 }`
316    /// can be transformed into events below:
317    /// `{ <0, +epoch1> <wmk1, -epoch1> <wmk1, +epoch2> <wmk2, -epoch2> <wmk2, +epoch3> <wmk3,
318    /// -epoch3> }`
319    /// Then we can get monotonic events (they are in order by user key) as below:
320    /// `{ <0, epoch1>, <wmk1, epoch2>, <wmk2, epoch3>, <wmk3, +inf> }`
321    /// which means that delete range of [0, wmk1) is epoch1, delete range of [wmk1, wmk2) if
322    /// epoch2, etc. In this example, at the event key wmk1 (5), delete range changes from
323    /// epoch1 to epoch2, thus the `new epoch` is epoch2. epoch2 will be used from the event
324    /// key wmk1 (5) and till the next event key wmk2 (7) (not inclusive).
325    /// If there is no range deletes between current event key and next event key, `new_epoch` will
326    /// be `HummockEpoch::MAX`.
327    #[deprecated]
328    pub monotonic_tombstone_events: Vec<MonotonicDeleteEvent>,
329    /// Format version, for further compatibility.
330    pub version: u32,
331}
332
333impl SstableMeta {
334    /// Format:
335    ///
336    /// ```plain
337    /// | N (4B) |
338    /// | block meta 0 | ... | block meta N-1 |
339    /// | SST filter len (4B) | SST filter |
340    /// | estimated size (4B) | key count (4B) |
341    /// | smallest key len (4B) | smallest key |
342    /// | largest key len (4B) | largest key |
343    /// | K (4B) |
344    /// | tombstone-event 0 | ... | tombstone-event K-1 |
345    /// | file offset of this meta block (8B) |
346    /// | checksum (8B) | version (4B) | magic (4B) |
347    /// ```
348    pub fn encode_to_bytes(&self) -> Vec<u8> {
349        let encoded_size = self.encoded_size();
350        let mut buf = Vec::with_capacity(encoded_size);
351        self.encode_to(&mut buf);
352        buf
353    }
354
355    pub fn encode_to(&self, mut buf: impl BufMut + AsRef<[u8]>) {
356        let start = buf.as_ref().len();
357
358        buf.put_u32_le(
359            utils::checked_into_u32(self.block_metas.len()).unwrap_or_else(|_| {
360                let tmp_full_key = FullKey::decode(&self.smallest_key);
361                panic!(
362                    "WARN overflow can't convert block_metas_len {} into u32 table {}",
363                    self.block_metas.len(),
364                    tmp_full_key.user_key.table_id,
365                )
366            }),
367        );
368        for block_meta in &self.block_metas {
369            block_meta.encode(&mut buf);
370        }
371        put_length_prefixed_slice(&mut buf, &self.bloom_filter);
372        buf.put_u32_le(self.estimated_size);
373        buf.put_u32_le(self.key_count);
374        put_length_prefixed_slice(&mut buf, &self.smallest_key);
375        put_length_prefixed_slice(&mut buf, &self.largest_key);
376        #[expect(deprecated)]
377        buf.put_u32_le(
378            utils::checked_into_u32(self.monotonic_tombstone_events.len()).unwrap_or_else(|_| {
379                let tmp_full_key = FullKey::decode(&self.smallest_key);
380                panic!(
381                    "WARN overflow can't convert monotonic_tombstone_events_len {} into u32 table {}",
382                    self.monotonic_tombstone_events.len(),
383                    tmp_full_key.user_key.table_id,
384                )
385            }),
386        );
387        #[expect(deprecated)]
388        for monotonic_tombstone_event in &self.monotonic_tombstone_events {
389            monotonic_tombstone_event.encode(&mut buf);
390        }
391        buf.put_u64_le(self.meta_offset);
392
393        let end = buf.as_ref().len();
394
395        let checksum = xxhash64_checksum(&buf.as_ref()[start..end]);
396        buf.put_u64_le(checksum);
397        buf.put_u32_le(VERSION);
398        buf.put_u32_le(MAGIC);
399    }
400
401    pub fn decode(buf: &[u8]) -> HummockResult<Self> {
402        let mut cursor = buf.len();
403
404        cursor -= 4;
405        let magic = (&buf[cursor..]).get_u32_le();
406        if magic != MAGIC {
407            return Err(HummockError::magic_mismatch(MAGIC, magic));
408        }
409
410        cursor -= 4;
411        let version = (&buf[cursor..cursor + 4]).get_u32_le();
412        if version != VERSION && version != OLD_VERSION {
413            return Err(HummockError::invalid_format_version(version));
414        }
415
416        cursor -= 8;
417        let checksum = (&buf[cursor..cursor + 8]).get_u64_le();
418        let buf = &mut &buf[..cursor];
419        xxhash64_verify(buf, checksum)?;
420
421        let block_meta_count = buf.get_u32_le() as usize;
422        let mut block_metas = Vec::with_capacity(block_meta_count);
423        if version == OLD_VERSION {
424            for _ in 0..block_meta_count {
425                block_metas.push(BlockMeta::decode_from_v1(buf));
426            }
427        } else {
428            for _ in 0..block_meta_count {
429                block_metas.push(BlockMeta::decode(buf));
430            }
431        }
432
433        let bloom_filter = get_length_prefixed_slice(buf);
434        let estimated_size = buf.get_u32_le();
435        let key_count = buf.get_u32_le();
436        let smallest_key = get_length_prefixed_slice(buf);
437        let largest_key = get_length_prefixed_slice(buf);
438        let tomb_event_count = buf.get_u32_le() as usize;
439        let mut monotonic_tombstone_events = Vec::with_capacity(tomb_event_count);
440        for _ in 0..tomb_event_count {
441            let monotonic_tombstone_event = MonotonicDeleteEvent::decode(buf);
442            monotonic_tombstone_events.push(monotonic_tombstone_event);
443        }
444        let meta_offset = buf.get_u64_le();
445
446        if !monotonic_tombstone_events.is_empty() {
447            warn!(
448                count = monotonic_tombstone_events.len(),
449                tables = ?monotonic_tombstone_events
450                    .iter()
451                    .map(|event| event.event_key.left_user_key.table_id)
452                    .collect::<HashSet<_>>(),
453                "read non-empty range tombstones");
454        }
455
456        #[expect(deprecated)]
457        Ok(Self {
458            block_metas,
459            bloom_filter,
460            estimated_size,
461            key_count,
462            smallest_key,
463            largest_key,
464            meta_offset,
465            monotonic_tombstone_events,
466            version,
467        })
468    }
469
470    #[inline]
471    pub fn encoded_size(&self) -> usize {
472        4 // block meta count
473            + self
474            .block_metas
475            .iter()
476            .map(|block_meta| block_meta.encoded_size())
477            .sum::<usize>()
478            + 4 // monotonic tombstone events len
479            + 4 // SST filter len
480            + self.bloom_filter.len()
481            + 4 // estimated size
482            + 4 // key count
483            + 4 // key len
484            + self.smallest_key.len()
485            + 4 // key len
486            + self.largest_key.len()
487            + 8 // footer
488            + 8 // checksum
489            + 4 // version
490            + 4 // magic
491    }
492
493    #[expect(
494        deprecated,
495        reason = "monotonic_tombstone_events is deprecated but still contributes to decoded meta heap size"
496    )]
497    fn estimated_heap_size(&self) -> usize {
498        self.block_metas.capacity() * std::mem::size_of::<BlockMeta>()
499            + self
500                .block_metas
501                .iter()
502                .map(BlockMeta::estimated_heap_size)
503                .sum::<usize>()
504            + self.bloom_filter.capacity()
505            + self.smallest_key.capacity()
506            + self.largest_key.capacity()
507            + self.monotonic_tombstone_events.capacity()
508                * std::mem::size_of::<MonotonicDeleteEvent>()
509    }
510}
511
512#[derive(Default)]
513pub struct SstableIteratorReadOptions {
514    pub cache_policy: CachePolicy,
515    pub must_iterated_end_user_key: Option<Bound<UserKey<KeyPayloadType>>>,
516    pub max_preload_retry_times: usize,
517    pub prefetch_for_large_query: bool,
518}
519
520impl SstableIteratorReadOptions {
521    pub fn from_read_options(read_options: &ReadOptions) -> Self {
522        Self {
523            cache_policy: read_options.cache_policy,
524            must_iterated_end_user_key: None,
525            max_preload_retry_times: 0,
526            prefetch_for_large_query: read_options.prefetch_options.for_large_query,
527        }
528    }
529}
530
531#[cfg(test)]
532mod tests {
533    use super::*;
534    use crate::hummock::HummockValue;
535    use crate::hummock::iterator::test_utils::{
536        default_builder_opt_for_test, iterator_test_key_of,
537    };
538    use crate::hummock::test_utils::gen_test_sstable_data;
539
540    #[test]
541    fn test_sstable_meta_enc_dec() {
542        #[expect(deprecated)]
543        let meta = SstableMeta {
544            block_metas: vec![
545                BlockMeta {
546                    smallest_key: b"0-smallest-key".to_vec(),
547                    len: 100,
548                    ..Default::default()
549                },
550                BlockMeta {
551                    smallest_key: b"5-some-key".to_vec(),
552                    offset: 100,
553                    len: 100,
554                    ..Default::default()
555                },
556            ],
557            bloom_filter: b"0123456789".to_vec(),
558            estimated_size: 123,
559            key_count: 123,
560            smallest_key: b"0-smallest-key".to_vec(),
561            largest_key: b"9-largest-key".to_vec(),
562            meta_offset: 123,
563            monotonic_tombstone_events: vec![],
564            version: VERSION,
565        };
566        let sz = meta.encoded_size();
567        let buf = meta.encode_to_bytes();
568        assert_eq!(sz, buf.len());
569        let decoded_meta = SstableMeta::decode(&buf[..]).unwrap();
570        assert_eq!(decoded_meta, meta);
571
572        println!("buf: {}", buf.len());
573    }
574
575    #[tokio::test]
576    async fn test_sstable_serde() {
577        let (_, meta) = gen_test_sstable_data(
578            default_builder_opt_for_test(),
579            (0..100).clone().map(|x| {
580                (
581                    iterator_test_key_of(x),
582                    HummockValue::put(format!("overlapped_new_{}", x).as_bytes().to_vec()),
583                )
584            }),
585        )
586        .await;
587
588        // skip sst serde
589        let sstable = Sstable::new(42.into(), meta.clone(), true);
590
591        let buffer = bincode::serialize(&sstable).unwrap();
592
593        let s: Sstable = bincode::deserialize(&buffer).unwrap();
594
595        assert_eq!(s.id, sstable.id);
596        assert_eq!(s.meta, sstable.meta);
597        assert!(!sstable.filter_reader.is_empty());
598        // The table filter reader is empty because the SST filter is skipped in serde.
599        assert!(s.filter_reader.is_empty());
600
601        // enable sst serde
602        let sstable = Sstable::new(42.into(), meta, false);
603
604        let buffer = bincode::serialize(&sstable).unwrap();
605
606        let s: Sstable = bincode::deserialize(&buffer).unwrap();
607
608        assert_eq!(s.id, sstable.id);
609        assert_eq!(s.meta, sstable.meta);
610        assert_eq!(
611            s.filter_reader.encode_to_bytes(),
612            sstable.filter_reader.encode_to_bytes()
613        );
614    }
615}
risingwave_storage/hummock/sstable/mod.rs

risingwave_storage/hummock/sstable/
mod.rs
