risingwave_meta_model/
table.rs

1// Copyright 2024 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 risingwave_common::catalog::OBJECT_ID_PLACEHOLDER;
16use risingwave_common::hash::VnodeCountCompat;
17use risingwave_common::id::JobId;
18use risingwave_pb::catalog::table::{CdcTableType as PbCdcTableType, PbEngine, PbTableType};
19use risingwave_pb::catalog::{PbHandleConflictBehavior, PbTable, PbVectorIndexInfo};
20use sea_orm::ActiveValue::Set;
21use sea_orm::NotSet;
22use sea_orm::entity::prelude::*;
23use serde::{Deserialize, Serialize};
24
25use crate::{
26    Cardinality, ColumnCatalogArray, ColumnOrderArray, FragmentId, I32Array, SourceId, TableId,
27    TableVersion, WebhookSourceInfo,
28};
29
30#[derive(
31    Clone, Debug, PartialEq, Hash, Copy, Eq, EnumIter, DeriveActiveEnum, Serialize, Deserialize,
32)]
33#[sea_orm(rs_type = "String", db_type = "string(None)")]
34pub enum TableType {
35    #[sea_orm(string_value = "TABLE")]
36    Table,
37    #[sea_orm(string_value = "MATERIALIZED_VIEW")]
38    MaterializedView,
39    #[sea_orm(string_value = "INDEX")]
40    Index,
41    #[sea_orm(string_value = "INTERNAL")]
42    Internal,
43    #[sea_orm(string_value = "VECTOR_INDEX")]
44    VectorIndex,
45}
46
47impl From<TableType> for PbTableType {
48    fn from(table_type: TableType) -> Self {
49        match table_type {
50            TableType::Table => Self::Table,
51            TableType::MaterializedView => Self::MaterializedView,
52            TableType::Index => Self::Index,
53            TableType::Internal => Self::Internal,
54            TableType::VectorIndex => Self::VectorIndex,
55        }
56    }
57}
58
59impl From<PbTableType> for TableType {
60    fn from(table_type: PbTableType) -> Self {
61        match table_type {
62            PbTableType::Table => Self::Table,
63            PbTableType::MaterializedView => Self::MaterializedView,
64            PbTableType::Index => Self::Index,
65            PbTableType::Internal => Self::Internal,
66            PbTableType::VectorIndex => Self::VectorIndex,
67            PbTableType::Unspecified => unreachable!("Unspecified table type"),
68        }
69    }
70}
71
72#[derive(Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
73#[sea_orm(rs_type = "String", db_type = "string(None)")]
74pub enum HandleConflictBehavior {
75    #[sea_orm(string_value = "OVERWRITE")]
76    Overwrite,
77    #[sea_orm(string_value = "IGNORE")]
78    Ignore,
79    #[sea_orm(string_value = "NO_CHECK")]
80    NoCheck,
81    #[sea_orm(string_value = "DO_UPDATE_IF_NOT_NULL")]
82    DoUpdateIfNotNull,
83}
84
85impl From<HandleConflictBehavior> for PbHandleConflictBehavior {
86    fn from(handle_conflict_behavior: HandleConflictBehavior) -> Self {
87        match handle_conflict_behavior {
88            HandleConflictBehavior::Overwrite => Self::Overwrite,
89            HandleConflictBehavior::Ignore => Self::Ignore,
90            HandleConflictBehavior::NoCheck => Self::NoCheck,
91            HandleConflictBehavior::DoUpdateIfNotNull => Self::DoUpdateIfNotNull,
92        }
93    }
94}
95
96impl From<PbHandleConflictBehavior> for HandleConflictBehavior {
97    fn from(handle_conflict_behavior: PbHandleConflictBehavior) -> Self {
98        match handle_conflict_behavior {
99            PbHandleConflictBehavior::Overwrite => Self::Overwrite,
100            PbHandleConflictBehavior::Ignore => Self::Ignore,
101            PbHandleConflictBehavior::NoCheck => Self::NoCheck,
102            PbHandleConflictBehavior::DoUpdateIfNotNull => Self::DoUpdateIfNotNull,
103            PbHandleConflictBehavior::Unspecified => {
104                unreachable!("Unspecified handle conflict behavior")
105            }
106        }
107    }
108}
109
110#[derive(Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
111#[sea_orm(rs_type = "String", db_type = "string(None)")]
112pub enum Engine {
113    #[sea_orm(string_value = "HUMMOCK")]
114    Hummock,
115    #[sea_orm(string_value = "ICEBERG")]
116    Iceberg,
117}
118
119impl From<Engine> for PbEngine {
120    fn from(engine: Engine) -> Self {
121        match engine {
122            Engine::Hummock => Self::Hummock,
123            Engine::Iceberg => Self::Iceberg,
124        }
125    }
126}
127
128impl From<PbEngine> for Engine {
129    fn from(engine: PbEngine) -> Self {
130        match engine {
131            PbEngine::Hummock => Self::Hummock,
132            PbEngine::Iceberg => Self::Iceberg,
133            PbEngine::Unspecified => Self::Hummock,
134        }
135    }
136}
137
138#[derive(Clone, Debug, PartialEq, Eq, Hash, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
139#[sea_orm(rs_type = "String", db_type = "string(None)")]
140pub enum CdcTableType {
141    #[sea_orm(string_value = "UNSPECIFIED")]
142    Unspecified,
143    #[sea_orm(string_value = "POSTGRES")]
144    Postgres,
145    #[sea_orm(string_value = "MYSQL")]
146    Mysql,
147    #[sea_orm(string_value = "MARIADB")]
148    Mariadb,
149    #[sea_orm(string_value = "SQLSERVER")]
150    #[serde(rename = "Sqlserver")] // keep the serde name used in meta backup snapshots
151    SqlServer,
152    #[sea_orm(string_value = "MONGO")]
153    #[serde(rename = "Mongo")] // keep the serde name used in meta backup snapshots
154    Mongodb,
155}
156
157impl From<CdcTableType> for PbCdcTableType {
158    fn from(cdc_table_type: CdcTableType) -> Self {
159        match cdc_table_type {
160            CdcTableType::Postgres => Self::Postgres,
161            CdcTableType::Mysql => Self::Mysql,
162            CdcTableType::Mariadb => Self::Mariadb,
163            CdcTableType::SqlServer => Self::SqlServer,
164            CdcTableType::Mongodb => Self::Mongodb,
165            CdcTableType::Unspecified => Self::Unspecified,
166        }
167    }
168}
169
170impl From<PbCdcTableType> for CdcTableType {
171    fn from(cdc_table_type: PbCdcTableType) -> Self {
172        match cdc_table_type {
173            PbCdcTableType::Postgres => Self::Postgres,
174            PbCdcTableType::Mysql => Self::Mysql,
175            PbCdcTableType::Mariadb => Self::Mariadb,
176            PbCdcTableType::SqlServer => Self::SqlServer,
177            PbCdcTableType::Mongodb => Self::Mongodb,
178            PbCdcTableType::Unspecified => Self::Unspecified,
179        }
180    }
181}
182
183crate::derive_from_blob!(VectorIndexInfo, PbVectorIndexInfo);
184
185#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
186#[sea_orm(table_name = "table")]
187pub struct Model {
188    #[sea_orm(primary_key, auto_increment = false)]
189    pub table_id: TableId,
190    pub name: String,
191    pub optional_associated_source_id: Option<SourceId>,
192    pub table_type: TableType,
193    pub belongs_to_job_id: Option<JobId>,
194    pub columns: ColumnCatalogArray,
195    pub pk: ColumnOrderArray,
196    pub distribution_key: I32Array,
197    pub stream_key: I32Array,
198    pub append_only: bool,
199    pub fragment_id: Option<FragmentId>,
200    pub vnode_col_index: Option<i32>,
201    pub row_id_index: Option<i32>,
202    pub value_indices: I32Array,
203    pub definition: String,
204    pub handle_pk_conflict_behavior: HandleConflictBehavior,
205    pub version_column_indices: Option<I32Array>,
206    pub read_prefix_len_hint: i32,
207    pub watermark_indices: I32Array,
208    pub dist_key_in_pk: I32Array,
209    pub dml_fragment_id: Option<FragmentId>,
210    pub cardinality: Option<Cardinality>,
211    pub cleaned_by_watermark: bool,
212    pub description: Option<String>,
213    pub version: Option<TableVersion>,
214    pub retention_seconds: Option<i32>,
215    pub cdc_table_id: Option<String>,
216    pub vnode_count: i32,
217    pub webhook_info: Option<WebhookSourceInfo>,
218    pub engine: Option<Engine>,
219    pub clean_watermark_index_in_pk: Option<i32>,
220    pub clean_watermark_indices: Option<I32Array>,
221    pub refreshable: bool,
222    pub vector_index_info: Option<VectorIndexInfo>,
223    pub cdc_table_type: Option<CdcTableType>,
224}
225
226#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
227pub enum Relation {
228    #[sea_orm(
229        belongs_to = "super::fragment::Entity",
230        from = "Column::DmlFragmentId",
231        to = "super::fragment::Column::FragmentId",
232        on_update = "NoAction",
233        on_delete = "NoAction"
234    )]
235    Fragment2,
236    #[sea_orm(
237        belongs_to = "super::fragment::Entity",
238        from = "Column::FragmentId",
239        to = "super::fragment::Column::FragmentId",
240        on_update = "NoAction",
241        on_delete = "NoAction"
242    )]
243    Fragment1,
244    #[sea_orm(
245        belongs_to = "super::object::Entity",
246        from = "Column::BelongsToJobId",
247        to = "super::object::Column::Oid",
248        on_update = "NoAction",
249        on_delete = "Cascade"
250    )]
251    Object2,
252    #[sea_orm(
253        belongs_to = "super::object::Entity",
254        from = "Column::TableId",
255        to = "super::object::Column::Oid",
256        on_update = "NoAction",
257        on_delete = "Cascade"
258    )]
259    Object1,
260    #[sea_orm(
261        belongs_to = "super::source::Entity",
262        from = "Column::OptionalAssociatedSourceId",
263        to = "super::source::Column::SourceId",
264        on_update = "NoAction",
265        on_delete = "NoAction"
266    )]
267    Source,
268
269    // To join object_dependency on the used_by column
270    #[sea_orm(
271        belongs_to = "super::object_dependency::Entity",
272        from = "Column::TableId",
273        to = "super::object_dependency::Column::UsedBy",
274        on_update = "NoAction",
275        on_delete = "Cascade"
276    )]
277    ObjectDependency,
278}
279
280impl Related<super::object::Entity> for Entity {
281    fn to() -> RelationDef {
282        Relation::Object1.def()
283    }
284}
285
286impl Related<super::source::Entity> for Entity {
287    fn to() -> RelationDef {
288        Relation::Source.def()
289    }
290}
291
292impl ActiveModelBehavior for ActiveModel {}
293
294impl Model {
295    pub fn job_id(&self) -> JobId {
296        self.belongs_to_job_id
297            .unwrap_or_else(|| self.table_id.as_job_id())
298    }
299}
300
301impl From<PbTable> for ActiveModel {
302    fn from(pb_table: PbTable) -> Self {
303        let table_type = pb_table.table_type();
304        let handle_pk_conflict_behavior = pb_table.handle_pk_conflict_behavior();
305
306        // `PbTable` here should be sourced from the wire, not from persistence.
307        // A placeholder `maybe_vnode_count` field should be treated as `NotSet`, instead of calling
308        // the compatibility code.
309        let vnode_count = pb_table
310            .vnode_count_inner()
311            .value_opt()
312            .map(|v| v as _)
313            .map_or(NotSet, Set);
314        let fragment_id = if pb_table.fragment_id == OBJECT_ID_PLACEHOLDER {
315            NotSet
316        } else {
317            Set(Some(pb_table.fragment_id))
318        };
319        let dml_fragment_id = pb_table
320            .dml_fragment_id
321            .map(|x| Set(Some(x)))
322            .unwrap_or_default();
323        let optional_associated_source_id =
324            if let Some(src_id) = pb_table.optional_associated_source_id {
325                Set(Some(src_id.into()))
326            } else {
327                NotSet
328            };
329
330        Self {
331            table_id: Set(pb_table.id),
332            name: Set(pb_table.name),
333            optional_associated_source_id,
334            table_type: Set(table_type.into()),
335            belongs_to_job_id: Set(pb_table.job_id),
336            columns: Set(pb_table.columns.into()),
337            pk: Set(pb_table.pk.into()),
338            distribution_key: Set(pb_table.distribution_key.into()),
339            stream_key: Set(pb_table.stream_key.into()),
340            append_only: Set(pb_table.append_only),
341            fragment_id,
342            vnode_col_index: Set(pb_table.vnode_col_index.map(|x| x as i32)),
343            row_id_index: Set(pb_table.row_id_index.map(|x| x as i32)),
344            value_indices: Set(pb_table.value_indices.into()),
345            definition: Set(pb_table.definition),
346            handle_pk_conflict_behavior: Set(handle_pk_conflict_behavior.into()),
347            version_column_indices: Set(Some(
348                pb_table
349                    .version_column_indices
350                    .iter()
351                    .map(|x| *x as i32)
352                    .collect::<Vec<_>>()
353                    .into(),
354            )),
355            read_prefix_len_hint: Set(pb_table.read_prefix_len_hint as _),
356            watermark_indices: Set(pb_table.watermark_indices.into()),
357            dist_key_in_pk: Set(pb_table.dist_key_in_pk.into()),
358            dml_fragment_id,
359            cardinality: Set(pb_table.cardinality.as_ref().map(|x| x.into())),
360            #[expect(deprecated)]
361            cleaned_by_watermark: Set(pb_table.cleaned_by_watermark),
362            description: Set(pb_table.description),
363            version: Set(pb_table.version.as_ref().map(|v| v.into())),
364            retention_seconds: Set(pb_table.retention_seconds.map(|i| i as _)),
365            cdc_table_id: Set(pb_table.cdc_table_id),
366            vnode_count,
367            webhook_info: Set(pb_table.webhook_info.as_ref().map(WebhookSourceInfo::from)),
368            engine: Set(pb_table
369                .engine
370                .map(|engine| Engine::from(PbEngine::try_from(engine).expect("Invalid engine")))),
371            #[expect(deprecated)]
372            clean_watermark_index_in_pk: Set(pb_table.clean_watermark_index_in_pk),
373            clean_watermark_indices: Set(if pb_table.clean_watermark_indices.is_empty() {
374                None
375            } else {
376                Some(
377                    pb_table
378                        .clean_watermark_indices
379                        .iter()
380                        .map(|x| *x as i32)
381                        .collect::<Vec<_>>()
382                        .into(),
383                )
384            }),
385            refreshable: Set(pb_table.refreshable),
386            vector_index_info: Set(pb_table
387                .vector_index_info
388                .as_ref()
389                .map(VectorIndexInfo::from)),
390            cdc_table_type: Set(pb_table.cdc_table_type.map(|cdc_table_type| {
391                match cdc_table_type {
392                    0 => CdcTableType::Postgres, // Map Unspecified to Postgres as default
393                    1 => CdcTableType::Postgres,
394                    2 => CdcTableType::Mysql,
395                    3 => CdcTableType::SqlServer,
396                    4 => CdcTableType::Mongodb,
397                    6 => CdcTableType::Mariadb,
398                    _ => panic!("Invalid CDC table type: {cdc_table_type}"),
399                }
400            })),
401        }
402    }
403}
404
405#[cfg(test)]
406mod tests {
407    use super::*;
408
409    #[test]
410    fn test_cdc_table_type_persisted_names() {
411        for (cdc_table_type, sea_orm_value, serde_name) in [
412            (CdcTableType::Unspecified, "UNSPECIFIED", "Unspecified"),
413            (CdcTableType::Postgres, "POSTGRES", "Postgres"),
414            (CdcTableType::Mysql, "MYSQL", "Mysql"),
415            (CdcTableType::Mariadb, "MARIADB", "Mariadb"),
416            (CdcTableType::SqlServer, "SQLSERVER", "Sqlserver"),
417            (CdcTableType::Mongodb, "MONGO", "Mongo"),
418        ] {
419            assert_eq!(cdc_table_type.to_value(), sea_orm_value);
420            assert_eq!(
421                serde_json::to_string(&cdc_table_type).unwrap(),
422                format!("\"{serde_name}\"")
423            );
424        }
425    }
426}
risingwave_meta_model/table.rs

risingwave_meta_model/
table.rs
