risingwave_connector/source/cdc/
split.rs

1// Copyright 2022 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::marker::PhantomData;
16
17use anyhow::Context;
18use risingwave_common::types::JsonbVal;
19use serde::{Deserialize, Serialize};
20
21use crate::error::ConnectorResult;
22use crate::source::cdc::external::DebeziumOffset;
23use crate::source::cdc::{CdcSourceType, CdcSourceTypeTrait, Mysql, Postgres, SqlServer};
24use crate::source::{SplitId, SplitMetaData};
25
26/// The base states of a CDC split, which will be persisted to checkpoint.
27/// CDC source only has single split, so we use the `source_id` to identify the split.
28#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
29pub struct CdcSplitBase {
30    pub split_id: u32,
31    pub start_offset: Option<String>,
32    pub snapshot_done: bool,
33}
34
35impl CdcSplitBase {
36    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
37        Self {
38            split_id,
39            start_offset,
40            snapshot_done: false,
41        }
42    }
43}
44
45trait CdcSplitTrait: Send + Sync {
46    fn split_id(&self) -> u32;
47    fn start_offset(&self) -> &Option<String>;
48    fn is_snapshot_done(&self) -> bool;
49    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()>;
50
51    // MySQL and MongoDB shares the same logic to extract the snapshot flag
52    fn extract_snapshot_flag(&self, start_offset: &str) -> ConnectorResult<bool> {
53        // if snapshot_done is already true, it won't be changed
54        let mut snapshot_done = self.is_snapshot_done();
55        if snapshot_done {
56            return Ok(snapshot_done);
57        }
58
59        let dbz_offset: DebeziumOffset = serde_json::from_str(start_offset).with_context(|| {
60            format!(
61                "invalid cdc offset: {}, split: {}",
62                start_offset,
63                self.split_id()
64            )
65        })?;
66
67        // heartbeat event should not update the `snapshot_done` flag
68        if !dbz_offset.is_heartbeat {
69            snapshot_done = match dbz_offset.source_offset.snapshot {
70                Some(val) => !val,
71                None => true,
72            };
73        }
74        Ok(snapshot_done)
75    }
76}
77
78#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
79pub struct MySqlCdcSplit {
80    pub inner: CdcSplitBase,
81}
82
83#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
84pub struct PostgresCdcSplit {
85    pub inner: CdcSplitBase,
86    // the hostname and port of a node that holding shard tables (for Citus)
87    pub server_addr: Option<String>,
88}
89
90#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
91pub struct MongoDbCdcSplit {
92    pub inner: CdcSplitBase,
93}
94
95#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
96pub struct SqlServerCdcSplit {
97    pub inner: CdcSplitBase,
98}
99
100impl MySqlCdcSplit {
101    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
102        let split = CdcSplitBase {
103            split_id,
104            start_offset,
105            snapshot_done: false,
106        };
107        Self { inner: split }
108    }
109
110    /// Extract MySQL CDC binlog offset (file sequence and position) from the offset JSON string.
111    ///
112    /// MySQL binlog offset format:
113    /// ```json
114    /// {
115    ///   "sourcePartition": { "server": "..." },
116    ///   "sourceOffset": {
117    ///     "file": "binlog.000123",
118    ///     "pos": 456789,
119    ///     ...
120    ///   }
121    /// }
122    /// ```
123    ///
124    /// Returns `Some((file_seq, position))` where:
125    /// - `file_seq`: the numeric part of binlog filename (e.g., 123 from "binlog.000123")
126    /// - `position`: the byte offset within the binlog file
127    pub fn mysql_binlog_offset(&self) -> Option<(u64, u64)> {
128        let offset_str = self.inner.start_offset.as_ref()?;
129        let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
130        let source_offset = offset.get("sourceOffset")?;
131
132        let file = source_offset.get("file")?.as_str()?;
133        let pos = source_offset.get("pos")?.as_u64()?;
134
135        let file_seq = extract_binlog_file_seq(file)?;
136
137        Some((file_seq, pos))
138    }
139}
140
141impl CdcSplitTrait for MySqlCdcSplit {
142    fn split_id(&self) -> u32 {
143        self.inner.split_id
144    }
145
146    fn start_offset(&self) -> &Option<String> {
147        &self.inner.start_offset
148    }
149
150    fn is_snapshot_done(&self) -> bool {
151        self.inner.snapshot_done
152    }
153
154    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
155        // if snapshot_done is already true, it won't be updated
156        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
157        self.inner.start_offset = Some(last_seen_offset);
158        Ok(())
159    }
160}
161
162impl PostgresCdcSplit {
163    pub fn new(split_id: u32, start_offset: Option<String>, server_addr: Option<String>) -> Self {
164        let split = CdcSplitBase {
165            split_id,
166            start_offset,
167            snapshot_done: false,
168        };
169        Self {
170            inner: split,
171            server_addr,
172        }
173    }
174
175    /// Extract PostgreSQL LSN value from the offset JSON string.
176    ///
177    /// This function parses the offset JSON and extracts the LSN value from the sourceOffset.lsn field.
178    /// Returns Some(lsn) if the LSN is found and can be parsed as u64, None otherwise.
179    pub fn pg_lsn(&self) -> Option<u64> {
180        let offset_str = self.inner.start_offset.as_ref()?;
181        let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
182        let source_offset = offset.get("sourceOffset")?;
183        let lsn = source_offset.get("lsn")?;
184        lsn.as_u64()
185    }
186}
187
188impl CdcSplitTrait for PostgresCdcSplit {
189    fn split_id(&self) -> u32 {
190        self.inner.split_id
191    }
192
193    fn start_offset(&self) -> &Option<String> {
194        &self.inner.start_offset
195    }
196
197    fn is_snapshot_done(&self) -> bool {
198        self.inner.snapshot_done
199    }
200
201    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
202        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
203        self.inner.start_offset = Some(last_seen_offset);
204        Ok(())
205    }
206
207    fn extract_snapshot_flag(&self, start_offset: &str) -> ConnectorResult<bool> {
208        // if snapshot_done is already true, it won't be changed
209        let mut snapshot_done = self.is_snapshot_done();
210        if snapshot_done {
211            return Ok(snapshot_done);
212        }
213
214        let dbz_offset: DebeziumOffset = serde_json::from_str(start_offset).with_context(|| {
215            format!(
216                "invalid postgres offset: {}, split: {}",
217                start_offset, self.inner.split_id
218            )
219        })?;
220
221        // heartbeat event should not update the `snapshot_done` flag
222        if !dbz_offset.is_heartbeat {
223            snapshot_done = dbz_offset
224                .source_offset
225                .last_snapshot_record
226                .unwrap_or(false);
227        }
228        Ok(snapshot_done)
229    }
230}
231
232impl MongoDbCdcSplit {
233    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
234        let split = CdcSplitBase {
235            split_id,
236            start_offset,
237            snapshot_done: false,
238        };
239        Self { inner: split }
240    }
241}
242
243impl CdcSplitTrait for MongoDbCdcSplit {
244    fn split_id(&self) -> u32 {
245        self.inner.split_id
246    }
247
248    fn start_offset(&self) -> &Option<String> {
249        &self.inner.start_offset
250    }
251
252    fn is_snapshot_done(&self) -> bool {
253        self.inner.snapshot_done
254    }
255
256    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
257        // if snapshot_done is already true, it will remain true
258        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
259        self.inner.start_offset = Some(last_seen_offset);
260        Ok(())
261    }
262}
263
264impl SqlServerCdcSplit {
265    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
266        let split = CdcSplitBase {
267            split_id,
268            start_offset,
269            snapshot_done: false,
270        };
271        Self { inner: split }
272    }
273
274    /// Extract SQL Server `change_lsn` value from the offset JSON string.
275    pub fn sql_server_change_lsn(&self) -> Option<u128> {
276        let offset_str = self.inner.start_offset.as_ref()?;
277        extract_sql_server_change_lsn_from_offset_str(offset_str)
278    }
279
280    /// Extract SQL Server `commit_lsn` value from the offset JSON string.
281    pub fn sql_server_commit_lsn(&self) -> Option<u128> {
282        let offset_str = self.inner.start_offset.as_ref()?;
283        extract_sql_server_commit_lsn_from_offset_str(offset_str)
284    }
285}
286
287impl CdcSplitTrait for SqlServerCdcSplit {
288    fn split_id(&self) -> u32 {
289        self.inner.split_id
290    }
291
292    fn start_offset(&self) -> &Option<String> {
293        &self.inner.start_offset
294    }
295
296    fn is_snapshot_done(&self) -> bool {
297        self.inner.snapshot_done
298    }
299
300    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
301        // if snapshot_done is already true, it will remain true
302        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
303        self.inner.start_offset = Some(last_seen_offset);
304        Ok(())
305    }
306}
307
308/// We use this struct to wrap the specific split, which act as an interface to other modules
309#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
310pub struct DebeziumCdcSplit<T: CdcSourceTypeTrait> {
311    pub mysql_split: Option<MySqlCdcSplit>,
312
313    #[serde(rename = "pg_split")] // backward compatibility
314    pub postgres_split: Option<PostgresCdcSplit>,
315    pub citus_split: Option<PostgresCdcSplit>,
316    pub mongodb_split: Option<MongoDbCdcSplit>,
317    pub sql_server_split: Option<SqlServerCdcSplit>,
318
319    #[serde(skip)]
320    pub _phantom: PhantomData<T>,
321}
322
323macro_rules! dispatch_cdc_split_inner {
324    ($dbz_split:expr, $as_type:tt, {$({$cdc_source_type:tt, $cdc_source_split:tt}),*}, $body:expr) => {
325        match T::source_type() {
326            $(
327                CdcSourceType::$cdc_source_type => {
328                    $crate::paste! {
329                        $dbz_split.[<$cdc_source_split>]
330                            .[<as_ $as_type>]()
331                            .expect(concat!(stringify!([<$cdc_source_type:lower>]), " split must exist"))
332                            .$body
333                    }
334                }
335            )*
336            CdcSourceType::Unspecified => {
337                unreachable!("invalid debezium split");
338            }
339        }
340    }
341}
342
343// call corresponding split method of the specific cdc source type
344macro_rules! dispatch_cdc_split {
345    ($dbz_split:expr, $as_type:tt, $body:expr) => {
346        dispatch_cdc_split_inner!($dbz_split, $as_type, {
347            {Mysql, mysql_split},
348            {Postgres, postgres_split},
349            {Citus, citus_split},
350            {Mongodb, mongodb_split},
351            {SqlServer, sql_server_split}
352        }, $body)
353    }
354}
355
356impl<T: CdcSourceTypeTrait> SplitMetaData for DebeziumCdcSplit<T> {
357    fn id(&self) -> SplitId {
358        format!("{}", self.split_id()).into()
359    }
360
361    fn encode_to_json(&self) -> JsonbVal {
362        serde_json::to_value(self.clone()).unwrap().into()
363    }
364
365    fn restore_from_json(value: JsonbVal) -> ConnectorResult<Self> {
366        serde_json::from_value(value.take()).map_err(Into::into)
367    }
368
369    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
370        self.update_offset_inner(last_seen_offset)
371    }
372}
373
374impl<T: CdcSourceTypeTrait> DebeziumCdcSplit<T> {
375    pub fn new(split_id: u32, start_offset: Option<String>, server_addr: Option<String>) -> Self {
376        let mut ret = Self {
377            mysql_split: None,
378            postgres_split: None,
379            citus_split: None,
380            mongodb_split: None,
381            sql_server_split: None,
382            _phantom: PhantomData,
383        };
384        match T::source_type() {
385            CdcSourceType::Mysql => {
386                let split = MySqlCdcSplit::new(split_id, start_offset);
387                ret.mysql_split = Some(split);
388            }
389            CdcSourceType::Postgres => {
390                let split = PostgresCdcSplit::new(split_id, start_offset, None);
391                ret.postgres_split = Some(split);
392            }
393            CdcSourceType::Citus => {
394                let split = PostgresCdcSplit::new(split_id, start_offset, server_addr);
395                ret.citus_split = Some(split);
396            }
397            CdcSourceType::Mongodb => {
398                let split = MongoDbCdcSplit::new(split_id, start_offset);
399                ret.mongodb_split = Some(split);
400            }
401            CdcSourceType::SqlServer => {
402                let split = SqlServerCdcSplit::new(split_id, start_offset);
403                ret.sql_server_split = Some(split);
404            }
405            CdcSourceType::Unspecified => {
406                unreachable!("invalid debezium split")
407            }
408        }
409        ret
410    }
411
412    pub fn split_id(&self) -> u32 {
413        dispatch_cdc_split!(self, ref, split_id())
414    }
415
416    pub fn start_offset(&self) -> &Option<String> {
417        dispatch_cdc_split!(self, ref, start_offset())
418    }
419
420    pub fn snapshot_done(&self) -> bool {
421        dispatch_cdc_split!(self, ref, is_snapshot_done())
422    }
423
424    pub fn update_offset_inner(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
425        dispatch_cdc_split!(self, mut, update_offset(last_seen_offset)?);
426        Ok(())
427    }
428}
429
430impl DebeziumCdcSplit<Postgres> {
431    /// Extract PostgreSQL LSN value from the current split offset.
432    ///
433    /// Returns Some(lsn) if the LSN is found and can be parsed as u64, None otherwise.
434    pub fn pg_lsn(&self) -> Option<u64> {
435        self.postgres_split.as_ref()?.pg_lsn()
436    }
437}
438
439impl DebeziumCdcSplit<Mysql> {
440    /// Extract MySQL CDC binlog offset (file sequence and position) from the current split offset.
441    ///
442    /// Returns `Some((file_seq, position))` where:
443    /// - `file_seq`: the numeric part of binlog filename (e.g., 123 from "binlog.000123")
444    /// - `position`: the byte offset within the binlog file
445    pub fn mysql_binlog_offset(&self) -> Option<(u64, u64)> {
446        self.mysql_split.as_ref()?.mysql_binlog_offset()
447    }
448}
449
450impl DebeziumCdcSplit<SqlServer> {
451    /// Extract SQL Server CDC `change_lsn` from the current split offset.
452    pub fn sql_server_change_lsn(&self) -> Option<u128> {
453        self.sql_server_split.as_ref()?.sql_server_change_lsn()
454    }
455
456    /// Extract SQL Server CDC `commit_lsn` from the current split offset.
457    pub fn sql_server_commit_lsn(&self) -> Option<u128> {
458        self.sql_server_split.as_ref()?.sql_server_commit_lsn()
459    }
460}
461
462/// Extract the numeric sequence from a MySQL binlog file name `<basename>.<sequence>`.
463///
464/// The basename is configurable (`binlog`, `mysql-bin`, `mysql-bin-changelog` on RDS/Aurora, ...),
465/// so we take the number after the last `.` rather than assuming a fixed prefix.
466pub fn extract_binlog_file_seq(file_name: &str) -> Option<u64> {
467    file_name.rsplit('.').next()?.parse::<u64>().ok()
468}
469
470/// Extract PostgreSQL LSN value from a CDC offset JSON string.
471///
472/// This is a standalone helper function that can be used when you only have the offset string
473/// (e.g., in callbacks) and don't have access to the Split object.
474///
475/// Returns Some(lsn) if the LSN is found and can be parsed as u64, None otherwise.
476pub fn extract_postgres_lsn_from_offset_str(offset_str: &str) -> Option<u64> {
477    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
478    let source_offset = offset.get("sourceOffset")?;
479    let lsn = source_offset.get("lsn")?;
480    lsn.as_u64()
481}
482
483/// Parse SQL Server LSN string (`XXXXXXXX:XXXXXXXX:XXXX`) into a comparable integer.
484pub fn parse_sql_server_lsn_str(lsn: &str) -> Option<u128> {
485    let mut parts = lsn.split(':');
486    let part0 = u32::from_str_radix(parts.next()?, 16).ok()? as u128;
487    let part1 = u32::from_str_radix(parts.next()?, 16).ok()? as u128;
488    let part2 = u16::from_str_radix(parts.next()?, 16).ok()? as u128;
489    if parts.next().is_some() {
490        return None;
491    }
492
493    Some((part0 << 48) | (part1 << 16) | part2)
494}
495
496/// Extract SQL Server `change_lsn` from a CDC offset JSON string.
497pub fn extract_sql_server_change_lsn_from_offset_str(offset_str: &str) -> Option<u128> {
498    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
499    let source_offset = offset.get("sourceOffset")?;
500    let lsn = source_offset.get("change_lsn")?.as_str()?;
501    parse_sql_server_lsn_str(lsn)
502}
503
504/// Extract SQL Server `commit_lsn` from a CDC offset JSON string.
505pub fn extract_sql_server_commit_lsn_from_offset_str(offset_str: &str) -> Option<u128> {
506    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
507    let source_offset = offset.get("sourceOffset")?;
508    let lsn = source_offset.get("commit_lsn")?.as_str()?;
509    parse_sql_server_lsn_str(lsn)
510}
511
512#[cfg(test)]
513mod tests {
514    use super::*;
515
516    #[test]
517    fn test_parse_sql_server_lsn_str() {
518        let lsn = "00000027:00000ac0:0002";
519        let parsed = parse_sql_server_lsn_str(lsn).unwrap();
520        let expected = ((0x00000027_u128) << 48) | ((0x00000ac0_u128) << 16) | (0x0002_u128);
521        assert_eq!(parsed, expected);
522    }
523
524    #[test]
525    fn test_extract_binlog_file_seq() {
526        // default MySQL 8.0 basename
527        assert_eq!(extract_binlog_file_seq("binlog.000123"), Some(123));
528        // `--log-bin=mysql-bin`
529        assert_eq!(extract_binlog_file_seq("mysql-bin.000123"), Some(123));
530        // `<hostname>-bin` on older versions
531        assert_eq!(extract_binlog_file_seq("my-host-bin.000001"), Some(1));
532        // RDS / Aurora MySQL
533        assert_eq!(
534            extract_binlog_file_seq("mysql-bin-changelog.037568"),
535            Some(37568)
536        );
537        // invalid suffix
538        assert_eq!(extract_binlog_file_seq("binlog.index"), None);
539        assert_eq!(extract_binlog_file_seq("no-extension"), None);
540    }
541
542    #[test]
543    fn test_mysql_binlog_offset() {
544        let offset = r#"{
545            "sourcePartition": {"server": "test"},
546            "sourceOffset": {
547                "file": "mysql-bin-changelog.037568",
548                "pos": 12345
549            },
550            "isHeartbeat": false
551        }"#;
552        let split = MySqlCdcSplit::new(1, Some(offset.to_owned()));
553        assert_eq!(split.mysql_binlog_offset(), Some((37568, 12345)));
554    }
555
556    #[test]
557    fn test_extract_sql_server_lsn_from_offset_str() {
558        let offset = r#"{
559            "sourcePartition": {"server":"RW_CDC_1001"},
560            "sourceOffset": {
561                "change_lsn":"00000027:00000ac0:0001",
562                "commit_lsn":"00000027:00000ac0:0002"
563            },
564            "isHeartbeat": false
565        }"#;
566
567        let change_lsn = extract_sql_server_change_lsn_from_offset_str(offset).unwrap();
568        let commit_lsn = extract_sql_server_commit_lsn_from_offset_str(offset).unwrap();
569        assert!(change_lsn < commit_lsn);
570    }
571}
risingwave_connector/source/cdc/split.rs

risingwave_connector/source/cdc/
split.rs
