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    /// Restores a catalog-only CDC table source descriptor, whose key ordering is resolved at creation.
85    pub fn from_protobuf(desc: &ExternalTableDesc) -> anyhow::Result<Self> {
86        let ordering = desc.get_pk_ordering()?;
87        anyhow::ensure!(
88            desc.pk.len() == ordering.columns.len(),
89            "CDC key ordering length mismatch"
90        );
91        let pk_comparisons = desc
92            .pk
93            .iter()
94            .zip_eq_fast(&ordering.columns)
95            .map(|(pk, column)| {
96                anyhow::ensure!(
97                    pk.column_index == column.pk_col_idx,
98                    "CDC key ordering index mismatch"
99                );
100                let comparison = column.get_comparison()?;
101                anyhow::ensure!(
102                    comparison != Comparison::Unspecified,
103                    "CDC key comparison is unspecified"
104                );
105                Ok(CdcKeyComparison::from_protobuf(comparison))
106            })
107            .collect::<anyhow::Result<Vec<_>>>()?;
108        Ok(Self {
109            table_id: desc.table_id,
110            source_id: desc.source_id,
111            external_table_name: desc.table_name.clone(),
112            pk: desc.pk.iter().map(ColumnOrder::from_protobuf).collect(),
113            pk_comparisons,
114            columns: desc.columns.iter().map(ColumnDesc::from).collect(),
115            stream_key: desc
116                .stream_key
117                .iter()
118                .map(|index| *index as usize)
119                .collect(),
120            connect_properties: desc.connect_properties.clone(),
121            secret_refs: desc.secret_refs.clone(),
122        })
123    }
124
125    pub fn to_protobuf(&self) -> ExternalTableDesc {
126        assert_eq!(self.pk.len(), self.pk_comparisons.len());
127        ExternalTableDesc {
128            table_id: self.table_id,
129            source_id: self.source_id,
130            columns: self.columns.iter().map(Into::into).collect(),
131            pk: self.pk.iter().map(|column| column.to_protobuf()).collect(),
132            pk_ordering: Some(PbCdcKeyOrdering {
133                columns: self
134                    .pk
135                    .iter()
136                    .zip_eq_fast(&self.pk_comparisons)
137                    .map(|(column_order, comparison)| PbCdcKeyColumn {
138                        pk_col_idx: column_order.column_index as _,
139                        comparison: comparison.to_protobuf() as _,
140                    })
141                    .collect(),
142            }),
143            table_name: self.external_table_name.clone(),
144            stream_key: self.stream_key.iter().map(|k| *k as _).collect(),
145            connect_properties: self.connect_properties.clone(),
146            secret_refs: self.secret_refs.clone(),
147        }
148    }
149
150    /// Helper function to create a mapping from `column id` to `column index`
151    pub fn get_id_to_op_idx_mapping(&self) -> HashMap<ColumnId, usize> {
152        ColumnDesc::get_id_to_op_idx_mapping(self.columns.as_slice(), None)
153    }
154}
155
156#[cfg(test)]
157mod tests {
158    use risingwave_pb::plan_common::cdc_key_ordering::Comparison;
159
160    use super::*;
161    use crate::util::sort_util::OrderType;
162
163    #[test]
164    fn test_cdc_key_comparisons_are_persisted_with_pk_indices() {
165        let table_desc = CdcTableDesc {
166            table_id: TableId::new(1),
167            source_id: SourceId::new(2),
168            external_table_name: "orders".to_owned(),
169            pk: vec![
170                ColumnOrder::new(3, OrderType::ascending()),
171                ColumnOrder::new(1, OrderType::ascending()),
172            ],
173            pk_comparisons: vec![CdcKeyComparison::UnsignedInt64, CdcKeyComparison::Native],
174            columns: vec![],
175            stream_key: vec![3, 1],
176            connect_properties: BTreeMap::new(),
177            secret_refs: BTreeMap::new(),
178        };
179
180        let protobuf = table_desc.to_protobuf();
181        assert_eq!(CdcTableDesc::from_protobuf(&protobuf).unwrap(), table_desc);
182        assert_eq!(protobuf.pk.len(), 2);
183        assert_eq!(protobuf.pk[0].column_index, 3);
184        assert_eq!(protobuf.pk[1].column_index, 1);
185
186        let pk_columns = protobuf.pk_ordering.unwrap().columns;
187        assert_eq!(pk_columns.len(), 2);
188        assert_eq!(pk_columns[0].pk_col_idx, 3);
189        assert_eq!(
190            pk_columns[0].get_comparison().unwrap(),
191            Comparison::UnsignedInt64
192        );
193        assert_eq!(pk_columns[1].pk_col_idx, 1);
194        assert_eq!(pk_columns[1].get_comparison().unwrap(), Comparison::Native);
195    }
196}
risingwave_common/catalog/external_table.rs

risingwave_common/catalog/
external_table.rs
