ForwardIterator::status() checks all child iterators

Summary:
Forward iterator only checked `status_` and `mutable_iter_->status()`, which is
not sufficient. For example, when reading exclusively from cache
(kBlockCacheTier), `mutable_iter_->status()` may return kOk (e.g. there's
nothing in the memtable), but one of immutable iterators could be in
kIncomplete. In this case, `ForwardIterator::status()` ought to return that
status instead of kOk.

This diff changes `status()` to also check `imm_iters_`, `l0_iters_`, and
`level_iters_`.

Test Plan:
  ROCKSDB_TESTS=TailingIteratorIncomplete ./db_test

Reviewers: ljin, igor

Reviewed By: igor

Subscribers: dhruba, leveldb

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

db/db_test.cc

@@ -6917,6 +6917,28 @@ TEST(DBTest, TailingIteratorPrefixSeek) {
   ASSERT_TRUE(!iter->Valid());
 }
 
+TEST(DBTest, TailingIteratorIncomplete) {
+  CreateAndReopenWithCF({"pikachu"});
+  ReadOptions read_options;
+  read_options.tailing = true;
+  read_options.read_tier = kBlockCacheTier;
+
+  std::string key("key");
+  std::string value("value");
+
+  ASSERT_OK(db_->Put(WriteOptions(), key, value));
+
+  std::unique_ptr<Iterator> iter(db_->NewIterator(read_options));
+  iter->SeekToFirst();
+  // we either see the entry or it's not in cache
+  ASSERT_TRUE(iter->Valid() || iter->status().IsIncomplete());
+
+  ASSERT_OK(db_->CompactRange(nullptr, nullptr));
+  iter->SeekToFirst();
+  // should still be true after compaction
+  ASSERT_TRUE(iter->Valid() || iter->status().IsIncomplete());
+}
+
 TEST(DBTest, BlockBasedTablePrefixIndexTest) {
   // create a DB with block prefix index
   BlockBasedTableOptions table_options;

db/forward_iterator.cc

@@ -87,7 +87,12 @@ class LevelIterator : public Iterator {
     return file_iter_->value();
   }
   Status status() const override {
+    if (!status_.ok()) {
       return status_;
+    } else if (file_iter_ && !file_iter_->status().ok()) {
+      return file_iter_->status();
+    }
+    return Status::OK();
   }
 
  private:
@@ -334,6 +339,23 @@ Status ForwardIterator::status() const {
   } else if (!mutable_iter_->status().ok()) {
     return mutable_iter_->status();
   }
+
+  for (auto *it : imm_iters_) {
+    if (it && !it->status().ok()) {
+      return it->status();
+    }
+  }
+  for (auto *it : l0_iters_) {
+    if (it && !it->status().ok()) {
+      return it->status();
+    }
+  }
+  for (auto *it : level_iters_) {
+    if (it && !it->status().ok()) {
+      return it->status();
+    }
+  }
+
   return Status::OK();
 }