risingwave_stream/executor/backfill/cdc/upstream_table/
external.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
15#[cfg(test)]
16use std::collections::VecDeque;
17#[cfg(test)]
18use std::sync::atomic::{AtomicUsize, Ordering};
19#[cfg(test)]
20use std::sync::{Arc, Mutex};
21
22use risingwave_common::catalog::{CdcKeyComparison, Schema, TableId};
23use risingwave_common::util::sort_util::OrderType;
24use risingwave_connector::error::ConnectorResult;
25use risingwave_connector::source::cdc::external::{
26    CdcOffset, ExternalCdcTableType, ExternalTableConfig, ExternalTableReader,
27    ExternalTableReaderImpl, SchemaTableName,
28};
29
30/// This struct represents an external table to be read during backfill
31#[derive(Debug, Clone)]
32pub struct ExternalStorageTable {
33    /// Id for this table.
34    table_id: TableId,
35
36    /// The normalized name of the table, e.g. `dbname.schema_name.table_name`.
37    table_name: String,
38
39    schema_name: String,
40
41    database_name: String,
42
43    config: ExternalTableConfig,
44
45    table_type: ExternalCdcTableType,
46
47    /// The schema of the output columns, i.e., this table VIEWED BY some executor like
48    /// `RowSeqScanExecutor`.
49    /// todo: the schema of the external table defined in the CREATE TABLE DDL
50    schema: Schema,
51
52    pk_order_types: Vec<OrderType>,
53
54    /// Comparison semantics persisted in the stream graph.
55    ///
56    /// `None` is only expected for legacy MySQL graphs with Int64 primary-key columns, whose
57    /// signedness must be recovered from a live external table reader.
58    pk_comparisons: Option<Vec<CdcKeyComparison>>,
59
60    /// Indices of primary key.
61    /// Note that the index is based on the all columns of the table.
62    pk_indices: Vec<usize>,
63
64    #[cfg(test)]
65    mock_snapshot_errors: Option<Arc<Mutex<VecDeque<usize>>>>,
66
67    #[cfg(test)]
68    mock_reader_create_count: Arc<AtomicUsize>,
69}
70
71impl ExternalStorageTable {
72    #[expect(clippy::too_many_arguments)]
73    pub fn new(
74        table_id: TableId,
75        SchemaTableName {
76            table_name,
77            schema_name,
78        }: SchemaTableName,
79        database_name: String,
80        config: ExternalTableConfig,
81        table_type: ExternalCdcTableType,
82        schema: Schema,
83        pk_order_types: Vec<OrderType>,
84        pk_comparisons: Option<Vec<CdcKeyComparison>>,
85        pk_indices: Vec<usize>,
86    ) -> Self {
87        assert_eq!(pk_order_types.len(), pk_indices.len());
88        if let Some(pk_comparisons) = &pk_comparisons {
89            assert_eq!(pk_order_types.len(), pk_comparisons.len());
90        }
91        Self {
92            table_id,
93            table_name,
94            schema_name,
95            database_name,
96            config,
97            table_type,
98            schema,
99            pk_order_types,
100            pk_comparisons,
101            pk_indices,
102            #[cfg(test)]
103            mock_snapshot_errors: None,
104            #[cfg(test)]
105            mock_reader_create_count: Arc::new(AtomicUsize::new(0)),
106        }
107    }
108
109    #[cfg(test)]
110    pub fn for_test_undefined() -> Self {
111        Self {
112            table_id: 1.into(),
113            table_name: "for_test_table_name".into(),
114            schema_name: "for_test_schema_name".into(),
115            database_name: "for_test_database_name".into(),
116            config: ExternalTableConfig::default(),
117            table_type: ExternalCdcTableType::Undefined,
118            schema: Schema::empty().to_owned(),
119            pk_order_types: vec![],
120            pk_comparisons: Some(vec![]),
121            pk_indices: vec![],
122            mock_snapshot_errors: None,
123            mock_reader_create_count: Arc::new(AtomicUsize::new(0)),
124        }
125    }
126
127    #[cfg(test)]
128    pub fn with_mock_snapshot_errors(
129        mut self,
130        snapshot_errors: impl IntoIterator<Item = usize>,
131    ) -> Self {
132        self.mock_snapshot_errors =
133            Some(Arc::new(Mutex::new(snapshot_errors.into_iter().collect())));
134
135        self
136    }
137
138    #[cfg(test)]
139    pub fn mock_reader_create_count(&self) -> usize {
140        self.mock_reader_create_count.load(Ordering::Relaxed)
141    }
142
143    pub fn table_id(&self) -> TableId {
144        self.table_id
145    }
146
147    pub fn pk_order_types(&self) -> &[OrderType] {
148        &self.pk_order_types
149    }
150
151    pub fn pk_comparisons(&self) -> Option<&[CdcKeyComparison]> {
152        self.pk_comparisons.as_deref()
153    }
154
155    pub fn schema(&self) -> &Schema {
156        &self.schema
157    }
158
159    pub fn pk_indices(&self) -> &[usize] {
160        &self.pk_indices
161    }
162
163    pub fn pk_names(&self) -> Vec<String> {
164        self.pk_indices
165            .iter()
166            .map(|&idx| self.schema.fields[idx].name.clone())
167            .collect()
168    }
169
170    pub fn schema_table_name(&self) -> SchemaTableName {
171        SchemaTableName {
172            schema_name: self.schema_name.clone(),
173            table_name: self.table_name.clone(),
174        }
175    }
176
177    pub async fn create_table_reader(&self) -> ConnectorResult<ExternalTableReaderImpl> {
178        #[cfg(test)]
179        if let Some(snapshot_errors) = &self.mock_snapshot_errors {
180            self.mock_reader_create_count
181                .fetch_add(1, Ordering::Relaxed);
182
183            let snapshot_errors = snapshot_errors
184                .lock()
185                .expect("mock snapshot error queue must not be poisoned")
186                .pop_front()
187                .unwrap_or_default();
188
189            return Ok(ExternalTableReaderImpl::Mock(
190                risingwave_connector::source::cdc::external::mock_external_table::MockExternalTableReader::new_with_snapshot_errors(snapshot_errors),
191            ));
192        }
193
194        self.table_type
195            .create_table_reader(
196                self.config.clone(),
197                self.schema.clone(),
198                self.pk_indices.clone(),
199                SchemaTableName {
200                    schema_name: self.schema_name.clone(),
201                    table_name: self.table_name.clone(),
202                },
203                self.table_id.as_raw_id(),
204            )
205            .await
206    }
207
208    pub fn qualified_table_name(&self) -> String {
209        format!("{}.{}", self.schema_name, self.table_name)
210    }
211
212    pub fn database_name(&self) -> &str {
213        self.database_name.as_str()
214    }
215
216    pub fn table_type(&self) -> &ExternalCdcTableType {
217        &self.table_type
218    }
219
220    pub async fn current_cdc_offset(
221        &self,
222        table_reader: &ExternalTableReaderImpl,
223    ) -> ConnectorResult<Option<CdcOffset>> {
224        let binlog = table_reader.current_cdc_offset().await?;
225        Ok(Some(binlog))
226    }
227}
risingwave_stream/executor/backfill/cdc/upstream_table/external.rs

risingwave_stream/executor/backfill/cdc/upstream_table/
external.rs
