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
101impl MySqlCdcSplit {
102    pub fn new(split_id: u32, start_offset: Option<String>) -> Self {
103        let split = CdcSplitBase {
104            split_id,
105            start_offset,
106            snapshot_done: false,
107        };
108        Self { inner: split }
109    }
110
111    /// Extract MySQL CDC binlog offset (file sequence and position) from the offset JSON string.
112    ///
113    /// MySQL binlog offset format:
114    /// ```json
115    /// {
116    ///   "sourcePartition": { "server": "..." },
117    ///   "sourceOffset": {
118    ///     "file": "binlog.000123",
119    ///     "pos": 456789,
120    ///     ...
121    ///   }
122    /// }
123    /// ```
124    ///
125    /// Returns `Some((file_seq, position))` where:
126    /// - `file_seq`: the numeric part of binlog filename (e.g., 123 from "binlog.000123")
127    /// - `position`: the byte offset within the binlog file
128    pub fn mysql_binlog_offset(&self) -> Option<(u64, u64)> {
129        let offset_str = self.inner.start_offset.as_ref()?;
130        let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
131        let source_offset = offset.get("sourceOffset")?;
132
133        let file = source_offset.get("file")?.as_str()?;
134        let pos = source_offset.get("pos")?.as_u64()?;
135
136        let file_seq = extract_binlog_file_seq(file)?;
137
138        Some((file_seq, pos))
139    }
140}
141
142impl CdcSplitTrait for MySqlCdcSplit {
143    fn split_id(&self) -> u32 {
144        self.inner.split_id
145    }
146
147    fn start_offset(&self) -> &Option<String> {
148        &self.inner.start_offset
149    }
150
151    fn is_snapshot_done(&self) -> bool {
152        self.inner.snapshot_done
153    }
154
155    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
156        // if snapshot_done is already true, it won't be updated
157        self.inner.snapshot_done = self.extract_snapshot_flag(last_seen_offset.as_str())?;
158        self.inner.start_offset = Some(last_seen_offset);
159        Ok(())
160    }
161}
162
163impl PostgresCdcSplit {
164    pub fn new(split_id: u32, start_offset: Option<String>, server_addr: Option<String>) -> Self {
165        let split = CdcSplitBase {
166            split_id,
167            start_offset,
168            snapshot_done: false,
169        };
170        Self {
171            inner: split,
172            server_addr,
173        }
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
375/// We use this struct to wrap the specific split, which act as an interface to other modules
376#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Hash)]
377pub struct DebeziumCdcSplit<T: CdcSourceTypeTrait> {
378    pub mysql_split: Option<MySqlCdcSplit>,
379
380    #[serde(rename = "pg_split")] // backward compatibility
381    pub postgres_split: Option<PostgresCdcSplit>,
382    pub citus_split: Option<PostgresCdcSplit>,
383    pub mongodb_split: Option<MongoDbCdcSplit>,
384    pub sql_server_split: Option<SqlServerCdcSplit>,
385
386    #[serde(skip)]
387    pub _phantom: PhantomData<T>,
388}
389
390macro_rules! dispatch_cdc_split_inner {
391    ($dbz_split:expr, $as_type:tt, {$({$cdc_source_type:tt, $cdc_source_split:tt}),*}, $body:expr) => {
392        match T::source_type() {
393            $(
394                CdcSourceType::$cdc_source_type => {
395                    $crate::paste! {
396                        $dbz_split.[<$cdc_source_split>]
397                            .[<as_ $as_type>]()
398                            .expect(concat!(stringify!([<$cdc_source_type:lower>]), " split must exist"))
399                            .$body
400                    }
401                }
402            )*
403            CdcSourceType::Unspecified => {
404                unreachable!("invalid debezium split");
405            }
406        }
407    }
408}
409
410// call corresponding split method of the specific cdc source type
411macro_rules! dispatch_cdc_split {
412    ($dbz_split:expr, $as_type:tt, $body:expr) => {
413        dispatch_cdc_split_inner!($dbz_split, $as_type, {
414            {Mysql, mysql_split},
415            {Postgres, postgres_split},
416            {Citus, citus_split},
417            {Mongodb, mongodb_split},
418            {SqlServer, sql_server_split}
419        }, $body)
420    }
421}
422
423impl<T: CdcSourceTypeTrait> SplitMetaData for DebeziumCdcSplit<T> {
424    fn id(&self) -> SplitId {
425        format!("{}", self.split_id()).into()
426    }
427
428    fn encode_to_json(&self) -> JsonbVal {
429        serde_json::to_value(self.clone()).unwrap().into()
430    }
431
432    fn restore_from_json(value: JsonbVal) -> ConnectorResult<Self> {
433        serde_json::from_value(value.take()).map_err(Into::into)
434    }
435
436    fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
437        self.update_offset_inner(last_seen_offset)
438    }
439}
440
441impl<T: CdcSourceTypeTrait> DebeziumCdcSplit<T> {
442    pub fn new(split_id: u32, start_offset: Option<String>, server_addr: Option<String>) -> Self {
443        let mut ret = Self {
444            mysql_split: None,
445            postgres_split: None,
446            citus_split: None,
447            mongodb_split: None,
448            sql_server_split: None,
449            _phantom: PhantomData,
450        };
451        match T::source_type() {
452            CdcSourceType::Mysql => {
453                let split = MySqlCdcSplit::new(split_id, start_offset);
454                ret.mysql_split = Some(split);
455            }
456            CdcSourceType::Postgres => {
457                let split = PostgresCdcSplit::new(split_id, start_offset, None);
458                ret.postgres_split = Some(split);
459            }
460            CdcSourceType::Citus => {
461                let split = PostgresCdcSplit::new(split_id, start_offset, server_addr);
462                ret.citus_split = Some(split);
463            }
464            CdcSourceType::Mongodb => {
465                let split = MongoDbCdcSplit::new(split_id, start_offset);
466                ret.mongodb_split = Some(split);
467            }
468            CdcSourceType::SqlServer => {
469                let split = SqlServerCdcSplit::new(split_id, start_offset);
470                ret.sql_server_split = Some(split);
471            }
472            CdcSourceType::Unspecified => {
473                unreachable!("invalid debezium split")
474            }
475        }
476        ret
477    }
478
479    pub fn split_id(&self) -> u32 {
480        dispatch_cdc_split!(self, ref, split_id())
481    }
482
483    pub fn start_offset(&self) -> &Option<String> {
484        dispatch_cdc_split!(self, ref, start_offset())
485    }
486
487    pub fn snapshot_done(&self) -> bool {
488        dispatch_cdc_split!(self, ref, is_snapshot_done())
489    }
490
491    pub fn update_offset_inner(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
492        dispatch_cdc_split!(self, mut, update_offset(last_seen_offset)?);
493        Ok(())
494    }
495}
496
497impl DebeziumCdcSplit<Postgres> {
498    /// Extract PostgreSQL LSN value from the current split offset.
499    ///
500    /// Returns Some(lsn) if the LSN is found and can be parsed as u64, None otherwise.
501    pub fn pg_lsn(&self) -> Option<u64> {
502        self.postgres_split.as_ref()?.pg_lsn()
503    }
504}
505
506impl DebeziumCdcSplit<Mysql> {
507    /// Extract MySQL CDC binlog offset (file sequence and position) from the current split offset.
508    ///
509    /// Returns `Some((file_seq, position))` where:
510    /// - `file_seq`: the numeric part of binlog filename (e.g., 123 from "binlog.000123")
511    /// - `position`: the byte offset within the binlog file
512    pub fn mysql_binlog_offset(&self) -> Option<(u64, u64)> {
513        self.mysql_split.as_ref()?.mysql_binlog_offset()
514    }
515}
516
517impl DebeziumCdcSplit<SqlServer> {
518    /// Extract SQL Server CDC `change_lsn` from the current split offset.
519    pub fn sql_server_change_lsn(&self) -> Option<u128> {
520        self.sql_server_split.as_ref()?.sql_server_change_lsn()
521    }
522
523    /// Extract SQL Server CDC `commit_lsn` from the current split offset.
524    pub fn sql_server_commit_lsn(&self) -> Option<u128> {
525        self.sql_server_split.as_ref()?.sql_server_commit_lsn()
526    }
527}
528
529/// Extract the numeric sequence from a MySQL binlog file name `<basename>.<sequence>`.
530///
531/// The basename is configurable (`binlog`, `mysql-bin`, `mysql-bin-changelog` on RDS/Aurora, ...),
532/// so we take the number after the last `.` rather than assuming a fixed prefix.
533pub fn extract_binlog_file_seq(file_name: &str) -> Option<u64> {
534    file_name.rsplit('.').next()?.parse::<u64>().ok()
535}
536
537/// Extract PostgreSQL LSN value from a CDC offset JSON string.
538///
539/// This is a standalone helper function that can be used when you only have the offset string
540/// (e.g., in callbacks) and don't have access to the Split object.
541///
542/// Returns Some(lsn) if the LSN is found and can be parsed as u64, None otherwise.
543pub fn extract_postgres_lsn_from_offset_str(offset_str: &str) -> Option<u64> {
544    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
545    let source_offset = offset.get("sourceOffset")?;
546    let lsn = source_offset.get("lsn")?;
547    lsn.as_u64()
548}
549
550fn parse_postgres_offset_from_offset_str(offset_str: &str) -> Option<PostgresOffset> {
551    PostgresOffset::parse_debezium_offset(offset_str).ok()
552}
553
554/// Extract PostgreSQL `lsn_commit` from a CDC offset JSON string.
555///
556/// `lsn_commit` is the WAL position of the last COMMIT message dbz has
557/// processed. It is the value dbz flushes to PG (`flushLsn`) to advance
558/// `confirmed_flush_lsn`, and the value `validateLogPosition` compares
559/// against `slot.restart_lsn` on restart. Must be monotonic across
560/// restarts -- a regression here can cause a fatal `DebeziumException`
561/// ("change stream ... no longer available").
562pub fn extract_postgres_lsn_commit_from_offset_str(offset_str: &str) -> Option<u64> {
563    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
564    let source_offset = offset.get("sourceOffset")?;
565    let lsn = source_offset.get("lsn_commit")?;
566    lsn.as_u64()
567}
568
569/// Extract PostgreSQL `lsn_proc` from a CDC offset JSON string.
570///
571/// `lsn_proc` is the WAL position of the last message dbz has fully
572/// processed. On restart dbz uses it as the start position passed to
573/// `startStreaming`, and as the dedup key for `WalPositionLocator` to
574/// filter replayed events. A regression can cause silent data loss or
575/// a hung resume if PG has already advanced past the rewound value.
576pub fn extract_postgres_lsn_proc_from_offset_str(offset_str: &str) -> Option<u64> {
577    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
578    let source_offset = offset.get("sourceOffset")?;
579    let lsn = source_offset.get("lsn_proc")?;
580    lsn.as_u64()
581}
582
583/// Parse SQL Server LSN string (`XXXXXXXX:XXXXXXXX:XXXX`) into a comparable integer.
584pub fn parse_sql_server_lsn_str(lsn: &str) -> Option<u128> {
585    let mut parts = lsn.split(':');
586    let part0 = u32::from_str_radix(parts.next()?, 16).ok()? as u128;
587    let part1 = u32::from_str_radix(parts.next()?, 16).ok()? as u128;
588    let part2 = u16::from_str_radix(parts.next()?, 16).ok()? as u128;
589    if parts.next().is_some() {
590        return None;
591    }
592
593    Some((part0 << 48) | (part1 << 16) | part2)
594}
595
596/// Extract SQL Server `change_lsn` from a CDC offset JSON string.
597pub fn extract_sql_server_change_lsn_from_offset_str(offset_str: &str) -> Option<u128> {
598    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
599    let source_offset = offset.get("sourceOffset")?;
600    let lsn = source_offset.get("change_lsn")?.as_str()?;
601    parse_sql_server_lsn_str(lsn)
602}
603
604/// Extract SQL Server `commit_lsn` from a CDC offset JSON string.
605pub fn extract_sql_server_commit_lsn_from_offset_str(offset_str: &str) -> Option<u128> {
606    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
607    let source_offset = offset.get("sourceOffset")?;
608    let lsn = source_offset.get("commit_lsn")?.as_str()?;
609    parse_sql_server_lsn_str(lsn)
610}
611
612#[cfg(test)]
613mod tests {
614    use super::*;
615
616    #[test]
617    fn test_parse_sql_server_lsn_str() {
618        let lsn = "00000027:00000ac0:0002";
619        let parsed = parse_sql_server_lsn_str(lsn).unwrap();
620        let expected = ((0x00000027_u128) << 48) | ((0x00000ac0_u128) << 16) | (0x0002_u128);
621        assert_eq!(parsed, expected);
622    }
623
624    /// Build a streaming-phase PG offset JSON. Sets `lsn`, `lsn_proc`,
625    /// and `lsn_commit` all to the same value -- this is the common
626    /// shape for the guard's basic monotonic tests.
627    fn pg_streaming_offset_json(lsn: u64) -> String {
628        pg_streaming_offset_json_full(lsn, lsn, lsn)
629    }
630
631    /// Build a streaming-phase PG offset JSON with independent control
632    /// over `lsn`, `lsn_proc`, and `lsn_commit`. Used to exercise
633    /// transaction-interleave and asymmetric-regression scenarios.
634    fn pg_streaming_offset_json_full(lsn: u64, lsn_proc: u64, lsn_commit: u64) -> String {
635        format!(
636            r#"{{
637                "sourcePartition": {{"server": "RW_CDC_1001"}},
638                "sourceOffset": {{
639                    "last_snapshot_record": false,
640                    "lsn": {lsn},
641                    "lsn_proc": {lsn_proc},
642                    "lsn_commit": {lsn_commit},
643                    "txId": 12345,
644                    "ts_usec": 1700000000000000
645                }},
646                "isHeartbeat": false
647            }}"#
648        )
649    }
650
651    #[test]
652    fn test_postgres_offset_monotonic_guard_rejects_backward_lsn() {
653        let mut split = PostgresCdcSplit::new(1, Some(pg_streaming_offset_json(200)), None);
654
655        // Push a backward LSN -- must be rejected, current offset kept.
656        split
657            .update_offset(pg_streaming_offset_json(150))
658            .expect("update_offset must not error on rejection");
659        assert_eq!(split.pg_lsn(), Some(200));
660    }
661
662    #[test]
663    fn test_postgres_offset_monotonic_guard_allows_forward_lsn() {
664        let mut split = PostgresCdcSplit::new(1, Some(pg_streaming_offset_json(200)), None);
665
666        split.update_offset(pg_streaming_offset_json(250)).unwrap();
667        assert_eq!(split.pg_lsn(), Some(250));
668    }
669
670    #[test]
671    fn test_postgres_offset_monotonic_guard_allows_equal_lsn() {
672        let mut split = PostgresCdcSplit::new(1, Some(pg_streaming_offset_json(200)), None);
673
674        // Equal is allowed (no regression).
675        split.update_offset(pg_streaming_offset_json(200)).unwrap();
676        assert_eq!(split.pg_lsn(), Some(200));
677    }
678
679    #[test]
680    fn test_postgres_offset_snapshot_phase_bypasses_guard() {
681        let snapshot_offset = r#"{
682            "sourcePartition": {"server": "RW_CDC_1001"},
683            "sourceOffset": {
684                "last_snapshot_record": false,
685                "lsn": 150,
686                "txId": 12345,
687                "ts_usec": 1700000000000000
688            },
689            "isHeartbeat": false
690        }"#;
691        let mut split = PostgresCdcSplit::new(1, Some(pg_streaming_offset_json(200)), None);
692
693        // Snapshot / incomplete offsets do not have a comparable streaming
694        // `lsn_proc`, so they pass through even when the plain `lsn` is lower.
695        split.update_offset(snapshot_offset.to_owned()).unwrap();
696        assert_eq!(split.pg_lsn(), Some(150));
697    }
698
699    #[test]
700    fn test_postgres_offset_lsn_commit_regression_rejected() {
701        // `lsn_commit` regressing alone must trigger reject -- this is
702        // the field validateLogPosition checks, regression here is the
703        // catastrophic case.
704        let initial = pg_streaming_offset_json_full(200, 200, 200);
705        let mut split = PostgresCdcSplit::new(1, Some(initial), None);
706
707        // lsn_commit 200 -> 150, but lsn / lsn_proc stay at 200.
708        let new = pg_streaming_offset_json_full(200, 200, 150);
709        split.update_offset(new).unwrap();
710
711        assert_eq!(
712            split.pg_lsn_commit(),
713            Some(200),
714            "lsn_commit must not regress"
715        );
716    }
717
718    #[test]
719    fn test_postgres_offset_lsn_proc_regression_rejected() {
720        // `lsn_proc` regressing alone must trigger reject -- this is the
721        // field dbz uses as the startStreaming resume position, and
722        // WalPositionLocator's dedup key on restart.
723        let initial = pg_streaming_offset_json_full(200, 200, 200);
724        let mut split = PostgresCdcSplit::new(1, Some(initial), None);
725
726        // lsn_proc 200 -> 150, but lsn / lsn_commit stay at 200.
727        let new = pg_streaming_offset_json_full(200, 150, 200);
728        split.update_offset(new).unwrap();
729
730        assert_eq!(split.pg_lsn_proc(), Some(200), "lsn_proc must not regress");
731    }
732
733    #[test]
734    fn test_postgres_offset_interleaved_tx_lsn_proc_drop_allowed_by_commit_lsn() {
735        // Concurrent overlapping transactions under PG logical
736        // replication produce a state-table chunk where the per-event
737        // `lsn` and `lsn_proc` move backwards while `lsn_commit`
738        // strictly advances (because PG streams events in commit
739        // order, and the earlier-begin transaction's INSERT has a
740        // smaller physical LSN even though it commits later).
741        //
742        // Use the same Postgres offset ordering as #22503: compare
743        // `lsn_commit` first, then `lsn_proc`. Since the commit LSN
744        // advances, this update is not a regression.
745        let initial = pg_streaming_offset_json_full(33_967_592, 33_967_592, 33_893_392);
746        let mut split = PostgresCdcSplit::new(1, Some(initial), None);
747
748        let new = pg_streaming_offset_json_full(33_918_336, 33_918_336, 33_969_832);
749        split.update_offset(new).unwrap();
750
751        assert_eq!(split.pg_lsn_commit(), Some(33_969_832));
752        assert_eq!(split.pg_lsn_proc(), Some(33_918_336));
753    }
754
755    #[test]
756    fn test_extract_binlog_file_seq() {
757        // default MySQL 8.0 basename
758        assert_eq!(extract_binlog_file_seq("binlog.000123"), Some(123));
759        // `--log-bin=mysql-bin`
760        assert_eq!(extract_binlog_file_seq("mysql-bin.000123"), Some(123));
761        // `<hostname>-bin` on older versions
762        assert_eq!(extract_binlog_file_seq("my-host-bin.000001"), Some(1));
763        // RDS / Aurora MySQL
764        assert_eq!(
765            extract_binlog_file_seq("mysql-bin-changelog.037568"),
766            Some(37568)
767        );
768        // invalid suffix
769        assert_eq!(extract_binlog_file_seq("binlog.index"), None);
770        assert_eq!(extract_binlog_file_seq("no-extension"), None);
771    }
772
773    #[test]
774    fn test_mysql_binlog_offset() {
775        let offset = r#"{
776            "sourcePartition": {"server": "test"},
777            "sourceOffset": {
778                "file": "mysql-bin-changelog.037568",
779                "pos": 12345
780            },
781            "isHeartbeat": false
782        }"#;
783        let split = MySqlCdcSplit::new(1, Some(offset.to_owned()));
784        assert_eq!(split.mysql_binlog_offset(), Some((37568, 12345)));
785    }
786
787    #[test]
788    fn test_extract_sql_server_lsn_from_offset_str() {
789        let offset = r#"{
790            "sourcePartition": {"server":"RW_CDC_1001"},
791            "sourceOffset": {
792                "change_lsn":"00000027:00000ac0:0001",
793                "commit_lsn":"00000027:00000ac0:0002"
794            },
795            "isHeartbeat": false
796        }"#;
797
798        let change_lsn = extract_sql_server_change_lsn_from_offset_str(offset).unwrap();
799        let commit_lsn = extract_sql_server_commit_lsn_from_offset_str(offset).unwrap();
800        assert!(change_lsn < commit_lsn);
801    }
802}
risingwave_connector/source/cdc/split.rs

risingwave_connector/source/cdc/
split.rs
