supports checkpoint (#721)

* support checkpoint

Signed-off-by: nolouch <[email protected]>

* suppport titan checkpoint

Signed-off-by: nolouch <[email protected]>

* address comments

Signed-off-by: nolouch <[email protected]>

* address comments

Signed-off-by: nolouch <[email protected]>

Signed-off-by: nolouch <[email protected]>

Author: nolouch

librocksdb_sys/crocksdb/c.cc

@@ -48,6 +48,7 @@
 #include "rocksdb/types.h"
 #include "rocksdb/universal_compaction.h"
 #include "rocksdb/utilities/backupable_db.h"
+#include "rocksdb/utilities/checkpoint.h"
 #include "rocksdb/utilities/db_ttl.h"
 #include "rocksdb/utilities/debug.h"
 #include "rocksdb/utilities/options_util.h"
@@ -57,6 +58,7 @@
 #include "table/block_based/block_based_table_factory.h"
 #include "table/sst_file_writer_collectors.h"
 #include "table/table_reader.h"
+#include "titan/checkpoint.h"
 #include "titan/db.h"
 #include "titan/options.h"
 #include "util/coding.h"
@@ -77,6 +79,7 @@ using rocksdb::BackupInfo;
 using rocksdb::BlockBasedTableOptions;
 using rocksdb::BlockCipher;
 using rocksdb::Cache;
+using rocksdb::Checkpoint;
 using rocksdb::ColumnFamilyDescriptor;
 using rocksdb::ColumnFamilyHandle;
 using rocksdb::ColumnFamilyOptions;
@@ -171,6 +174,7 @@ using rocksdb::ExternalSstFilePropertyNames;
 using rocksdb::IOStatsContext;
 using rocksdb::LDBTool;
 using rocksdb::LevelMetaData;
+using rocksdb::MemoryAllocator;
 using rocksdb::PerfContext;
 using rocksdb::PerfLevel;
 using rocksdb::PutFixed64;
@@ -193,8 +197,7 @@ using rocksdb::titandb::TitanDB;
 using rocksdb::titandb::TitanDBOptions;
 using rocksdb::titandb::TitanOptions;
 using rocksdb::titandb::TitanReadOptions;
-
-using rocksdb::MemoryAllocator;
+using TitanCheckpoint = rocksdb::titandb::Checkpoint;
 
 #ifdef OPENSSL
 using rocksdb::encryption::EncryptionMethod;
@@ -223,6 +226,9 @@ struct crocksdb_backup_engine_t {
 struct crocksdb_backup_engine_info_t {
   std::vector<BackupInfo> rep;
 };
+struct crocksdb_checkpoint_t {
+  Checkpoint* rep;
+};
 struct crocksdb_restore_options_t {
   RestoreOptions rep;
 };
@@ -756,6 +762,29 @@ void crocksdb_resume(crocksdb_t* db, char** errptr) {
   SaveError(errptr, db->rep->Resume());
 }
 
+crocksdb_checkpoint_t* crocksdb_checkpoint_object_create(crocksdb_t* db,
+                                                         char** errptr) {
+  Checkpoint* checkpoint;
+  if (SaveError(errptr, Checkpoint::Create(db->rep, &checkpoint))) {
+    return nullptr;
+  }
+  crocksdb_checkpoint_t* result = new crocksdb_checkpoint_t;
+  result->rep = checkpoint;
+  return result;
+}
+
+void crocksdb_checkpoint_create(crocksdb_checkpoint_t* checkpoint,
+                                const char* checkpoint_dir,
+                                uint64_t log_size_for_flush, char** errptr) {
+  SaveError(errptr, checkpoint->rep->CreateCheckpoint(
+                        std::string(checkpoint_dir), log_size_for_flush));
+}
+
+void crocksdb_checkpoint_object_destroy(crocksdb_checkpoint_t* checkpoint) {
+  delete checkpoint->rep;
+  delete checkpoint;
+}
+
 crocksdb_backup_engine_t* crocksdb_backup_engine_open(
     const crocksdb_options_t* options, const char* path, char** errptr) {
   BackupEngine* be;
@@ -6383,10 +6412,40 @@ void crocksdb_run_sst_dump_tool(int argc, char** argv,
 }
 
 /* Titan */
+struct ctitandb_checkpoint_t {
+  TitanCheckpoint* rep;
+};
+
 struct ctitandb_options_t {
   TitanOptions rep;
 };
 
+ctitandb_checkpoint_t* ctitandb_checkpoint_object_create(crocksdb_t* db,
+                                                         char** errptr) {
+  TitanCheckpoint* checkpoint;
+  if (SaveError(errptr, TitanCheckpoint::Create(static_cast<TitanDB*>(db->rep),
+                                                &checkpoint))) {
+    return nullptr;
+  }
+  ctitandb_checkpoint_t* result = new ctitandb_checkpoint_t;
+  result->rep = checkpoint;
+  return result;
+}
+
+void ctitandb_checkpoint_create(ctitandb_checkpoint_t* checkpoint,
+                                const char* basedb_checkpoint_dir,
+                                const char* titan_checkpoint_dir,
+                                uint64_t log_size_for_flush, char** errptr) {
+  SaveError(errptr, checkpoint->rep->CreateCheckpoint(
+                        std::string(basedb_checkpoint_dir),
+                        std::string(titan_checkpoint_dir), log_size_for_flush));
+}
+
+void ctitandb_checkpoint_object_destroy(ctitandb_checkpoint_t* checkpoint) {
+  delete checkpoint->rep;
+  delete checkpoint;
+}
+
 crocksdb_t* ctitandb_open_column_families(
     const char* name, const ctitandb_options_t* tdb_options,
     int num_column_families, const char** column_family_names,

librocksdb_sys/crocksdb/crocksdb/c.h

@@ -80,6 +80,7 @@ typedef struct crocksdb_lru_cache_options_t crocksdb_lru_cache_options_t;
 typedef struct crocksdb_cache_t crocksdb_cache_t;
 typedef struct crocksdb_memory_allocator_t crocksdb_memory_allocator_t;
 typedef struct crocksdb_compactionfilter_t crocksdb_compactionfilter_t;
+typedef struct crocksdb_checkpoint_t crocksdb_checkpoint_t;
 enum {
   crocksdb_table_file_creation_reason_flush = 0,
   crocksdb_table_file_creation_reason_compaction = 1,
@@ -245,6 +246,16 @@ extern C_ROCKSDB_LIBRARY_API void crocksdb_status_ptr_get_error(
 
 extern C_ROCKSDB_LIBRARY_API void rocksdb_resume(crocksdb_t* db, char** errptr);
 
+extern C_ROCKSDB_LIBRARY_API crocksdb_checkpoint_t*
+crocksdb_checkpoint_object_create(crocksdb_t* db, char** errptr);
+
+extern C_ROCKSDB_LIBRARY_API void crocksdb_checkpoint_create(
+    crocksdb_checkpoint_t* checkpoint, const char* checkpoint_dir,
+    uint64_t log_size_for_flush, char** errptr);
+
+extern C_ROCKSDB_LIBRARY_API void crocksdb_checkpoint_object_destroy(
+    crocksdb_checkpoint_t* checkpoint);
+
 extern C_ROCKSDB_LIBRARY_API crocksdb_backup_engine_t*
 crocksdb_backup_engine_open(const crocksdb_options_t* options, const char* path,
                             char** errptr);
@@ -2533,6 +2544,7 @@ struct ctitandb_blob_index_t {
 typedef struct ctitandb_options_t ctitandb_options_t;
 typedef struct ctitandb_readoptions_t ctitandb_readoptions_t;
 typedef struct ctitandb_blob_index_t ctitandb_blob_index_t;
+typedef struct ctitandb_checkpoint_t ctitandb_checkpoint_t;
 
 extern C_ROCKSDB_LIBRARY_API crocksdb_t* ctitandb_open_column_families(
     const char* name, const ctitandb_options_t* tdb_options,
@@ -2545,6 +2557,17 @@ ctitandb_create_column_family(
     crocksdb_t* db, const ctitandb_options_t* titan_column_family_options,
     const char* column_family_name, char** errptr);
 
+extern C_ROCKSDB_LIBRARY_API ctitandb_checkpoint_t*
+ctitandb_checkpoint_object_create(crocksdb_t* db, char** errptr);
+
+extern C_ROCKSDB_LIBRARY_API void ctitandb_checkpoint_create(
+    ctitandb_checkpoint_t* checkpoint, const char* basedb_checkpoint_dir,
+    const char* titan_checkpoint_dir, uint64_t log_size_for_flush,
+    char** errptr);
+
+extern C_ROCKSDB_LIBRARY_API void ctitandb_checkpoint_object_destroy(
+    ctitandb_checkpoint_t* checkpoint);
+
 /* TitanDBOptions */
 
 extern C_ROCKSDB_LIBRARY_API ctitandb_options_t* ctitandb_options_create();

librocksdb_sys/src/lib.rs

@@ -85,7 +85,8 @@ pub struct DBCompactionFilter(c_void);
 pub struct DBCompactionFilterFactory(c_void);
 #[repr(C)]
 pub struct DBCompactionFilterContext(c_void);
-
+#[repr(C)]
+pub struct DBCheckpoint(c_void);
 #[repr(C)]
 pub struct EnvOptions(c_void);
 #[repr(C)]
@@ -2335,6 +2336,20 @@ extern "C" {
 
     pub fn crocksdb_keyversions_value(kvs: *mut DBKeyVersions, index: usize) -> *const c_char;
 
+    pub fn crocksdb_checkpoint_object_create(
+        db: *mut DBInstance,
+        errptr: *mut *mut c_char,
+    ) -> *mut DBCheckpoint;
+
+    pub fn crocksdb_checkpoint_create(
+        check_point: *mut DBCheckpoint,
+        check_point_dir: *const c_char,
+        log_size_for_flush: u64,
+        errptr: *mut *mut c_char,
+    );
+
+    pub fn crocksdb_checkpoint_object_destroy(check_point: *mut DBCheckpoint);
+
     pub fn crocksdb_keyversions_seq(kvs: *mut DBKeyVersions, index: usize) -> u64;
 
     pub fn crocksdb_keyversions_type(kvs: *mut DBKeyVersions, index: usize) -> c_int;
@@ -2813,6 +2828,21 @@ extern "C" {
         include_end: bool,
         errptr: *mut *mut c_char,
     );
+
+    pub fn ctitandb_checkpoint_object_create(
+        db: *mut DBInstance,
+        errptr: *mut *mut c_char,
+    ) -> *mut DBCheckpoint;
+
+    pub fn ctitandb_checkpoint_create(
+        checkpoint: *mut DBCheckpoint,
+        basedb_checkpoint_dir: *const c_char,
+        titan_checkpoint_dir: *const c_char,
+        log_size_for_flush: u64,
+        errptr: *mut *mut c_char,
+    );
+
+    pub fn ctitandb_checkpoint_object_destroy(check_point: *mut DBCheckpoint);
 }
 
 #[cfg(test)]

src/checkpoint.rs

@@ -0,0 +1,106 @@
+// Copyright 2022 TiKV Project Authors. Licensed under Apache-2.0.
+
+use std::ffi::CString;
+use std::path::Path;
+
+pub struct Checkpointer {
+    ptr: *mut librocksdb_sys::DBCheckpoint,
+    is_titan: bool,
+}
+
+impl Checkpointer {
+    /// Creates new checkpoint object for specific DB.
+    pub(crate) fn new(
+        db: *mut librocksdb_sys::DBInstance,
+        is_titan: bool,
+    ) -> Result<Checkpointer, String> {
+        let ptr = if is_titan {
+            unsafe { ffi_try!(ctitandb_checkpoint_object_create(db)) }
+        } else {
+            unsafe { ffi_try!(crocksdb_checkpoint_object_create(db)) }
+        };
+        Ok(Checkpointer { ptr, is_titan })
+    }
+    /// Creates new physical DB checkpoint in directory specified by `path`.
+    ///
+    /// Builds an openable snapshot of RocksDB on the same disk, which
+    /// accepts an output directory on the same disk, and under the directory
+    /// (1) hard-linked SST files pointing to existing live SST files
+    /// SST files will be copied if output directory is on a different filesystem
+    /// (2) a copied manifest files and other files
+    /// The directory should not already exist and will be created by this API.
+    /// The directory will be an absolute path
+    /// log_size_for_flush: if the total log file size is equal or larger than
+    /// this value, then a flush is triggered for all the column families. The
+    /// default value is 0, which means flush is always triggered. If you move
+    /// away from the default, the checkpoint may not contain up-to-date data
+    /// if WAL writing is not always enabled.
+    /// Flush will always trigger if it is 2PC.
+    ///
+    /// basedb_out_dir: the checkpoint path about rocksdb
+    /// titan_out_dir: the checkpoint path about titan's files. if titan_out_dir
+    /// is None, the path will be "{basedb_out_dir}/titandb".
+    pub fn create_at(
+        &mut self,
+        basedb_out_dir: &Path,
+        titan_out_dir: Option<&Path>,
+        log_size_for_flush: u64,
+    ) -> Result<(), String> {
+        let basedb_out_dir = match basedb_out_dir.to_str().and_then(|s| CString::new(s).ok()) {
+            Some(s) => s,
+            None => {
+                return Err(format!(
+                    "{} is not a valid directory",
+                    basedb_out_dir.display()
+                ))
+            }
+        };
+        let mut titan_out_dir_str = CString::new("").ok().unwrap();
+        if let Some(titan_out_dir) = titan_out_dir {
+            match titan_out_dir.to_str().and_then(|s| CString::new(s).ok()) {
+                Some(s) => titan_out_dir_str = s,
+                None => {
+                    return Err(format!(
+                        "{} is not a valid directory",
+                        titan_out_dir.display()
+                    ))
+                }
+            };
+        }
+
+        if self.is_titan {
+            unsafe {
+                ffi_try!(ctitandb_checkpoint_create(
+                    self.ptr,
+                    basedb_out_dir.as_ptr(),
+                    titan_out_dir_str.as_ptr(),
+                    log_size_for_flush
+                ));
+            }
+        } else {
+            unsafe {
+                ffi_try!(crocksdb_checkpoint_create(
+                    self.ptr,
+                    basedb_out_dir.as_ptr(),
+                    log_size_for_flush
+                ));
+            }
+        }
+
+        Ok(())
+    }
+}
+
+impl Drop for Checkpointer {
+    fn drop(&mut self) {
+        if self.is_titan {
+            unsafe {
+                librocksdb_sys::ctitandb_checkpoint_object_destroy(self.ptr);
+            }
+            return;
+        }
+        unsafe {
+            librocksdb_sys::crocksdb_checkpoint_object_destroy(self.ptr);
+        }
+    }
+}

src/compaction_filter.rs

@@ -523,7 +523,6 @@ mod tests {
         db.compact_range_cf(cfh, None, None);
         let sk = rx.recv().unwrap();
         let ek = rx.recv().unwrap();
-        println!("sk:{:?} ek:{:?}", sk, ek);
         let sk = str::from_utf8(&sk).unwrap();
         let ek = str::from_utf8(&ek).unwrap();
         assert_eq!("key0", sk);

src/lib.rs

@@ -27,6 +27,7 @@ extern crate tempfile;
 #[macro_use]
 extern crate lazy_static;
 
+pub use checkpoint::Checkpointer;
 pub use compaction_filter::{
     new_compaction_filter, new_compaction_filter_factory, new_compaction_filter_raw,
     CompactionFilter, CompactionFilterContext, CompactionFilterDecision, CompactionFilterFactory,
@@ -83,6 +84,7 @@ pub use write_batch::{WriteBatch, WriteBatchIter, WriteBatchRef};
 #[allow(deprecated)]
 pub use rocksdb::Kv;
 
+mod checkpoint;
 mod compaction_filter;
 pub mod comparator;
 #[cfg(feature = "encryption")]

src/rocksdb.rs

@@ -1856,6 +1856,10 @@ impl DB {
         DB::open_default(restore_db_path)
     }
 
+    pub fn new_checkpointer(&self) -> Result<crate::Checkpointer, String> {
+        crate::Checkpointer::new(self.inner, self.is_titan())
+    }
+
     pub fn get_block_cache_usage(&self) -> u64 {
         self.get_options().get_block_cache_usage()
     }

tests/cases/mod.rs

@@ -1,3 +1,4 @@
+mod test_checkpoint;
 mod test_column_family;
 mod test_compact_range;
 mod test_compaction_filter;