fix(cpp): preserve vertex IDs for nonzero builder starts

cpp/src/graphar/high-level/vertices_builder.cc

@@ -35,17 +35,9 @@ Status VerticesBuilder::validate(const Vertex& v, IdType index,
   if (validate_level == ValidateLevel::default_validate) {
     validate_level = validate_level_;
   }
-  // no validate
-  if (validate_level == ValidateLevel::no_validate) {
-    return Status::OK();
-  }
-
-  // weak validate
-  // can not add new vertices after dumping
-  if (is_saved_) {
-    return Status::Invalid(
-        "The vertices builder has been saved, can not add "
-        "new vertices any more");
+  if (start_vertex_index_ < 0) {
+    return Status::IndexError("The start vertex index ", start_vertex_index_,
+                              " is smaller than 0");
   }
   // the start vertex index must be aligned with the chunk size
   if (start_vertex_index_ % vertex_info_->GetChunkSize() != 0) {
@@ -59,7 +51,18 @@ Status VerticesBuilder::validate(const Vertex& v, IdType index,
                               " is smaller than the start index ",
                               start_vertex_index_);
   }
+  // no validate
+  if (validate_level == ValidateLevel::no_validate) {
+    return Status::OK();
+  }
 
+  // weak validate
+  // can not add new vertices after dumping
+  if (is_saved_) {
+    return Status::Invalid(
+        "The vertices builder has been saved, can not add "
+        "new vertices any more");
+  }
   // strong validate
   if (validate_level == ValidateLevel::strong_validate) {
     for (auto& property : v.GetProperties()) {

cpp/src/graphar/high-level/vertices_builder.h

@@ -338,17 +338,14 @@ class VerticesBuilder {
       ValidateLevel validate_level = ValidateLevel::default_validate) {
     // validate
     GAR_RETURN_NOT_OK(validate(v, index, validate_level));
-    // add a vertex
-    if (index == -1) {
-      v.SetId(vertices_.size());
-      vertices_.push_back(v);
-    } else {
-      v.SetId(index);
-      if (index >= static_cast<IdType>(vertices_.size())) {
-        vertices_.resize(index + 1);
-      }
-      vertices_[index] = v;
+    const IdType local_index = index == -1
+                                   ? static_cast<IdType>(vertices_.size())
+                                   : index - start_vertex_index_;
+    v.SetId(start_vertex_index_ + local_index);
+    if (local_index >= static_cast<IdType>(vertices_.size())) {
+      vertices_.resize(local_index + 1);
     }
+    vertices_[local_index] = v;
     num_vertices_++;
     return Status::OK();
   }

cpp/test/test_builder.cc

@@ -101,6 +101,76 @@ TEST_CASE_METHOD(GlobalFixture, "Test_vertices_builder") {
   REQUIRE(indexed_vertex.GetId() == 7);
   REQUIRE(indexed_vertex.Empty());
 
+  SECTION("nonzero start vertex index") {
+    const IdType nonzero_start_index = vertex_info->GetChunkSize();
+    auto nonzero_builder =
+        builder::VerticesBuilder::Make(vertex_info, "/tmp/nonzero/",
+                                       nonzero_start_index)
+            .value();
+    nonzero_builder->SetValidateLevel(ValidateLevel::weak_validate);
+
+    builder::Vertex auto_indexed_vertex;
+    auto_indexed_vertex.AddProperty("id", int64_t{10});
+    REQUIRE(nonzero_builder->AddVertex(auto_indexed_vertex).ok());
+    REQUIRE(auto_indexed_vertex.GetId() == nonzero_start_index);
+
+    builder::Vertex explicitly_indexed_vertex;
+    explicitly_indexed_vertex.AddProperty("id", int64_t{11});
+    REQUIRE(nonzero_builder
+                ->AddVertex(explicitly_indexed_vertex, nonzero_start_index + 1)
+                .ok());
+    REQUIRE(explicitly_indexed_vertex.GetId() == nonzero_start_index + 1);
+    REQUIRE(nonzero_builder->Dump().ok());
+
+    auto nonzero_chunk =
+        "/tmp/nonzero/vertex/person/id/chunk" +
+        std::to_string(nonzero_start_index / vertex_info->GetChunkSize());
+    std::unique_ptr<parquet::arrow::FileReader> nonzero_reader;
+    REQUIRE(graphar::util::OpenParquetArrowReader(
+                nonzero_chunk, arrow::default_memory_pool(), &nonzero_reader)
+                .ok());
+    auto maybe_nonzero_table = ReadParquetTable(nonzero_reader.get());
+    REQUIRE(maybe_nonzero_table.ok());
+    auto nonzero_table = maybe_nonzero_table.ValueOrDie();
+    REQUIRE(nonzero_table->num_rows() == 2);
+    auto vertex_index_array = std::static_pointer_cast<arrow::Int64Array>(
+        nonzero_table->GetColumnByName("_graphArVertexIndex")->chunk(0));
+    REQUIRE(vertex_index_array->Value(0) == nonzero_start_index);
+    REQUIRE(vertex_index_array->Value(1) == nonzero_start_index + 1);
+
+    auto no_validate_builder =
+        builder::VerticesBuilder::Make(vertex_info, "/tmp/nonzero-no-validate/",
+                                       nonzero_start_index)
+            .value();
+    builder::Vertex below_start_vertex;
+    below_start_vertex.AddProperty("id", int64_t{9});
+    REQUIRE(
+        no_validate_builder->AddVertex(below_start_vertex, 5).IsIndexError());
+    REQUIRE_FALSE(below_start_vertex.HasId());
+    REQUIRE(no_validate_builder->GetNum() == 0);
+
+    auto unaligned_builder = builder::VerticesBuilder::Make(
+                                 vertex_info, "/tmp/unaligned-no-validate/",
+                                 nonzero_start_index + 50)
+                                 .value();
+    builder::Vertex unaligned_vertex;
+    unaligned_vertex.AddProperty("id", int64_t{10});
+    REQUIRE(unaligned_builder->AddVertex(unaligned_vertex).IsIndexError());
+    REQUIRE_FALSE(unaligned_vertex.HasId());
+    REQUIRE(unaligned_builder->GetNum() == 0);
+
+    auto negative_start_builder =
+        builder::VerticesBuilder::Make(
+            vertex_info, "/tmp/negative-no-validate/", -nonzero_start_index)
+            .value();
+    builder::Vertex negative_start_vertex;
+    negative_start_vertex.AddProperty("id", int64_t{10});
+    REQUIRE(negative_start_builder->AddVertex(negative_start_vertex)
+                .IsIndexError());
+    REQUIRE_FALSE(negative_start_vertex.HasId());
+    REQUIRE(negative_start_builder->GetNum() == 0);
+  }
+
   // add vertices
   std::ifstream fp(test_data_dir + "/ldbc_sample/person_0_0.csv");
   std::string line;