Repository: risingwavelabs/risingwave
File: src/stream/src/executor/source/source_executor.rs (Rust)

Code before:

/// This function parses the offset JSON and extracts the LSN value from the sourceOffset.lsn field.
/// Returns Some(lsn) if the LSN is found and can be parsed as u64, None otherwise.
fn extract_pg_cdc_lsn_from_offset(offset_str: &str) -> Option<u64> {
    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
    let source_offset = offset.get("sourceOffset")?;
    let lsn = source_offset.get("lsn")?;
    lsn.as_u64()
}

/// Extract MySQL CDC binlog offset (file sequence and position) from the offset JSON string.
///
/// MySQL binlog offset format:
/// {
///   "sourcePartition": { "server": "..." },
///   "sourceOffset": {
///     "file": "binlog.000123",
///     "pos": 456789,
///     ...
///   }
/// }
///
/// Returns `Some((file_seq, position))` where:
/// - `file_seq`: the numeric part of binlog filename (e.g., 123 from "binlog.000123")
/// - `position`: the byte offset within the binlog file
fn extract_mysql_cdc_binlog_offset(offset_str: &str) -> Option<(u64, u64)> {
    let offset = serde_json::from_str::<serde_json::Value>(offset_str).ok()?;
    let source_offset = offset.get("sourceOffset")?;

    let file = source_offset.get("file")?.as_str()?;
    let pos = source_offset.get("pos")?.as_u64()?;

    // Extract numeric sequence from "binlog.NNNNNN"
    let file_seq = file.strip_prefix("binlog.")?.parse::<u64>().ok()?;

    Some((file_seq, pos))
}

pub struct SourceExecutor<S: StateStore> {
    actor_ctx: ActorContextRef,

    /// Streaming source for external
    stream_source_core: StreamSourceCore<S>,

    /// Metrics for monitor.
    metrics: Arc<StreamingMetrics>,

    /// Receiver of barrier channel.
    barrier_receiver: Option<UnboundedReceiver<Barrier>>,

    /// System parameter reader to read barrier interval

Review comment (refactor):
`source_executor.rs` is already huge. Let's try to avoid putting connector specific logics in this file as much as possible.

Let's put these methods in https://github.com/risingwavelabs/risingwave/blob/d052368d0f54631b5649dce648b70ac323634954/src/connector/src/source/cdc/split.rs#L256

Same for `extract_pg_cdc_lsn_from_offset`

Example:
```
impl DebeziumCdcSplit<Postgres> {
  pub fn pg_lsn(&self) -> Option<u64> {
     ...
  }
}

impl DebeziumCdcSplit<Mysql> {
  pub fn mysql_binlog_offset(&self) -> Option<(u64, u64)> {
     ...
  }
}
```

Code after:
    /// System parameter reader to read barrier interval
    system_params: SystemParamsReaderRef,

    /// Rate limit in rows/s.
    rate_limit_rps: Option<u32>,

    is_shared_non_cdc: bool,

    /// Local barrier manager for reporting source load finished events
    _barrier_manager: LocalBarrierManager,
}

impl<S: StateStore> SourceExecutor<S> {
    #[expect(clippy::too_many_arguments)]
    pub fn new(
        actor_ctx: ActorContextRef,
        stream_source_core: StreamSourceCore<S>,
        metrics: Arc<StreamingMetrics>,
        barrier_receiver: UnboundedReceiver<Barrier>,
        system_params: SystemParamsReaderRef,
        rate_limit_rps: Option<u32>,
        is_shared_non_cdc: bool,
        barrier_manager: LocalBarrierManager,
    ) -> Self {
        Self {
            actor_ctx,
            stream_source_core,
            metrics,
            barrier_receiver: Some(barrier_receiver),
            system_params,
            rate_limit_rps,
            is_shared_non_cdc,
            _barrier_manager: barrier_manager,
        }
    }

    fn stream_reader_builder(&self, source_desc: SourceDesc) -> StreamReaderBuilder {
        StreamReaderBuilder {
            source_desc,
            rate_limit: self.rate_limit_rps,
            source_id: self.stream_source_core.source_id,
            source_name: self.stream_source_core.source_name.clone(),
            is_auto_schema_change_enable: self.is_auto_schema_change_enable(),
            actor_ctx: self.actor_ctx.clone(),
            reader_stream: None,
        }
    }

    async fn spawn_wait_checkpoint_worker(
        core: &StreamSourceCore<S>,
        source_reader: SourceReader,