risingwave_stream/executor/backfill/cdc/
state.rs

1// Copyright 2023 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 anyhow::anyhow;
16use risingwave_common::id::TableId;
17use risingwave_common::row;
18use risingwave_common::row::{OwnedRow, Row};
19use risingwave_common::types::{Datum, JsonbVal, ScalarImpl};
20use risingwave_common::util::epoch::EpochPair;
21use risingwave_connector::source::cdc::external::CdcOffset;
22use risingwave_storage::StateStore;
23
24use crate::common::table::state_table::StateTable;
25use crate::executor::StreamExecutorResult;
26
27#[derive(Debug, Default)]
28pub struct CdcStateRecord {
29    pub current_pk_pos: Option<OwnedRow>,
30    pub is_finished: bool,
31    /// The last cdc offset that has been consumed by the cdc backfill executor
32    pub last_cdc_offset: Option<CdcOffset>,
33    pub row_count: i64,
34}
35
36/// state schema: | `split_id` | `pk...` | `backfill_finished` | `row_count` | `cdc_offset` |
37pub struct CdcBackfillState<S: StateStore> {
38    /// Id of the backfilling table, will be the key of the state
39    split_id: String,
40    state_table: StateTable<S>,
41
42    cached_state: Vec<Datum>,
43}
44
45impl<S: StateStore> CdcBackfillState<S> {
46    pub fn new(table_id: TableId, state_table: StateTable<S>, state_len: usize) -> Self {
47        Self {
48            split_id: table_id.to_string(),
49            state_table,
50            cached_state: vec![None; state_len],
51        }
52    }
53
54    pub async fn init_epoch(&mut self, epoch: EpochPair) -> StreamExecutorResult<()> {
55        self.state_table.init_epoch(epoch).await
56    }
57
58    /// Restore the backfill state from storage
59    pub async fn restore_state(&mut self) -> StreamExecutorResult<CdcStateRecord> {
60        let key = Some(self.split_id.clone());
61        match self
62            .state_table
63            .get_row(row::once(key.map(ScalarImpl::from)))
64            .await?
65        {
66            Some(row) => {
67                tracing::info!("restored cdc backfill state: {:?}", row);
68                self.cached_state = row.into_inner().into_vec();
69                let state = self.cached_state.as_slice();
70                let state_len = state.len();
71                // schema: | `split_id` | `pk...` | `backfill_finished` | `row_count` | `cdc_offset` |
72                let row_count = match state[state_len - 2] {
73                    Some(ScalarImpl::Int64(val)) => val,
74                    _ => return Err(anyhow!("invalid backfill state: row_count").into()),
75                };
76                let is_finished = match state[state_len - 3] {
77                    Some(ScalarImpl::Bool(val)) => val,
78                    _ => return Err(anyhow!("invalid backfill state: backfill_finished").into()),
79                };
80                let cdc_offset = match state[state_len - 1] {
81                    Some(ScalarImpl::Jsonb(ref jsonb)) => {
82                        serde_json::from_value(jsonb.clone().take()).unwrap()
83                    }
84                    None if is_finished => None,
85                    None => {
86                        return Err(anyhow!(
87                            "invalid backfill state: unfinished row has null cdc_offset"
88                        )
89                        .into());
90                    }
91                    _ => return Err(anyhow!("invalid backfill state: cdc_offset").into()),
92                };
93
94                let current_pk_pos = state[1..state_len - 3].to_vec();
95                Ok(CdcStateRecord {
96                    current_pk_pos: Some(OwnedRow::new(current_pk_pos)),
97                    is_finished,
98                    last_cdc_offset: cdc_offset,
99                    row_count,
100                })
101            }
102            None => Ok(CdcStateRecord::default()),
103        }
104    }
105
106    /// Modify the state of the corresponding split (currently only supports single split)
107    pub async fn mutate_state(
108        &mut self,
109        current_pk_pos: Option<OwnedRow>,
110        last_cdc_offset: Option<CdcOffset>,
111        row_count: u64,
112        is_finished: bool,
113    ) -> StreamExecutorResult<()> {
114        // schema: | `split_id` | `pk...` | `backfill_finished` | `row_count` | `cdc_offset` |
115        let state = self.cached_state.as_mut_slice();
116        let split_id = Some(ScalarImpl::from(self.split_id.clone()));
117        let state_len = state.len();
118        state[0].clone_from(&split_id);
119        if let Some(current_pk_pos) = &current_pk_pos {
120            state[1..=current_pk_pos.len()].clone_from_slice(current_pk_pos.as_inner());
121        }
122        state[state_len - 3] = Some(is_finished.into());
123        state[state_len - 2] = Some((row_count as i64).into());
124        state[state_len - 1] = last_cdc_offset.clone().map(|cdc_offset| {
125            let json = serde_json::to_value(cdc_offset).unwrap();
126            ScalarImpl::Jsonb(JsonbVal::from(json))
127        });
128
129        match self.state_table.get_row(row::once(split_id)).await? {
130            Some(prev_row) => {
131                self.state_table
132                    .update(prev_row, self.cached_state.as_slice());
133            }
134            None => {
135                self.state_table.insert(self.cached_state.as_slice());
136            }
137        }
138        Ok(())
139    }
140
141    /// Persist the state to storage
142    pub async fn commit_state(&mut self, new_epoch: EpochPair) -> StreamExecutorResult<()> {
143        self.state_table
144            .commit_assert_no_update_vnode_bitmap(new_epoch)
145            .await
146    }
147}
risingwave_stream/executor/backfill/cdc/state.rs

risingwave_stream/executor/backfill/cdc/
state.rs
