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

risingwave_connector/source/cdc/
split.rs
