From 30742e1692ed43c9f2fe4637d255c5c37c06d025 Mon Sep 17 00:00:00 2001
From: Abhishek Kona <abhishekk@fb.com>
Date: Mon, 12 Nov 2012 16:45:45 -0800
Subject: [PATCH] LDB can read WAL.

Summary:
Add option to read WAL and print a summary for each record.
facebook task => #1885013

E.G. Output :
./ldb dump_wal --walfile=/tmp/leveldbtest-5907/dbbench/026122.log --header
Sequence,Count,ByteSize
49981,1,100033
49981,1,100033
49982,1,100033
49981,1,100033
49982,1,100033
49983,1,100033
49981,1,100033
49982,1,100033
49983,1,100033
49984,1,100033
49981,1,100033
49982,1,100033

Test Plan:
Works run
./ldb read_wal --wal-file=/tmp/leveldbtest-5907/dbbench/000078.log --header

Reviewers: dhruba, heyongqiang

Reviewed By: dhruba

CC: emayanke, leveldb, zshao

Differential Revision: https://reviews.facebook.net/D6675
---
 tools/ldb.cc    | 12 ++++++---
 util/ldb_cmd.cc | 68 +++++++++++++++++++++++++++++++++++++++++++++++++
 util/ldb_cmd.h  | 27 ++++++++++++++++++--
 3 files changed, 102 insertions(+), 5 deletions(-)

diff --git a/tools/ldb.cc b/tools/ldb.cc
index d594d500e..c4d7c714b 100644
--- a/tools/ldb.cc
+++ b/tools/ldb.cc
@@ -26,6 +26,10 @@ public:
     ret.append(" reduce_levels ");
     ReduceDBLevels::Help(ret);
 
+    ret.append("\n---dump_wal----:\n");
+    ret.append(exec_name);
+    ret.append(" dump_wal ");
+    WALDumper::Help(ret);
     fprintf(stderr, "%s\n", ret.c_str());
   }
 
@@ -46,15 +50,17 @@ public:
     }
 
     LDBCommand* cmdObj = NULL;
-    if (strncmp(cmd, "compact", strlen("compact")) == 0) {
+    if (strcmp(cmd, "compact") == 0) {
       // run compactor
       cmdObj = new Compactor(db_name, args);
-    } else if (strncmp(cmd, "dump", strlen("dump")) == 0) {
+    } else if (strcmp(cmd, "dump") == 0) {
       // run dump
       cmdObj = new DBDumper(db_name, args);
-    } else if (strncmp(cmd, "reduce_levels", strlen("reduce_levels")) == 0) {
+    } else if (strcmp(cmd, "reduce_levels") == 0) {
       // reduce db levels
       cmdObj = new ReduceDBLevels(db_name, args);
+    } else if (strcmp(cmd, "dump_wal") == 0) {
+      cmdObj = new WALDumper(args);
     } else {
       fprintf(stderr, "Unknown command: %s\n", cmd);
       PrintHelp(argv[0]);
diff --git a/util/ldb_cmd.cc b/util/ldb_cmd.cc
index 4b266a7df..60c3f9b43 100644
--- a/util/ldb_cmd.cc
+++ b/util/ldb_cmd.cc
@@ -2,6 +2,11 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+
+#include "leveldb/write_batch.h"
+#include "db/dbformat.h"
+#include "db/log_reader.h"
+#include "db/write_batch_internal.h"
 #include "util/ldb_cmd.h"
 
 namespace leveldb {
@@ -340,4 +345,67 @@ void ReduceDBLevels::DoCommand() {
   }
 }
 
+const char* WALDumper::WAL_FILE_ARG = "--walfile=";
+WALDumper::WALDumper(std::vector<std::string>& args) :
+  LDBCommand(args), print_header_(false) {
+  wal_file_.clear();
+  for (unsigned int i = 0; i < args.size(); i++) {
+    std::string& arg = args.at(i);
+    if (arg.find("--header") == 0) {
+      print_header_ = true;
+    } else if (arg.find(WAL_FILE_ARG) == 0) {
+      wal_file_ = arg.substr(strlen(WAL_FILE_ARG));
+    } else {
+      exec_state_ = LDBCommandExecuteResult::FAILED("Unknown argument " + arg);
+    }
+  }
+  if (wal_file_.empty()) {
+    exec_state_ = LDBCommandExecuteResult::FAILED("Argument --walfile reqd.");
+  }
+}
+
+void WALDumper::Help(std::string& ret) {
+  ret.append("--walfile write_ahead_log ");
+  ret.append("[--header print's a header] ");
+}
+
+void WALDumper::DoCommand() {
+  struct StdErrReporter : public log::Reader::Reporter {
+    virtual void Corruption(size_t bytes, const Status& s) {
+      std::cerr<<"Corruption detected in log file "<<s.ToString()<<"\n";
+    }
+  };
+
+  SequentialFile* file;
+  Env* env_ = Env::Default();
+  Status status = env_->NewSequentialFile(wal_file_, &file);
+  if (!status.ok()) {
+    exec_state_ = LDBCommandExecuteResult::FAILED("Failed to open WAL file " +
+      status.ToString());
+  } else {
+    StdErrReporter reporter;
+    log::Reader reader(file, &reporter, true, 0);
+    std::string scratch;
+    WriteBatch batch;
+    Slice record;
+    std::stringstream row;
+    if (print_header_) {
+      std::cout<<"Sequence,Count,ByteSize\n";
+    }
+    while(reader.ReadRecord(&record, &scratch)) {
+      row.clear();
+      if (record.size() < 12) {
+        reporter.Corruption(
+            record.size(), Status::Corruption("log record too small"));
+      } else {
+        WriteBatchInternal::SetContents(&batch, record);
+        row<<WriteBatchInternal::Sequence(&batch)<<",";
+        row<<WriteBatchInternal::Count(&batch)<<",";
+        row<<WriteBatchInternal::ByteSize(&batch)<<"\n";
+      }
+      std::cout<<row.str();
+    }
+  }
+}
+
 }
diff --git a/util/ldb_cmd.h b/util/ldb_cmd.h
index 45cd20398..ab56c4d68 100644
--- a/util/ldb_cmd.h
+++ b/util/ldb_cmd.h
@@ -13,6 +13,7 @@
 #include <stdio.h>
 
 #include "leveldb/db.h"
+#include "leveldb/env.h"
 #include "leveldb/options.h"
 #include "leveldb/iterator.h"
 #include "leveldb/slice.h"
@@ -41,7 +42,6 @@ public:
     std::string ret;
     switch (state_) {
     case EXEC_SUCCEED:
-      ret.append("Succeeded.");
       break;
     case EXEC_FAILED:
       ret.append("Failed.");
@@ -97,6 +97,11 @@ public:
     db_(NULL) {
   }
 
+  LDBCommand(std::vector<std::string>& args) :
+    db_path_(""),
+    db_(NULL) {
+  }
+
   virtual leveldb::Options PrepareOptionsForOpenDB()  {
     leveldb::Options opt;
     opt.create_if_missing = false;
@@ -262,6 +267,24 @@ private:
   Status GetOldNumOfLevels(leveldb::Options& opt, int* levels);
 };
 
-}
+class WALDumper : public LDBCommand {
+public:
+
+  WALDumper (std::vector<std::string>& args);
+
+  ~WALDumper() {}
+
+  virtual bool  NoDBOpen() {
+    return true;
+  }
+
+  static void Help(std::string& ret);
+  virtual void DoCommand();
+private:
+  bool print_header_;
+  std::string wal_file_;
 
+  static const char* WAL_FILE_ARG;
+};
+}
 #endif