mito2/region/
catchup.rs

1// Copyright 2023 Greptime Team
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::sync::Arc;
16use std::time::Instant;
17
18use common_telemetry::{info, warn};
19use snafu::ensure;
20use store_api::logstore::LogStore;
21
22use crate::error::{self, Result};
23use crate::region::MitoRegion;
24use crate::region::opener::{check_wal_history, replay_memtable};
25use crate::wal::Wal;
26use crate::wal::entry_distributor::WalEntryReceiver;
27
28pub struct RegionCatchupTask<S> {
29    entry_receiver: Option<WalEntryReceiver>,
30    region: Arc<MitoRegion>,
31    replay_checkpoint_entry_id: Option<u64>,
32    expected_last_entry_id: Option<u64>,
33    allow_stale_entries: bool,
34    wal: Wal<S>,
35}
36
37impl<S: LogStore> RegionCatchupTask<S> {
38    pub fn new(region: Arc<MitoRegion>, wal: Wal<S>, allow_stale_entries: bool) -> Self {
39        Self {
40            entry_receiver: None,
41            region,
42            replay_checkpoint_entry_id: None,
43            expected_last_entry_id: None,
44            allow_stale_entries,
45            wal,
46        }
47    }
48
49    /// Sets the expected last entry id.
50    pub(crate) fn with_expected_last_entry_id(
51        mut self,
52        expected_last_entry_id: Option<u64>,
53    ) -> Self {
54        self.expected_last_entry_id = expected_last_entry_id;
55        self
56    }
57
58    /// Sets the entry receiver.
59    pub(crate) fn with_entry_receiver(mut self, entry_receiver: Option<WalEntryReceiver>) -> Self {
60        self.entry_receiver = entry_receiver;
61        self
62    }
63
64    /// Sets the replay checkpoint entry id.
65    pub(crate) fn with_replay_checkpoint_entry_id(
66        mut self,
67        replay_checkpoint_entry_id: Option<u64>,
68    ) -> Self {
69        self.replay_checkpoint_entry_id = replay_checkpoint_entry_id;
70        self
71    }
72
73    pub async fn run(&mut self) -> Result<()> {
74        if self.region.provider.is_remote_wal() {
75            self.remote_wal_catchup().await
76        } else {
77            self.local_wal_catchup().await
78        }
79    }
80
81    async fn remote_wal_catchup(&mut self) -> Result<()> {
82        let flushed_entry_id = self.region.version_control.current().last_entry_id;
83        let replay_from_entry_id = self
84            .replay_checkpoint_entry_id
85            .unwrap_or(flushed_entry_id)
86            .max(flushed_entry_id);
87        let region_id = self.region.region_id;
88        info!(
89            "Trying to replay memtable for region: {region_id}, provider: {:?}, replay from entry id: {replay_from_entry_id}, flushed entry id: {flushed_entry_id}",
90            self.region.provider
91        );
92        check_wal_history(
93            &self.wal,
94            &self.region.provider,
95            &self.region.version_control.current().version,
96        )
97        .await?;
98        let timer = Instant::now();
99        let wal_entry_reader = self
100            .entry_receiver
101            .take()
102            .map(|r| Box::new(r) as _)
103            .unwrap_or_else(|| self.wal.wal_entry_reader(&self.region.provider, region_id));
104        let on_region_opened = self.wal.on_region_opened();
105        let last_entry_id = replay_memtable(
106            &self.region.provider,
107            wal_entry_reader,
108            region_id,
109            replay_from_entry_id,
110            &self.region.version_control,
111            self.allow_stale_entries,
112            on_region_opened,
113        )
114        .await?;
115        info!(
116            "Elapsed: {:?}, region: {region_id}, provider: {:?} catchup finished. replay from entry id: {replay_from_entry_id}, flushed entry id: {flushed_entry_id}, last entry id: {last_entry_id}, expected: {:?}.",
117            timer.elapsed(),
118            self.region.provider,
119            self.expected_last_entry_id
120        );
121        if let Some(expected_last_entry_id) = self.expected_last_entry_id {
122            ensure!(
123                // The replayed last entry id may be greater than the `expected_last_entry_id`.
124                last_entry_id >= expected_last_entry_id,
125                error::UnexpectedSnafu {
126                    reason: format!(
127                        "Failed to catchup region {}, it was expected to replay to {}, but actually replayed to {}",
128                        region_id, expected_last_entry_id, last_entry_id,
129                    ),
130                }
131            )
132        }
133        Ok(())
134    }
135
136    async fn local_wal_catchup(&mut self) -> Result<()> {
137        let version = self.region.version_control.current();
138        let mut flushed_entry_id = version.last_entry_id;
139        let region_id = self.region.region_id;
140        let latest_entry_id = self
141            .wal
142            .store()
143            .latest_entry_id(&self.region.provider)
144            .unwrap_or_default();
145        info!(
146            "Skips to replay memtable for region: {}, flushed entry id: {}, latest entry id: {}",
147            region_id, flushed_entry_id, latest_entry_id
148        );
149
150        if latest_entry_id > flushed_entry_id {
151            warn!(
152                "Found latest entry id is greater than flushed entry id, using latest entry id as flushed entry id, region: {}, latest entry id: {}, flushed entry id: {}",
153                region_id, latest_entry_id, flushed_entry_id
154            );
155            flushed_entry_id = latest_entry_id;
156            self.region.version_control.set_entry_id(flushed_entry_id);
157        }
158        let on_region_opened = self.wal.on_region_opened();
159        on_region_opened(region_id, flushed_entry_id, &self.region.provider).await?;
160        Ok(())
161    }
162}
mito2/region/catchup.rs

mito2/region/
catchup.rs
