Upgrade vineyard to v0.1.5. (#73)

.github/workflows/ci.yml

@@ -28,7 +28,7 @@ jobs:
       run:
         shell: scl enable devtoolset-7 -- bash --noprofile --norc -eo pipefail {0}
     container:
-      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:latest
+      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.1.5
       options:
         --shm-size 4096m
     strategy:
@@ -133,7 +133,7 @@ jobs:
     defaults:
       run:
         shell: scl enable devtoolset-7 -- bash --noprofile --norc -eo pipefail {0}
-    container: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:latest
+    container: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.1.5
     steps:
     - name: Install Dependencies
       run: |
@@ -173,7 +173,7 @@ jobs:
     defaults:
       run:
         shell: scl enable devtoolset-7 -- bash --noprofile --norc -eo pipefail {0}
-    container: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:latest
+    container: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.1.5
     steps:
     - name: Install Dependencies
       run: |
@@ -284,10 +284,14 @@ jobs:
     - uses: actions/[email protected]
 
     - uses: actions/download-artifact@v2
-    - name: Prepare environment
+
+    - name: Prepare environment for dataset
       run: |
         export GS_TEST_DIR='/root/gstest'
         git clone -b master --single-branch --depth=1 https://github.com/7br/gstest.git ${GS_TEST_DIR}
+
+    - name: Prepare environment for deps
+      run: |
         python3 -m pip install pytest-cov
         tar -xf ./gae-${{ github.sha }}/gae.tar
         tar -xf ./gle-${{ github.sha }}/gle.tar
@@ -302,11 +306,16 @@ jobs:
     - name: App Test with Expr Off
       if: matrix.experimental == 'OFF'
       run: |
+        # prelaunch the etcd
+        /usr/local/bin/etcd --data-dir=/dev/shm/default.etcd &
+
         export GS_TEST_DIR='/root/gstest'
         cd python
         python3 -m pytest --cov=graphscope --cov-config=.coveragerc --cov-report=xml \
                           --cov-report=term --exitfirst -s -v tests
 
+        pkill -TERM etcd || true
+
     - name: Upload Coverage
       if: matrix.experimental == 'OFF'
       uses: codecov/codecov-action@v1
@@ -317,12 +326,17 @@ jobs:
     - name: App Test with Expr On
       if: matrix.experimental == 'ON'
       run: |
+        # prelaunch the etcd
+        /usr/local/bin/etcd --data-dir=/dev/shm/default.etcd &
+
         export GS_TEST_DIR='/root/gstest'
         python3 -m pytest --exitfirst -s -v python/graphscope/experimental/nx/tests/classes
         python3 -m pytest --exitfirst -s -v python/graphscope/experimental/nx/tests/test_nx.py
         python3 -m pytest --exitfirst -s -v python/graphscope/experimental/nx/tests/algorithms/builtin \
                                             python/graphscope/experimental/nx/tests/test_utils.py
 
+        pkill -TERM etcd || true
+
   k8s-test:
     runs-on: self-hosted
     needs: [build-gae, build-gie, build-gle, build-manager]
@@ -368,9 +382,9 @@ jobs:
                           --network=host \
                           -f ./manager.Dockerfile .
         popd
-        sudo python3 -m pip install -r python/requirements.txt
-        sudo python3 -m pip install -r python/requirements-dev.txt
-        sudo python3 -m pip install pytest-cov
+        python3 -m pip install -r python/requirements.txt --user
+        python3 -m pip install -r python/requirements-dev.txt --user
+        python3 -m pip install pytest-cov --user
         pushd python && sudo -E python3 setup.py develop && popd
 
     - name: Kubernetes test
@@ -446,9 +460,9 @@ jobs:
                           --network=host \
                           -f ./manager.Dockerfile .
         popd
-        sudo python3 -m pip install -r python/requirements.txt
-        sudo python3 -m pip install -r python/requirements-dev.txt
-        sudo python3 -m pip install pytest-cov
+        python3 -m pip install -r python/requirements.txt
+        python3 -m pip install -r python/requirements-dev.txt
+        python3 -m pip install pytest-cov
         pushd python && sudo -E python3 setup.py develop && popd
 
     - name: Run function test

analytical_engine/CMakeLists.txt

@@ -105,7 +105,7 @@ endif ()
 find_package(libgrapelite REQUIRED)
 include_directories(${LIBGRAPELITE_INCLUDE_DIRS})
 
-find_package(vineyard 0.1.3 REQUIRED)
+find_package(vineyard 0.1.5 REQUIRED)
 include_directories(${VINEYARD_INCLUDE_DIRS})
 add_compile_options(-DENABLE_SELECTOR)
 

analytical_engine/core/object/fragment_wrapper.h

@@ -43,7 +43,7 @@ inline void set_graph_def(
     rpc::GraphDef& graph_def) {
   auto& meta = fragment->meta();
   graph_def.set_graph_type(rpc::ARROW_PROPERTY);
-  graph_def.set_directed(meta.GetKeyValue<bool>("directed"));
+  graph_def.set_directed(static_cast<bool>(meta.GetKeyValue<int>("directed")));
 
   auto* schema_def = graph_def.mutable_schema_def();
   schema_def->set_oid_type(

analytical_engine/frame/project_frame.cc

@@ -93,7 +93,8 @@ class ProjectFrame<gs::ArrowProjectedFragment<OID_T, VID_T, VDATA_T, EDATA_T>> {
     auto oid_type = parent_meta.GetKeyValue("oid_type");
     auto vid_type = parent_meta.GetKeyValue("vid_type");
 
-    graph_def.set_directed(parent_meta.template GetKeyValue<bool>("directed"));
+    graph_def.set_directed(
+        static_cast<bool>(parent_meta.template GetKeyValue<int>("directed")));
     auto* schema_def = graph_def.mutable_schema_def();
     schema_def->set_oid_type(oid_type);
     schema_def->set_vid_type(vid_type);

interactive_engine/src/executor/runtime/native/CMakeLists.txt

@@ -44,7 +44,7 @@ find_package(Threads REQUIRED)
 #  we need edge src/dst ids in etable.
 add_definitions(-DENDPOINT_LISTS)
 
-find_package(vineyard 0.1.3 REQUIRED)
+find_package(vineyard 0.1.5 REQUIRED)
 add_library(native_store global_store_ffi.cc
                          htap_ds_impl.cc
                          graph_builder_ffi.cc

interactive_engine/src/executor/runtime/native/graph_builder_ffi.cc

@@ -57,13 +57,13 @@ void get_builder_id(GraphBuilder builder, ObjectId *object_id,
 void launch_property_graph_loader(vineyard::Client &client,
                                   vineyard::ObjectID global_stream_id,
                                   size_t size, InstanceId *instance_ids) {
-  std::map<vineyard::InstanceID, vineyard::ptree> cluster;
+  std::map<vineyard::InstanceID, vineyard::json> cluster;
   VINEYARD_CHECK_OK(client.ClusterInfo(cluster));
   std::set<std::string> host_list;
   for (size_t idx = 0; idx < size; ++idx) {
     auto iter = cluster.find(instance_ids[idx]);
     VINEYARD_ASSERT(iter != cluster.end());
-    host_list.emplace(iter->second.get<std::string>("hostname"));
+    host_list.emplace(iter->second["hostname"].get_ref<std::string const &>());
   }
   std::string hosts = boost::algorithm::join(host_list, ",");
   std::string loader_path;
@@ -157,9 +157,9 @@ GraphBuilder get_graph_builder(const char *graph_name, const int index) {
   LOG(INFO) << "get name " << graph_name << " yields ID "
             << vineyard::VYObjectIDToString(id);
 #endif
-#ifndef NDEBUG
   vineyard::ObjectMeta meta;
   VINEYARD_CHECK_OK(client.GetMetaData(id, meta, true));
+#ifndef NDEBUG
   meta.PrintMeta();
 #endif
   auto gstream =

interactive_engine/src/executor/runtime/native/graph_schema.cc

@@ -36,7 +36,7 @@ limitations under the License.
 #include "arrow/api.h"
 #include "glog/logging.h"
 
-#include "common/util/ptree.h"
+#include "common/util/json.h"
 
 namespace vineyard {
 
@@ -95,8 +95,6 @@ PropertyType PropertyTypeFromString(const std::string& type) {
 
 }  // namespace detail
 
-using boost::property_tree::ptree;
-
 MGPropertyGraphSchema::PropertyId MGPropertyGraphSchema::GetPropertyId(
     const std::string& name) {
   PropertyId id;
@@ -196,51 +194,47 @@ MGPropertyGraphSchema::Entry* MGPropertyGraphSchema::CreateEntry(
   }
 }
 
-void MGPropertyGraphSchema::ToJSON(ptree& root) const {
-  root.put("partitionNum", fnum_);
-  ptree types;
+void MGPropertyGraphSchema::ToJSON(vineyard::json& root) const {
+  root["partitionNum"] = fnum_;
+  vineyard::json types = vineyard::json::array();
   for (auto const& entry : vertex_entries_) {
-    types.push_back(std::make_pair("", entry.ToJSON()));
+    types.emplace_back(entry.ToJSON());
   }
   for (auto const& entry : edge_entries_) {
-    types.push_back(std::make_pair("", entry.ToJSON()));
+    types.emplace_back(entry.ToJSON());
   }
-  root.add_child("types", types);
+  root["types"] = types;
   if (!unique_property_names_.empty()) {
     vineyard::put_container(root, "uniquePropertyNames",
                             unique_property_names_);
   }
 }
 
-void MGPropertyGraphSchema::FromJSON(ptree const& root) {
-  fnum_ = root.get<size_t>("partitionNum");
-  for (auto const& kv : root.get_child("types")) {
+void MGPropertyGraphSchema::FromJSON(vineyard::json const& root) {
+  fnum_ = root["partitionNum"].get<size_t>();
+  for (auto const& item : root["types"]) {
     Entry entry;
-    entry.FromJSON(kv.second);
+    entry.FromJSON(item);
     if (entry.type == "VERTEX") {
       vertex_entries_.push_back(std::move(entry));
     } else {
       edge_entries_.push_back(std::move(entry));
     }
   }
-  if (root.get_optional<std::string>("uniquePropertyNames")) {
+  if (root.contains("uniquePropertyNames")) {
     vineyard::get_container(root, "uniquePropertyNames",
                             unique_property_names_);
   }
 }
 
 std::string MGPropertyGraphSchema::ToJSONString() const {
-  std::stringstream ss;
-  ptree root;
+  vineyard::json root;
   ToJSON(root);
-  boost::property_tree::write_json(ss, root, false);
-  return ss.str();
+  return vineyard::json_to_string(root);
 }
 
 void MGPropertyGraphSchema::FromJSONString(std::string const& schema) {
-  ptree root;
-  std::istringstream iss(schema);
-  boost::property_tree::read_json(iss, root);
+  vineyard::json root = vineyard::json::parse(schema);
   FromJSON(root);
 }
 

interactive_engine/src/executor/runtime/native/graph_schema.h

@@ -29,11 +29,10 @@
 
 #include "arrow/api.h"
 #include "boost/leaf/all.hpp"
-#include "boost/property_tree/json_parser.hpp"
-#include "boost/property_tree/ptree.hpp"
 
-#include "graph/fragment/property_graph_types.h"
-#include "graph/fragment/graph_schema.h"
+#include "vineyard/common/util/json.h"
+#include "vineyard/graph/fragment/property_graph_types.h"
+#include "vineyard/graph/fragment/graph_schema.h"
 
 namespace vineyard {
 
@@ -70,8 +69,8 @@ class MGPropertyGraphSchema {
     }
   }
 
-  void ToJSON(boost::property_tree::ptree& root) const;
-  void FromJSON(boost::property_tree::ptree const& root);
+  void ToJSON(vineyard::json& root) const;
+  void FromJSON(vineyard::json const& root);
 
   std::string ToJSONString() const;
   void FromJSONString(std::string const& schema);

interactive_engine/src/executor/runtime/native/htap_ds_impl.cc

@@ -109,7 +109,7 @@ void get_graph_handle(ObjectId id, PartitionId channel_num,
       if (handle->schema == NULL) {
         auto schema = handle->fragments[fid].schema();
         vineyard::MGPropertyGraphSchema mgschema;
-        boost::property_tree::ptree schema_json;
+        vineyard::json schema_json;
         schema.ToJSON(schema_json);
         mgschema.FromJSON(schema_json);
         handle->schema =

interactive_engine/src/executor/runtime/native/property_graph_stream.cc

@@ -212,7 +212,7 @@ void PropertyGraphOutStream::AddVertex(VertexId id, LabelId labelid,
       batch_chunk->RemoveColumn(vertex_primary_key_column_[labelid]));
 #endif
   }
-  this->buildTableChunk(batch_chunk, vertex_writer_, 1,
+  this->buildTableChunk(batch_chunk, vertex_stream_, vertex_writer_, 1,
                         vertex_property_id_mapping_[labelid]);
 }
 
@@ -252,7 +252,7 @@ void PropertyGraphOutStream::AddEdge(EdgeId edge_id, VertexId src_id,
   appender->Apply(builder, edge_id, src_id, dst_id, src_label, dst_label,
                   property_size, properties,
                   edge_property_id_mapping_[label], batch_chunk);
-  this->buildTableChunk(batch_chunk, edge_writer_, 2,
+  this->buildTableChunk(batch_chunk, edge_stream_, edge_writer_, 2,
                         edge_property_id_mapping_[label]);
 }
 
@@ -404,7 +404,8 @@ void PropertyGraphOutStream::initialTables() {
 
 void PropertyGraphOutStream::buildTableChunk(
     std::shared_ptr<arrow::RecordBatch> batch,
-    std::shared_ptr<vineyard::DataframeStreamWriter>& stream_writer,
+    std::shared_ptr<vineyard::DataframeStream> &output_stream,
+    std::unique_ptr<vineyard::DataframeStreamWriter>& stream_writer,
     int const property_offset,
     std::map<int, int> const& property_id_mapping) {
 #ifndef NDEBUG
@@ -413,6 +414,11 @@ void PropertyGraphOutStream::buildTableChunk(
   if (batch == nullptr) {
     return;
   }
+
+  if (stream_writer == nullptr) {
+    VINEYARD_CHECK_OK(this->Open(output_stream, stream_writer));
+  }
+
 #ifndef NDEBUG
   LOG(INFO) << "chunk schema: batch is " << batch->schema()->ToString();
 #endif
@@ -444,7 +450,7 @@ void PropertyGraphOutStream::FinishAllVertices() {
         batch->RemoveColumn(vertex_primary_key_column_[vertices.first]));
 #endif
     }
-    buildTableChunk(batch, vertex_writer_, 1,
+    buildTableChunk(batch, vertex_stream_, vertex_writer_, 1,
                     vertex_property_id_mapping_[vertices.first]);
   }
   VINEYARD_CHECK_OK(vertex_writer_->Finish());
@@ -465,7 +471,7 @@ void PropertyGraphOutStream::FinishAllEdges() {
 #ifndef NDEBUG
     LOG(INFO) << "finish edges: " << batch;
 #endif
-      buildTableChunk(batch, edge_writer_, 2,
+      buildTableChunk(batch, edge_stream_, edge_writer_, 2,
                       edge_property_id_mapping_[edges.first]);
     }
   }
@@ -496,16 +502,14 @@ std::shared_ptr<Object> GlobalPGStreamBuilder::_Seal(Client& client) {
   auto gstream = std::make_shared<GlobalPGStream>();
   gstream->total_stream_chunks_ = total_stream_chunks_;
   gstream->meta_.SetTypeName(type_name<GlobalPGStream>());
-  gstream->meta_.AddKeyValue("total_stream_chunks",
-                             std::to_string(total_stream_chunks_));
+  gstream->meta_.AddKeyValue("total_stream_chunks", total_stream_chunks_);
 
   for (size_t idx = 0; idx < stream_chunks_.size(); ++idx) {
     gstream->meta_.AddMember("stream_chunk_" + std::to_string(idx),
                              stream_chunks_[idx]);
   }
 
   VINEYARD_CHECK_OK(client.CreateMetaData(gstream->meta_, gstream->id_));
-  VINEYARD_CHECK_OK(client.Persist(gstream->id_));
   return std::dynamic_pointer_cast<Object>(gstream);
 }