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

risingwave_storage/hummock/event_handler/
mod.rs
