risingwave_storage/hummock/event_handler/
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
15use std::collections::{HashMap, HashSet};
16use std::sync::Arc;
17
18use itertools::Itertools;
19use parking_lot::{RwLock, RwLockReadGuard};
20use risingwave_common::bitmap::Bitmap;
21use risingwave_common::catalog::TableId;
22use risingwave_hummock_sdk::{HummockEpoch, HummockRawObjectId};
23use thiserror_ext::AsReport;
24use tokio::sync::oneshot;
25
26use crate::hummock::HummockResult;
27use crate::hummock::shared_buffer::shared_buffer_batch::{SharedBufferBatch, SharedBufferBatchId};
28use crate::mem_table::ImmutableMemtable;
29use crate::store::SealCurrentEpochOptions;
30
31pub mod hummock_event_handler;
32pub mod refiller;
33pub mod uploader;
34
35pub(crate) use hummock_event_handler::HummockEventHandler;
36use risingwave_hummock_sdk::vector_index::VectorIndexAdd;
37use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
38use risingwave_pb::meta::PbTableRefillRuntimeConfig;
39use risingwave_pb::meta::subscribe_response::Operation;
40
41use super::store::version::HummockReadVersion;
42use crate::hummock::event_handler::hummock_event_handler::HummockEventSender;
43use crate::hummock::event_handler::refiller::TableCacheRefillMonitorSnapshot;
44use crate::hummock::event_handler::uploader::SyncedData;
45use crate::hummock::utils::MemoryTracker;
46
47#[derive(Debug)]
48pub struct BufferWriteRequest {
49    pub batch: SharedBufferBatch,
50    pub epoch: HummockEpoch,
51    pub grant_sender: oneshot::Sender<()>,
52}
53
54#[derive(Debug)]
55pub enum HummockVersionUpdate {
56    VersionDeltas(Vec<HummockVersionDelta>),
57    PinnedVersion(Box<HummockVersion>),
58}
59
60#[derive(Debug)]
61pub enum HummockObserverEvent {
62    VersionUpdate(HummockVersionUpdate),
63    TableRefillRuntimeConfig(Operation, PbTableRefillRuntimeConfig),
64}
65
66pub enum HummockEvent {
67    /// Notify that we may flush the shared buffer.
68    BufferMayFlush,
69
70    /// An epoch is going to be synced. Once the event is processed, there will be no more flush
71    /// task on this epoch. Previous concurrent flush task join handle will be returned by the join
72    /// handle sender.
73    SyncEpoch {
74        sync_result_sender: oneshot::Sender<HummockResult<SyncedData>>,
75        sync_table_epochs: Vec<(HummockEpoch, HashSet<TableId>)>,
76    },
77
78    /// Clear shared buffer and reset all states
79    Clear(oneshot::Sender<()>, Option<HashSet<TableId>>),
80
81    Shutdown,
82
83    ImmToUploader {
84        instance_id: SharedBufferBatchId,
85        imms: Vec<(ImmutableMemtable, MemoryTracker)>,
86    },
87
88    StartEpoch {
89        epoch: HummockEpoch,
90        table_ids: HashSet<TableId>,
91    },
92
93    InitEpoch {
94        instance_id: LocalInstanceId,
95        init_epoch: HummockEpoch,
96    },
97
98    LocalSealEpoch {
99        instance_id: LocalInstanceId,
100        next_epoch: HummockEpoch,
101        opts: SealCurrentEpochOptions,
102    },
103
104    #[cfg(any(test, feature = "test"))]
105    /// Flush all previous event. When all previous events has been consumed, the event handler
106    /// will notify
107    FlushEvent(oneshot::Sender<()>),
108
109    RegisterReadVersion {
110        table_id: TableId,
111        new_read_version_sender: oneshot::Sender<(HummockReadVersionRef, LocalInstanceGuard)>,
112        is_replicated: bool,
113        vnodes: Arc<Bitmap>,
114    },
115
116    DestroyReadVersion {
117        instance_id: LocalInstanceId,
118    },
119
120    RegisterVectorWriter {
121        table_id: TableId,
122        init_epoch: HummockEpoch,
123    },
124
125    VectorWriterSealEpoch {
126        table_id: TableId,
127        next_epoch: HummockEpoch,
128        add: Option<VectorIndexAdd>,
129    },
130
131    DropVectorWriter {
132        table_id: TableId,
133    },
134
135    GetMinUncommittedObjectId {
136        result_tx: oneshot::Sender<Option<HummockRawObjectId>>,
137    },
138
139    GetTableCacheRefillMonitorSnapshot {
140        result_tx: oneshot::Sender<TableCacheRefillMonitorSnapshot>,
141    },
142}
143
144impl HummockEvent {
145    fn to_debug_string(&self) -> String {
146        match self {
147            HummockEvent::BufferMayFlush => "BufferMayFlush".to_owned(),
148
149            HummockEvent::SyncEpoch {
150                sync_result_sender: _,
151                sync_table_epochs,
152            } => format!("AwaitSyncEpoch epoch {:?}", sync_table_epochs),
153
154            HummockEvent::Clear(_, table_ids) => {
155                format!("Clear {:?}", table_ids)
156            }
157
158            HummockEvent::Shutdown => "Shutdown".to_owned(),
159
160            HummockEvent::StartEpoch { epoch, table_ids } => {
161                format!("StartEpoch {} {:?}", epoch, table_ids)
162            }
163
164            HummockEvent::InitEpoch {
165                instance_id,
166                init_epoch,
167            } => {
168                format!("InitEpoch {} {}", instance_id, init_epoch)
169            }
170
171            HummockEvent::ImmToUploader { instance_id, imms } => {
172                format!(
173                    "ImmToUploader {} {:?}",
174                    instance_id,
175                    imms.iter().map(|(imm, _)| imm.batch_id()).collect_vec()
176                )
177            }
178
179            HummockEvent::LocalSealEpoch {
180                instance_id,
181                next_epoch,
182                opts,
183            } => {
184                format!(
185                    "LocalSealEpoch next_epoch: {}, instance_id: {}, opts: {:?}",
186                    next_epoch, instance_id, opts
187                )
188            }
189
190            HummockEvent::RegisterReadVersion {
191                table_id,
192                new_read_version_sender: _,
193                is_replicated,
194                vnodes: _,
195            } => format!(
196                "RegisterReadVersion table_id {:?}, is_replicated: {:?}",
197                table_id, is_replicated
198            ),
199
200            HummockEvent::DestroyReadVersion { instance_id } => {
201                format!("DestroyReadVersion instance_id {:?}", instance_id)
202            }
203
204            #[cfg(any(test, feature = "test"))]
205            HummockEvent::FlushEvent(_) => "FlushEvent".to_owned(),
206            HummockEvent::GetMinUncommittedObjectId { .. } => {
207                "GetMinUncommittedObjectId".to_owned()
208            }
209            HummockEvent::GetTableCacheRefillMonitorSnapshot { .. } => {
210                "GetTableCacheRefillMonitorSnapshot".to_owned()
211            }
212            HummockEvent::RegisterVectorWriter { .. } => "RegisterVectorWriter".to_owned(),
213            HummockEvent::VectorWriterSealEpoch { .. } => "VectorWriterSealEpoch".to_owned(),
214            HummockEvent::DropVectorWriter { .. } => "DropVectorWriter".to_owned(),
215        }
216    }
217
218    pub fn event_name(&self) -> &'static str {
219        match self {
220            HummockEvent::BufferMayFlush => "BufferMayFlush",
221            HummockEvent::SyncEpoch { .. } => "SyncEpoch",
222            HummockEvent::Clear(..) => "Clear",
223            HummockEvent::Shutdown => "Shutdown",
224            HummockEvent::ImmToUploader { .. } => "ImmToUploader",
225            HummockEvent::StartEpoch { .. } => "StartEpoch",
226            HummockEvent::InitEpoch { .. } => "InitEpoch",
227            HummockEvent::LocalSealEpoch { .. } => "LocalSealEpoch",
228            #[cfg(any(test, feature = "test"))]
229            HummockEvent::FlushEvent(_) => "FlushEvent",
230            HummockEvent::RegisterReadVersion { .. } => "RegisterReadVersion",
231            HummockEvent::DestroyReadVersion { .. } => "DestroyReadVersion",
232            HummockEvent::RegisterVectorWriter { .. } => "RegisterVectorWriter",
233            HummockEvent::VectorWriterSealEpoch { .. } => "VectorWriterSealEpoch",
234            HummockEvent::DropVectorWriter { .. } => "DropVectorWriter",
235            HummockEvent::GetMinUncommittedObjectId { .. } => "GetMinUncommittedObjectId",
236            HummockEvent::GetTableCacheRefillMonitorSnapshot { .. } => {
237                "GetTableCacheRefillMonitorSnapshot"
238            }
239        }
240    }
241}
242
243impl std::fmt::Debug for HummockEvent {
244    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
245        f.debug_struct("HummockEvent")
246            .field("debug_string", &self.to_debug_string())
247            .finish()
248    }
249}
250
251pub type LocalInstanceId = u64;
252pub const TEST_LOCAL_INSTANCE_ID: LocalInstanceId = 233;
253pub type HummockReadVersionRef = Arc<RwLock<HummockReadVersion>>;
254pub type ReadVersionMappingType = HashMap<TableId, HashMap<LocalInstanceId, HummockReadVersionRef>>;
255pub type ReadOnlyReadVersionMapping = ReadOnlyRwLockRef<ReadVersionMappingType>;
256
257pub struct ReadOnlyRwLockRef<T>(Arc<RwLock<T>>);
258
259impl<T> Clone for ReadOnlyRwLockRef<T> {
260    fn clone(&self) -> Self {
261        Self(self.0.clone())
262    }
263}
264
265impl<T> ReadOnlyRwLockRef<T> {
266    pub fn new(inner: Arc<RwLock<T>>) -> Self {
267        Self(inner)
268    }
269
270    pub fn read(&self) -> RwLockReadGuard<'_, T> {
271        self.0.read()
272    }
273}
274
275pub struct LocalInstanceGuard {
276    pub table_id: TableId,
277    pub instance_id: LocalInstanceId,
278    // Only send destroy event when event_sender when is_some
279    event_sender: Option<HummockEventSender>,
280}
281
282impl Drop for LocalInstanceGuard {
283    fn drop(&mut self) {
284        if let Some(sender) = self.event_sender.take() {
285            // If sending fails, it means that event_handler and event_channel have been destroyed, no
286            // need to handle failure
287            sender
288                .send(HummockEvent::DestroyReadVersion {
289                    instance_id: self.instance_id,
290                })
291                .unwrap_or_else(|err| {
292                    tracing::debug!(
293                        error = %err.as_report(),
294                        table_id = %self.table_id,
295                        instance_id = self.instance_id,
296                        "LocalInstanceGuard Drop SendError",
297                    )
298                })
299        }
300    }
301}
risingwave_storage/hummock/event_handler/mod.rs

risingwave_storage/hummock/event_handler/
mod.rs
