refactor(metactl-import): when importing, also write the data version (#18080)

Before this commit, when importing, the data version (such as `V004`)
is not written to disk, and the data version is assumed to be the
oldest compatible version, and the on-disk data then will be upgraded to the
latest version. For example, when `databend-metactl` imports `V004`
data, the data on disk is left as `V002`, and then the data is upgraded
from `V002` to `V003` and finally to `V004`.

This is an unnecessary burden, and the upgrading process invokes the legacy
`sled-db` based storage, which leads to problems on NFS (`device busy`
error when deleting sled from disk after closing it).

This commit refines this behavior by directly writing the data
version such as `V004` to disk after importing. Thus there won't be an
unnecessary burden of upgrading the data, and it gets rid of the
necessity to access sled DB, which is a workaround for running on NFS.

Author: drmingdrmer

src/meta/control/src/import_v004.rs

@@ -19,6 +19,8 @@ use std::sync::Mutex;
 use databend_common_meta_raft_store::config::RaftConfig;
 use databend_common_meta_raft_store::key_spaces::RaftStoreEntry;
 use databend_common_meta_raft_store::key_spaces::SMEntry;
+use databend_common_meta_raft_store::ondisk::DataVersion;
+use databend_common_meta_raft_store::ondisk::Header;
 use databend_common_meta_raft_store::ondisk::OnDisk;
 use databend_common_meta_raft_store::raft_log_v004;
 use databend_common_meta_raft_store::raft_log_v004::RaftLogV004;
@@ -38,6 +40,8 @@ pub async fn import_v004(
     raft_config: RaftConfig,
     lines: impl IntoIterator<Item = Result<String, io::Error>>,
 ) -> anyhow::Result<Option<LogId>> {
+    let data_version = DataVersion::V004;
+
     OnDisk::ensure_dirs(&raft_config.raft_dir)?;
 
     let mut n = 0;
@@ -50,7 +54,7 @@ pub async fn import_v004(
         raft_log_v004::Importer::new(raft_log)
     };
 
-    let snapshot_store = SnapshotStoreV004::new(raft_config);
+    let snapshot_store = SnapshotStoreV004::new(raft_config.clone());
     let writer = snapshot_store.new_writer()?;
     let (tx, join_handle) = writer.spawn_writer_thread("import_v004");
 
@@ -104,11 +108,24 @@ pub async fn import_v004(
     let db = temp_snapshot_data.move_to_final_path(snapshot_id.to_string())?;
 
     eprintln!(
-        "Imported {} records, snapshot: {}; snapshot_path: {}; snapshot_stat: {}",
+        "{data_version}: Imported {} records, snapshot: {}; snapshot_path: {}; snapshot_stat: {}",
         n,
         snapshot_id,
         db.path(),
         db.stat()
     );
+
+    let on_disk = OnDisk::new(
+        Header {
+            version: DataVersion::V004,
+            upgrading: None,
+            cleaning: false,
+        },
+        &raft_config,
+    );
+
+    on_disk.write_header(&on_disk.header)?;
+    eprintln!("{data_version}: written on-disk header: {}", on_disk.header);
+
     Ok(max_log_id)
 }

src/meta/raft-store/src/ondisk/mod.rs

@@ -121,7 +121,7 @@ impl OnDisk {
         Ok(OnDisk::new(header, config))
     }
 
-    fn new(header: Header, config: &RaftConfig) -> Self {
+    pub fn new(header: Header, config: &RaftConfig) -> Self {
         let min_compatible = DATA_VERSION.min_compatible_data_version();
 
         if header.version < min_compatible {
@@ -358,7 +358,7 @@ impl OnDisk {
         Ok(())
     }
 
-    fn write_header(&self, header: &Header) -> Result<(), MetaStorageError> {
+    pub fn write_header(&self, header: &Header) -> Result<(), MetaStorageError> {
         Self::write_header_to_fs(&self.config, header)?;
         Ok(())
     }