Improved CompactionFilter api: pass in a opaque argument to CompactionFilter invocation.

Summary:
There are applications that operate on multiple leveldb instances.
These applications will like to pass in an opaque type for each
leveldb instance and this type should be passed back to the application
with every invocation of the CompactionFilter api.

Test Plan: Enehanced unit test for opaque parameter to CompactionFilter.

Reviewers: heyongqiang

Reviewed By: heyongqiang

CC: MarkCallaghan, sheki, emayanke

Differential Revision: https://reviews.facebook.net/D6711

db/db_impl.cc

@@ -1164,7 +1164,8 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
         // it. If this key is not visible via any snapshot and the
         // return value of the compaction filter is true and then
         // drop this key from the output.
-        drop = options_.CompactionFilter(compact->compaction->level(),
+        drop = options_.CompactionFilter(options_.compaction_filter_args,
+                         compact->compaction->level(),
                          ikey.user_key, value, &compaction_filter_value);
 
         if (drop) {

db/db_test.cc

@@ -1198,18 +1198,21 @@ TEST(DBTest, RepeatedWritesToSameKey) {
 // kvs during the compaction process.
 static int cfilter_count;
 static std::string NEW_VALUE = "NewValue";
-static bool keep_filter(int level, const Slice& key,
+static bool keep_filter(void* arg, int level, const Slice& key,
   const Slice& value, Slice** new_value) {
+  assert(arg == NULL);
   cfilter_count++;
   return false;
 }
-static bool delete_filter(int level, const Slice& key,
+static bool delete_filter(void*argv, int level, const Slice& key,
   const Slice& value, Slice** new_value) {
+  assert(arg == NULL);
   cfilter_count++;
   return true;
 }
-static bool change_filter(int level, const Slice& key,
+static bool change_filter(void*argv, int level, const Slice& key,
   const Slice& value, Slice** new_value) {
+  assert(argv == (void*)100);
   assert(new_value != NULL);
   *new_value = new Slice(NEW_VALUE);
   return false;
@@ -1320,6 +1323,7 @@ TEST(DBTest, CompactionFilterWithValueChange) {
   Options options = CurrentOptions();
   options.num_levels = 3;
   options.max_mem_compaction_level = 0;
+  options.compaction_filter_args = (void *)100;
   options.CompactionFilter = change_filter;
   Reopen(&options);
 

include/leveldb/options.h

@@ -313,7 +313,11 @@ struct Options {
   // should allocate memory for the Slice object that is used to
   // return the new value and the leveldb framework will
   // free up that memory.
-  bool (*CompactionFilter)(int level, const Slice& key, 
+  // The compaction_filter_args, if specified here, are passed
+  // back to the invocation of the CompactionFilter.
+  void* compaction_filter_args;
+  bool (*CompactionFilter)(void* compaction_filter_args, 
+         int level, const Slice& key, 
          const Slice& existing_value, Slice** new_value);
 };
 

util/options.cc

@@ -48,6 +48,7 @@ Options::Options()
       rate_limit(0.0),
       no_block_cache(false),
       table_cache_numshardbits(4),
+      compaction_filter_args(NULL),
       CompactionFilter(NULL) {
 }
 
@@ -124,6 +125,8 @@ Options::Dump(
         delete_obsolete_files_period_micros);
     Log(log,"                             Options.rate_limit: %.2f",
         rate_limit);
+    Log(log,"                 Options.compaction_filter_args: %p",
+        compaction_filter_args);
     Log(log,"                       Options.CompactionFilter: %p",
         CompactionFilter);
 }   // Options::Dump