avoid calling FindFile twice in TwoLevelIterator for PlainTable

Summary:
this is to reclaim the regression introduced in
https://reviews.facebook.net/D17853

Test Plan: make all check

Reviewers: igor, haobo, sdong, dhruba, yhchiang

Reviewed By: haobo

CC: leveldb

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

db/db_test.cc

@@ -4421,6 +4421,8 @@ TEST(DBTest, HiddenValuesAreRemoved) {
 
 TEST(DBTest, CompactBetweenSnapshots) {
   do {
+    Options options = CurrentOptions();
+    options.disable_auto_compactions = true;
     CreateAndReopenWithCF({"pikachu"});
     Random rnd(301);
     FillLevels("a", "z", 1);

db/table_cache.cc

@@ -190,33 +190,6 @@ Status TableCache::GetTableProperties(
   return s;
 }
 
-bool TableCache::PrefixMayMatch(const ReadOptions& options,
-    const InternalKeyComparator& icomparator,
-    const FileMetaData& file_meta,
-    const Slice& internal_key, bool* table_io) {
-  bool may_match = true;
-  auto table_reader = file_meta.table_reader;
-  Cache::Handle* table_handle = nullptr;
-  if (table_reader == nullptr) {
-    // Need to get table handle from file number
-    Status s = FindTable(storage_options_, icomparator, file_meta.number,
-        file_meta.file_size, &table_handle, table_io);
-    if (!s.ok()) {
-      return may_match;
-    }
-    table_reader = GetTableReaderFromHandle(table_handle);
-  }
-
-  may_match = table_reader->PrefixMayMatch(internal_key);
-
-  if (table_handle != nullptr) {
-    // Need to release handle if it is generated from here.
-    ReleaseHandle(table_handle);
-  }
-
-  return may_match;
-}
-
 void TableCache::Evict(Cache* cache, uint64_t file_number) {
   cache->Erase(GetSliceForFileNumber(&file_number));
 }

db/table_cache.h

@@ -56,13 +56,6 @@ class TableCache {
                                    const Slice&, bool),
              bool* table_io, void (*mark_key_may_exist)(void*) = nullptr);
 
-  // Determine whether the table may contain the specified prefix.  If
-  // the table index or blooms are not in memory, this may cause an I/O
-  bool PrefixMayMatch(const ReadOptions& options,
-                      const InternalKeyComparator& internal_comparator,
-                      const FileMetaData& file_meta,
-                      const Slice& internal_prefix, bool* table_io);
-
   // Evict any entry for the specified file number
   static void Evict(Cache* cache, uint64_t file_number);
 

db/version_set.cc

@@ -31,6 +31,7 @@
 #include "table/merger.h"
 #include "table/two_level_iterator.h"
 #include "table/format.h"
+#include "table/plain_table_factory.h"
 #include "table/meta_blocks.h"
 #include "util/coding.h"
 #include "util/logging.h"
@@ -308,13 +309,13 @@ Status Version::GetPropertiesOfAllTables(TablePropertiesCollection* props) {
   return Status::OK();
 }
 
-void Version::AddIterators(const ReadOptions& options,
+void Version::AddIterators(const ReadOptions& read_options,
                            const EnvOptions& soptions,
                            std::vector<Iterator*>* iters) {
   // Merge all level zero files together since they may overlap
   for (const FileMetaData* file : files_[0]) {
     iters->push_back(cfd_->table_cache()->NewIterator(
-        options, soptions, cfd_->internal_comparator(), *file));
+        read_options, soptions, cfd_->internal_comparator(), *file));
   }
 
   // For levels > 0, we can use a concatenating iterator that sequentially
@@ -323,7 +324,7 @@ void Version::AddIterators(const ReadOptions& options,
   for (int level = 1; level < num_levels_; level++) {
     if (!files_[level].empty()) {
       iters->push_back(NewTwoLevelIterator(new LevelFileIteratorState(
-          cfd_->table_cache(), options, soptions,
+          cfd_->table_cache(), read_options, soptions,
           cfd_->internal_comparator(), false /* for_compaction */,
           cfd_->options()->prefix_extractor != nullptr),
         new LevelFileNumIterator(cfd_->internal_comparator(), &files_[level])));

table/block_based_table_reader.h

@@ -63,7 +63,7 @@ class BlockBasedTable : public TableReader {
                      unique_ptr<RandomAccessFile>&& file, uint64_t file_size,
                      unique_ptr<TableReader>* table_reader);
 
-  bool PrefixMayMatch(const Slice& internal_key) override;
+  bool PrefixMayMatch(const Slice& internal_key);
 
   // Returns a new iterator over the table contents.
   // The result of NewIterator() is initially invalid (caller must

table/plain_table_factory.h

@@ -2,8 +2,9 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
-#ifndef ROCKSDB_LITE
 #pragma once
+
+#ifndef ROCKSDB_LITE
 #include <memory>
 #include <stdint.h>
 

table/table_reader.h

@@ -25,14 +25,6 @@ class TableReader {
  public:
   virtual ~TableReader() {}
 
-  // Determine whether there is a chance that the current table file
-  // contains the key a key starting with iternal_prefix. The specific
-  // table implementation can use bloom filter and/or other heuristic
-  // to filter out this table as a whole.
-  virtual bool PrefixMayMatch(const Slice& internal_prefix) {
-    return true;
-  }
-
   // Returns a new iterator over the table contents.
   // The result of NewIterator() is initially invalid (caller must
   // call one of the Seek methods on the iterator before using it).

table/two_level_iterator.cc

@@ -77,16 +77,18 @@ TwoLevelIterator::TwoLevelIterator(TwoLevelIteratorState* state,
   : state_(state), first_level_iter_(first_level_iter) {}
 
 void TwoLevelIterator::Seek(const Slice& target) {
-  if (state_->prefix_enabled && !state_->PrefixMayMatch(target)) {
+  if (state_->check_prefix_may_match &&
+      !state_->PrefixMayMatch(target)) {
     SetSecondLevelIterator(nullptr);
-  } else {
+    return;
+  }
   first_level_iter_.Seek(target);
+
   InitDataBlock();
   if (second_level_iter_.iter() != nullptr) {
     second_level_iter_.Seek(target);
   }
   SkipEmptyDataBlocksForward();
-  }
 }
 
 void TwoLevelIterator::SeekToFirst() {

table/two_level_iterator.h

@@ -18,14 +18,15 @@ struct ReadOptions;
 class InternalKeyComparator;
 
 struct TwoLevelIteratorState {
-  explicit TwoLevelIteratorState(bool prefix_enabled)
-    : prefix_enabled(prefix_enabled) {}
+  explicit TwoLevelIteratorState(bool check_prefix_may_match)
+    : check_prefix_may_match(check_prefix_may_match) {}
 
   virtual ~TwoLevelIteratorState() {}
   virtual Iterator* NewSecondaryIterator(const Slice& handle) = 0;
   virtual bool PrefixMayMatch(const Slice& internal_key) = 0;
 
-  bool prefix_enabled;
+  // If call PrefixMayMatch()
+  bool check_prefix_may_match;
 };