*: add bottommost level compaction (#212)

Expose the bottommost level compaction option to make it more flexible.

librocksdb_sys/crocksdb/c.cc

@@ -138,6 +138,7 @@ using rocksdb::SstFileMetaData;
 using rocksdb::CompactionOptions;
 using rocksdb::PerfLevel;
 using rocksdb::PerfContext;
+using rocksdb::BottommostLevelCompaction;
 
 using std::shared_ptr;
 
@@ -3033,6 +3034,12 @@ void crocksdb_compactoptions_set_max_subcompactions(
   opt->rep.max_subcompactions = v;
 }
 
+void crocksdb_compactoptions_set_bottommost_level_compaction(
+    crocksdb_compactoptions_t* opt,
+    int v) {
+  opt->rep.bottommost_level_compaction = static_cast<BottommostLevelCompaction>(v);
+}
+
 crocksdb_flushoptions_t* crocksdb_flushoptions_create() {
   return new crocksdb_flushoptions_t;
 }

librocksdb_sys/crocksdb/crocksdb/c.h

@@ -1234,6 +1234,8 @@ extern C_ROCKSDB_LIBRARY_API void crocksdb_compactoptions_set_target_level(
     crocksdb_compactoptions_t*, int);
 extern C_ROCKSDB_LIBRARY_API void
 crocksdb_compactoptions_set_max_subcompactions(crocksdb_compactoptions_t*, int);
+extern C_ROCKSDB_LIBRARY_API void
+crocksdb_compactoptions_set_bottommost_level_compaction(crocksdb_compactoptions_t*, int);
 
 /* Flush options */
 

librocksdb_sys/rocksdb/db/compaction_picker.cc

@@ -677,7 +677,7 @@ Compaction* CompactionPicker::CompactRange(
       mutable_cf_options.max_compaction_bytes, output_path_id,
       GetCompressionType(ioptions_, vstorage, mutable_cf_options, output_level,
                          vstorage->base_level()),
-      /* max_subcompactions */ 0, std::move(grandparents),
+      max_subcompactions, std::move(grandparents),
       /* is manual compaction */ true);
 
   TEST_SYNC_POINT_CALLBACK("CompactionPicker::CompactRange:Return", compaction);

librocksdb_sys/src/lib.rs

@@ -292,6 +292,18 @@ pub enum DBTableProperty {
     CompressionName = 17,
 }
 
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+#[repr(C)]
+pub enum DBBottommostLevelCompaction {
+    // Skip bottommost level compaction
+    Skip = 0,
+    // Compact bottommost level if there is a compaction filter
+    // This is the default option
+    IfHaveCompactionFilter = 1,
+    // Force bottommost level compaction
+    Force = 2,
+}
+
 pub fn error_message(ptr: *mut c_char) -> String {
     let c_str = unsafe { CStr::from_ptr(ptr) };
     let s = format!("{}", c_str.to_string_lossy());
@@ -1021,6 +1033,10 @@ extern "C" {
     pub fn crocksdb_compactoptions_set_change_level(opt: *mut DBCompactOptions, v: bool);
     pub fn crocksdb_compactoptions_set_target_level(opt: *mut DBCompactOptions, v: i32);
     pub fn crocksdb_compactoptions_set_max_subcompactions(opt: *mut DBCompactOptions, v: i32);
+    pub fn crocksdb_compactoptions_set_bottommost_level_compaction(
+        opt: *mut DBCompactOptions,
+        v: DBBottommostLevelCompaction,
+    );
 
     pub fn crocksdb_fifo_compaction_options_create() -> *mut DBFifoCompactionOptions;
     pub fn crocksdb_fifo_compaction_options_set_max_table_files_size(

src/lib.rs

@@ -37,8 +37,9 @@ mod table_properties_collector_factory;
 pub use compaction_filter::CompactionFilter;
 pub use event_listener::{CompactionJobInfo, EventListener, FlushJobInfo, IngestionInfo};
 pub use librocksdb_sys::{self as crocksdb_ffi, new_bloom_filter, CompactionPriority,
-                         DBCompactionStyle, DBCompressionType, DBEntryType, DBInfoLogLevel,
-                         DBRecoveryMode, DBStatisticsHistogramType, DBStatisticsTickerType};
+                         DBBottommostLevelCompaction, DBCompactionStyle, DBCompressionType,
+                         DBEntryType, DBInfoLogLevel, DBRecoveryMode, DBStatisticsHistogramType,
+                         DBStatisticsTickerType};
 pub use merge_operator::MergeOperands;
 pub use metadata::{ColumnFamilyMetaData, LevelMetaData, SstFileMetaData};
 pub use perf_context::{get_perf_level, set_perf_level, PerfContext, PerfLevel};

src/rocksdb_options.rs

@@ -15,11 +15,11 @@
 
 use compaction_filter::{new_compaction_filter, CompactionFilter, CompactionFilterHandle};
 use comparator::{self, compare_callback, ComparatorCallback};
-use crocksdb_ffi::{self, DBBlockBasedTableOptions, DBCompactOptions, DBCompactionOptions,
-                   DBCompressionType, DBFifoCompactionOptions, DBFlushOptions, DBInfoLogLevel,
-                   DBInstance, DBRateLimiter, DBReadOptions, DBRecoveryMode, DBRestoreOptions,
-                   DBSnapshot, DBStatisticsHistogramType, DBStatisticsTickerType, DBWriteOptions,
-                   Options};
+use crocksdb_ffi::{self, DBBlockBasedTableOptions, DBBottommostLevelCompaction, DBCompactOptions,
+                   DBCompactionOptions, DBCompressionType, DBFifoCompactionOptions,
+                   DBFlushOptions, DBInfoLogLevel, DBInstance, DBRateLimiter, DBReadOptions,
+                   DBRecoveryMode, DBRestoreOptions, DBSnapshot, DBStatisticsHistogramType,
+                   DBStatisticsTickerType, DBWriteOptions, Options};
 use event_listener::{new_event_listener, EventListener};
 use libc::{self, c_double, c_int, c_uchar, c_void, size_t};
 use merge_operator::MergeFn;
@@ -487,6 +487,12 @@ impl CompactOptions {
             crocksdb_ffi::crocksdb_compactoptions_set_max_subcompactions(self.inner, v);
         }
     }
+
+    pub fn set_bottommost_level_compaction(&mut self, v: DBBottommostLevelCompaction) {
+        unsafe {
+            crocksdb_ffi::crocksdb_compactoptions_set_bottommost_level_compaction(self.inner, v);
+        }
+    }
 }
 
 impl Drop for CompactOptions {

tests/test_compact_range.rs

@@ -11,7 +11,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use rocksdb::{ColumnFamilyOptions, CompactOptions, DBOptions, Range, Writable, DB};
+use rocksdb::{ColumnFamilyOptions, CompactOptions, DBBottommostLevelCompaction, DBOptions, Range,
+              Writable, DB};
 use tempdir::TempDir;
 
 #[test]
@@ -78,3 +79,52 @@ fn test_compact_range_change_level() {
     let name = format!("rocksdb.num-files-at-level{}", compact_level);
     assert_eq!(db.get_property_int(&name).unwrap(), samples.len() as u64);
 }
+
+#[test]
+fn test_compact_range_bottommost_level_compaction() {
+    let path = TempDir::new("test_compact_range_bottommost_level_compaction").expect("");
+    let mut opts = DBOptions::new();
+    opts.create_if_missing(true);
+
+    let db = DB::open(opts, path.path().to_str().unwrap()).unwrap();
+    db.put(&[0], &[0]).unwrap();
+    db.flush(true).unwrap();
+
+    // Compact to bottommost level
+    let cf_handle = db.cf_handle("default").unwrap();
+    let cf_opts = db.get_options_cf(cf_handle);
+    let bottommost_level = (cf_opts.get_num_levels() - 1) as i32;
+    let mut compact_opts = CompactOptions::new();
+    compact_opts.set_change_level(true);
+    compact_opts.set_target_level(bottommost_level);
+    db.compact_range_cf_opt(cf_handle, &compact_opts, None, None);
+
+    let bottommost_files = db.get_column_family_meta_data(cf_handle)
+        .get_levels()
+        .last()
+        .unwrap()
+        .get_files();
+    assert_eq!(bottommost_files.len(), 1);
+    let bottommost_filename = bottommost_files[0].get_name();
+
+    // Skip bottommost level compaction
+    compact_opts.set_bottommost_level_compaction(DBBottommostLevelCompaction::Skip);
+    db.compact_range_cf_opt(cf_handle, &compact_opts, None, None);
+    let bottommost_files = db.get_column_family_meta_data(cf_handle)
+        .get_levels()
+        .last()
+        .unwrap()
+        .get_files();
+    assert_eq!(bottommost_files.len(), 1);
+    assert_eq!(bottommost_filename, bottommost_files[0].get_name());
+    // Force bottommost level compaction
+    compact_opts.set_bottommost_level_compaction(DBBottommostLevelCompaction::Force);
+    db.compact_range_cf_opt(cf_handle, &compact_opts, None, None);
+    let bottommost_files = db.get_column_family_meta_data(cf_handle)
+        .get_levels()
+        .last()
+        .unwrap()
+        .get_files();
+    assert_eq!(bottommost_files.len(), 1);
+    assert_ne!(bottommost_filename, bottommost_files[0].get_name());
+}