add more api for CompactionJobInfo (#183)

librocksdb_sys/crocksdb/c.cc

@@ -1712,6 +1712,26 @@ int crocksdb_compactionjobinfo_output_level(
   return info->rep.output_level;
 }
 
+uint64_t crocksdb_compactionjobinfo_input_records(
+    const crocksdb_compactionjobinfo_t* info) {
+  return info->rep.stats.num_input_records;
+}
+
+uint64_t crocksdb_compactionjobinfo_output_records(
+    const crocksdb_compactionjobinfo_t* info) {
+  return info->rep.stats.num_output_records;
+}
+
+uint64_t crocksdb_compactionjobinfo_total_input_bytes(
+    const crocksdb_compactionjobinfo_t* info) {
+  return info->rep.stats.total_input_bytes;
+}
+
+uint64_t crocksdb_compactionjobinfo_total_output_bytes(
+    const crocksdb_compactionjobinfo_t* info) {
+  return info->rep.stats.total_output_bytes;
+}
+
 /* ExternalFileIngestionInfo */
 
 const char* crocksdb_externalfileingestioninfo_cf_name(

librocksdb_sys/crocksdb/crocksdb/c.h

@@ -644,6 +644,18 @@ crocksdb_compactionjobinfo_num_corrupt_keys(
 extern C_ROCKSDB_LIBRARY_API int
 crocksdb_compactionjobinfo_output_level(
     const crocksdb_compactionjobinfo_t*);
+extern C_ROCKSDB_LIBRARY_API uint64_t
+crocksdb_compactionjobinfo_input_records(
+    const crocksdb_compactionjobinfo_t*);
+extern C_ROCKSDB_LIBRARY_API uint64_t
+crocksdb_compactionjobinfo_output_records(
+    const crocksdb_compactionjobinfo_t*);
+extern C_ROCKSDB_LIBRARY_API uint64_t
+crocksdb_compactionjobinfo_total_input_bytes(
+    const crocksdb_compactionjobinfo_t* info);
+extern C_ROCKSDB_LIBRARY_API uint64_t
+crocksdb_compactionjobinfo_total_output_bytes(
+    const crocksdb_compactionjobinfo_t* info);
 
 /* External file ingestion info */
 

librocksdb_sys/src/lib.rs

@@ -1462,6 +1462,10 @@ extern "C" {
         info: *const DBCompactionJobInfo,
     ) -> uint64_t;
     pub fn crocksdb_compactionjobinfo_output_level(info: *const DBCompactionJobInfo) -> c_int;
+    pub fn crocksdb_compactionjobinfo_input_records(info: *const DBCompactionJobInfo) -> uint64_t;
+    pub fn crocksdb_compactionjobinfo_output_records(info: *const DBCompactionJobInfo) -> uint64_t;
+    pub fn crocksdb_compactionjobinfo_total_input_bytes(info: *const DBCompactionJobInfo) -> uint64_t;
+    pub fn crocksdb_compactionjobinfo_total_output_bytes(info: *const DBCompactionJobInfo) -> uint64_t;
 
     pub fn crocksdb_externalfileingestioninfo_cf_name(
         info: *const DBIngestionInfo,

src/event_listener.rs

@@ -92,6 +92,22 @@ impl CompactionJobInfo {
     pub fn output_level(&self) -> i32 {
         unsafe { crocksdb_ffi::crocksdb_compactionjobinfo_output_level(&self.0) }
     }
+
+    pub fn input_records(&self) -> u64 {
+        unsafe { crocksdb_ffi::crocksdb_compactionjobinfo_input_records(&self.0) }
+    }
+
+    pub fn output_records(&self) -> u64 {
+        unsafe { crocksdb_ffi::crocksdb_compactionjobinfo_output_records(&self.0) }
+    }
+
+    pub fn total_input_bytes(&self) -> u64 {
+        unsafe { crocksdb_ffi::crocksdb_compactionjobinfo_total_input_bytes(&self.0) }
+    }
+
+    pub fn total_output_bytes(&self) -> u64 {
+        unsafe { crocksdb_ffi::crocksdb_compactionjobinfo_total_output_bytes(&self.0) }
+    }
 }
 
 pub struct IngestionInfo(DBIngestionInfo);

tests/test_event_listener.rs

@@ -25,6 +25,10 @@ struct EventCounter {
     compaction: Arc<AtomicUsize>,
     ingestion: Arc<AtomicUsize>,
     drop_count: Arc<AtomicUsize>,
+    input_records: Arc<AtomicUsize>,
+    output_records: Arc<AtomicUsize>,
+    input_bytes: Arc<AtomicUsize>,
+    output_bytes: Arc<AtomicUsize>,
 }
 
 impl Drop for EventCounter {
@@ -65,6 +69,14 @@ impl EventListener for EventCounter {
         assert!(info.output_level() >= 0);
 
         self.compaction.fetch_add(1, Ordering::SeqCst);
+        self.input_records
+            .fetch_add(info.input_records() as usize, Ordering::SeqCst);
+        self.output_records
+            .fetch_add(info.output_records() as usize, Ordering::SeqCst);
+        self.input_bytes
+            .fetch_add(info.total_input_bytes() as usize, Ordering::SeqCst);
+        self.output_bytes
+            .fetch_add(info.total_output_bytes() as usize, Ordering::SeqCst);
     }
 
     fn on_external_file_ingested(&self, info: &IngestionInfo) {
@@ -109,6 +121,13 @@ fn test_event_listener_basic() {
     assert_ne!(counter.compaction.load(Ordering::SeqCst), 0);
     drop(db);
     assert_eq!(counter.drop_count.load(Ordering::SeqCst), 1);
+    assert!(
+        counter.input_records.load(Ordering::SeqCst) >
+            counter.output_records.load(Ordering::SeqCst)
+    );
+    assert!(
+        counter.input_bytes.load(Ordering::SeqCst) > counter.output_bytes.load(Ordering::SeqCst)
+    );
 }
 
 #[test]