risingwave_connector/source/nexmark/
split.rs1use risingwave_common::types::JsonbVal;
16use serde::{Deserialize, Serialize};
17
18use crate::error::ConnectorResult;
19use crate::source::{SplitId, SplitMetaData};
20
21#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Hash)]
22pub struct NexmarkSplit {
23 pub(crate) split_index: i32,
24 pub(crate) split_num: i32,
25 pub(crate) start_offset: Option<u64>,
26}
27
28impl SplitMetaData for NexmarkSplit {
29 fn id(&self) -> SplitId {
30 format!("{}-{}", self.split_num, self.split_index).into()
32 }
33
34 fn restore_from_json(value: JsonbVal) -> ConnectorResult<Self> {
35 serde_json::from_value(value.take()).map_err(Into::into)
36 }
37
38 fn encode_to_json(&self) -> JsonbVal {
39 serde_json::to_value(self.clone()).unwrap().into()
40 }
41
42 fn update_offset(&mut self, last_seen_offset: String) -> ConnectorResult<()> {
43 self.start_offset = Some(last_seen_offset.as_str().parse::<u64>().unwrap());
44 Ok(())
45 }
46}
47
48impl NexmarkSplit {
49 pub fn new(split_index: i32, split_num: i32, start_offset: Option<u64>) -> NexmarkSplit {
50 NexmarkSplit {
51 split_index,
52 split_num,
53 start_offset,
54 }
55 }
56}