fix 1

db/db_impl/db_impl.h

@@ -2071,15 +2071,18 @@ class DBImpl : public DB {
       bool read_only, int job_id, SequenceNumber* next_sequence,
       bool* stop_replay_for_corruption, bool* stop_replay_by_wal_filter,
       uint64_t* corrupted_wal_number, bool* corrupted_wal_found,
-      std::unordered_map<int, VersionEdit>* version_edits, bool* flushed);
+      std::unordered_map<int, VersionEdit>* version_edits, bool* flushed,
+      PredecessorWALInfo& predecessor_wal_info);
 
   void SetupLogFileProcessing(uint64_t wal_number);
 
-  Status InitializeLogReader(uint64_t wal_number, bool is_retry,
-                             std::string& fname, bool* const old_log_record,
-                             Status* const reporter_status,
-                             DBOpenLogReporter* reporter,
-                             std::unique_ptr<log::Reader>& reader);
+  Status InitializeLogReader(
+      uint64_t wal_number, bool is_retry, std::string& fname,
+
+      bool stop_replay_for_corruption, uint64_t min_wal_number,
+      const PredecessorWALInfo& predecessor_wal_info,
+      bool* const old_log_record, Status* const reporter_status,
+      DBOpenLogReporter* reporter, std::unique_ptr<log::Reader>& reader);
   Status ProcessLogRecord(
       Slice record, const std::unique_ptr<log::Reader>& reader,
       const UnorderedMap<uint32_t, size_t>& running_ts_sz, uint64_t wal_number,
@@ -2116,8 +2119,13 @@ class DBImpl : public DB {
       bool* stop_replay_for_corruption, uint64_t* corrupted_wal_number,
       bool* corrupted_wal_found);
 
-  void FinishLogFileProcessing(SequenceNumber const* const next_sequence,
-                               const Status& status);
+  Status UpdatePredecessorWALInfo(uint64_t wal_number,
+                                  const SequenceNumber* next_sequence,
+                                  const std::string& fname,
+                                  PredecessorWALInfo& predecessor_wal_info);
+
+  void FinishLogFileProcessing(const Status& status,
+                               const SequenceNumber* next_sequence);
 
   // Return `Status::Corruption()` when `stop_replay_for_corruption == true` and
   // exits inconsistency between SST and WAL data
@@ -2554,6 +2562,7 @@ class DBImpl : public DB {
 
   IOStatus CreateWAL(const WriteOptions& write_options, uint64_t log_file_num,
                      uint64_t recycle_log_number, size_t preallocate_block_size,
+                     const PredecessorWALInfo& predecessor_wal_info,
                      log::Writer** new_log);
 
   // Validate self-consistency of DB options

db/db_impl/db_impl_open.cc

@@ -754,6 +754,11 @@ Status DBImpl::Recover(
       }
     }
 
+    if (!is_new_db && immutable_db_options_.track_and_verify_wals &&
+        wal_files.empty()) {
+      return Status::Corruption("Opening an existing DB with no WAL files");
+    }
+
     if (immutable_db_options_.track_and_verify_wals_in_manifest) {
       if (!immutable_db_options_.best_efforts_recovery) {
         // Verify WALs in MANIFEST.
@@ -1190,14 +1195,15 @@ Status DBImpl::ProcessLogFiles(
   bool stop_replay_for_corruption = false;
   bool flushed = false;
   uint64_t corrupted_wal_number = kMaxSequenceNumber;
+  PredecessorWALInfo predecessor_wal_info;
 
   for (auto wal_number : wal_numbers) {
     if (status.ok()) {
-      status =
-          ProcessLogFile(wal_number, min_wal_number, is_retry, read_only,
-                         job_id, next_sequence, &stop_replay_for_corruption,
-                         &stop_replay_by_wal_filter, &corrupted_wal_number,
-                         corrupted_wal_found, version_edits, &flushed);
+      status = ProcessLogFile(
+          wal_number, min_wal_number, is_retry, read_only, job_id,
+          next_sequence, &stop_replay_for_corruption,
+          &stop_replay_by_wal_filter, &corrupted_wal_number,
+          corrupted_wal_found, version_edits, &flushed, predecessor_wal_info);
     }
   }
 
@@ -1218,7 +1224,8 @@ Status DBImpl::ProcessLogFile(
     int job_id, SequenceNumber* next_sequence, bool* stop_replay_for_corruption,
     bool* stop_replay_by_wal_filter, uint64_t* corrupted_wal_number,
     bool* corrupted_wal_found,
-    std::unordered_map<int, VersionEdit>* version_edits, bool* flushed) {
+    std::unordered_map<int, VersionEdit>* version_edits, bool* flushed,
+    PredecessorWALInfo& predecessor_wal_info) {
   assert(stop_replay_by_wal_filter);
 
   // Variable initialization starts
@@ -1265,14 +1272,19 @@ Status DBImpl::ProcessLogFile(
   }
 
   Status init_status = InitializeLogReader(
-      wal_number, is_retry, fname, &old_log_record, &status, &reporter, reader);
+      wal_number, is_retry, fname, *stop_replay_for_corruption, min_wal_number,
+      predecessor_wal_info, &old_log_record, &status, &reporter, reader);
+
+  // FIXME(hx235): Consolidate `!init_status.ok()` and `reader == nullptr` cases
   if (!init_status.ok()) {
     assert(status.ok());
     status.PermitUncheckedError();
     return init_status;
   } else if (reader == nullptr) {
     // TODO(hx235): remove this case since it's confusing
     assert(status.ok());
+    // Fail initializing log reader for one log file with an ok status.
+    // Try next one.
     return status;
   }
 
@@ -1312,13 +1324,19 @@ Status DBImpl::ProcessLogFile(
                  "Recovered to log #%" PRIu64 " seq #%" PRIu64, wal_number,
                  *next_sequence);
 
+  if (status.ok()) {
+    status = UpdatePredecessorWALInfo(wal_number, next_sequence, fname,
+                                      predecessor_wal_info);
+  }
+
   if (!status.ok() || old_log_record) {
     status = HandleNonOkStatusOrOldLogRecord(
         wal_number, next_sequence, status, &old_log_record,
         stop_replay_for_corruption, corrupted_wal_number, corrupted_wal_found);
   }
 
-  FinishLogFileProcessing(next_sequence, status);
+  FinishLogFileProcessing(status, next_sequence);
+
   return status;
 }
 
@@ -1333,12 +1351,12 @@ void DBImpl::SetupLogFileProcessing(uint64_t wal_number) {
                  static_cast<int>(immutable_db_options_.wal_recovery_mode));
 }
 
-Status DBImpl::InitializeLogReader(uint64_t wal_number, bool is_retry,
-                                   std::string& fname,
-                                   bool* const old_log_record,
-                                   Status* const reporter_status,
-                                   DBOpenLogReporter* reporter,
-                                   std::unique_ptr<log::Reader>& reader) {
+Status DBImpl::InitializeLogReader(
+    uint64_t wal_number, bool is_retry, std::string& fname,
+    bool stop_replay_for_corruption, uint64_t min_wal_number,
+    const PredecessorWALInfo& predecessor_wal_info, bool* const old_log_record,
+    Status* const reporter_status, DBOpenLogReporter* reporter,
+    std::unique_ptr<log::Reader>& reader) {
   assert(old_log_record);
   assert(reporter_status);
   assert(reporter);
@@ -1376,9 +1394,11 @@ Status DBImpl::InitializeLogReader(uint64_t wal_number, bool is_retry,
   // paranoid_checks==false so that corruptions cause entire commits
   // to be skipped instead of propagating bad information (like overly
   // large sequence numbers).
-  reader.reset(new log::Reader(immutable_db_options_.info_log,
-                               std::move(file_reader), reporter,
-                               true /*checksum*/, wal_number));
+  reader.reset(new log::Reader(
+      immutable_db_options_.info_log, std::move(file_reader), reporter,
+      true /*checksum*/, wal_number,
+      immutable_db_options_.track_and_verify_wals, stop_replay_for_corruption,
+      min_wal_number, predecessor_wal_info));
   return status;
 }
 
@@ -1631,8 +1651,26 @@ Status DBImpl::HandleNonOkStatusOrOldLogRecord(
     return status;
   }
 }
-void DBImpl::FinishLogFileProcessing(SequenceNumber const* const next_sequence,
-                                     const Status& status) {
+
+Status DBImpl::UpdatePredecessorWALInfo(
+    uint64_t wal_number, const SequenceNumber* next_sequence,
+    const std::string& fname, PredecessorWALInfo& predecessor_wal_info) {
+  uint64_t bytes;
+
+  Status s = env_->GetFileSize(fname, &bytes);
+  if (!s.ok()) {
+    return s;
+  }
+
+  assert(next_sequence);
+  predecessor_wal_info = PredecessorWALInfo(
+      wal_number, bytes,
+      *next_sequence == kMaxSequenceNumber ? 0 : *next_sequence - 1);
+  return s;
+}
+
+void DBImpl::FinishLogFileProcessing(const Status& status,
+                                     const SequenceNumber* next_sequence) {
   if (status.ok()) {
     assert(next_sequence);
     flush_scheduler_.Clear();
@@ -2194,6 +2232,7 @@ Status DB::OpenAndTrimHistory(
 IOStatus DBImpl::CreateWAL(const WriteOptions& write_options,
                            uint64_t log_file_num, uint64_t recycle_log_number,
                            size_t preallocate_block_size,
+                           const PredecessorWALInfo& predecessor_wal_info,
                            log::Writer** new_log) {
   IOStatus io_s;
   std::unique_ptr<FSWritableFile> lfile;
@@ -2237,9 +2276,15 @@ IOStatus DBImpl::CreateWAL(const WriteOptions& write_options,
     *new_log = new log::Writer(std::move(file_writer), log_file_num,
                                immutable_db_options_.recycle_log_file_num > 0,
                                immutable_db_options_.manual_wal_flush,
-                               immutable_db_options_.wal_compression);
+                               immutable_db_options_.wal_compression,
+                               immutable_db_options_.track_and_verify_wals);
     io_s = (*new_log)->AddCompressionTypeRecord(write_options);
+    if (io_s.ok()) {
+      io_s = (*new_log)->MaybeAddPredecessorWALInfo(write_options,
+                                                    predecessor_wal_info);
+    }
   }
+
   return io_s;
 }
 
@@ -2332,8 +2377,14 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
     log::Writer* new_log = nullptr;
     const size_t preallocate_block_size =
         impl->GetWalPreallocateBlockSize(max_write_buffer_size);
+    // TODO(hx235): Pass in the correct `predecessor_wal_info` for the first WAL
+    // created during DB open with predecessor WALs from previous DB session due
+    // to `avoid_flush_during_recovery == true`. This can protect the last WAL
+    // recovered.
     s = impl->CreateWAL(write_options, new_log_number, 0 /*recycle_log_number*/,
-                        preallocate_block_size, &new_log);
+                        preallocate_block_size,
+                        PredecessorWALInfo() /* predecessor_wal_info */,
+                        &new_log);
     if (s.ok()) {
       // Prevent log files created by previous instance from being recycled.
       // They might be in alive_log_file_, and might get recycled otherwise.

db/db_impl/db_impl_write.cc

@@ -1584,7 +1584,8 @@ IOStatus DBImpl::WriteToWAL(const WriteBatch& merged_batch,
   if (!io_s.ok()) {
     return io_s;
   }
-  io_s = log_writer->AddRecord(write_options, log_entry);
+  io_s = log_writer->AddRecord(write_options, log_entry,
+                               WriteBatchInternal::Sequence(&merged_batch));
 
   if (UNLIKELY(needs_locking)) {
     log_write_mutex_.Unlock();
@@ -2445,8 +2446,16 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
   if (creating_new_log) {
     // TODO: Write buffer size passed in should be max of all CF's instead
     // of mutable_cf_options.write_buffer_size.
+    PredecessorWALInfo info;
+    if (!logs_.empty()) {
+      log::Writer* cur_log_writer = logs_.back().writer;
+      info = PredecessorWALInfo(cur_log_writer->get_log_number(),
+                                cur_log_writer->file()->GetFileSize(),
+                                cur_log_writer->GetLastSeqnoRecorded());
+    }
+
     io_s = CreateWAL(write_options, new_log_number, recycle_log_number,
-                     preallocate_block_size, &new_log);
+                     preallocate_block_size, info, &new_log);
     if (s.ok()) {
       s = io_s;
     }

db/dbformat.h

@@ -1179,4 +1179,54 @@ struct ParsedInternalKeyComparator {
   const InternalKeyComparator* cmp;
 };
 
+class PredecessorWALInfo {
+ public:
+  PredecessorWALInfo()
+      : log_number_(0),
+        size_bytes_(0),
+        last_seqno_recorded_(0),
+        initialized_(false) {}
+
+  explicit PredecessorWALInfo(uint64_t log_number, uint64_t size_bytes,
+                              SequenceNumber last_seqno_recorded)
+      : log_number_(log_number),
+        size_bytes_(size_bytes),
+        last_seqno_recorded_(last_seqno_recorded),
+        initialized_(true) {}
+
+  uint64_t GetLogNumber() const { return log_number_; }
+
+  uint64_t GetSizeBytes() const { return size_bytes_; }
+
+  SequenceNumber GetLastSeqnoRecorded() const { return last_seqno_recorded_; }
+
+  bool IsInitialized() const { return initialized_; }
+
+  inline void EncodeTo(std::string* dst) const {
+    assert(dst != nullptr);
+    PutFixed64(dst, log_number_);
+    PutFixed64(dst, size_bytes_);
+    PutFixed64(dst, last_seqno_recorded_);
+  }
+
+  inline Status DecodeFrom(Slice* src) {
+    if (!GetFixed64(src, &log_number_)) {
+      return Status::Corruption("Error decoding log number");
+    }
+    if (!GetFixed64(src, &size_bytes_)) {
+      return Status::Corruption("Error decoding size bytes");
+    }
+    if (!GetFixed64(src, &last_seqno_recorded_)) {
+      return Status::Corruption("Error decoding last seqno recorded");
+    }
+    initialized_ = true;
+    return Status::OK();
+  }
+
+ private:
+  uint64_t log_number_;
+  uint64_t size_bytes_;
+  SequenceNumber last_seqno_recorded_;
+  bool initialized_;
+};
 }  // namespace ROCKSDB_NAMESPACE

db/log_format.h

@@ -40,10 +40,15 @@ enum RecordType : int {
   kUserDefinedTimestampSizeType = 10,
   kRecyclableUserDefinedTimestampSizeType = 11,
 
-  // Reserve kRecordTypeSafeIgnoreMask = 16
+  // Reserve kRecordTypeSafeIgnoreMask = 32
+  // Avoid kMaxRecordType + 6
+
+  // For WAL verification
+  kPredecessorWALInfoType = 101,
+  kRecyclePredecessorWALInfoType = 102,
 };
-const int kRecordTypeSafeIgnoreMask = 1 << 4;
-constexpr int kMaxRecordType = kRecyclableUserDefinedTimestampSizeType;
+const int kRecordTypeSafeIgnoreMask = 1 << 5;
+constexpr int kMaxRecordType = kRecyclePredecessorWALInfoType;
 
 constexpr unsigned int kBlockSize = 32768;