forward iterator

Summary:
Forward iterator puts everything together in a flat structure instead of
a hierarchy of nested iterators. this should simplify the code and
provide better performance. It also enables more optimization since all
information are accessiable in one place.
Init evaluation shows about 6% improvement

Test Plan: db_test and db_bench

Reviewers: dhruba, igor, tnovak, sdong, haobo

Reviewed By: haobo

Subscribers: sdong, leveldb

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

db/column_family.h

@@ -180,7 +180,7 @@ class ColumnFamilyData {
   void SetCurrent(Version* current);
   void CreateNewMemtable();
 
-  TableCache* table_cache() { return table_cache_.get(); }
+  TableCache* table_cache() const { return table_cache_.get(); }
 
   // See documentation in compaction_picker.h
   Compaction* PickCompaction(LogBuffer* log_buffer);

db/db_bench.cc

@@ -1928,7 +1928,7 @@ class Benchmark {
     }
 
     char msg[100];
-    snprintf(msg, sizeof(msg), "(%" PRIu64 " of %" PRIu64 " found)",
+    snprintf(msg, sizeof(msg), "(%" PRIu64 " of %" PRIu64 " found)\n",
              found, read);
 
     thread->stats.AddMessage(msg);
@@ -2056,9 +2056,12 @@ class Benchmark {
     }
 
     char msg[100];
-    snprintf(msg, sizeof(msg), "(%" PRIu64 " of %" PRIu64 " found)",
+    snprintf(msg, sizeof(msg), "(%" PRIu64 " of %" PRIu64 " found)\n",
              found, read);
     thread->stats.AddMessage(msg);
+    if (FLAGS_perf_level > 0) {
+      thread->stats.AddMessage(perf_context.ToString());
+    }
   }
 
   void SeekRandomWhileWriting(ThreadState* thread) {

db/db_impl.cc

@@ -36,6 +36,7 @@
 #include "db/table_cache.h"
 #include "db/table_properties_collector.h"
 #include "db/tailing_iter.h"
+#include "db/forward_iterator.h"
 #include "db/transaction_log_impl.h"
 #include "db/version_set.h"
 #include "db/write_batch_internal.h"
@@ -2578,7 +2579,7 @@ Status DBImpl::ProcessKeyValueCompaction(
           cfd->user_comparator()->Compare(ikey.user_key,
                                           current_user_key.GetKey()) != 0) {
         // First occurrence of this user key
-        current_user_key.SetUserKey(ikey.user_key);
+        current_user_key.SetKey(ikey.user_key);
         has_current_user_key = true;
         last_sequence_for_key = kMaxSequenceNumber;
         visible_in_snapshot = kMaxSequenceNumber;
@@ -3538,7 +3539,11 @@ Iterator* DBImpl::NewIterator(const ReadOptions& options,
     // not supported in lite version
     return nullptr;
 #else
-    iter = new TailingIterator(env_, this, options, cfd);
+    // TODO(ljin): remove tailing iterator
+    iter = new ForwardIterator(env_, this, options, cfd);
+    iter = NewDBIterator(env_, *cfd->options(),
+        cfd->user_comparator(), iter, kMaxSequenceNumber);
+    //iter = new TailingIterator(env_, this, options, cfd);
 #endif
   } else {
     SequenceNumber latest_snapshot = versions_->LastSequence();

db/db_impl.h

@@ -285,6 +285,7 @@ class DBImpl : public DB {
   friend class InternalStats;
 #ifndef ROCKSDB_LITE
   friend class TailingIterator;
+  friend class ForwardIterator;
 #endif
   friend struct SuperVersion;
   struct CompactionState;

db/db_iter.cc

@@ -211,18 +211,18 @@ void DBIter::FindNextUserEntryInternal(bool skipping) {
           case kTypeDeletion:
             // Arrange to skip all upcoming entries for this key since
             // they are hidden by this deletion.
-            saved_key_.SetUserKey(ikey.user_key);
+            saved_key_.SetKey(ikey.user_key);
             skipping = true;
             num_skipped = 0;
             PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
             break;
           case kTypeValue:
             valid_ = true;
-            saved_key_.SetUserKey(ikey.user_key);
+            saved_key_.SetKey(ikey.user_key);
             return;
           case kTypeMerge:
             // By now, we are sure the current ikey is going to yield a value
-            saved_key_.SetUserKey(ikey.user_key);
+            saved_key_.SetKey(ikey.user_key);
             current_entry_is_merged_ = true;
             valid_ = true;
             MergeValuesNewToOld();  // Go to a different state machine
@@ -331,7 +331,7 @@ void DBIter::Prev() {
     // iter_ is pointing at the current entry.  Scan backwards until
     // the key changes so we can use the normal reverse scanning code.
     assert(iter_->Valid());  // Otherwise valid_ would have been false
-    saved_key_.SetUserKey(ExtractUserKey(iter_->key()));
+    saved_key_.SetKey(ExtractUserKey(iter_->key()));
     while (true) {
       iter_->Prev();
       if (!iter_->Valid()) {
@@ -377,7 +377,7 @@ void DBIter::FindPrevUserEntry() {
             std::string empty;
             swap(empty, saved_value_);
           }
-          saved_key_.SetUserKey(ExtractUserKey(iter_->key()));
+          saved_key_.SetKey(ExtractUserKey(iter_->key()));
           saved_value_.assign(raw_value.data(), raw_value.size());
         }
       } else {

db/dbformat.h

@@ -256,10 +256,10 @@ class IterKey {
 
   void Clear() { key_size_ = 0; }
 
-  void SetUserKey(const Slice& user_key) {
-    size_t size = user_key.size();
+  void SetKey(const Slice& key) {
+    size_t size = key.size();
     EnlargeBufferIfNeeded(size);
-    memcpy(key_, user_key.data(), size);
+    memcpy(key_, key.data(), size);
     key_size_ = size;
   }