risingwave_meta/barrier/
mod.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::collections::{HashMap, HashSet};
16
17use anyhow::anyhow;
18use risingwave_common::catalog::{DatabaseId, TableId};
19use risingwave_pb::catalog::Database;
20use risingwave_pb::hummock::HummockVersionStats;
21use risingwave_pb::meta::PbRecoveryStatus;
22use tokio::sync::oneshot::Sender;
23
24use self::notifier::Notifier;
25use crate::barrier::info::BarrierInfo;
26use crate::manager::ActiveStreamingWorkerNodes;
27use crate::model::{ActorId, BackfillUpstreamType, FragmentId, StreamActor, SubscriptionId};
28use crate::{MetaError, MetaResult};
29
30mod backfill_order_control;
31pub mod cdc_progress;
32mod checkpoint;
33mod command;
34pub use command::RescheduleContext;
35mod complete_task;
36pub(super) mod context;
37mod edge_builder;
38mod info;
39mod manager;
40mod notifier;
41mod partial_graph;
42mod progress;
43mod rpc;
44mod schedule;
45#[cfg(test)]
46mod tests;
47mod trace;
48mod utils;
49mod worker;
50
51pub use backfill_order_control::{BackfillNode, BackfillOrderState};
52use risingwave_common::id::JobId;
53use risingwave_pb::ddl_service::PbBackfillType;
54
55pub use self::command::{
56    BarrierKind, BatchRefreshInfo, Command, CreateStreamingJobCommandInfo, CreateStreamingJobType,
57    ReplaceStreamJobPlan, Reschedule, ReschedulePlan, ResumeBackfillTarget, SinceEpochInfo,
58    SnapshotBackfillInfo,
59};
60pub(crate) use self::info::{SharedActorInfos, SharedFragmentInfo};
61pub use self::manager::{BarrierManagerRef, GlobalBarrierManager};
62pub(crate) use self::rpc::to_partial_graph_id;
63pub use self::schedule::BarrierScheduler;
64pub use self::trace::TracedEpoch;
65use crate::barrier::cdc_progress::CdcProgress;
66use crate::barrier::context::recovery::LoadedRecoveryContext;
67use crate::controller::fragment::InflightFragmentInfo;
68use crate::stream::cdc::CdcTableSnapshotSplits;
69
70/// The reason why the cluster is recovering.
71enum RecoveryReason {
72    /// After bootstrap.
73    Bootstrap,
74    /// After failure.
75    Failover(MetaError),
76    /// Manually triggered
77    Adhoc,
78}
79
80/// Status of barrier manager.
81enum BarrierManagerStatus {
82    /// Barrier manager is starting.
83    Starting,
84    /// Barrier manager is under recovery.
85    Recovering(RecoveryReason),
86    /// Barrier manager is running.
87    Running,
88}
89
90/// Scheduled command with its notifiers.
91struct Scheduled {
92    database_id: DatabaseId,
93    command: Command,
94    notifiers: Vec<Notifier>,
95    span: tracing::Span,
96}
97
98impl From<&BarrierManagerStatus> for PbRecoveryStatus {
99    fn from(status: &BarrierManagerStatus) -> Self {
100        match status {
101            BarrierManagerStatus::Starting => Self::StatusStarting,
102            BarrierManagerStatus::Recovering(reason) => match reason {
103                RecoveryReason::Bootstrap => Self::StatusStarting,
104                RecoveryReason::Failover(_) | RecoveryReason::Adhoc => Self::StatusRecovering,
105            },
106            BarrierManagerStatus::Running => Self::StatusRunning,
107        }
108    }
109}
110
111pub(crate) struct BackfillProgress {
112    pub(crate) progress: String,
113    pub(crate) backfill_type: PbBackfillType,
114}
115
116#[derive(Debug, Clone, Copy)]
117pub(crate) struct FragmentBackfillProgress {
118    pub(crate) job_id: JobId,
119    pub(crate) fragment_id: FragmentId,
120    pub(crate) consumed_rows: u64,
121    pub(crate) done: bool,
122    pub(crate) upstream_type: BackfillUpstreamType,
123}
124
125pub(crate) struct UpdateDatabaseBarrierRequest {
126    pub database_id: DatabaseId,
127    pub barrier_interval_ms: Option<u32>,
128    pub checkpoint_frequency: Option<u64>,
129    pub sender: Sender<()>,
130}
131
132pub(crate) enum BarrierManagerRequest {
133    GetBackfillProgress(Sender<MetaResult<HashMap<JobId, BackfillProgress>>>),
134    GetFragmentBackfillProgress(Sender<MetaResult<Vec<FragmentBackfillProgress>>>),
135    GetCdcProgress(Sender<MetaResult<HashMap<JobId, CdcProgress>>>),
136    AdhocRecovery(Sender<()>),
137    UpdateDatabaseBarrier(UpdateDatabaseBarrierRequest),
138}
139
140#[derive(Debug)]
141struct BarrierWorkerRuntimeInfoSnapshot {
142    active_streaming_nodes: ActiveStreamingWorkerNodes,
143    recovery_context: LoadedRecoveryContext,
144    state_table_committed_epochs: HashMap<TableId, u64>,
145    /// `table_id` -> (`Vec<non-checkpoint epoch>`, checkpoint epoch)
146    state_table_log_epochs: HashMap<TableId, Vec<(Vec<u64>, u64)>>,
147    mv_depended_subscriptions: HashMap<TableId, HashMap<SubscriptionId, u64>>,
148    background_jobs: HashSet<JobId>,
149    hummock_version_stats: HummockVersionStats,
150    database_infos: Vec<Database>,
151    cdc_table_snapshot_splits: HashMap<JobId, CdcTableSnapshotSplits>,
152}
153
154impl BarrierWorkerRuntimeInfoSnapshot {
155    fn validate_database_info(
156        database_id: DatabaseId,
157        database_jobs: &HashMap<JobId, HashMap<FragmentId, InflightFragmentInfo>>,
158        active_streaming_nodes: &ActiveStreamingWorkerNodes,
159        stream_actors: &HashMap<ActorId, StreamActor>,
160        state_table_committed_epochs: &HashMap<TableId, u64>,
161    ) -> MetaResult<()> {
162        {
163            for fragment in database_jobs.values().flat_map(|job| job.values()) {
164                for (actor_id, actor) in &fragment.actors {
165                    if !active_streaming_nodes
166                        .current()
167                        .contains_key(&actor.worker_id)
168                    {
169                        return Err(anyhow!(
170                            "worker_id {} of actor {} do not exist",
171                            actor.worker_id,
172                            actor_id
173                        )
174                        .into());
175                    }
176                    if !stream_actors.contains_key(actor_id) {
177                        return Err(anyhow!("cannot find StreamActor of actor {}", actor_id).into());
178                    }
179                }
180                for state_table_id in &fragment.state_table_ids {
181                    if !state_table_committed_epochs.contains_key(state_table_id) {
182                        return Err(anyhow!(
183                            "state table {} is not registered to hummock",
184                            state_table_id
185                        )
186                        .into());
187                    }
188                }
189            }
190            for (job_id, fragments) in database_jobs {
191                let mut committed_epochs =
192                    InflightFragmentInfo::existing_table_ids(fragments.values()).map(|table_id| {
193                        (
194                            table_id,
195                            *state_table_committed_epochs
196                                .get(&table_id)
197                                .expect("checked exist"),
198                        )
199                    });
200                let (first_table, first_epoch) = committed_epochs.next().ok_or_else(|| {
201                    anyhow!(
202                        "job {} in database {} has no state table after recovery",
203                        job_id,
204                        database_id
205                    )
206                })?;
207                for (table_id, epoch) in committed_epochs {
208                    if epoch != first_epoch {
209                        return Err(anyhow!(
210                            "job {} in database {} has tables with different table ids. {}:{}, {}:{}",
211                            job_id,
212                            database_id,
213                            first_table,
214                            first_epoch,
215                            table_id,
216                            epoch
217                        )
218                        .into());
219                    }
220                }
221            }
222        }
223        Ok(())
224    }
225}
226
227#[derive(Debug)]
228struct DatabaseRuntimeInfoSnapshot {
229    recovery_context: LoadedRecoveryContext,
230    state_table_committed_epochs: HashMap<TableId, u64>,
231    /// `table_id` -> (`Vec<non-checkpoint epoch>`, checkpoint epoch)
232    state_table_log_epochs: HashMap<TableId, Vec<(Vec<u64>, u64)>>,
233    mv_depended_subscriptions: HashMap<TableId, HashMap<SubscriptionId, u64>>,
234    background_jobs: HashSet<JobId>,
235    cdc_table_snapshot_splits: HashMap<JobId, CdcTableSnapshotSplits>,
236}
risingwave_meta/barrier/mod.rs

risingwave_meta/barrier/
mod.rs
