risingwave_hummock_trace/
opts.rs

1// Copyright 2025 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 bincode::{Decode, Encode};
16use foyer::Hint;
17use risingwave_common::bitmap::Bitmap;
18use risingwave_common::cache::CachePriority;
19use risingwave_common::catalog::{TableId, TableOption};
20use risingwave_common::util::epoch::EpochPair;
21use risingwave_hummock_sdk::{HummockReadEpoch, HummockVersionId};
22use risingwave_pb::common::PbBuffer;
23
24use crate::TracedBytes;
25
26#[derive(Encode, Decode, PartialEq, Eq, Debug, Clone)]
27pub struct TracedPrefetchOptions {
28    pub prefetch: bool,
29    pub for_large_query: bool,
30}
31
32#[derive(Encode, Decode, PartialEq, Eq, Debug, Clone)]
33pub enum TracedCachePolicy {
34    Disable,
35    Fill(TracedCachePriority),
36    NotFill,
37}
38
39#[derive(Encode, Decode, PartialEq, Eq, Debug, Clone)]
40pub enum TracedCachePriority {
41    High,
42    Low,
43}
44
45impl From<CachePriority> for TracedCachePriority {
46    fn from(value: CachePriority) -> Self {
47        match value {
48            CachePriority::High => Self::High,
49            CachePriority::Low => Self::Low,
50        }
51    }
52}
53
54impl From<TracedCachePriority> for CachePriority {
55    fn from(value: TracedCachePriority) -> Self {
56        match value {
57            TracedCachePriority::High => Self::High,
58            TracedCachePriority::Low => Self::Low,
59        }
60    }
61}
62
63impl From<Hint> for TracedCachePriority {
64    fn from(value: Hint) -> Self {
65        match value {
66            Hint::Normal => Self::High,
67            Hint::Low => Self::Low,
68        }
69    }
70}
71
72impl From<TracedCachePriority> for Hint {
73    fn from(value: TracedCachePriority) -> Self {
74        match value {
75            TracedCachePriority::High => Self::Normal,
76            TracedCachePriority::Low => Self::Low,
77        }
78    }
79}
80
81#[derive(Encode, Decode, PartialEq, Eq, Debug, Clone)]
82pub struct TracedTableId {
83    pub table_id: u32,
84}
85
86impl From<TableId> for TracedTableId {
87    fn from(value: TableId) -> Self {
88        Self {
89            table_id: value.table_id,
90        }
91    }
92}
93
94impl From<TracedTableId> for TableId {
95    fn from(value: TracedTableId) -> Self {
96        Self {
97            table_id: value.table_id,
98        }
99    }
100}
101
102#[derive(Encode, Decode, PartialEq, Eq, Debug, Clone)]
103pub struct TracedReadOptions {
104    pub prefix_hint: Option<TracedBytes>,
105    pub prefetch_options: TracedPrefetchOptions,
106    pub cache_policy: TracedCachePolicy,
107
108    pub retention_seconds: Option<u32>,
109    pub table_id: TracedTableId,
110    pub read_version_from_backup: bool,
111    pub read_committed: bool,
112}
113
114impl TracedReadOptions {
115    pub fn for_test(table_id: u32) -> Self {
116        Self {
117            prefix_hint: Some(TracedBytes::from(vec![0])),
118            prefetch_options: TracedPrefetchOptions {
119                prefetch: true,
120                for_large_query: true,
121            },
122            cache_policy: TracedCachePolicy::Disable,
123            retention_seconds: None,
124            table_id: TracedTableId { table_id },
125            read_version_from_backup: false,
126            read_committed: false,
127        }
128    }
129}
130
131#[derive(Encode, Decode, PartialEq, Eq, Debug, Clone)]
132pub struct TracedTableOption {
133    pub retention_seconds: Option<u32>,
134}
135
136impl From<TableOption> for TracedTableOption {
137    fn from(value: TableOption) -> Self {
138        Self {
139            retention_seconds: value.retention_seconds,
140        }
141    }
142}
143
144impl From<TracedTableOption> for TableOption {
145    fn from(value: TracedTableOption) -> Self {
146        Self {
147            retention_seconds: value.retention_seconds,
148        }
149    }
150}
151
152#[derive(Encode, Decode, PartialEq, Eq, Debug, Clone)]
153pub enum TracedOpConsistencyLevel {
154    Inconsistent,
155    ConsistentOldValue,
156}
157
158#[derive(Encode, Decode, PartialEq, Eq, Debug, Clone)]
159pub struct TracedNewLocalOptions {
160    pub table_id: TracedTableId,
161    pub op_consistency_level: TracedOpConsistencyLevel,
162    pub table_option: TracedTableOption,
163    pub is_replicated: bool,
164    pub vnodes: TracedBitmap,
165    pub upload_on_flush: bool,
166}
167
168#[derive(Encode, Decode, PartialEq, Debug, Clone)]
169pub struct TracedTryWaitEpochOptions {
170    pub table_id: TracedTableId,
171}
172
173#[cfg(test)]
174impl TracedNewLocalOptions {
175    pub(crate) fn for_test(table_id: u32) -> Self {
176        use risingwave_common::hash::VirtualNode;
177
178        Self {
179            table_id: TracedTableId { table_id },
180            op_consistency_level: TracedOpConsistencyLevel::Inconsistent,
181            table_option: TracedTableOption {
182                retention_seconds: None,
183            },
184            is_replicated: false,
185            vnodes: TracedBitmap::from(Bitmap::ones(VirtualNode::COUNT_FOR_TEST)),
186            upload_on_flush: true,
187        }
188    }
189}
190
191pub type TracedHummockEpoch = u64;
192
193#[derive(Debug, Clone, PartialEq, Eq, Decode, Encode)]
194pub enum TracedHummockReadEpoch {
195    Committed(TracedHummockEpoch),
196    BatchQueryReadCommitted(TracedHummockEpoch, u64),
197    NoWait(TracedHummockEpoch),
198    Backup(TracedHummockEpoch),
199    TimeTravel(TracedHummockEpoch),
200}
201
202impl From<HummockReadEpoch> for TracedHummockReadEpoch {
203    fn from(value: HummockReadEpoch) -> Self {
204        match value {
205            HummockReadEpoch::Committed(epoch) => Self::Committed(epoch),
206            HummockReadEpoch::BatchQueryCommitted(epoch, version_id) => {
207                Self::BatchQueryReadCommitted(epoch, version_id.to_u64())
208            }
209            HummockReadEpoch::NoWait(epoch) => Self::NoWait(epoch),
210            HummockReadEpoch::Backup(epoch) => Self::Backup(epoch),
211            HummockReadEpoch::TimeTravel(epoch) => Self::TimeTravel(epoch),
212        }
213    }
214}
215
216impl From<TracedHummockReadEpoch> for HummockReadEpoch {
217    fn from(value: TracedHummockReadEpoch) -> Self {
218        match value {
219            TracedHummockReadEpoch::Committed(epoch) => Self::Committed(epoch),
220            TracedHummockReadEpoch::BatchQueryReadCommitted(epoch, version_id) => {
221                Self::BatchQueryCommitted(epoch, HummockVersionId::new(version_id))
222            }
223            TracedHummockReadEpoch::NoWait(epoch) => Self::NoWait(epoch),
224            TracedHummockReadEpoch::Backup(epoch) => Self::Backup(epoch),
225            TracedHummockReadEpoch::TimeTravel(epoch) => Self::TimeTravel(epoch),
226        }
227    }
228}
229
230#[derive(Debug, Clone, PartialEq, Eq, Decode, Encode)]
231pub struct TracedEpochPair {
232    pub curr: TracedHummockEpoch,
233    pub prev: TracedHummockEpoch,
234}
235
236impl From<EpochPair> for TracedEpochPair {
237    fn from(value: EpochPair) -> Self {
238        TracedEpochPair {
239            curr: value.curr,
240            prev: value.prev,
241        }
242    }
243}
244
245impl From<TracedEpochPair> for EpochPair {
246    fn from(value: TracedEpochPair) -> Self {
247        EpochPair {
248            curr: value.curr,
249            prev: value.prev,
250        }
251    }
252}
253
254#[derive(Debug, Clone, PartialEq, Eq, Decode, Encode)]
255pub struct TracedInitOptions {
256    pub epoch: TracedEpochPair,
257}
258
259#[derive(Debug, Clone, PartialEq, Eq, Decode, Encode)]
260pub struct TracedSealCurrentEpochOptions {
261    // The watermark is serialized into protobuf
262    pub table_watermarks: Option<(bool, Vec<Vec<u8>>, bool)>,
263    pub switch_op_consistency_level: Option<bool>,
264}
265
266#[derive(Debug, Clone, PartialEq, Eq, Decode, Encode)]
267pub struct TracedBitmap {
268    pub compression: i32,
269    pub body: Vec<u8>,
270}
271
272impl From<Bitmap> for TracedBitmap {
273    fn from(value: Bitmap) -> Self {
274        let pb = value.to_protobuf();
275        Self {
276            compression: pb.compression,
277            body: pb.body,
278        }
279    }
280}
281
282impl From<TracedBitmap> for Bitmap {
283    fn from(value: TracedBitmap) -> Self {
284        let pb = PbBuffer {
285            compression: value.compression,
286            body: value.body,
287        };
288        Bitmap::from(&pb)
289    }
290}
risingwave_hummock_trace/opts.rs

risingwave_hummock_trace/
opts.rs
