risingwave_common/catalog/
external_table.rs

1// Copyright 2023 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::{BTreeMap, HashMap};
16
17use risingwave_pb::plan_common::cdc_key_ordering::{Column as PbCdcKeyColumn, Comparison};
18use risingwave_pb::plan_common::{CdcKeyOrdering as PbCdcKeyOrdering, ExternalTableDesc};
19use risingwave_pb::secret::PbSecretRef;
20
21use super::{ColumnDesc, ColumnId, TableId};
22use crate::id::SourceId;
23use crate::util::iter_util::ZipEqFast;
24use crate::util::sort_util::ColumnOrder;
25
26/// A resolved comparison rule for a CDC primary-key column.
27///
28/// There is no `Unspecified` variant: `ExternalStorageTable` represents unresolved legacy
29/// comparison metadata with `None` in its `Option<Vec<CdcKeyComparison>>` instead.
30#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
31pub enum CdcKeyComparison {
32    #[default]
33    Native,
34    UnsignedInt64,
35}
36
37impl CdcKeyComparison {
38    pub fn from_protobuf(comparison: Comparison) -> Self {
39        match comparison {
40            Comparison::Unspecified => unreachable!("comparison must be specified"),
41            Comparison::Native => Self::Native,
42            Comparison::UnsignedInt64 => Self::UnsignedInt64,
43        }
44    }
45
46    fn to_protobuf(self) -> Comparison {
47        match self {
48            Self::Native => Comparison::Native,
49            Self::UnsignedInt64 => Comparison::UnsignedInt64,
50        }
51    }
52}
53
54/// Necessary information for compute node to access data in the external database.
55/// Compute node will use this information to connect to the external database and scan the table.
56#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
57pub struct CdcTableDesc {
58    /// Id of the table in RW
59    pub table_id: TableId,
60
61    /// Id of the upstream source in sharing cdc mode
62    pub source_id: SourceId,
63
64    /// The full name of the table in external database, e.g. `database_name.table_name` in MySQL
65    /// and `schema_name.table_name` in the Postgres.
66    pub external_table_name: String,
67    /// The key used to sort in storage.
68    pub pk: Vec<ColumnOrder>,
69    /// Comparison semantics for each primary-key column.
70    pub pk_comparisons: Vec<CdcKeyComparison>,
71    /// All columns in the table, noticed it is NOT sorted by columnId in the vec.
72    pub columns: Vec<ColumnDesc>,
73
74    /// Column indices for primary keys.
75    pub stream_key: Vec<usize>,
76
77    /// properties will be passed into the `StreamScanNode`
78    pub connect_properties: BTreeMap<String, String>,
79    /// Secret refs
80    pub secret_refs: BTreeMap<String, PbSecretRef>,
81}
82
83impl CdcTableDesc {
84    pub fn to_protobuf(&self) -> ExternalTableDesc {
85        assert_eq!(self.pk.len(), self.pk_comparisons.len());
86        ExternalTableDesc {
87            table_id: self.table_id,
88            source_id: self.source_id,
89            columns: self.columns.iter().map(Into::into).collect(),
90            pk: self.pk.iter().map(|column| column.to_protobuf()).collect(),
91            pk_ordering: Some(PbCdcKeyOrdering {
92                columns: self
93                    .pk
94                    .iter()
95                    .zip_eq_fast(&self.pk_comparisons)
96                    .map(|(column_order, comparison)| PbCdcKeyColumn {
97                        pk_col_idx: column_order.column_index as _,
98                        comparison: comparison.to_protobuf() as _,
99                    })
100                    .collect(),
101            }),
102            table_name: self.external_table_name.clone(),
103            stream_key: self.stream_key.iter().map(|k| *k as _).collect(),
104            connect_properties: self.connect_properties.clone(),
105            secret_refs: self.secret_refs.clone(),
106        }
107    }
108
109    /// Helper function to create a mapping from `column id` to `column index`
110    pub fn get_id_to_op_idx_mapping(&self) -> HashMap<ColumnId, usize> {
111        ColumnDesc::get_id_to_op_idx_mapping(self.columns.as_slice(), None)
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use risingwave_pb::plan_common::cdc_key_ordering::Comparison;
118
119    use super::*;
120    use crate::util::sort_util::OrderType;
121
122    #[test]
123    fn test_cdc_key_comparisons_are_persisted_with_pk_indices() {
124        let table_desc = CdcTableDesc {
125            table_id: TableId::new(1),
126            source_id: SourceId::new(2),
127            external_table_name: "orders".to_owned(),
128            pk: vec![
129                ColumnOrder::new(3, OrderType::ascending()),
130                ColumnOrder::new(1, OrderType::ascending()),
131            ],
132            pk_comparisons: vec![CdcKeyComparison::UnsignedInt64, CdcKeyComparison::Native],
133            columns: vec![],
134            stream_key: vec![3, 1],
135            connect_properties: BTreeMap::new(),
136            secret_refs: BTreeMap::new(),
137        };
138
139        let protobuf = table_desc.to_protobuf();
140        assert_eq!(protobuf.pk.len(), 2);
141        assert_eq!(protobuf.pk[0].column_index, 3);
142        assert_eq!(protobuf.pk[1].column_index, 1);
143
144        let pk_columns = protobuf.pk_ordering.unwrap().columns;
145        assert_eq!(pk_columns.len(), 2);
146        assert_eq!(pk_columns[0].pk_col_idx, 3);
147        assert_eq!(
148            pk_columns[0].get_comparison().unwrap(),
149            Comparison::UnsignedInt64
150        );
151        assert_eq!(pk_columns[1].pk_col_idx, 1);
152        assert_eq!(pk_columns[1].get_comparison().unwrap(), Comparison::Native);
153    }
154}
risingwave_common/catalog/external_table.rs

risingwave_common/catalog/
external_table.rs
