dont create backup_input if compaction filter v2 is not used

Summary:
Compaction creates backup_input iterator even though it only needed
when compaction filter v2 is enabled

Test Plan: make all check

Reviewers: sdong, yhchiang, igor

Reviewed By: igor

Subscribers: leveldb

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

db/db_impl.cc

@@ -3117,9 +3117,6 @@ Status DBImpl::DoCompactionWork(CompactionState* compact,
   const uint64_t start_micros = env_->NowMicros();
   unique_ptr<Iterator> input(versions_->MakeInputIterator(compact->compaction));
   input->SeekToFirst();
-  shared_ptr<Iterator> backup_input(
-      versions_->MakeInputIterator(compact->compaction));
-  backup_input->SeekToFirst();
 
   Status status;
   ParsedInternalKey ikey;
@@ -3132,6 +3129,20 @@ Status DBImpl::DoCompactionWork(CompactionState* compact,
   auto compaction_filter_v2 =
     compaction_filter_from_factory_v2.get();
 
+  if (!compaction_filter_v2) {
+    status = ProcessKeyValueCompaction(
+      is_snapshot_supported,
+      visible_at_tip,
+      earliest_snapshot,
+      latest_snapshot,
+      deletion_state,
+      bottommost_level,
+      imm_micros,
+      input.get(),
+      compact,
+      false,
+      log_buffer);
+  } else {
     // temp_backup_input always point to the start of the current buffer
     // temp_backup_input = backup_input;
     // iterate through input,
@@ -3139,7 +3150,9 @@ Status DBImpl::DoCompactionWork(CompactionState* compact,
     // 2) send value_buffer to compaction filter and alternate the values;
     // 3) merge value_buffer with ineligible_value_buffer;
     // 4) run the modified "compaction" using the old for loop.
-  if (compaction_filter_v2) {
+    shared_ptr<Iterator> backup_input(
+        versions_->MakeInputIterator(compact->compaction));
+    backup_input->SeekToFirst();
     while (backup_input->Valid() && !shutting_down_.Acquire_Load() &&
            !cfd->IsDropped()) {
       // FLUSH preempts compaction
@@ -3267,21 +3280,6 @@ Status DBImpl::DoCompactionWork(CompactionState* compact,
         log_buffer);
   }  // checking for compaction filter v2
 
-  if (!compaction_filter_v2) {
-    status = ProcessKeyValueCompaction(
-      is_snapshot_supported,
-      visible_at_tip,
-      earliest_snapshot,
-      latest_snapshot,
-      deletion_state,
-      bottommost_level,
-      imm_micros,
-      input.get(),
-      compact,
-      false,
-      log_buffer);
-  }
-
   if (status.ok() && (shutting_down_.Acquire_Load() || cfd->IsDropped())) {
     status = Status::ShutdownInProgress(
         "Database shutdown or Column family drop during compaction");