diff --git a/src/storage/src/error.rs b/src/storage/src/error.rs index cba5e67fa7b..f4026410e4c 100644 --- a/src/storage/src/error.rs +++ b/src/storage/src/error.rs @@ -210,6 +210,12 @@ pub enum Error { message: String, backtrace: Backtrace, }, + + #[snafu(display("Region version not found in manifest, the region: {}", region_name))] + VersionNotFound { + region_name: String, + backtrace: Backtrace, + }, } pub type Result = std::result::Result; @@ -235,7 +241,8 @@ impl ErrorExt for Error { | Readline { .. } | InvalidParquetSchema { .. } | SequenceColumnNotFound { .. } - | WalDataCorrupted { .. } => StatusCode::Unexpected, + | WalDataCorrupted { .. } + | VersionNotFound { .. } => StatusCode::Unexpected, FlushIo { .. } | InitBackend { .. } diff --git a/src/storage/src/flush.rs b/src/storage/src/flush.rs index 86c20f08712..9a5d18ae8ad 100644 --- a/src/storage/src/flush.rs +++ b/src/storage/src/flush.rs @@ -4,8 +4,7 @@ use async_trait::async_trait; use common_telemetry::logging; use common_time::RangeMillis; use store_api::logstore::LogStore; -use store_api::manifest::Manifest; -use store_api::manifest::ManifestVersion; +use store_api::manifest::{Manifest, ManifestVersion, MetaAction}; use store_api::storage::SequenceNumber; use uuid::Uuid; @@ -200,18 +199,23 @@ impl FlushJob { async fn write_to_manifest(&self, file_metas: &[FileMeta]) -> Result { let edit = RegionEdit { - region_id: self.shared.id, region_version: self.shared.version_control.metadata().version, - flush_sequence: self.flush_sequence, + flushed_sequence: self.flush_sequence, files_to_add: file_metas.to_vec(), files_to_remove: Vec::default(), }; - logging::debug!("Write region edit: {:?} to manifest.", edit); - self.manifest - .update(RegionMetaActionList::with_action(RegionMetaAction::Edit( - edit, - ))) - .await + let prev_version = self.shared.version_control.current_manifest_version(); + + logging::debug!( + "Write region edit: {:?} to manifest, prev_version: {}.", + edit, + prev_version, + ); + + let mut action_list = RegionMetaActionList::with_action(RegionMetaAction::Edit(edit)); + action_list.set_prev_version(prev_version); + + self.manifest.update(action_list).await } /// Generates random SST file name in format: `^[a-f\d]{8}(-[a-f\d]{4}){3}-[a-f\d]{12}.parquet$` diff --git a/src/storage/src/manifest.rs b/src/storage/src/manifest.rs index 0db6375124a..3c9ae3cabdc 100644 --- a/src/storage/src/manifest.rs +++ b/src/storage/src/manifest.rs @@ -3,3 +3,5 @@ pub(crate) mod action; pub(crate) mod checkpoint; pub mod region; pub(crate) mod storage; +#[cfg(test)] +pub mod test_utils; diff --git a/src/storage/src/manifest/action.rs b/src/storage/src/manifest/action.rs index e278bd1b101..8b356afb281 100644 --- a/src/storage/src/manifest/action.rs +++ b/src/storage/src/manifest/action.rs @@ -8,7 +8,6 @@ use store_api::manifest::action::ProtocolAction; use store_api::manifest::action::ProtocolVersion; use store_api::manifest::ManifestVersion; use store_api::manifest::MetaAction; -use store_api::manifest::Metadata; use store_api::storage::RegionId; use store_api::storage::SequenceNumber; @@ -31,20 +30,12 @@ pub struct RegionRemove { #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] pub struct RegionEdit { - pub region_id: RegionId, pub region_version: VersionNumber, - pub flush_sequence: SequenceNumber, + pub flushed_sequence: SequenceNumber, pub files_to_add: Vec, pub files_to_remove: Vec, } -#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] -pub struct RegionManifestData { - pub region_meta: RegionMetadataRef, - // TODO(dennis): [open_region] 1. load version metadata 2. The `region_meta` field could be removed if we - // have a `version` field. -} - #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] pub struct RegionMetaActionList { pub actions: Vec, @@ -157,8 +148,6 @@ impl RegionMetaActionList { } } -impl Metadata for RegionManifestData {} - impl MetaAction for RegionMetaActionList { fn set_prev_version(&mut self, version: ManifestVersion) { self.prev_version = version; @@ -170,6 +159,7 @@ mod tests { use common_telemetry::logging; use super::*; + use crate::manifest::test_utils; #[test] fn test_encode_decode_action_list() { @@ -178,32 +168,18 @@ mod tests { protocol.min_reader_version = 1; let mut action_list = RegionMetaActionList::new(vec![ RegionMetaAction::Protocol(protocol.clone()), - RegionMetaAction::Edit(RegionEdit { - region_id: 1, - region_version: 10, - flush_sequence: 99, - files_to_add: vec![ - FileMeta { - file_name: "test1".to_string(), - level: 1, - }, - FileMeta { - file_name: "test2".to_string(), - level: 2, - }, - ], - files_to_remove: vec![FileMeta { - file_name: "test0".to_string(), - level: 0, - }], - }), + RegionMetaAction::Edit(test_utils::build_region_edit( + 99, + &["test1", "test2"], + &["test3"], + )), ]); action_list.set_prev_version(3); let bs = action_list.encode().unwrap(); // {"prev_version":3} // {"Protocol":{"min_reader_version":1,"min_writer_version":0}} - // {"Edit":{"region_id":1,"region_version":10,"flush_sequence":99,"files_to_add":[{"file_name":"test1","level":1},{"file_name":"test2","level":2}],"files_to_remove":[{"file_name":"test0","level":0}]}} + // {"Edit":{"region_version":0,"flush_sequence":99,"files_to_add":[{"file_name":"test1","level":1},{"file_name":"test2","level":2}],"files_to_remove":[{"file_name":"test0","level":0}]}} logging::debug!( "Encoded action list: \r\n{}", diff --git a/src/storage/src/manifest/region.rs b/src/storage/src/manifest/region.rs index 559f2077c16..e3f83338f2e 100644 --- a/src/storage/src/manifest/region.rs +++ b/src/storage/src/manifest/region.rs @@ -28,48 +28,38 @@ impl RegionManifest { inner: Arc::new(RegionManifestInner::new(manifest_dir, object_store)), } } + + /// Update inner state. + pub fn update_state(&self, version: ManifestVersion, protocol: Option) { + self.inner.update_state(version, protocol); + } } #[async_trait] impl Manifest for RegionManifest { type Error = Error; type MetaAction = RegionMetaActionList; - type Metadata = RegionManifestData; + type MetaActionIterator = RegionMetaActionListIterator; async fn update(&self, action_list: RegionMetaActionList) -> Result { self.inner.save(action_list).await } - async fn load(&self) -> Result> { - let last_version = self.inner.last_version(); - - let start_bound = if last_version == MIN_VERSION { - // No actions have ever saved - MIN_VERSION - } else { - last_version - 1 - }; - - let mut iter = self.inner.scan(start_bound, MAX_VERSION).await?; - - // TODO(yingwen): [open_region] 1. Create Version from metadata 2. Load VersionEdits - // and apply to the Version by `Version::apply_edit`. - while let Some((_v, action_list)) = iter.next_action().await? { - for action in action_list.actions { - if let RegionMetaAction::Change(c) = action { - return Ok(Some(RegionManifestData { - region_meta: c.metadata, - })); - } - } - } - - Ok(None) + async fn scan( + &self, + start: ManifestVersion, + end: ManifestVersion, + ) -> Result { + self.inner.scan(start, end).await } async fn checkpoint(&self) -> Result { unimplemented!(); } + + fn last_version(&self) -> ManifestVersion { + self.inner.last_version() + } } #[derive(Debug)] @@ -83,18 +73,33 @@ struct RegionManifestInner { supported_writer_version: ProtocolVersion, } -struct RegionMetaActionListIterator { +pub struct RegionMetaActionListIterator { log_iter: ObjectStoreLogIterator, reader_version: ProtocolVersion, + last_protocol: Option, } impl RegionMetaActionListIterator { + pub fn last_protocol(&self) -> &Option { + &self.last_protocol + } +} + +#[async_trait] +impl MetaActionIterator for RegionMetaActionListIterator { + type Error = Error; + type MetaAction = RegionMetaActionList; + async fn next_action(&mut self) -> Result> { match self.log_iter.next_log().await? { Some((v, bytes)) => { - //TODO(dennis): save protocol into inner's protocol when recovering - let (action_list, _protocol) = + let (action_list, protocol) = RegionMetaActionList::decode(&bytes, self.reader_version)?; + + if protocol.is_some() { + self.last_protocol = protocol; + } + Ok(Some((v, action_list))) } None => Ok(None), @@ -108,7 +113,6 @@ impl RegionManifestInner { Self { store: Arc::new(ManifestObjectStore::new(manifest_dir, object_store)), - // TODO(dennis): recover the last version from history version: AtomicU64::new(0), protocol: ArcSwap::new(Arc::new(ProtocolAction::new())), supported_reader_version: reader_version, @@ -121,6 +125,13 @@ impl RegionManifestInner { self.version.fetch_add(1, Ordering::Relaxed) } + fn update_state(&self, version: ManifestVersion, protocol: Option) { + self.version.store(version, Ordering::Relaxed); + if let Some(p) = protocol { + self.protocol.store(Arc::new(p)); + } + } + #[inline] fn last_version(&self) -> ManifestVersion { self.version.load(Ordering::Relaxed) @@ -158,19 +169,18 @@ impl RegionManifestInner { Ok(RegionMetaActionListIterator { log_iter: self.store.scan(start, end).await?, reader_version: self.supported_reader_version, + last_protocol: None, }) } } #[cfg(test)] mod tests { - use datatypes::type_id::LogicalTypeId; use object_store::{backend::fs, ObjectStore}; use tempdir::TempDir; use super::*; - use crate::metadata::RegionMetadata; - use crate::test_util::descriptor_util::RegionDescBuilder; + use crate::manifest::test_utils::*; #[tokio::test] async fn test_region_manifest() { @@ -186,16 +196,16 @@ mod tests { let manifest = RegionManifest::new("/manifest/", object_store); - let region_name = "region-0"; - let desc = RegionDescBuilder::new(region_name) - .id(0) - .push_key_column(("k1", LogicalTypeId::Int32, false)) - .push_value_column(("v1", LogicalTypeId::Float32, true)) - .build(); - let metadata: RegionMetadata = desc.try_into().unwrap(); - let region_meta = Arc::new(metadata); + let region_meta = Arc::new(build_region_meta()); - assert!(manifest.load().await.unwrap().is_none()); + assert!(manifest + .scan(0, MAX_VERSION) + .await + .unwrap() + .next_action() + .await + .unwrap() + .is_none()); manifest .update(RegionMetaActionList::with_action(RegionMetaAction::Change( @@ -206,30 +216,48 @@ mod tests { .await .unwrap(); - let manifest_data = manifest.load().await.unwrap().unwrap(); - assert_eq!(manifest_data.region_meta, region_meta); + let mut iter = manifest.scan(0, MAX_VERSION).await.unwrap(); - // save another metadata - let region_name = "region-0"; - let desc = RegionDescBuilder::new(region_name) - .id(0) - .push_key_column(("k1", LogicalTypeId::Int32, false)) - .push_key_column(("k2", LogicalTypeId::Int64, false)) - .push_value_column(("v1", LogicalTypeId::Float32, true)) - .push_value_column(("bool", LogicalTypeId::Boolean, true)) - .build(); - let metadata: RegionMetadata = desc.try_into().unwrap(); - let region_meta = Arc::new(metadata); + let (v, action_list) = iter.next_action().await.unwrap().unwrap(); + assert_eq!(0, v); + assert_eq!(1, action_list.actions.len()); + let action = &action_list.actions[0]; + + match action { + RegionMetaAction::Change(c) => { + assert_eq!(c.metadata, region_meta); + } + _ => unreachable!(), + } + + // Save some actions manifest - .update(RegionMetaActionList::with_action(RegionMetaAction::Change( - RegionChange { - metadata: region_meta.clone(), - }, - ))) + .update(RegionMetaActionList::new(vec![ + RegionMetaAction::Edit(build_region_edit(1, &["f1"], &[])), + RegionMetaAction::Edit(build_region_edit(2, &["f2", "f3"], &[])), + ])) .await .unwrap(); - let manifest_data = manifest.load().await.unwrap().unwrap(); - assert_eq!(manifest_data.region_meta, region_meta); + let mut iter = manifest.scan(0, MAX_VERSION).await.unwrap(); + let (v, action_list) = iter.next_action().await.unwrap().unwrap(); + assert_eq!(0, v); + assert_eq!(1, action_list.actions.len()); + let action = &action_list.actions[0]; + match action { + RegionMetaAction::Change(c) => { + assert_eq!(c.metadata, region_meta); + } + _ => unreachable!(), + } + + let (v, action_list) = iter.next_action().await.unwrap().unwrap(); + assert_eq!(1, v); + assert_eq!(2, action_list.actions.len()); + assert!(matches!(&action_list.actions[0], RegionMetaAction::Edit(_))); + assert!(matches!(&action_list.actions[1], RegionMetaAction::Edit(_))); + + // Reach end + assert!(iter.next_action().await.unwrap().is_none()); } } diff --git a/src/storage/src/manifest/test_utils.rs b/src/storage/src/manifest/test_utils.rs new file mode 100644 index 00000000000..154165c7ff5 --- /dev/null +++ b/src/storage/src/manifest/test_utils.rs @@ -0,0 +1,42 @@ +use datatypes::type_id::LogicalTypeId; +use store_api::storage::SequenceNumber; + +use crate::manifest::action::*; +use crate::metadata::RegionMetadata; +use crate::sst::FileMeta; +use crate::test_util::descriptor_util::RegionDescBuilder; + +pub fn build_region_meta() -> RegionMetadata { + let region_name = "region-0"; + let desc = RegionDescBuilder::new(region_name) + .id(0) + .push_key_column(("k1", LogicalTypeId::Int32, false)) + .push_value_column(("v1", LogicalTypeId::Float32, true)) + .build(); + desc.try_into().unwrap() +} + +pub fn build_region_edit( + sequence: SequenceNumber, + files_to_add: &[&str], + files_to_remove: &[&str], +) -> RegionEdit { + RegionEdit { + region_version: 0, + flushed_sequence: sequence, + files_to_add: files_to_add + .iter() + .map(|f| FileMeta { + file_name: f.to_string(), + level: 0, + }) + .collect(), + files_to_remove: files_to_remove + .iter() + .map(|f| FileMeta { + file_name: f.to_string(), + level: 0, + }) + .collect(), + } +} diff --git a/src/storage/src/region.rs b/src/storage/src/region.rs index dee349cec8d..75db4473a44 100644 --- a/src/storage/src/region.rs +++ b/src/storage/src/region.rs @@ -6,21 +6,22 @@ use std::sync::Arc; use async_trait::async_trait; use datatypes::schema::SchemaRef; -use snafu::ensure; +use snafu::{ensure, OptionExt}; use store_api::logstore::LogStore; -use store_api::manifest::Manifest; +use store_api::manifest::{self, Manifest, ManifestVersion, MetaActionIterator}; use store_api::storage::{ OpenOptions, ReadContext, Region, RegionId, RegionMeta, WriteContext, WriteResponse, }; use crate::error::{self, Error, Result}; use crate::flush::{FlushSchedulerRef, FlushStrategyRef}; -use crate::manifest::region::RegionManifest; +use crate::manifest::{action::RegionMetaAction, region::RegionManifest}; use crate::memtable::MemtableBuilderRef; use crate::metadata::{RegionMetaImpl, RegionMetadata}; pub use crate::region::writer::{RegionWriter, RegionWriterRef, WriterContext}; use crate::snapshot::SnapshotImpl; use crate::sst::AccessLayerRef; +use crate::version::VersionEdit; use crate::version::{Version, VersionControl, VersionControlRef}; use crate::wal::Wal; use crate::write_batch::WriteBatch; @@ -115,7 +116,7 @@ impl RegionImpl { opts: &OpenOptions, ) -> Result> { // Load version meta data from manifest. - let version = Self::recover_from_manifest(&store_config.manifest).await?; + let version = Self::recover_from_manifest(&name, &store_config.manifest).await?; let metadata = version.metadata().clone(); let version_control = Arc::new(VersionControl::with_version(version)); let wal = Wal::new(name.clone(), store_config.log_store); @@ -140,11 +141,76 @@ impl RegionImpl { unimplemented!() } - async fn recover_from_manifest(manifest: &RegionManifest) -> Result { - let _metadata = manifest.load().await?; + async fn recover_from_manifest( + region_name: &str, + manifest: &RegionManifest, + ) -> Result { + let (start, end) = Self::manifest_scan_range(); + let mut iter = manifest.scan(start, end).await?; - // TODO(yingwen): [open_region] Get version from metadata. - unimplemented!() + let mut version = None; + let mut actions = Vec::new(); + let mut last_manifest_version = manifest::MIN_VERSION; + while let Some((manifest_version, action_list)) = iter.next_action().await? { + last_manifest_version = manifest_version; + + for action in action_list.actions { + match (action, version) { + (RegionMetaAction::Change(c), None) => { + version = Some(Version::new(c.metadata)); + for (manifest_version, action) in actions.drain(..) { + version = Self::replay_edit(manifest_version, action, version); + } + } + (RegionMetaAction::Change(_), Some(_)) => { + unimplemented!("alter schema is not implemented") + } + (action, None) => { + actions.push((manifest_version, action)); + version = None; + } + (action, Some(v)) => { + version = Self::replay_edit(manifest_version, action, Some(v)); + } + } + } + } + + assert!(actions.is_empty() || version.is_none()); + + if version.is_some() { + // update manifest state after recovering + let protocol = iter.last_protocol(); + manifest.update_state(last_manifest_version + 1, protocol.clone()); + } + + version.context(error::VersionNotFoundSnafu { region_name }) + } + + fn manifest_scan_range() -> (ManifestVersion, ManifestVersion) { + // TODO(dennis): use manifest version in WAL + (manifest::MIN_VERSION, manifest::MAX_VERSION) + } + + fn replay_edit( + manifest_version: ManifestVersion, + action: RegionMetaAction, + version: Option, + ) -> Option { + if let RegionMetaAction::Edit(e) = action { + let edit = VersionEdit { + files_to_add: e.files_to_add, + flushed_sequence: Some(e.flushed_sequence), + manifest_version, + max_memtable_id: None, + }; + version.map(|mut v| { + v.apply_edit(edit); + v + }) + } else { + version + } } } diff --git a/src/storage/src/region/tests.rs b/src/storage/src/region/tests.rs index 6521cf88882..4de85dc0cb5 100644 --- a/src/storage/src/region/tests.rs +++ b/src/storage/src/region/tests.rs @@ -4,10 +4,15 @@ mod flush; mod read_write; use datatypes::type_id::LogicalTypeId; +use log_store::fs::noop::NoopLogStore; +use object_store::{backend::fs, ObjectStore}; use store_api::storage::consts; use tempdir::TempDir; use super::*; +use crate::manifest::action::RegionChange; +use crate::manifest::action::RegionMetaActionList; +use crate::manifest::test_utils::*; use crate::test_util::{self, config_util, descriptor_util::RegionDescBuilder, schema_util}; #[tokio::test] @@ -43,3 +48,65 @@ async fn test_new_region() { assert_eq!(region_name, region.name()); assert_eq!(expect_schema, *region.in_memory_metadata().schema()); } + +#[tokio::test] +async fn test_recover_region_manifets() { + let tmp_dir = TempDir::new("test_new_region").unwrap(); + + let object_store = ObjectStore::new( + fs::Backend::build() + .root(&tmp_dir.path().to_string_lossy()) + .finish() + .await + .unwrap(), + ); + + let manifest = RegionManifest::new("/manifest/", object_store); + let region_name = "region-0"; + let region_meta = Arc::new(build_region_meta()); + + // Recover from empty + assert!( + RegionImpl::::recover_from_manifest(region_name, &manifest) + .await + .is_err() + ); + + { + // save some actions into region_meta + manifest + .update(RegionMetaActionList::with_action(RegionMetaAction::Change( + RegionChange { + metadata: region_meta.clone(), + }, + ))) + .await + .unwrap(); + + manifest + .update(RegionMetaActionList::new(vec![ + RegionMetaAction::Edit(build_region_edit(1, &["f1"], &[])), + RegionMetaAction::Edit(build_region_edit(2, &["f2", "f3"], &[])), + ])) + .await + .unwrap(); + } + + // try to recover + let version = RegionImpl::::recover_from_manifest(region_name, &manifest) + .await + .unwrap(); + assert_eq!(*version.metadata(), region_meta); + assert_eq!(version.flushed_sequence(), 2); + assert_eq!(version.manifest_version(), 1); + let ssts = version.ssts(); + let files = ssts.levels()[0].files(); + assert_eq!(3, files.len()); + for (i, file) in files.iter().enumerate() { + assert_eq!(format!("f{}", i + 1), file.file_name()); + } + assert!(version.mutable_memtables().is_empty()); + + // check manifest state + assert_eq!(2, manifest.last_version()); +} diff --git a/src/storage/src/sst.rs b/src/storage/src/sst.rs index 673a81da23a..dc306b2d2cc 100644 --- a/src/storage/src/sst.rs +++ b/src/storage/src/sst.rs @@ -70,6 +70,11 @@ impl LevelMetas { Ok(()) } + + #[cfg(test)] + pub fn levels(&self) -> &[LevelMeta] { + &self.levels + } } impl Default for LevelMetas { @@ -96,6 +101,11 @@ impl LevelMeta { fn visit_level(&self, visitor: &mut V) -> Result<()> { visitor.visit(self.level.into(), &self.files) } + + #[cfg(test)] + pub fn files(&self) -> &[FileHandle] { + &self.files + } } fn new_level_meta_vec() -> LevelMetaVec { diff --git a/src/storage/src/version.rs b/src/storage/src/version.rs index ab44f28c75f..50f1ebe01e5 100644 --- a/src/storage/src/version.rs +++ b/src/storage/src/version.rs @@ -39,7 +39,7 @@ impl VersionControl { /// Construct a new version control from `metadata`. pub fn new(metadata: RegionMetadata) -> VersionControl { VersionControl { - version: CowCell::new(Version::new(metadata)), + version: CowCell::new(Version::new(Arc::new(metadata))), committed_sequence: AtomicU64::new(0), } } @@ -58,6 +58,11 @@ impl VersionControl { self.version.get() } + #[inline] + pub fn current_manifest_version(&self) -> ManifestVersion { + self.current().manifest_version + } + /// Metadata of current version. pub fn metadata(&self) -> RegionMetadataRef { let version = self.current(); @@ -147,9 +152,9 @@ pub struct Version { } impl Version { - pub fn new(metadata: RegionMetadata) -> Version { + pub fn new(metadata: RegionMetadataRef) -> Version { Version { - metadata: Arc::new(metadata), + metadata, memtables: Arc::new(MemtableVersion::new()), ssts: Arc::new(LevelMetas::new()), flushed_sequence: 0, diff --git a/src/store-api/src/manifest.rs b/src/store-api/src/manifest.rs index 4ca31d2c9c9..095f2a3f912 100644 --- a/src/store-api/src/manifest.rs +++ b/src/store-api/src/manifest.rs @@ -12,24 +12,38 @@ pub type ManifestVersion = u64; pub const MIN_VERSION: u64 = 0; pub const MAX_VERSION: u64 = u64::MAX; -pub trait Metadata: Clone {} - pub trait MetaAction: Serialize + DeserializeOwned { fn set_prev_version(&mut self, version: ManifestVersion); } +#[async_trait] +pub trait MetaActionIterator { + type MetaAction: MetaAction; + type Error: ErrorExt + Send + Sync; + + async fn next_action( + &mut self, + ) -> Result, Self::Error>; +} + /// Manifest service #[async_trait] pub trait Manifest: Send + Sync + Clone + 'static { type Error: ErrorExt + Send + Sync; type MetaAction: MetaAction; - type Metadata: Metadata; + type MetaActionIterator: MetaActionIterator; /// Update metadata by the action async fn update(&self, action: Self::MetaAction) -> Result; - /// Retrieve the latest metadata - async fn load(&self) -> Result, Self::Error>; + /// Scan actions which version in range [start, end) + async fn scan( + &self, + start: ManifestVersion, + end: ManifestVersion, + ) -> Result; async fn checkpoint(&self) -> Result; + + fn last_version(&self) -> ManifestVersion; }