risingwave_storage/hummock/
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 is the state store of the streaming system.
16
17use std::ops::Bound;
18use std::sync::Arc;
19
20use bytes::Bytes;
21use risingwave_hummock_sdk::key::{FullKey, TableKey, UserKeyRangeRef};
22use risingwave_hummock_sdk::sstable_info::SstableInfo;
23use risingwave_hummock_sdk::{HummockEpoch, *};
24
25pub mod block_cache;
26pub use block_cache::*;
27
28pub mod sstable;
29pub use sstable::*;
30
31pub mod compactor;
32mod error;
33pub mod hummock_meta_client;
34pub mod iterator;
35pub mod shared_buffer;
36pub mod sstable_store;
37#[cfg(any(test, feature = "test"))]
38pub mod test_utils;
39pub mod utils;
40pub use utils::MemoryLimiter;
41pub mod backup_reader;
42pub mod event_handler;
43pub mod local_version;
44pub mod observer_manager;
45pub mod store;
46pub(crate) mod table_cache_metadata;
47pub use store::*;
48mod validator;
49pub mod value;
50pub mod write_limiter;
51
52pub mod recent_filter;
53pub use recent_filter::*;
54
55pub mod block_stream;
56mod time_travel_version_cache;
57
58pub(crate) mod vector;
59
60mod object_id_manager;
61pub use error::*;
62pub use object_id_manager::*;
63pub use risingwave_common::cache::{CacheableEntry, LookupResult, LruCache};
64pub use validator::*;
65use value::*;
66
67use self::iterator::HummockIterator;
68pub use self::sstable_store::*;
69use crate::mem_table::ImmutableMemtable;
70use crate::monitor::StoreLocalStatistic;
71use crate::store::ReadOptions;
72
73pub(in crate::hummock) struct IteratorStatsGuard<'a, TI: HummockIterator> {
74    iter: Option<TI>,
75    parent_stats: &'a mut StoreLocalStatistic,
76}
77
78impl<'a, TI: HummockIterator> IteratorStatsGuard<'a, TI> {
79    pub(in crate::hummock) fn new(iter: TI, parent_stats: &'a mut StoreLocalStatistic) -> Self {
80        Self {
81            iter: Some(iter),
82            parent_stats,
83        }
84    }
85
86    pub(in crate::hummock) fn iter(&self) -> &TI {
87        self.iter.as_ref().expect("iterator must be present")
88    }
89
90    pub(in crate::hummock) fn iter_mut(&mut self) -> &mut TI {
91        self.iter.as_mut().expect("iterator must be present")
92    }
93
94    fn collect(&mut self) {
95        if let Some(iter) = &self.iter {
96            iter.collect_local_statistic(self.parent_stats);
97        }
98    }
99
100    pub(in crate::hummock) fn handoff(mut self) -> TI {
101        self.iter.take().expect("iterator must be present")
102    }
103
104    pub(in crate::hummock) fn collect_and_take(mut self) -> TI {
105        self.collect();
106        self.iter.take().expect("iterator must be present")
107    }
108}
109
110impl<TI: HummockIterator> Drop for IteratorStatsGuard<'_, TI> {
111    fn drop(&mut self) {
112        self.collect();
113    }
114}
115
116pub async fn get_from_sstable_info(
117    sstable_store_ref: SstableStoreRef,
118    sstable_info: &SstableInfo,
119    full_key: FullKey<&[u8]>,
120    read_options: &ReadOptions,
121    dist_key_hash: Option<u64>,
122    local_stats: &mut StoreLocalStatistic,
123) -> HummockResult<Option<impl HummockIterator>> {
124    let sstable = sstable_store_ref.sstable(sstable_info, local_stats).await?;
125
126    // SST filter key is the distribution key, which does not need to be the prefix of pk, and does not
127    // contain `TablePrefix` and `VnodePrefix`.
128    if let Some(hash) = dist_key_hash
129        && !hit_sstable_filter(
130            &sstable,
131            &(
132                Bound::Included(full_key.user_key),
133                Bound::Included(full_key.user_key),
134            ),
135            hash,
136            local_stats,
137        )
138    {
139        return Ok(None);
140    }
141
142    let mut iterator_read_options = SstableIteratorReadOptions::from_read_options(read_options);
143    iterator_read_options.read_table_id = Some(full_key.user_key.table_id);
144
145    let mut iter = IteratorStatsGuard::new(
146        SstableIterator::create(
147            sstable,
148            sstable_store_ref.clone(),
149            Arc::new(iterator_read_options),
150            sstable_info,
151        ),
152        local_stats,
153    );
154    iter.iter_mut().seek(full_key).await?;
155    // Iterator has sought passed the borders.
156    if !iter.iter().is_valid() {
157        return Ok(None);
158    }
159
160    // Iterator gets us the key, we tell if it's the key we want
161    // or key next to it.
162    let value = if iter.iter().key().user_key == full_key.user_key {
163        Some(iter.collect_and_take())
164    } else {
165        None
166    };
167
168    Ok(value)
169}
170
171pub fn hit_sstable_filter(
172    sstable_ref: &Sstable,
173    user_key_range: &UserKeyRangeRef<'_>,
174    prefix_hash: u64,
175    local_stats: &mut StoreLocalStatistic,
176) -> bool {
177    local_stats.bloom_filter_check_counts += 1;
178    let may_exist = sstable_ref.may_match_hash(user_key_range, prefix_hash);
179    if !may_exist {
180        local_stats.bloom_filter_true_negative_counts += 1;
181    }
182    may_exist
183}
184
185/// Get `user_value` from `ImmutableMemtable`
186pub fn get_from_batch<'a>(
187    imm: &'a ImmutableMemtable,
188    table_key: TableKey<&[u8]>,
189    read_epoch: HummockEpoch,
190    read_options: &ReadOptions,
191    local_stats: &mut StoreLocalStatistic,
192) -> Option<(HummockValue<&'a Bytes>, EpochWithGap)> {
193    imm.get(table_key, read_epoch, read_options).inspect(|_| {
194        local_stats.get_shared_buffer_hit_counts += 1;
195    })
196}
197
198#[cfg(test)]
199mod tests {
200    use bytes::Bytes;
201    use risingwave_common::catalog::TableId;
202    use risingwave_common::hash::VirtualNode;
203    use risingwave_common::util::epoch::test_epoch;
204    use risingwave_hummock_sdk::EpochWithGap;
205    use risingwave_hummock_sdk::key::{FullKey, TableKey, UserKey};
206
207    use super::{CachePolicy, get_from_sstable_info};
208    use crate::hummock::iterator::test_utils::{iterator_test_key_of, mock_sstable_store};
209    use crate::hummock::test_utils::{
210        default_builder_opt_for_test, gen_test_sstable_info, gen_test_sstable_with_table_ids,
211        test_value_of,
212    };
213    use crate::hummock::value::HummockValue;
214    use crate::monitor::StoreLocalStatistic;
215    use crate::store::ReadOptions;
216
217    #[tokio::test]
218    async fn test_get_collects_stats_when_seek_passes_sst_end() {
219        let sstable_store = mock_sstable_store().await;
220        let sstable_info = gen_test_sstable_info(
221            default_builder_opt_for_test(),
222            1,
223            (0..10).map(|idx| {
224                (
225                    iterator_test_key_of(idx),
226                    HummockValue::put(format!("value_{idx}").into_bytes()),
227                )
228            }),
229            sstable_store.clone(),
230        )
231        .await;
232        let mut stats = StoreLocalStatistic::default();
233        let key = iterator_test_key_of(10);
234        let read_options = ReadOptions::default();
235
236        let result = get_from_sstable_info(
237            sstable_store,
238            &sstable_info,
239            key.to_ref(),
240            &read_options,
241            None,
242            &mut stats,
243        )
244        .await
245        .unwrap();
246
247        assert!(result.is_none());
248        drop(result);
249        assert_eq!(stats.cache_data_block_total, 1);
250    }
251
252    #[tokio::test]
253    async fn test_point_get_reads_only_requested_table_blocks() {
254        let sstable_store = mock_sstable_store().await;
255        let mut builder_options = default_builder_opt_for_test();
256        builder_options.block_capacity = 128;
257
258        let test_user_key = |table_id, key: &str| {
259            UserKey::new(
260                TableId::new(table_id),
261                TableKey(Bytes::from(
262                    [VirtualNode::ZERO.to_be_bytes().as_slice(), key.as_bytes()].concat(),
263                )),
264            )
265        };
266        let test_key = |table_id, key: &str| FullKey {
267            user_key: test_user_key(table_id, key),
268            epoch_with_gap: EpochWithGap::new_from_epoch(test_epoch(1)),
269        };
270        let kv_pairs = (1..=2).flat_map(|table_id| {
271            (0..8).map(move |idx| {
272                (
273                    test_key(table_id, &format!("key_{idx:05}")),
274                    HummockValue::put(Bytes::from(test_value_of(idx))),
275                )
276            })
277        });
278        let (sstable, sstable_info) = gen_test_sstable_with_table_ids(
279            builder_options,
280            10,
281            kv_pairs,
282            sstable_store.clone(),
283            vec![1, 2],
284        )
285        .await;
286        let table_2_block_start = sstable
287            .meta
288            .block_metas
289            .partition_point(|block_meta| block_meta.table_id() < TableId::new(2));
290        assert!(table_2_block_start > 0);
291
292        // The queried table-2 key sorts before the first table-2 block. Without a point-get table
293        // id filter, the SST iterator seeks to the previous table-1 block first.
294        let full_key = test_key(2, "key");
295        let mut local_stats = StoreLocalStatistic::default();
296        let read_options = ReadOptions {
297            cache_policy: CachePolicy::Disable,
298            ..Default::default()
299        };
300
301        {
302            let result = get_from_sstable_info(
303                sstable_store,
304                &sstable_info,
305                full_key.to_ref(),
306                &read_options,
307                None,
308                &mut local_stats,
309            )
310            .await
311            .unwrap();
312            assert!(result.is_none());
313        }
314        assert_eq!(local_stats.cache_data_block_total, 1);
315    }
316}
risingwave_storage/hummock/mod.rs

risingwave_storage/hummock/
mod.rs
