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 use store::*;
47mod validator;
48pub mod value;
49pub mod write_limiter;
50
51pub mod recent_filter;
52pub use recent_filter::*;
53
54pub mod block_stream;
55mod time_travel_version_cache;
56
57pub(crate) mod vector;
58
59mod object_id_manager;
60pub use error::*;
61pub use object_id_manager::*;
62pub use risingwave_common::cache::{CacheableEntry, LookupResult, LruCache};
63pub use validator::*;
64use value::*;
65
66use self::iterator::HummockIterator;
67pub use self::sstable_store::*;
68use crate::mem_table::ImmutableMemtable;
69use crate::monitor::StoreLocalStatistic;
70use crate::store::ReadOptions;
71
72pub(in crate::hummock) struct IteratorStatsGuard<'a, TI: HummockIterator> {
73    iter: Option<TI>,
74    parent_stats: &'a mut StoreLocalStatistic,
75}
76
77impl<'a, TI: HummockIterator> IteratorStatsGuard<'a, TI> {
78    pub(in crate::hummock) fn new(iter: TI, parent_stats: &'a mut StoreLocalStatistic) -> Self {
79        Self {
80            iter: Some(iter),
81            parent_stats,
82        }
83    }
84
85    pub(in crate::hummock) fn iter(&self) -> &TI {
86        self.iter.as_ref().expect("iterator must be present")
87    }
88
89    pub(in crate::hummock) fn iter_mut(&mut self) -> &mut TI {
90        self.iter.as_mut().expect("iterator must be present")
91    }
92
93    fn collect(&mut self) {
94        if let Some(iter) = &self.iter {
95            iter.collect_local_statistic(self.parent_stats);
96        }
97    }
98
99    pub(in crate::hummock) fn handoff(mut self) -> TI {
100        self.iter.take().expect("iterator must be present")
101    }
102
103    pub(in crate::hummock) fn collect_and_take(mut self) -> TI {
104        self.collect();
105        self.iter.take().expect("iterator must be present")
106    }
107}
108
109impl<TI: HummockIterator> Drop for IteratorStatsGuard<'_, TI> {
110    fn drop(&mut self) {
111        self.collect();
112    }
113}
114
115pub async fn get_from_sstable_info(
116    sstable_store_ref: SstableStoreRef,
117    sstable_info: &SstableInfo,
118    full_key: FullKey<&[u8]>,
119    read_options: &ReadOptions,
120    dist_key_hash: Option<u64>,
121    local_stats: &mut StoreLocalStatistic,
122) -> HummockResult<Option<impl HummockIterator>> {
123    let sstable = sstable_store_ref.sstable(sstable_info, local_stats).await?;
124
125    // SST filter key is the distribution key, which does not need to be the prefix of pk, and does not
126    // contain `TablePrefix` and `VnodePrefix`.
127    if let Some(hash) = dist_key_hash
128        && !hit_sstable_filter(
129            &sstable,
130            &(
131                Bound::Included(full_key.user_key),
132                Bound::Included(full_key.user_key),
133            ),
134            hash,
135            local_stats,
136        )
137    {
138        return Ok(None);
139    }
140
141    let mut iter = IteratorStatsGuard::new(
142        SstableIterator::create(
143            sstable,
144            sstable_store_ref.clone(),
145            Arc::new(SstableIteratorReadOptions::from_read_options(read_options)),
146            sstable_info,
147        ),
148        local_stats,
149    );
150    iter.iter_mut().seek(full_key).await?;
151    // Iterator has sought passed the borders.
152    if !iter.iter().is_valid() {
153        return Ok(None);
154    }
155
156    // Iterator gets us the key, we tell if it's the key we want
157    // or key next to it.
158    let value = if iter.iter().key().user_key == full_key.user_key {
159        Some(iter.collect_and_take())
160    } else {
161        None
162    };
163
164    Ok(value)
165}
166
167pub fn hit_sstable_filter(
168    sstable_ref: &Sstable,
169    user_key_range: &UserKeyRangeRef<'_>,
170    prefix_hash: u64,
171    local_stats: &mut StoreLocalStatistic,
172) -> bool {
173    local_stats.bloom_filter_check_counts += 1;
174    let may_exist = sstable_ref.may_match_hash(user_key_range, prefix_hash);
175    if !may_exist {
176        local_stats.bloom_filter_true_negative_counts += 1;
177    }
178    may_exist
179}
180
181/// Get `user_value` from `ImmutableMemtable`
182pub fn get_from_batch<'a>(
183    imm: &'a ImmutableMemtable,
184    table_key: TableKey<&[u8]>,
185    read_epoch: HummockEpoch,
186    read_options: &ReadOptions,
187    local_stats: &mut StoreLocalStatistic,
188) -> Option<(HummockValue<&'a Bytes>, EpochWithGap)> {
189    imm.get(table_key, read_epoch, read_options).inspect(|_| {
190        local_stats.get_shared_buffer_hit_counts += 1;
191    })
192}
193
194#[cfg(test)]
195mod tests {
196    use super::get_from_sstable_info;
197    use crate::hummock::iterator::test_utils::{iterator_test_key_of, mock_sstable_store};
198    use crate::hummock::test_utils::{default_builder_opt_for_test, gen_test_sstable_info};
199    use crate::hummock::value::HummockValue;
200    use crate::monitor::StoreLocalStatistic;
201    use crate::store::ReadOptions;
202
203    #[tokio::test]
204    async fn test_get_collects_stats_when_seek_passes_sst_end() {
205        let sstable_store = mock_sstable_store().await;
206        let sstable_info = gen_test_sstable_info(
207            default_builder_opt_for_test(),
208            1,
209            (0..10).map(|idx| {
210                (
211                    iterator_test_key_of(idx),
212                    HummockValue::put(format!("value_{idx}").into_bytes()),
213                )
214            }),
215            sstable_store.clone(),
216        )
217        .await;
218        let mut stats = StoreLocalStatistic::default();
219        let key = iterator_test_key_of(10);
220        let read_options = ReadOptions::default();
221
222        let result = get_from_sstable_info(
223            sstable_store,
224            &sstable_info,
225            key.to_ref(),
226            &read_options,
227            None,
228            &mut stats,
229        )
230        .await
231        .unwrap();
232
233        assert!(result.is_none());
234        drop(result);
235        assert_eq!(stats.cache_data_block_total, 1);
236    }
237}
risingwave_storage/hummock/mod.rs

risingwave_storage/hummock/
mod.rs
