[Backport 2.6] [#8773] PITR: Add DDL log

Summary:
It is a frequent case then we would like to restore the DB state to some time right before DDL operation.
For instance before the drop table.

This diff adds a DDL log, which could be seen using yb-admin.
So user could easily pick up time for restore snapshot schedule operation using this log.

Also changed UI to render time using local timezone instead of UTC.

Original commit: D11744 / a4ea9e4

Test Plan:
ybd --cxx-test yb-admin-test --gtest_filter AdminCliTest.DdlLog
Jenkins: rebase: 2.6

Reviewers: bogdan

Reviewed By: bogdan

Subscribers: ybase

Differential Revision: https://phabricator.dev.yugabyte.com/D11952

ent/src/yb/master/catalog_manager_ent.cc

@@ -735,7 +735,6 @@ Status CatalogManager::ImportSnapshotPreprocess(const SysSnapshotEntryPB& snapsh
       case SysRowEntry::TABLET: // Preprocess original tablets.
         RETURN_NOT_OK(PreprocessTabletEntry(entry, tables_data));
         break;
-      case SysRowEntry::UNKNOWN: FALLTHROUGH_INTENDED;
       case SysRowEntry::CLUSTER_CONFIG: FALLTHROUGH_INTENDED;
       case SysRowEntry::REDIS_CONFIG: FALLTHROUGH_INTENDED;
       case SysRowEntry::UDTYPE: FALLTHROUGH_INTENDED;
@@ -744,7 +743,9 @@ Status CatalogManager::ImportSnapshotPreprocess(const SysSnapshotEntryPB& snapsh
       case SysRowEntry::CDC_STREAM: FALLTHROUGH_INTENDED;
       case SysRowEntry::UNIVERSE_REPLICATION: FALLTHROUGH_INTENDED;
       case SysRowEntry::SNAPSHOT:  FALLTHROUGH_INTENDED;
-      case SysRowEntry::SNAPSHOT_SCHEDULE:
+      case SysRowEntry::SNAPSHOT_SCHEDULE: FALLTHROUGH_INTENDED;
+      case SysRowEntry::DDL_LOG_ENTRY: FALLTHROUGH_INTENDED;
+      case SysRowEntry::UNKNOWN:
         FATAL_INVALID_ENUM_VALUE(SysRowEntry::Type, entry.type());
     }
   }

ent/src/yb/tools/yb-admin_cli_ent.cc

@@ -38,25 +38,26 @@ using strings::Substitute;
 
 namespace {
 
-Result<HybridTime> ParseHybridTime(const string& timestamp) {
+Result<HybridTime> ParseHybridTime(string input) {
   // Acceptable system time formats:
   //  1. HybridTime Timestamp (in Microseconds)
   //  2. -Interval
   //  3. Human readable string
-  auto ts = boost::trim_copy(timestamp);
+  boost::trim(input);
 
   HybridTime ht;
   // The HybridTime is given in milliseconds and will contain 16 chars.
   static const std::regex int_regex("[0-9]{16}");
-  if (std::regex_match(ts, int_regex)) {
-    return HybridTime::FromMicros(std::stoul(ts));
+  if (std::regex_match(input, int_regex)) {
+    return HybridTime::FromMicros(std::stoul(input));
   }
-  if (!ts.empty() && ts[0] == '-') {
+  if (!input.empty() && input[0] == '-') {
     return HybridTime::FromMicros(
         VERIFY_RESULT(WallClock()->Now()).time_point -
-        VERIFY_RESULT(DateTime::IntervalFromString(ts.substr(1))).ToMicroseconds());
+        VERIFY_RESULT(DateTime::IntervalFromString(input.substr(1))).ToMicroseconds());
   }
-  return HybridTime::FromMicros(VERIFY_RESULT(DateTime::TimestampFromString(ts)).ToInt64());;
+  auto ts = VERIFY_RESULT(DateTime::TimestampFromString(input, DateTime::HumanReadableInputFormat));
+  return HybridTime::FromMicros(ts.ToInt64());
 }
 
 const string kMinus = "minus";
@@ -74,20 +75,6 @@ Result<T> GetOptionalArg(const Args& args, size_t idx) {
   return VERIFY_RESULT(T::FromString(args[idx]));
 }
 
-CHECKED_STATUS CheckArgumentsCount(int count, int min, int max) {
-  if (count < min) {
-    return STATUS_FORMAT(
-        InvalidArgument, "Too few arguments $0, should be in range [$1, $2]", count, min, max);
-  }
-
-  if (count > max) {
-    return STATUS_FORMAT(
-        InvalidArgument, "Too many arguments $0, should be in range [$1, $2]", count, min, max);
-  }
-
-  return Status::OK();
-}
-
 } // namespace
 
 void ClusterAdminCli::RegisterCommandHandlers(ClusterAdminClientClass* client) {

ent/src/yb/tools/yb-admin_client_ent.cc

@@ -94,21 +94,6 @@ using master::SysSnapshotEntryPB;
 
 PB_ENUM_FORMATTERS(yb::master::SysSnapshotEntryPB::State);
 
-namespace {
-
-void AddStringField(
-    const char* name, const std::string& value, rapidjson::Value* out,
-    rapidjson::Value::AllocatorType* allocator) {
-  rapidjson::Value json_value(value.c_str(), *allocator);
-  out->AddMember(rapidjson::StringRef(name), json_value, *allocator);
-}
-
-string HybridTimeToString(HybridTime ht) {
-  return Timestamp(ht.GetPhysicalValueMicros()).ToFormattedString();
-}
-
-} // namespace
-
 Status ClusterAdminClient::ListSnapshots(bool show_details, bool show_restored, bool show_deleted) {
   RpcController rpc;
   rpc.set_timeout(timeout_);
@@ -381,14 +366,14 @@ Result<rapidjson::Document> ClusterAdminClient::ListSnapshotSchedules(
       AddStringField("id", VERIFY_RESULT(FullyDecodeTxnSnapshotId(snapshot.id())).ToString(),
                      &json_snapshot, &result.GetAllocator());
       auto snapshot_ht = HybridTime::FromPB(snapshot.entry().snapshot_hybrid_time());
-      AddStringField("snapshot_time_utc",
+      AddStringField("snapshot_time",
                      HybridTimeToString(snapshot_ht),
                      &json_snapshot, &result.GetAllocator());
       auto previous_snapshot_ht = HybridTime::FromPB(
           snapshot.entry().previous_snapshot_hybrid_time());
       if (previous_snapshot_ht) {
         AddStringField(
-            "previous_snapshot_time_utc",
+            "previous_snapshot_time",
             HybridTimeToString(previous_snapshot_ht),
             &json_snapshot, &result.GetAllocator());
       }

src/yb/common/json_util.cc

@@ -173,25 +173,19 @@ Status ConvertQLValuePBToRapidJson(const QLValuePB& ql_value_pb,
   return Status::OK();
 }
 
-std::string WriteRapidJsonToString(const rapidjson::Document& document) {
+std::string WriteRapidJsonToString(const rapidjson::Value& document) {
   rapidjson::StringBuffer buffer;
   rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
   document.Accept(writer);
   return std::string(buffer.GetString());
 }
 
-std::string PrettyWriteRapidJsonToString(const rapidjson::Document& document) {
+std::string PrettyWriteRapidJsonToString(const rapidjson::Value& document) {
   rapidjson::StringBuffer buffer;
   rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(buffer);
   document.Accept(writer);
   return std::string(buffer.GetString());
 }
 
-std::string WriteRapidJsonToString(const rapidjson::Value& value) {
-  rapidjson::Document document;
-  document.CopyFrom(value, document.GetAllocator());
-  return WriteRapidJsonToString(document);
-}
-
 } // namespace common
 } // namespace yb

src/yb/common/json_util.h

@@ -33,9 +33,8 @@ inline CHECKED_STATUS ConvertQLValuePBToRapidJson(const QLValuePB& value_pb,
   return ConvertQLValuePBToRapidJson(value_pb, document, &document->GetAllocator());
 }
 
-std::string WriteRapidJsonToString(const rapidjson::Document& document);
 std::string WriteRapidJsonToString(const rapidjson::Value& value);
-std::string PrettyWriteRapidJsonToString(const rapidjson::Document& document);
+std::string PrettyWriteRapidJsonToString(const rapidjson::Value& document);
 
 } // namespace common
 } // namespace yb

src/yb/integration-tests/cql_test_util.h

@@ -239,6 +239,11 @@ class CassandraSession {
 
   CHECKED_STATUS ExecuteQuery(const std::string& query);
 
+  template <class... Args>
+  CHECKED_STATUS ExecuteQueryFormat(const std::string& query, Args&&... args) {
+    return ExecuteQuery(Format(query, std::forward<Args>(args)...));
+  }
+
   Result<CassandraResult> ExecuteWithResult(const std::string& query);
 
   Result<std::string> ExecuteAndRenderToString(const std::string& statement);

src/yb/master/catalog_entity_info.cc

@@ -37,6 +37,8 @@
 #include "yb/util/format.h"
 #include "yb/util/locks.h"
 #include "yb/gutil/strings/substitute.h"
+
+#include "yb/common/doc_hybrid_time.h"
 #include "yb/common/wire_protocol.h"
 
 using std::string;
@@ -802,5 +804,31 @@ string UDTypeInfo::ToString() const {
   return Format("$0 [id=$1] {metadata=$2} ", name(), udtype_id_, l->pb);
 }
 
+DdlLogEntry::DdlLogEntry(
+    HybridTime time, const TableId& table_id, const SysTablesEntryPB& table,
+    const std::string& action) {
+  pb_.set_time(time.ToUint64());
+  pb_.set_table_type(table.table_type());
+  pb_.set_namespace_name(table.namespace_name());
+  pb_.set_namespace_id(table.namespace_id());
+  pb_.set_table_name(table.name());
+  pb_.set_table_id(table_id);
+  pb_.set_action(action);
+}
+
+const DdlLogEntryPB& DdlLogEntry::old_pb() const {
+  // Since DDL log entry are always added, we don't have previous PB for the same entry.
+  static const DdlLogEntryPB kEmpty;
+  return kEmpty;
+}
+
+const DdlLogEntryPB& DdlLogEntry::new_pb() const {
+  return pb_;
+}
+
+std::string DdlLogEntry::id() const {
+  return DocHybridTime(HybridTime(pb_.time()), kMaxWriteId).EncodedInDocDbFormat();
+}
+
 }  // namespace master
 }  // namespace yb

src/yb/master/catalog_entity_info.h

@@ -113,9 +113,9 @@ template <class PersistentDataEntryPB>
 class MetadataCowWrapper {
  public:
   // Type declaration for use in the Lock classes.
-  typedef PersistentDataEntryPB cow_state;
-  typedef CowWriteLock<cow_state> WriteLock;
-  typedef CowReadLock<cow_state> ReadLock;
+  typedef PersistentDataEntryPB CowState;
+  typedef CowWriteLock<CowState> WriteLock;
+  typedef CowReadLock<CowState> ReadLock;
 
   // This method should return the id to be written into the sys_catalog id column.
   virtual const std::string& id() const = 0;
@@ -139,6 +139,18 @@ class MetadataCowWrapper {
     return WriteLock(mutable_metadata());
   }
 
+  const auto& old_pb() const {
+    return metadata_.state().pb;
+  }
+
+  const auto& new_pb() const {
+    return metadata_.dirty().pb;
+  }
+
+  static auto type() {
+    return CowState::type();
+  }
+
  protected:
   virtual ~MetadataCowWrapper() = default;
   CowObject<PersistentDataEntryPB> metadata_;
@@ -806,6 +818,30 @@ class SysConfigInfo : public RefCountedThreadSafe<SysConfigInfo>,
   DISALLOW_COPY_AND_ASSIGN(SysConfigInfo);
 };
 
+class DdlLogEntry {
+ public:
+  // time - when DDL operation was started.
+  // table_id - modified table id.
+  // table - what table was modified during DDL.
+  // action - string description of DDL.
+  DdlLogEntry(
+      HybridTime time, const TableId& table_id, const SysTablesEntryPB& table,
+      const std::string& action);
+
+  static SysRowEntry::Type type() {
+    return SysRowEntry::DDL_LOG_ENTRY;
+  }
+
+  std::string id() const;
+
+  // Used by sys catalog writer. It requires 2 protobuf to check whether entry was actually changed.
+  const DdlLogEntryPB& new_pb() const;
+  const DdlLogEntryPB& old_pb() const;
+
+ protected:
+  DdlLogEntryPB pb_;
+};
+
 // Convenience typedefs.
 // Table(t)InfoMap ordered for deterministic locking.
 typedef std::map<TabletId, scoped_refptr<TabletInfo>> TabletInfoMap;