add db-upgrade V3

src/tools/db-upgrade/CMakeLists.txt

@@ -5,6 +5,7 @@ nebula_add_executable(
         DbUpgraderTool.cpp
         NebulaKeyUtilsV1.cpp
         NebulaKeyUtilsV2.cpp
+        NebulaKeyUtilsV3.cpp
         DbUpgrader.cpp
     OBJECTS
         $<TARGET_OBJECTS:meta_service_handler>

src/tools/db-upgrade/DbUpgrader.cpp

@@ -9,8 +9,10 @@
 #include "common/fs/FileUtils.h"
 #include "common/utils/IndexKeyUtils.h"
 #include "common/utils/NebulaKeyUtils.h"
+#include "rocksdb/sst_file_writer.h"
 #include "tools/db-upgrade/NebulaKeyUtilsV1.h"
 #include "tools/db-upgrade/NebulaKeyUtilsV2.h"
+#include "tools/db-upgrade/NebulaKeyUtilsV3.h"
 
 DEFINE_string(src_db_path,
               "",
@@ -882,6 +884,107 @@ std::string UpgraderSpace::encodeRowVal(const RowReader* reader,
   return std::move(rowWrite).moveEncodedStr();
 }
 
+void UpgraderSpace::runPartV3() {
+  std::chrono::milliseconds take_dura{10};
+  if (auto pId = partQueue_.try_take_for(take_dura)) {
+    PartitionID partId = *pId;
+    // Handle vertex and edge, if there is an index, generate index data
+    LOG(INFO) << "Start to handle vertex/edge/index data in space id " << spaceId_ << " part id "
+              << partId;
+    auto prefix = NebulaKeyUtilsV3::partTagPrefix(partId);
+    std::unique_ptr<kvstore::KVIterator> iter;
+    auto retCode = readEngine_->prefix(prefix, &iter);
+    if (retCode != nebula::cpp2::ErrorCode::SUCCEEDED) {
+      LOG(ERROR) << "Space id " << spaceId_ << " part " << partId << " no found!";
+      LOG(ERROR) << "Handle vertex/edge/index data in space id " << spaceId_ << " part id "
+                 << partId << " failed";
+
+      auto unFinishedPart = --unFinishedPart_;
+      if (unFinishedPart == 0) {
+        // all parts has finished
+        LOG(INFO) << "Handle last part: " << partId << " vertex/edge/index data in space id "
+                  << spaceId_ << " finished";
+      } else {
+        pool_->add(std::bind(&UpgraderSpace::runPartV3, this));
+      }
+      return;
+    }
+    auto write_sst = [&, this](const std::vector<kvstore::KV>& data) {
+      ::rocksdb::Options option;
+      option.create_if_missing = true;
+      option.compression = ::rocksdb::CompressionType::kNoCompression;
+      ::rocksdb::SstFileWriter sst_file_writer(::rocksdb::EnvOptions(), option);
+      std::string file = ::fmt::format(
+          ".nebula_upgrade.space-{}.part-{}.{}.sst", spaceId_, partId, std::time(nullptr));
+      ::rocksdb::Status s = sst_file_writer.Open(file);
+      if (!s.ok()) {
+        LOG(FATAL) << "Faild upgrade V3 of space " << spaceId_ << ", part " << partId << ":"
+                   << s.code();
+      }
+      for (auto item : data) {
+        sst_file_writer.Put(item.first, item.second);
+      }
+      s = sst_file_writer.Finish();
+      if (!s.ok()) {
+        LOG(FATAL) << "Faild upgrade V3 of space " << spaceId_ << ", part " << partId << ":"
+                   << s.code();
+      }
+      std::lock_guard<std::mutex> lck(this->ingest_sst_file_mut_);
+      ingest_sst_file_.push_back(file);
+    };
+    std::vector<kvstore::KV> data;
+    std::string lastVertexKey = "";
+    while (iter && iter->valid()) {
+      auto vertex = NebulaKeyUtilsV3::getVertexKey(iter->key());
+      if (vertex == lastVertexKey) {
+        continue;
+      }
+      data.emplace_back(vertex, "");
+      lastVertexKey = vertex;
+      if (data.size() >= 100000) {
+        write_sst(data);
+        data.clear();
+      }
+    }
+    if (!data.empty()) {
+      write_sst(data);
+      data.clear();
+    }
+    LOG(INFO) << "Handle vertex/edge/index data in space id " << spaceId_ << " part id " << partId
+              << " succeed";
+
+    auto unFinishedPart = --unFinishedPart_;
+    if (unFinishedPart == 0) {
+      // all parts has finished
+      LOG(INFO) << "Handle last part: " << partId << " vertex/edge/index data in space id "
+                << spaceId_ << " finished.";
+    } else {
+      pool_->add(std::bind(&UpgraderSpace::runPartV3, this));
+    }
+  } else {
+    LOG(INFO) << "Handle vertex/edge/index of parts data in space id " << spaceId_ << " finished";
+  }
+}
+void UpgraderSpace::doProcessV3() {
+  LOG(INFO) << "Start to handle data in space id " << spaceId_;
+  // Parallel process part
+  auto partConcurrency = std::min(static_cast<size_t>(FLAGS_max_concurrent_parts), parts_.size());
+  LOG(INFO) << "Max concurrent parts: " << partConcurrency;
+  unFinishedPart_ = parts_.size();
+
+  LOG(INFO) << "Start to handle vertex/edge/index of parts data in space id " << spaceId_;
+  for (size_t i = 0; i < partConcurrency; ++i) {
+    pool_->add(std::bind(&UpgraderSpace::runPartV3, this));
+  }
+
+  while (unFinishedPart_ != 0) {
+    sleep(10);
+  }
+  auto code = readEngine_->ingest(ingest_sst_file_, true);
+  if (code != ::nebula::cpp2::ErrorCode::SUCCEEDED) {
+    LOG(FATAL) << "Faild upgrade V3 when ingest sst file:" << static_cast<int>(code);
+  }
+}
 std::vector<std::string> UpgraderSpace::indexVertexKeys(
     PartitionID partId,
     VertexID& vId,
@@ -1096,8 +1199,10 @@ void DbUpgrader::doSpace() {
               << " begin";
     if (FLAGS_upgrade_version == 1) {
       upgraderSpaceIter->doProcessV1();
-    } else {
+    } else if (FLAGS_upgrade_version == 2) {
       upgraderSpaceIter->doProcessV2();
+    } else {
+      upgraderSpaceIter->doProcessV3();
     }
 
     auto ret = upgraderSpaceIter->copyWal();

src/tools/db-upgrade/DbUpgrader.h

@@ -55,6 +55,9 @@ class UpgraderSpace {
   // Processing v2 Rc data upgrade to v2 Ga
   void doProcessV2();
 
+  // Processing v2 Ga data upgrade to v3
+  void doProcessV3();
+
   // Perform manual compact
   void doCompaction();
 
@@ -111,6 +114,8 @@ class UpgraderSpace {
 
   void runPartV2();
 
+  void runPartV3();
+
  public:
   // Source data path
   std::string srcPath_;
@@ -159,6 +164,9 @@ class UpgraderSpace {
   folly::UnboundedBlockingQueue<PartitionID> partQueue_;
 
   std::atomic<size_t> unFinishedPart_;
+
+  std::mutex ingest_sst_file_mut_;
+  std::vector<std::string> ingest_sst_file_;
 };
 
 // Upgrade one data path in storage conf

src/tools/db-upgrade/DbUpgraderTool.cpp

@@ -39,10 +39,11 @@ void printHelp() {
          A list of meta severs' ip:port separated by comma.
          Default: 127.0.0.1:45500
 
-       --upgrade_version=<1|2>
-         This tool can only upgrade 1.x data or 2.0 RC data.
+       --upgrade_version=<1|2|3>
+         This tool can only upgrade 1.x data, 2.0 RC, or 2.0 GA data.
          When the value is 1, upgrade the data from 1.x to 2.0 GA.
          When the value is 2, upgrade the data from 2.0 RC to 2.0 GA.
+         When the Value is 3, upgrade the data from 2.0 GA to 3.0
          Default: 0
 
  optional:
@@ -164,9 +165,9 @@ int main(int argc, char* argv[]) {
   CHECK_NOTNULL(schemaMan);
   CHECK_NOTNULL(indexMan);
 
-  if (FLAGS_upgrade_version != 1 && FLAGS_upgrade_version != 2) {
+  if (FLAGS_upgrade_version != 1 && FLAGS_upgrade_version != 2 && FLAGS_upgrade_version != 3) {
     LOG(ERROR) << "Flag upgrade_version : " << FLAGS_upgrade_version
-               << " illegal, upgrade_version can only be 1 or 2";
+               << " illegal, upgrade_version can only be 1,2,3";
     return EXIT_FAILURE;
   }
   LOG(INFO) << "Prepare phase end";

src/tools/db-upgrade/NebulaKeyUtilsV3.cpp

@@ -0,0 +1,22 @@
+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#include "tools/db-upgrade/NebulaKeyUtilsV3.h"
+
+namespace nebula {
+std::string NebulaKeyUtilsV3::partTagPrefix(PartitionID partId) {
+  PartitionID item = (partId << kPartitionOffset) | static_cast<uint32_t>(kTag_);
+  std::string key;
+  key.reserve(sizeof(PartitionID));
+  key.append(reinterpret_cast<const char*>(&item), sizeof(PartitionID));
+  return key;
+}
+std::string NebulaKeyUtilsV3::getVertexKey(folly::StringPiece tagKey) {
+  std::string key = tagKey.toString();
+  key[3] = static_cast<uint32_t>(kVertex);
+  key.resize(key.size() - sizeof(TagID));
+  return key;
+}
+}  // namespace nebula

src/tools/db-upgrade/NebulaKeyUtilsV3.h

@@ -0,0 +1,17 @@
+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+#pragma once
+#include "common/utils/Types.h"
+namespace nebula {
+class NebulaKeyUtilsV3 {
+ public:
+  static std::string partTagPrefix(PartitionID partId);
+  static std::string getVertexKey(folly::StringPiece tagKey);
+
+ private:
+  enum NebulaKeyTypeV3 : uint32_t { kTag_ = 0x00000001, kVertex = 0x00000007 };
+};
+
+}  // namespace nebula