risingwave_pb/
id.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 std::any::type_name;
16use std::borrow::Borrow;
17use std::fmt::Formatter;
18use std::iter::Step;
19use std::num::TryFromIntError;
20use std::ops::{Add, AddAssign, Sub};
21use std::str::FromStr;
22
23use sea_orm::sea_query::{ArrayType, ValueTypeErr};
24use sea_orm::{ColIdx, ColumnType, DbErr, QueryResult, TryGetError};
25use serde::{Deserialize, Deserializer, Serialize, Serializer};
26use thiserror_ext::AsReport;
27use tracing::warn;
28
29use crate::catalog::source::OptionalAssociatedTableId;
30use crate::catalog::table::OptionalAssociatedSourceId;
31
32pub const OBJECT_ID_PLACEHOLDER: u32 = u32::MAX - 1;
33
34#[derive(Clone, Copy, Default, Hash, PartialOrd, PartialEq, Eq, Ord)]
35#[repr(transparent)]
36pub struct TypedId<const N: usize, P>(pub(crate) P);
37
38// SAFETY: `TypedId` is `#[repr(transparent)]` over `P`, so it is layout-compatible with `P`.
39unsafe impl<const N: usize, P> ::prost::TransparentOver for TypedId<N, P> {
40    type Inner = P;
41}
42
43impl<const N: usize, P: std::fmt::Debug> std::fmt::Debug for TypedId<N, P> {
44    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
45        <P as std::fmt::Debug>::fmt(&self.0, f)
46    }
47}
48
49impl<const N: usize, P: std::fmt::Display> std::fmt::Display for TypedId<N, P> {
50    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
51        <P as std::fmt::Display>::fmt(&self.0, f)
52    }
53}
54
55impl<const N: usize, P: std::fmt::UpperHex> std::fmt::UpperHex for TypedId<N, P> {
56    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
57        <P as std::fmt::UpperHex>::fmt(&self.0, f)
58    }
59}
60
61impl<const N: usize, P: PartialEq> PartialEq<P> for TypedId<N, P> {
62    fn eq(&self, other: &P) -> bool {
63        self.0 == *other
64    }
65}
66
67impl<const N: usize, P: FromStr> FromStr for TypedId<N, P> {
68    type Err = <P as FromStr>::Err;
69
70    fn from_str(s: &str) -> Result<Self, Self::Err> {
71        Ok(Self(<P as FromStr>::from_str(s)?))
72    }
73}
74
75impl<const N: usize, P> Borrow<P> for TypedId<N, P> {
76    fn borrow(&self) -> &P {
77        // Safety: transparent repr
78        unsafe { std::mem::transmute(self) }
79    }
80}
81
82impl<const N: usize, P> TypedId<N, P> {
83    pub const fn new(inner: P) -> Self {
84        TypedId(inner)
85    }
86}
87
88impl<const N: usize, P> TypedId<N, P>
89where
90    Self: UniqueTypedIdDeclaration,
91{
92    #[expect(clippy::wrong_self_convention)]
93    pub fn as_raw_id(self) -> P {
94        self.0
95    }
96}
97
98type TypedU32Id<const N: usize> = TypedId<N, u32>;
99
100impl<const N: usize> TypedU32Id<N> {
101    pub const fn placeholder() -> Self {
102        Self(OBJECT_ID_PLACEHOLDER)
103    }
104
105    pub fn is_placeholder(&self) -> bool {
106        self.0 == OBJECT_ID_PLACEHOLDER
107    }
108}
109
110impl<const N: usize, P> From<P> for TypedId<N, P> {
111    fn from(id: P) -> Self {
112        Self(id)
113    }
114}
115
116/// Implements `from_$signed` and `to_$signed` conversion methods for TypedId types.
117/// - `$unsigned`: the unsigned primitive type (u32 or u64)
118/// - `$signed`: the corresponding signed type for DB storage (i32 or i64)
119macro_rules! impl_typed_id_conversion {
120    ($unsigned:ty, $signed:ty) => {
121        paste::paste! {
122            impl<const N: usize> TypedId<N, $unsigned>
123            {
124                fn [< from_ $signed >](inner: $signed) -> Self {
125                    Self(inner.try_into().unwrap_or_else(|e: TryFromIntError| {
126                        if cfg!(debug_assertions) {
127                            panic!(
128                                "invalid {} id {} for {}: {:?}",
129                                stringify!($signed),
130                                inner,
131                                type_name::<Self>(),
132                                e.as_report()
133                            );
134                        } else {
135                            warn!(
136                                "invalid {} id {} for {}: {:?}",
137                                stringify!($signed),
138                                inner,
139                                type_name::<Self>(),
140                                e.as_report()
141                            );
142                            inner as _
143                        }
144                    }))
145                }
146
147                fn [< to_ $signed >](self) -> $signed {
148                    self.0.try_into().unwrap_or_else(|e: TryFromIntError| {
149                        if cfg!(debug_assertions) {
150                            panic!(
151                                "invalid {} id {} for {}: {:?}",
152                                stringify!($unsigned),
153                                self.0,
154                                type_name::<Self>(),
155                                e.as_report()
156                            );
157                        } else {
158                            warn!(
159                                "invalid {} id {} for {}: {:?}",
160                                stringify!($unsigned),
161                                self.0,
162                                type_name::<Self>(),
163                                e.as_report()
164                            );
165                            self.0 as _
166                        }
167                    })
168                }
169
170                pub fn [< as_ $signed _id >](self) -> $signed {
171                    self.[< to_ $signed >]()
172                }
173            }
174        }
175    };
176}
177
178impl_typed_id_conversion!(u32, i32);
179impl_typed_id_conversion!(u64, i64);
180
181/// Implements SeaORM traits for TypedId types.
182/// - `$unsigned`: the unsigned primitive type (u32 or u64)
183/// - `$signed`: the corresponding signed type for DB storage (i32 or i64)
184macro_rules! impl_sea_orm_for_typed_id {
185    ($unsigned:ty, $signed:ty) => {
186        paste::paste! {
187            impl<const N: usize> From<TypedId<N, $unsigned>> for sea_orm::Value
188            where
189                TypedId<N, $unsigned>: UniqueTypedIdDeclaration,
190            {
191                fn from(value: TypedId<N, $unsigned>) -> Self {
192                    sea_orm::Value::from(value.[< to_ $signed >]())
193                }
194            }
195
196            impl<const N: usize> sea_orm::sea_query::ValueType for TypedId<N, $unsigned>
197            where
198                Self: UniqueTypedIdDeclaration,
199            {
200                fn try_from(v: sea_orm::Value) -> Result<Self, ValueTypeErr> {
201                    let inner = <$signed as sea_orm::sea_query::ValueType>::try_from(v)?;
202                    Ok(Self::[< from_ $signed >](inner))
203                }
204
205                fn type_name() -> String {
206                    <$signed as sea_orm::sea_query::ValueType>::type_name()
207                }
208
209                fn array_type() -> ArrayType {
210                    <$signed as sea_orm::sea_query::ValueType>::array_type()
211                }
212
213                fn column_type() -> ColumnType {
214                    <$signed as sea_orm::sea_query::ValueType>::column_type()
215                }
216            }
217
218            impl<const N: usize> sea_orm::sea_query::Nullable for TypedId<N, $unsigned>
219            where
220                Self: UniqueTypedIdDeclaration,
221            {
222                fn null() -> sea_orm::Value {
223                    <$signed as sea_orm::sea_query::Nullable>::null()
224                }
225            }
226
227            impl<const N: usize> sea_orm::TryGetable for TypedId<N, $unsigned>
228            where
229                Self: UniqueTypedIdDeclaration,
230            {
231                fn try_get_by<I: ColIdx>(res: &QueryResult, index: I) -> Result<Self, TryGetError> {
232                    let inner = <$signed as sea_orm::TryGetable>::try_get_by(res, index)?;
233                    Ok(Self::[< from_ $signed >](inner))
234                }
235            }
236
237            impl<const N: usize> sea_orm::TryFromU64 for TypedId<N, $unsigned>
238            where
239                Self: UniqueTypedIdDeclaration,
240            {
241                fn try_from_u64(n: u64) -> Result<Self, DbErr> {
242                    Ok(Self::[< from_ $signed >](<$signed as sea_orm::TryFromU64>::try_from_u64(n)?))
243                }
244            }
245        }
246    };
247}
248
249impl_sea_orm_for_typed_id!(u32, i32);
250impl_sea_orm_for_typed_id!(u64, i64);
251
252impl<const N: usize, P: Serialize> Serialize for TypedId<N, P> {
253    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
254    where
255        S: Serializer,
256    {
257        <P as Serialize>::serialize(&self.0, serializer)
258    }
259}
260
261impl<'de, const N: usize, P: Deserialize<'de>> Deserialize<'de> for TypedId<N, P> {
262    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
263    where
264        D: Deserializer<'de>,
265    {
266        Ok(Self(<P as Deserialize>::deserialize(deserializer)?))
267    }
268}
269
270impl<const N: usize, P: Sub<Output = P>> Sub for TypedId<N, P> {
271    type Output = P;
272
273    fn sub(self, rhs: Self) -> Self::Output {
274        self.0 - rhs.0
275    }
276}
277
278impl<const N: usize, P: Step> Step for TypedId<N, P> {
279    fn steps_between(start: &Self, end: &Self) -> (usize, Option<usize>) {
280        P::steps_between(&start.0, &end.0)
281    }
282
283    fn forward_checked(start: Self, count: usize) -> Option<Self> {
284        P::forward_checked(start.0, count).map(Self)
285    }
286
287    fn backward_checked(start: Self, count: usize) -> Option<Self> {
288        P::backward_checked(start.0, count).map(Self)
289    }
290}
291
292macro_rules! impl_add {
293    ($type_name:ty) => {
294        impl<const N: usize> Add<$type_name> for TypedId<N, $type_name> {
295            type Output = Self;
296
297            fn add(self, rhs: $type_name) -> Self::Output {
298                Self(self.0.checked_add(rhs).unwrap())
299            }
300        }
301
302        impl<const N: usize> AddAssign<$type_name> for TypedId<N, $type_name> {
303            fn add_assign(&mut self, rhs: $type_name) {
304                self.0 = self.0.checked_add(rhs).unwrap()
305            }
306        }
307
308        impl<const N: usize> PartialEq<TypedId<N, $type_name>> for $type_name {
309            fn eq(&self, other: &TypedId<N, $type_name>) -> bool {
310                *self == other.0
311            }
312        }
313    };
314}
315
316impl_add!(u32);
317impl_add!(u64);
318
319#[expect(dead_code)]
320pub trait UniqueTypedIdDeclaration {}
321
322macro_rules! declare_id_type {
323    ($name:ident, $primitive:ty, $type_id:expr) => {
324        pub type $name = TypedId<{ $type_id }, $primitive>;
325        impl UniqueTypedIdDeclaration for $name {}
326    };
327}
328
329macro_rules! declare_id_types {
330    ($primitive:ty $(, $name:ident)+) => {
331        declare_id_types! {
332            $primitive, 0 $(, $name)+
333        }
334    };
335    ($primitive:ty, $next_type_id:expr) => {};
336    ($primitive:ty, $next_type_id:expr, $name:ident $(, $rest:ident)*) => {
337        declare_id_type! { $name, $primitive, $next_type_id }
338        declare_id_types! {
339            $primitive, $next_type_id + 1  $(, $rest)*
340        }
341    };
342    ($($invalid:tt)+) => {
343        compile_error!(stringify!($($invalid)+))
344    }
345}
346
347declare_id_types!(
348    u32,
349    TableId,
350    JobId,
351    DatabaseId,
352    SchemaId,
353    FragmentId,
354    ActorId,
355    WorkerId,
356    SinkId,
357    SourceId,
358    SubscriptionId,
359    IndexId,
360    ViewId,
361    FunctionId,
362    ConnectionId,
363    SecretId,
364    SubscriberId,
365    LocalOperatorId,
366    UserId,
367    RelationId
368);
369
370declare_id_type!(ObjectId, u32, 256);
371
372declare_id_types!(
373    u64,
374    GlobalOperatorId,
375    StreamNodeLocalOperatorId,
376    ExecutorId,
377    PartialGraphId,
378    HummockRawObjectId,
379    HummockSstableObjectId,
380    HummockSstableId,
381    HummockVectorFileId,
382    HummockHnswGraphFileId,
383    HummockVersionId,
384    CompactionGroupId,
385    IcebergCompactionTaskId
386);
387
388macro_rules! impl_as {
389    (@func $target_id_name:ident, $alias_name:ident) => {
390        paste::paste! {
391            pub fn [< as_ $alias_name >](self) -> $target_id_name {
392                $target_id_name::new(self.0)
393            }
394        }
395    };
396    (@func $target_id_name:ident) => {
397        paste::paste! {
398            impl_as! { @func $target_id_name, [< $target_id_name:snake >] }
399        }
400    };
401    ($src_id_name:ident $(,$target_id_name:ident)* $(,{$orig_target_id_name:ident , $alias_name:ident})*) => {
402        impl $src_id_name {
403            $(
404                impl_as! { @func $target_id_name }
405            )*
406            $(
407                impl_as! { @func $orig_target_id_name, $alias_name }
408            )*
409        }
410    }
411}
412
413impl JobId {
414    pub fn is_mv_table_id(self, table_id: TableId) -> bool {
415        self.0 == table_id.0
416    }
417}
418
419impl_as!(JobId, SinkId, IndexId, SubscriberId, {TableId, mv_table_id}, {SourceId, shared_source_id});
420impl_as!(TableId, JobId);
421
422impl From<StreamNodeLocalOperatorId> for LocalOperatorId {
423    fn from(value: StreamNodeLocalOperatorId) -> Self {
424        assert!(
425            value.0 <= u32::MAX as u64,
426            "oversized operator id {} in stream node",
427            value.0
428        );
429        Self(value.0 as u32)
430    }
431}
432
433impl From<LocalOperatorId> for StreamNodeLocalOperatorId {
434    fn from(value: LocalOperatorId) -> Self {
435        Self(value.0 as u64)
436    }
437}
438
439impl From<OptionalAssociatedTableId> for TableId {
440    fn from(value: OptionalAssociatedTableId) -> Self {
441        let OptionalAssociatedTableId::AssociatedTableId(table_id) = value;
442        table_id
443    }
444}
445
446impl From<TableId> for OptionalAssociatedTableId {
447    fn from(value: TableId) -> Self {
448        OptionalAssociatedTableId::AssociatedTableId(value)
449    }
450}
451
452impl_as!(SinkId, JobId);
453impl_as!(IndexId, JobId);
454impl_as!(SourceId, {JobId, share_source_job_id}, {TableId, cdc_table_id});
455impl_as!(SubscriptionId, SubscriberId);
456
457impl From<OptionalAssociatedSourceId> for SourceId {
458    fn from(value: OptionalAssociatedSourceId) -> Self {
459        let OptionalAssociatedSourceId::AssociatedSourceId(source_id) = value;
460        source_id
461    }
462}
463
464impl From<SourceId> for OptionalAssociatedSourceId {
465    fn from(value: SourceId) -> Self {
466        OptionalAssociatedSourceId::AssociatedSourceId(value)
467    }
468}
469
470macro_rules! impl_into_object {
471    ($mod_prefix:ty, $($type_name:ident),+) => {
472        $(
473            impl From<$type_name> for $mod_prefix {
474                fn from(value: $type_name) -> Self {
475                    <$mod_prefix>::$type_name(value)
476                }
477            }
478        )+
479    };
480}
481
482impl_into_object!(
483    crate::user::grant_privilege::Object,
484    DatabaseId,
485    TableId,
486    SchemaId,
487    SinkId,
488    SourceId,
489    SubscriptionId,
490    ViewId,
491    FunctionId,
492    ConnectionId,
493    SecretId
494);
495
496impl_into_object!(
497    crate::ddl_service::alter_name_request::Object,
498    DatabaseId,
499    TableId,
500    SchemaId,
501    SinkId,
502    SourceId,
503    SubscriptionId,
504    IndexId,
505    ViewId
506);
507
508impl_into_object!(
509    crate::ddl_service::alter_owner_request::Object,
510    DatabaseId,
511    TableId,
512    SchemaId,
513    SinkId,
514    SourceId,
515    SubscriptionId,
516    ViewId,
517    ConnectionId,
518    FunctionId,
519    SecretId
520);
521
522impl_into_object!(
523    crate::ddl_service::alter_set_schema_request::Object,
524    TableId,
525    ViewId,
526    SourceId,
527    SinkId,
528    SubscriptionId,
529    FunctionId,
530    ConnectionId
531);
532
533macro_rules! impl_into_rename_object {
534    ($($type_name:ident),+) => {
535        paste::paste! {
536            $(
537                impl From<([<$type_name Id>], [<$type_name Id>])> for crate::ddl_service::alter_swap_rename_request::Object {
538                    fn from((src_object_id, dst_object_id): ([<$type_name Id>], [<$type_name Id>])) -> Self {
539                        crate::ddl_service::alter_swap_rename_request::Object::$type_name(crate::ddl_service::alter_swap_rename_request::ObjectNameSwapPair {
540                            src_object_id: src_object_id.as_object_id(),
541                            dst_object_id: dst_object_id.as_object_id(),
542                        })
543                    }
544                }
545            )+
546        }
547    };
548}
549
550impl_into_rename_object!(Table, View, Source, Sink, Subscription);
551
552macro_rules! impl_object_id_conversion {
553    ($($type_name:ident),+) => {
554        $(
555            impl From<$type_name> for ObjectId {
556                fn from(value: $type_name) -> Self {
557                    Self::new(value.0)
558                }
559            }
560
561            impl $type_name {
562                pub fn as_object_id(self) -> ObjectId {
563                    ObjectId::new(self.0)
564                }
565            }
566        )+
567
568        paste::paste! {
569            impl ObjectId {
570                $(
571                    pub fn [< as_ $type_name:snake>](self) -> $type_name {
572                        $type_name::new(self.0)
573                    }
574                )+
575            }
576        }
577    };
578}
579
580impl_object_id_conversion!(
581    DatabaseId,
582    TableId,
583    SchemaId,
584    SinkId,
585    SourceId,
586    JobId,
587    SubscriptionId,
588    IndexId,
589    ViewId,
590    FunctionId,
591    ConnectionId,
592    SecretId
593);
594
595macro_rules! declare_relation {
596    ($($id_name:ident),+) => {
597        $(
598            impl $id_name {
599                pub fn as_relation_id(self) -> RelationId {
600                    RelationId::new(self.0)
601                }
602            }
603        )+
604    };
605}
606
607declare_relation!(TableId, SourceId, SinkId, IndexId, ViewId, SubscriptionId);
608
609macro_rules! impl_hummock_object_id {
610    ($type_name:ty) => {
611        impl $type_name {
612            pub fn as_raw(&self) -> HummockRawObjectId {
613                HummockRawObjectId::new(self.0)
614            }
615        }
616
617        impl From<HummockRawObjectId> for $type_name {
618            fn from(id: HummockRawObjectId) -> Self {
619                Self(id.0)
620            }
621        }
622    };
623}
624
625impl_hummock_object_id!(HummockSstableObjectId);
626impl_hummock_object_id!(HummockVectorFileId);
627impl_hummock_object_id!(HummockHnswGraphFileId);
risingwave_pb/id.rs

risingwave_pb/
id.rs
