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

risingwave_connector/source/cdc/
split.rs
