Add ReadOptions to TransactionLogIterator.

Summary:
Add an optional input parameter ReadOptions to DB::GetUpdateSince(),
which allows the verification of checksums to be disabled by setting
ReadOptions::verify_checksums to false.

Test Plan: Tests are done off-line and will not be included in the regular unit test.

Reviewers: igor

Reviewed By: igor

CC: leveldb, xjin, dhruba

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

db/db_impl.cc

@@ -1490,8 +1490,9 @@ SequenceNumber DBImpl::GetLatestSequenceNumber() const {
   return versions_->LastSequence();
 }
 
-Status DBImpl::GetUpdatesSince(SequenceNumber seq,
-                               unique_ptr<TransactionLogIterator>* iter) {
+Status DBImpl::GetUpdatesSince(
+    SequenceNumber seq, unique_ptr<TransactionLogIterator>* iter,
+    const TransactionLogIterator::ReadOptions& read_options) {
 
   RecordTick(options_.statistics.get(), GET_UPDATES_SINCE_CALLS);
   if (seq > versions_->LastSequence()) {
@@ -1511,13 +1512,9 @@ Status DBImpl::GetUpdatesSince(SequenceNumber seq,
   if (!s.ok()) {
     return s;
   }
-  iter->reset(
-    new TransactionLogIteratorImpl(options_.wal_dir,
-                                   &options_,
-                                   storage_options_,
-                                   seq,
-                                   std::move(wal_files),
-                                   this));
+  iter->reset(new TransactionLogIteratorImpl(options_.wal_dir, &options_,
+                                             read_options, storage_options_,
+                                             seq, std::move(wal_files), this));
   return (*iter)->status();
 }
 

db/db_impl.h

@@ -85,8 +85,10 @@ class DBImpl : public DB {
                               bool flush_memtable = true);
   virtual Status GetSortedWalFiles(VectorLogPtr& files);
   virtual SequenceNumber GetLatestSequenceNumber() const;
-  virtual Status GetUpdatesSince(SequenceNumber seq_number,
-                                 unique_ptr<TransactionLogIterator>* iter);
+  virtual Status GetUpdatesSince(
+      SequenceNumber seq_number, unique_ptr<TransactionLogIterator>* iter,
+      const TransactionLogIterator::ReadOptions&
+          read_options = TransactionLogIterator::ReadOptions());
   virtual Status DeleteFile(std::string name);
 
   virtual void GetLiveFilesMetaData(

db/db_test.cc

@@ -5005,8 +5005,10 @@ class ModelDB: public DB {
   virtual SequenceNumber GetLatestSequenceNumber() const {
     return 0;
   }
-  virtual Status GetUpdatesSince(rocksdb::SequenceNumber,
-                                 unique_ptr<rocksdb::TransactionLogIterator>*) {
+  virtual Status GetUpdatesSince(
+      rocksdb::SequenceNumber, unique_ptr<rocksdb::TransactionLogIterator>*,
+      const TransactionLogIterator::ReadOptions&
+          read_options = TransactionLogIterator::ReadOptions()) {
     return Status::NotSupported("Not supported in Model DB");
   }
 

db/transaction_log_impl.cc

@@ -9,23 +9,22 @@
 namespace rocksdb {
 
 TransactionLogIteratorImpl::TransactionLogIteratorImpl(
-                           const std::string& dir,
-                           const Options* options,
-                           const EnvOptions& soptions,
-                           const SequenceNumber seq,
-                           std::unique_ptr<VectorLogPtr> files,
-                           DBImpl const * const dbimpl) :
-    dir_(dir),
-    options_(options),
-    soptions_(soptions),
-    startingSequenceNumber_(seq),
-    files_(std::move(files)),
-    started_(false),
-    isValid_(false),
-    currentFileIndex_(0),
-    currentBatchSeq_(0),
-    currentLastSeq_(0),
-    dbimpl_(dbimpl) {
+    const std::string& dir, const Options* options,
+    const TransactionLogIterator::ReadOptions& read_options,
+    const EnvOptions& soptions, const SequenceNumber seq,
+    std::unique_ptr<VectorLogPtr> files, DBImpl const* const dbimpl)
+    : dir_(dir),
+      options_(options),
+      read_options_(read_options),
+      soptions_(soptions),
+      startingSequenceNumber_(seq),
+      files_(std::move(files)),
+      started_(false),
+      isValid_(false),
+      currentFileIndex_(0),
+      currentBatchSeq_(0),
+      currentLastSeq_(0),
+      dbimpl_(dbimpl) {
   assert(files_ != nullptr);
   assert(dbimpl_ != nullptr);
 
@@ -253,9 +252,8 @@ Status TransactionLogIteratorImpl::OpenLogReader(const LogFile* logFile) {
     return status;
   }
   assert(file);
-  currentLogReader_.reset(
-    new log::Reader(std::move(file), &reporter_, true, 0)
-  );
+  currentLogReader_.reset(new log::Reader(std::move(file), &reporter_,
+                                          read_options_.verify_checksums_, 0));
   return Status::OK();
 }
 }  //  namespace rocksdb

db/transaction_log_impl.h

@@ -66,12 +66,11 @@ class LogFileImpl : public LogFile {
 
 class TransactionLogIteratorImpl : public TransactionLogIterator {
  public:
-  TransactionLogIteratorImpl(const std::string& dir,
-                             const Options* options,
-                             const EnvOptions& soptions,
-                             const SequenceNumber seqNum,
-                             std::unique_ptr<VectorLogPtr> files,
-                             DBImpl const * const dbimpl);
+  TransactionLogIteratorImpl(
+      const std::string& dir, const Options* options,
+      const TransactionLogIterator::ReadOptions& read_options,
+      const EnvOptions& soptions, const SequenceNumber seqNum,
+      std::unique_ptr<VectorLogPtr> files, DBImpl const* const dbimpl);
 
   virtual bool Valid();
 
@@ -84,6 +83,7 @@ class TransactionLogIteratorImpl : public TransactionLogIterator {
  private:
   const std::string& dir_;
   const Options* options_;
+  const TransactionLogIterator::ReadOptions read_options_;
   const EnvOptions& soptions_;
   SequenceNumber startingSequenceNumber_;
   std::unique_ptr<VectorLogPtr> files_;

include/rocksdb/db.h

@@ -300,8 +300,10 @@ class DB {
   // use this api, else the WAL files will get
   // cleared aggressively and the iterator might keep getting invalid before
   // an update is read.
-  virtual Status GetUpdatesSince(SequenceNumber seq_number,
-                                 unique_ptr<TransactionLogIterator>* iter) = 0;
+  virtual Status GetUpdatesSince(
+      SequenceNumber seq_number, unique_ptr<TransactionLogIterator>* iter,
+      const TransactionLogIterator::ReadOptions&
+          read_options = TransactionLogIterator::ReadOptions()) = 0;
 
   // Delete the file name from the db directory and update the internal state to
   // reflect that. Supports deletion of sst and log files only. 'name' must be

include/rocksdb/transaction_log.h

@@ -85,6 +85,19 @@ class TransactionLogIterator {
   // earliest transaction contained in the batch.
   // ONLY use if Valid() is true and status() is OK.
   virtual BatchResult GetBatch() = 0;
+
+  // The read options for TransactionLogIterator.
+  struct ReadOptions {
+    // If true, all data read from underlying storage will be
+    // verified against corresponding checksums.
+    // Default: true
+    bool verify_checksums_;
+
+    ReadOptions() : verify_checksums_(true) {}
+
+    explicit ReadOptions(bool verify_checksums)
+        : verify_checksums_(verify_checksums) {}
+  };
 };
 } //  namespace rocksdb
 

include/utilities/stackable_db.h

@@ -152,10 +152,10 @@ class StackableDB : public DB {
     return db_->GetPropertiesOfAllTables(props);
   }
 
-  virtual Status GetUpdatesSince(SequenceNumber seq_number,
-                                 unique_ptr<TransactionLogIterator>* iter)
-    override {
-      return db_->GetUpdatesSince(seq_number, iter);
+  virtual Status GetUpdatesSince(
+      SequenceNumber seq_number, unique_ptr<TransactionLogIterator>* iter,
+      const TransactionLogIterator::ReadOptions& read_options) override {
+    return db_->GetUpdatesSince(seq_number, iter, read_options);
   }
 
  protected: