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::external::postgres::PostgresOffset;
24use crate::source::cdc::{CdcSourceType, CdcSourceTypeTrait, Mariadb, Mysql, Postgres, SqlServer};
25use crate::source::{SplitId, SplitMetaData};
26
27/// The base states of a CDC split, which will be persisted to checkpoint.
28/// CDC source only has single split, so we use the `source_id` to identify the split.
29#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
30pub struct CdcSplitBase {
31    pub split_id: u32,
32    pub start_offset: Option<String>,
33    pub snapshot_done: bool,
34}
35
36impl CdcSplitBase {
37    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
38        Self {
39            split_id,
40            start_offset,
41            snapshot_done: false,
42        }
43    }
44}
45
46trait CdcSplitTrait: Send + Sync {
47    fn split_id(&self) -> u32;
48    fn start_offset(&self) -> &Option<String>;
49    fn is_snapshot_done(&self) -> bool;
50    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()>;
51
52    // MySQL and MongoDB shares the same logic to extract the snapshot flag
53    fn extract_snapshot_flag(&self, start_offset: &str) -> ConnectorResult<bool> {
54        // if snapshot_done is already true, it won't be changed
55        let mut snapshot_done = self.is_snapshot_done();
56        if snapshot_done {
57            return Ok(snapshot_done);
58        }
59
60        let dbz_offset: DebeziumOffset = serde_json::from_str(start_offset).with_context(|| {
61            format!(
62                "invalid cdc offset: {}, split: {}",
63                start_offset,
64                self.split_id()
65            )
66        })?;
67
68        // heartbeat event should not update the `snapshot_done` flag
69        if !dbz_offset.is_heartbeat {
70            snapshot_done = match dbz_offset.source_offset.snapshot {
71                Some(val) => !val,
72                None => true,
73            };
74        }
75        Ok(snapshot_done)
76    }
77}
78
79#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
80pub struct MysqlCdcSplit {
81    pub inner: CdcSplitBase,
82}
83
84#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
85pub struct MariadbCdcSplit {
86    pub inner: CdcSplitBase,
87}
88
89#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
90pub struct PostgresCdcSplit {
91    pub inner: CdcSplitBase,
92}
93
94#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
95pub struct MongodbCdcSplit {
96    pub inner: CdcSplitBase,
97}
98
99#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
100pub struct SqlServerCdcSplit {
101    pub inner: CdcSplitBase,
102}
103
104#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
105pub struct OracleCdcSplit {
106    pub inner: CdcSplitBase,
107}
108
109impl MysqlCdcSplit {
110    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
111        let split = CdcSplitBase {
112            split_id,
113            start_offset,
114            snapshot_done: false,
115        };
116        Self { inner: split }
117    }
118
119    /// Extract MySQL CDC binlog offset (file sequence and position) from the offset JSON string.
120    ///
121    /// MySQL binlog offset format:
122    /// ```json
123    /// {
124    ///   "sourcePartition": { "server": "..." },
125    ///   "sourceOffset": {
126    ///     "file": "binlog.000123",
127    ///     "pos": 456789,
128    ///     ...
129    ///   }
130    /// }
131    /// ```
132    ///
133    /// Returns `Some((file_seq, position))` where:
134    /// - `file_seq`: the numeric part of binlog filename (e.g., 123 from "binlog.000123")
135    /// - `position`: the byte offset within the binlog file
136    pub fn mysql_binlog_offset(&self) -> Option<(u64, u64)> {
137        let offset_str = self.inner.start_offset.as_ref()?;
138        let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
139        let source_offset = offset.get("sourceOffset")?;
140
141        let file = source_offset.get("file")?.as_str()?;
142        let pos = source_offset.get("pos")?.as_u64()?;
143
144        let file_seq = extract_binlog_file_seq(file)?;
145
146        Some((file_seq, pos))
147    }
148}
149
150impl MariadbCdcSplit {
151    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
152        Self {
153            inner: CdcSplitBase::new(split_id, start_offset),
154        }
155    }
156
157    pub fn mariadb_binlog_offset(&self) -> Option<(u64, u64)> {
158        binlog_offset(&self.inner)
159    }
160}
161
162impl CdcSplitTrait for MariadbCdcSplit {
163    fn split_id(&self) -> u32 {
164        self.inner.split_id
165    }
166
167    fn start_offset(&self) -> &Option<String> {
168        &self.inner.start_offset
169    }
170
171    fn is_snapshot_done(&self) -> bool {
172        self.inner.snapshot_done
173    }
174
175    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
176        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
177        self.inner.start_offset = Some(last_seen_offset);
178        Ok(())
179    }
180}
181
182impl CdcSplitTrait for MysqlCdcSplit {
183    fn split_id(&self) -> u32 {
184        self.inner.split_id
185    }
186
187    fn start_offset(&self) -> &Option<String> {
188        &self.inner.start_offset
189    }
190
191    fn is_snapshot_done(&self) -> bool {
192        self.inner.snapshot_done
193    }
194
195    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
196        // if snapshot_done is already true, it won't be updated
197        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
198        self.inner.start_offset = Some(last_seen_offset);
199        Ok(())
200    }
201}
202
203impl PostgresCdcSplit {
204    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
205        let split = CdcSplitBase {
206            split_id,
207            start_offset,
208            snapshot_done: false,
209        };
210        Self { inner: split }
211    }
212
213    /// Extract PostgreSQL LSN value from the offset JSON string.
214    ///
215    /// This function parses the offset JSON and extracts the LSN value from the sourceOffset.lsn field.
216    /// Returns Some(lsn) if the LSN is found and can be parsed as u64, None otherwise.
217    pub fn pg_lsn(&self) -> Option<u64> {
218        let offset_str = self.inner.start_offset.as_ref()?;
219        let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
220        let source_offset = offset.get("sourceOffset")?;
221        let lsn = source_offset.get("lsn")?;
222        lsn.as_u64()
223    }
224
225    /// Extract PostgreSQL `lsn_commit` from the current offset (last commit position).
226    pub fn pg_lsn_commit(&self) -> Option<u64> {
227        let offset_str = self.inner.start_offset.as_ref()?;
228        extract_postgres_lsn_commit_from_offset_str(offset_str)
229    }
230
231    /// Extract PostgreSQL `lsn_proc` from the current offset (last completely processed position).
232    pub fn pg_lsn_proc(&self) -> Option<u64> {
233        let offset_str = self.inner.start_offset.as_ref()?;
234        extract_postgres_lsn_proc_from_offset_str(offset_str)
235    }
236}
237
238impl CdcSplitTrait for PostgresCdcSplit {
239    fn split_id(&self) -> u32 {
240        self.inner.split_id
241    }
242
243    fn start_offset(&self) -> &Option<String> {
244        &self.inner.start_offset
245    }
246
247    fn is_snapshot_done(&self) -> bool {
248        self.inner.snapshot_done
249    }
250
251    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
252        let new_snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
253
254        // Monotonicity guard for streaming-phase Postgres CDC offsets.
255        //
256        // Debezium's task-level auto-restart (BaseSourceTask) reloads
257        // effectiveOffset from ConfigurableOffsetBackingStore, which lags
258        // dbz's in-memory position by one barrier + async commit cycle.
259        // After a broken pipe, dbz re-emits older WAL events whose LSNs
260        // are smaller than the freshest entry we already persisted.
261        // Without this guard those stale LSNs would overwrite the split
262        // state, the store would then be pushed backwards via the next
263        // barrier, and a subsequent restart would load the stale value.
264        //
265        // We guard both `lsn_commit` and `lsn_proc` because dbz consumes
266        // them for two different things on resume:
267        //   - `lsn_commit` is what `validateLogPosition` checks against
268        //     PG slot.restart_lsn; regression here causes a fatal
269        //     `DebeziumException` ("change stream ... no longer available").
270        //   - `lsn_proc` is the start position for `startStreaming` and
271        //     the key `WalPositionLocator` uses to deduplicate replayed
272        //     events on restart; regression here can lead to silent
273        //     data loss / a hung resume if PG has already advanced past
274        //     the rewound value.
275        //
276        // The per-message `lsn` field is intentionally NOT used as a
277        // monotonicity key: under concurrent overlapping transactions
278        // PG streams events in commit order, so the per-event physical
279        // lsn can legitimately move backwards when a long-running
280        // earlier transaction commits later than a shorter newer one.
281        // Only `lsn_commit` and `lsn_proc` are monotonic under correct
282        // dbz behavior.
283        //
284        // If both offsets are parseable as streaming Postgres offsets, use
285        // the same `(lsn_commit, lsn_proc)` ordering as snapshot backfill
286        // deduplication. Snapshot / heartbeat / incomplete offsets are not
287        // comparable and pass through.
288        if let (Some(old_offset), Some(new_offset)) = (
289            self.inner
290                .start_offset
291                .as_deref()
292                .and_then(parse_postgres_offset_from_offset_str),
293            parse_postgres_offset_from_offset_str(&last_seen_offset),
294        ) && new_offset < old_offset
295        {
296            tracing::warn!(
297                split_id = self.inner.split_id,
298                ?old_offset,
299                ?new_offset,
300                "Rejecting backward Postgres CDC offset update; \
301                 keeping current state to prevent state-table regression."
302            );
303            return Ok(());
304        }
305
306        self.inner.snapshot_done = new_snapshot_done;
307        self.inner.start_offset = Some(last_seen_offset);
308        Ok(())
309    }
310
311    fn extract_snapshot_flag(&self, start_offset: &str) -> ConnectorResult<bool> {
312        // if snapshot_done is already true, it won't be changed
313        let mut snapshot_done = self.is_snapshot_done();
314        if snapshot_done {
315            return Ok(snapshot_done);
316        }
317
318        let dbz_offset: DebeziumOffset = serde_json::from_str(start_offset).with_context(|| {
319            format!(
320                "invalid postgres offset: {}, split: {}",
321                start_offset, self.inner.split_id
322            )
323        })?;
324
325        // heartbeat event should not update the `snapshot_done` flag
326        if !dbz_offset.is_heartbeat {
327            snapshot_done = dbz_offset
328                .source_offset
329                .last_snapshot_record
330                .unwrap_or(false);
331        }
332        Ok(snapshot_done)
333    }
334}
335
336impl MongodbCdcSplit {
337    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
338        let split = CdcSplitBase {
339            split_id,
340            start_offset,
341            snapshot_done: false,
342        };
343        Self { inner: split }
344    }
345}
346
347impl CdcSplitTrait for MongodbCdcSplit {
348    fn split_id(&self) -> u32 {
349        self.inner.split_id
350    }
351
352    fn start_offset(&self) -> &Option<String> {
353        &self.inner.start_offset
354    }
355
356    fn is_snapshot_done(&self) -> bool {
357        self.inner.snapshot_done
358    }
359
360    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
361        // if snapshot_done is already true, it will remain true
362        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
363        self.inner.start_offset = Some(last_seen_offset);
364        Ok(())
365    }
366}
367
368impl SqlServerCdcSplit {
369    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
370        let split = CdcSplitBase {
371            split_id,
372            start_offset,
373            snapshot_done: false,
374        };
375        Self { inner: split }
376    }
377
378    /// Extract SQL Server `change_lsn` value from the offset JSON string.
379    pub fn sql_server_change_lsn(&self) -> Option<u128> {
380        let offset_str = self.inner.start_offset.as_ref()?;
381        extract_sql_server_change_lsn_from_offset_str(offset_str)
382    }
383
384    /// Extract SQL Server `commit_lsn` value from the offset JSON string.
385    pub fn sql_server_commit_lsn(&self) -> Option<u128> {
386        let offset_str = self.inner.start_offset.as_ref()?;
387        extract_sql_server_commit_lsn_from_offset_str(offset_str)
388    }
389}
390
391impl CdcSplitTrait for SqlServerCdcSplit {
392    fn split_id(&self) -> u32 {
393        self.inner.split_id
394    }
395
396    fn start_offset(&self) -> &Option<String> {
397        &self.inner.start_offset
398    }
399
400    fn is_snapshot_done(&self) -> bool {
401        self.inner.snapshot_done
402    }
403
404    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
405        // if snapshot_done is already true, it will remain true
406        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
407        self.inner.start_offset = Some(last_seen_offset);
408        Ok(())
409    }
410}
411
412impl OracleCdcSplit {
413    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
414        Self {
415            inner: CdcSplitBase::new(split_id, start_offset),
416        }
417    }
418}
419
420impl CdcSplitTrait for OracleCdcSplit {
421    fn split_id(&self) -> u32 {
422        self.inner.split_id
423    }
424
425    fn start_offset(&self) -> &Option<String> {
426        &self.inner.start_offset
427    }
428
429    fn is_snapshot_done(&self) -> bool {
430        self.inner.snapshot_done
431    }
432
433    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
434        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
435        self.inner.start_offset = Some(last_seen_offset);
436        Ok(())
437    }
438}
439
440/// We use this struct to wrap the specific split, which act as an interface to other modules
441#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
442pub struct DebeziumCdcSplit<T: CdcSourceTypeTrait> {
443    pub mysql_split: Option<MysqlCdcSplit>,
444    pub mariadb_split: Option<MariadbCdcSplit>,
445
446    #[serde(rename = "pg_split")] // backward compatibility
447    pub postgres_split: Option<PostgresCdcSplit>,
448    pub mongodb_split: Option<MongodbCdcSplit>,
449    pub sql_server_split: Option<SqlServerCdcSplit>,
450    pub oracle_split: Option<OracleCdcSplit>,
451
452    #[serde(skip)]
453    pub _phantom: PhantomData<T>,
454}
455
456macro_rules! dispatch_cdc_split_inner {
457    ($dbz_split:expr, $as_type:tt, {$({$cdc_source_type:tt, $cdc_source_split:tt}),*}, $body:expr) => {
458        match T::source_type() {
459            $(
460                CdcSourceType::$cdc_source_type => {
461                    $crate::paste! {
462                        $dbz_split.[<$cdc_source_split>]
463                            .[<as_ $as_type>]()
464                            .expect(concat!(stringify!([<$cdc_source_type:lower>]), " split must exist"))
465                            .$body
466                    }
467                }
468            )*
469            CdcSourceType::Unspecified => {
470                unreachable!("invalid debezium split");
471            }
472        }
473    }
474}
475
476// call corresponding split method of the specific cdc source type
477macro_rules! dispatch_cdc_split {
478    ($dbz_split:expr, $as_type:tt, $body:expr) => {
479        dispatch_cdc_split_inner!($dbz_split, $as_type, {
480            {Mysql, mysql_split},
481            {Mariadb, mariadb_split},
482            {Postgres, postgres_split},
483            {Mongodb, mongodb_split},
484            {SqlServer, sql_server_split},
485            {Oracle, oracle_split}
486        }, $body)
487    }
488}
489
490impl<T: CdcSourceTypeTrait> SplitMetaData for DebeziumCdcSplit<T> {
491    fn id(&self) -> SplitId {
492        format!("{}", self.split_id()).into()
493    }
494
495    fn encode_to_json(&self) -> JsonbVal {
496        serde_json::to_value(self.clone()).unwrap().into()
497    }
498
499    fn restore_from_json(value: JsonbVal) -> ConnectorResult<Self> {
500        serde_json::from_value(value.take()).map_err(Into::into)
501    }
502
503    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
504        self.update_offset_inner(last_seen_offset)
505    }
506}
507
508impl<T: CdcSourceTypeTrait> DebeziumCdcSplit<T> {
509    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
510        let mut ret = Self {
511            mysql_split: None,
512            mariadb_split: None,
513            postgres_split: None,
514            mongodb_split: None,
515            sql_server_split: None,
516            oracle_split: None,
517            _phantom: PhantomData,
518        };
519        match T::source_type() {
520            CdcSourceType::Mysql => {
521                let split = MysqlCdcSplit::new(split_id, start_offset);
522                ret.mysql_split = Some(split);
523            }
524            CdcSourceType::Mariadb => {
525                let split = MariadbCdcSplit::new(split_id, start_offset);
526                ret.mariadb_split = Some(split);
527            }
528            CdcSourceType::Postgres => {
529                let split = PostgresCdcSplit::new(split_id, start_offset);
530                ret.postgres_split = Some(split);
531            }
532            CdcSourceType::Mongodb => {
533                let split = MongodbCdcSplit::new(split_id, start_offset);
534                ret.mongodb_split = Some(split);
535            }
536            CdcSourceType::SqlServer => {
537                let split = SqlServerCdcSplit::new(split_id, start_offset);
538                ret.sql_server_split = Some(split);
539            }
540            CdcSourceType::Oracle => {
541                let split = OracleCdcSplit::new(split_id, start_offset);
542                ret.oracle_split = Some(split);
543            }
544            CdcSourceType::Unspecified => {
545                unreachable!("invalid debezium split")
546            }
547        }
548        ret
549    }
550
551    pub fn split_id(&self) -> u32 {
552        dispatch_cdc_split!(self, ref, split_id())
553    }
554
555    pub fn start_offset(&self) -> &Option<String> {
556        dispatch_cdc_split!(self, ref, start_offset())
557    }
558
559    pub fn snapshot_done(&self) -> bool {
560        dispatch_cdc_split!(self, ref, is_snapshot_done())
561    }
562
563    pub fn update_offset_inner(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
564        dispatch_cdc_split!(self, mut, update_offset(last_seen_offset)?);
565        Ok(())
566    }
567}
568
569impl DebeziumCdcSplit<Postgres> {
570    /// Extract PostgreSQL LSN value from the current split offset.
571    ///
572    /// Returns Some(lsn) if the LSN is found and can be parsed as u64, None otherwise.
573    pub fn pg_lsn(&self) -> Option<u64> {
574        self.postgres_split.as_ref()?.pg_lsn()
575    }
576}
577
578impl DebeziumCdcSplit<Mysql> {
579    /// Extract MySQL CDC binlog offset (file sequence and position) from the current split offset.
580    ///
581    /// Returns `Some((file_seq, position))` where:
582    /// - `file_seq`: the numeric part of binlog filename (e.g., 123 from "binlog.000123")
583    /// - `position`: the byte offset within the binlog file
584    pub fn mysql_binlog_offset(&self) -> Option<(u64, u64)> {
585        self.mysql_split.as_ref()?.mysql_binlog_offset()
586    }
587}
588
589impl DebeziumCdcSplit<Mariadb> {
590    pub fn mariadb_binlog_offset(&self) -> Option<(u64, u64)> {
591        self.mariadb_split.as_ref()?.mariadb_binlog_offset()
592    }
593}
594
595fn binlog_offset(split: &CdcSplitBase) -> Option<(u64, u64)> {
596    let offset_str = split.start_offset.as_ref()?;
597    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
598    let source_offset = offset.get("sourceOffset")?;
599    let file = source_offset.get("file")?.as_str()?;
600    let pos = source_offset.get("pos")?.as_u64()?;
601    Some((extract_binlog_file_seq(file)?, pos))
602}
603
604impl DebeziumCdcSplit<SqlServer> {
605    /// Extract SQL Server CDC `change_lsn` from the current split offset.
606    pub fn sql_server_change_lsn(&self) -> Option<u128> {
607        self.sql_server_split.as_ref()?.sql_server_change_lsn()
608    }
609
610    /// Extract SQL Server CDC `commit_lsn` from the current split offset.
611    pub fn sql_server_commit_lsn(&self) -> Option<u128> {
612        self.sql_server_split.as_ref()?.sql_server_commit_lsn()
613    }
614}
615
616/// Extract the numeric sequence from a MySQL binlog file name `<basename>.<sequence>`.
617///
618/// The basename is configurable (`binlog`, `mysql-bin`, `mysql-bin-changelog` on RDS/Aurora, ...),
619/// so we take the number after the last `.` rather than assuming a fixed prefix.
620pub fn extract_binlog_file_seq(file_name: &str) -> Option<u64> {
621    file_name.rsplit('.').next()?.parse::<u64>().ok()
622}
623
624/// Extract PostgreSQL LSN value from a CDC offset JSON string.
625///
626/// This is a standalone helper function that can be used when you only have the offset string
627/// (e.g., in callbacks) and don't have access to the Split object.
628///
629/// Returns Some(lsn) if the LSN is found and can be parsed as u64, None otherwise.
630pub fn extract_postgres_lsn_from_offset_str(offset_str: &str) -> Option<u64> {
631    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
632    let source_offset = offset.get("sourceOffset")?;
633    let lsn = source_offset.get("lsn")?;
634    lsn.as_u64()
635}
636
637fn parse_postgres_offset_from_offset_str(offset_str: &str) -> Option<PostgresOffset> {
638    PostgresOffset::parse_debezium_offset(offset_str).ok()
639}
640
641/// Extract PostgreSQL `lsn_commit` from a CDC offset JSON string.
642///
643/// `lsn_commit` is the WAL position of the last COMMIT message dbz has
644/// processed. It is the value dbz flushes to PG (`flushLsn`) to advance
645/// `confirmed_flush_lsn`, and the value `validateLogPosition` compares
646/// against `slot.restart_lsn` on restart. Must be monotonic across
647/// restarts -- a regression here can cause a fatal `DebeziumException`
648/// ("change stream ... no longer available").
649pub fn extract_postgres_lsn_commit_from_offset_str(offset_str: &str) -> Option<u64> {
650    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
651    let source_offset = offset.get("sourceOffset")?;
652    let lsn = source_offset.get("lsn_commit")?;
653    lsn.as_u64()
654}
655
656/// Extract PostgreSQL `lsn_proc` from a CDC offset JSON string.
657///
658/// `lsn_proc` is the WAL position of the last message dbz has fully
659/// processed. On restart dbz uses it as the start position passed to
660/// `startStreaming`, and as the dedup key for `WalPositionLocator` to
661/// filter replayed events. A regression can cause silent data loss or
662/// a hung resume if PG has already advanced past the rewound value.
663pub fn extract_postgres_lsn_proc_from_offset_str(offset_str: &str) -> Option<u64> {
664    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
665    let source_offset = offset.get("sourceOffset")?;
666    let lsn = source_offset.get("lsn_proc")?;
667    lsn.as_u64()
668}
669
670/// Parse SQL Server LSN string (`XXXXXXXX:XXXXXXXX:XXXX`) into a comparable integer.
671pub fn parse_sql_server_lsn_str(lsn: &str) -> Option<u128> {
672    let mut parts = lsn.split(':');
673    let part0 = u32::from_str_radix(parts.next()?, 16).ok()? as u128;
674    let part1 = u32::from_str_radix(parts.next()?, 16).ok()? as u128;
675    let part2 = u16::from_str_radix(parts.next()?, 16).ok()? as u128;
676    if parts.next().is_some() {
677        return None;
678    }
679
680    Some((part0 << 48) | (part1 << 16) | part2)
681}
682
683/// Extract SQL Server `change_lsn` from a CDC offset JSON string.
684pub fn extract_sql_server_change_lsn_from_offset_str(offset_str: &str) -> Option<u128> {
685    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
686    let source_offset = offset.get("sourceOffset")?;
687    let lsn = source_offset.get("change_lsn")?.as_str()?;
688    parse_sql_server_lsn_str(lsn)
689}
690
691/// Extract SQL Server `commit_lsn` from a CDC offset JSON string.
692pub fn extract_sql_server_commit_lsn_from_offset_str(offset_str: &str) -> Option<u128> {
693    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
694    let source_offset = offset.get("sourceOffset")?;
695    let lsn = source_offset.get("commit_lsn")?.as_str()?;
696    parse_sql_server_lsn_str(lsn)
697}
698
699#[cfg(test)]
700mod tests {
701    use super::*;
702
703    #[test]
704    fn mariadb_split_round_trip_and_binlog_offset() {
705        let offset = r#"{"sourcePartition":{"server":"RW_CDC_1"},"sourceOffset":{"file":"mariadb-bin.000123","pos":456},"isHeartbeat":false}"#;
706        let split = DebeziumCdcSplit::<Mariadb>::new(1, Some(offset.to_owned()));
707        assert_eq!(split.mariadb_binlog_offset(), Some((123, 456)));
708
709        let json = split.encode_to_json();
710        let restored = DebeziumCdcSplit::<Mariadb>::restore_from_json(json).unwrap();
711        assert_eq!(restored, split);
712    }
713
714    #[test]
715    fn test_parse_sql_server_lsn_str() {
716        let lsn = "00000027:00000ac0:0002";
717        let parsed = parse_sql_server_lsn_str(lsn).unwrap();
718        let expected = ((0x00000027_u128) << 48) | ((0x00000ac0_u128) << 16) | (0x0002_u128);
719        assert_eq!(parsed, expected);
720    }
721
722    /// Build a streaming-phase PG offset JSON. Sets `lsn`, `lsn_proc`,
723    /// and `lsn_commit` all to the same value -- this is the common
724    /// shape for the guard's basic monotonic tests.
725    fn pg_streaming_offset_json(lsn: u64) -> String {
726        pg_streaming_offset_json_full(lsn, lsn, lsn)
727    }
728
729    /// Build a streaming-phase PG offset JSON with independent control
730    /// over `lsn`, `lsn_proc`, and `lsn_commit`. Used to exercise
731    /// transaction-interleave and asymmetric-regression scenarios.
732    fn pg_streaming_offset_json_full(lsn: u64, lsn_proc: u64, lsn_commit: u64) -> String {
733        format!(
734            r#"{{
735                "sourcePartition": {{"server": "RW_CDC_1001"}},
736                "sourceOffset": {{
737                    "last_snapshot_record": false,
738                    "lsn": {lsn},
739                    "lsn_proc": {lsn_proc},
740                    "lsn_commit": {lsn_commit},
741                    "txId": 12345,
742                    "ts_usec": 1700000000000000
743                }},
744                "isHeartbeat": false
745            }}"#
746        )
747    }
748
749    #[test]
750    fn test_postgres_offset_monotonic_guard_rejects_backward_lsn() {
751        let mut split = PostgresCdcSplit::new(1, Some(pg_streaming_offset_json(200)));
752
753        // Push a backward LSN -- must be rejected, current offset kept.
754        split
755            .update_offset(pg_streaming_offset_json(150))
756            .expect("update_offset must not error on rejection");
757        assert_eq!(split.pg_lsn(), Some(200));
758    }
759
760    #[test]
761    fn test_postgres_offset_monotonic_guard_allows_forward_lsn() {
762        let mut split = PostgresCdcSplit::new(1, Some(pg_streaming_offset_json(200)));
763
764        split.update_offset(pg_streaming_offset_json(250)).unwrap();
765        assert_eq!(split.pg_lsn(), Some(250));
766    }
767
768    #[test]
769    fn test_postgres_offset_monotonic_guard_allows_equal_lsn() {
770        let mut split = PostgresCdcSplit::new(1, Some(pg_streaming_offset_json(200)));
771
772        // Equal is allowed (no regression).
773        split.update_offset(pg_streaming_offset_json(200)).unwrap();
774        assert_eq!(split.pg_lsn(), Some(200));
775    }
776
777    #[test]
778    fn test_postgres_offset_snapshot_phase_bypasses_guard() {
779        let snapshot_offset = r#"{
780            "sourcePartition": {"server": "RW_CDC_1001"},
781            "sourceOffset": {
782                "last_snapshot_record": false,
783                "lsn": 150,
784                "txId": 12345,
785                "ts_usec": 1700000000000000
786            },
787            "isHeartbeat": false
788        }"#;
789        let mut split = PostgresCdcSplit::new(1, Some(pg_streaming_offset_json(200)));
790
791        // Snapshot / incomplete offsets do not have a comparable streaming
792        // `lsn_proc`, so they pass through even when the plain `lsn` is lower.
793        split.update_offset(snapshot_offset.to_owned()).unwrap();
794        assert_eq!(split.pg_lsn(), Some(150));
795    }
796
797    #[test]
798    fn test_postgres_offset_lsn_commit_regression_rejected() {
799        // `lsn_commit` regressing alone must trigger reject -- this is
800        // the field validateLogPosition checks, regression here is the
801        // catastrophic case.
802        let initial = pg_streaming_offset_json_full(200, 200, 200);
803        let mut split = PostgresCdcSplit::new(1, Some(initial));
804
805        // lsn_commit 200 -> 150, but lsn / lsn_proc stay at 200.
806        let new = pg_streaming_offset_json_full(200, 200, 150);
807        split.update_offset(new).unwrap();
808
809        assert_eq!(
810            split.pg_lsn_commit(),
811            Some(200),
812            "lsn_commit must not regress"
813        );
814    }
815
816    #[test]
817    fn test_postgres_offset_lsn_proc_regression_rejected() {
818        // `lsn_proc` regressing alone must trigger reject -- this is the
819        // field dbz uses as the startStreaming resume position, and
820        // WalPositionLocator's dedup key on restart.
821        let initial = pg_streaming_offset_json_full(200, 200, 200);
822        let mut split = PostgresCdcSplit::new(1, Some(initial));
823
824        // lsn_proc 200 -> 150, but lsn / lsn_commit stay at 200.
825        let new = pg_streaming_offset_json_full(200, 150, 200);
826        split.update_offset(new).unwrap();
827
828        assert_eq!(split.pg_lsn_proc(), Some(200), "lsn_proc must not regress");
829    }
830
831    #[test]
832    fn test_postgres_offset_interleaved_tx_lsn_proc_drop_allowed_by_commit_lsn() {
833        // Concurrent overlapping transactions under PG logical
834        // replication produce a state-table chunk where the per-event
835        // `lsn` and `lsn_proc` move backwards while `lsn_commit`
836        // strictly advances (because PG streams events in commit
837        // order, and the earlier-begin transaction's INSERT has a
838        // smaller physical LSN even though it commits later).
839        //
840        // Use the same Postgres offset ordering as #22503: compare
841        // `lsn_commit` first, then `lsn_proc`. Since the commit LSN
842        // advances, this update is not a regression.
843        let initial = pg_streaming_offset_json_full(33_967_592, 33_967_592, 33_893_392);
844        let mut split = PostgresCdcSplit::new(1, Some(initial));
845
846        let new = pg_streaming_offset_json_full(33_918_336, 33_918_336, 33_969_832);
847        split.update_offset(new).unwrap();
848
849        assert_eq!(split.pg_lsn_commit(), Some(33_969_832));
850        assert_eq!(split.pg_lsn_proc(), Some(33_918_336));
851    }
852
853    #[test]
854    fn test_extract_binlog_file_seq() {
855        // default MySQL 8.0 basename
856        assert_eq!(extract_binlog_file_seq("binlog.000123"), Some(123));
857        // `--log-bin=mysql-bin`
858        assert_eq!(extract_binlog_file_seq("mysql-bin.000123"), Some(123));
859        // `<hostname>-bin` on older versions
860        assert_eq!(extract_binlog_file_seq("my-host-bin.000001"), Some(1));
861        // RDS / Aurora MySQL
862        assert_eq!(
863            extract_binlog_file_seq("mysql-bin-changelog.037568"),
864            Some(37568)
865        );
866        // invalid suffix
867        assert_eq!(extract_binlog_file_seq("binlog.index"), None);
868        assert_eq!(extract_binlog_file_seq("no-extension"), None);
869    }
870
871    #[test]
872    fn test_mysql_binlog_offset() {
873        let offset = r#"{
874            "sourcePartition": {"server": "test"},
875            "sourceOffset": {
876                "file": "mysql-bin-changelog.037568",
877                "pos": 12345
878            },
879            "isHeartbeat": false
880        }"#;
881        let split = MysqlCdcSplit::new(1, Some(offset.to_owned()));
882        assert_eq!(split.mysql_binlog_offset(), Some((37568, 12345)));
883    }
884
885    #[test]
886    fn test_extract_sql_server_lsn_from_offset_str() {
887        let offset = r#"{
888            "sourcePartition": {"server":"RW_CDC_1001"},
889            "sourceOffset": {
890                "change_lsn":"00000027:00000ac0:0001",
891                "commit_lsn":"00000027:00000ac0:0002"
892            },
893            "isHeartbeat": false
894        }"#;
895
896        let change_lsn = extract_sql_server_change_lsn_from_offset_str(offset).unwrap();
897        let commit_lsn = extract_sql_server_commit_lsn_from_offset_str(offset).unwrap();
898        assert!(change_lsn < commit_lsn);
899    }
900}
risingwave_connector/source/cdc/split.rs

risingwave_connector/source/cdc/
split.rs
