fix(interactive): Support Add Vertex and Edge Operations in Interactive (#4591)

<!--
Thanks for your contribution! please review
https://github.com/alibaba/GraphScope/blob/main/CONTRIBUTING.md before
opening an issue.
-->

## What do these changes do?

Support the following Cypher queries to add vertex and edge data by the
Interactive Python SDK:

1. Add Vertex
```
CREATE (charlie:Person {name: 'Charlie', id: 211, age: 32})
```
2. Add Edge
```
CREATE (charlie:Person {id: 211})-[:ACTED_IN]->(movie:Movie {name: 'Wall Street'});
```

Continuous `Add` operations will be batched, and leverage the batch api
provided by Python SDK to implement.

<!-- Please give a short brief about these changes. -->

## Related issue number

<!-- Are there any issues opened that will be resolved by merging this
change? -->

Fixes

---------

Co-authored-by: xiaolei.zl <[email protected]>

Author: shirly121

.github/workflows/interactive.yml

@@ -341,6 +341,13 @@ jobs:
            ${GITHUB_WORKSPACE}/flex/interactive/examples/modern_graph/graph.yaml ${INTERACTIVE_ENGINE_HOME}/compiler/src/test/resources/statistics/modern_statistics.json \
           "MATCH(n) return count(n);" /tmp/physical_plan_gen_config.yaml
 
+    - name: Tests cypher write
+      env:
+        INTERACTIVE_WORKSPACE: /tmp/interactive_workspace
+      run: |
+        cd ${GITHUB_WORKSPACE}/flex/tests/hqps
+        bash hqps_cypher_write_test.sh ${INTERACTIVE_WORKSPACE} modern_graph ./interactive_config_test.yaml
+
     - name: Run End-to-End cypher adhoc ldbc query test
       env:
         GS_TEST_DIR: ${{ github.workspace }}/gstest

flex/engines/graph_db/database/graph_db_operations.cc

@@ -29,13 +29,17 @@ namespace gs {
 bool check_primary_key_value_valid(
     const rapidjson::Value& vertex_json,
     const std::string& pk_filed_name = "primary_key_values") {
-  if (!vertex_json.HasMember(pk_filed_name) ||
-      !vertex_json[pk_filed_name].IsArray() ||
-      vertex_json[pk_filed_name].Size() != 1 ||
-      !vertex_json[pk_filed_name][0].HasMember("value")) {
-    return false;
+  LOG(INFO) << "check_primary_key_value_valid for " << pk_filed_name << " "
+            << gs::rapidjson_stringify(vertex_json);
+  if (vertex_json.HasMember(pk_filed_name)) {
+    if (vertex_json[pk_filed_name].IsArray()) {
+      return vertex_json[pk_filed_name].Size() == 1 &&
+             vertex_json[pk_filed_name][0].HasMember("value");
+    }
+    return true;
   }
-  return true;
+  LOG(INFO) << "check_primary_key_value_valid failed";
+  return false;
 }
 
 Result<std::string> GraphDBOperations::CreateVertex(
@@ -260,12 +264,17 @@ VertexData GraphDBOperations::inputVertex(const rapidjson::Value& vertex_json,
                                           GraphDBSession& session) {
   VertexData vertex;
   std::string label = jsonToString(vertex_json["label"]);
-  if (!check_primary_key_value_valid(vertex_json)) {
-    throw std::runtime_error("primary_key_values is invalid");
+  if (!check_primary_key_value_valid(vertex_json, "primary_key_values") &&
+      !check_primary_key_value_valid(vertex_json, "primary_key_value")) {
+    throw std::runtime_error("primary_key_values/primary_key_value is invalid");
   }
 
-  vertex.pk_value =
-      Any(jsonToString(vertex_json["primary_key_values"][0]["value"]));
+  if (vertex_json.HasMember("primary_key_values")) {
+    vertex.pk_value =
+        Any(jsonToString(vertex_json["primary_key_values"][0]["value"]));
+  } else {
+    vertex.pk_value = Any(jsonToString(vertex_json["primary_key_value"]));
+  }
   std::unordered_set<std::string> property_names;
   std::vector<std::string> property_names_arr;
   for (auto& property : vertex_json["properties"].GetArray()) {
@@ -294,16 +303,27 @@ EdgeData GraphDBOperations::inputEdge(const rapidjson::Value& edge_json,
   std::string src_label = jsonToString(edge_json["src_label"]);
   std::string dst_label = jsonToString(edge_json["dst_label"]);
   std::string edge_label = jsonToString(edge_json["edge_label"]);
-  if (!check_primary_key_value_valid(edge_json, "src_primary_key_values")) {
-    throw std::runtime_error("src_primary_key_values is invalid");
+  if (!check_primary_key_value_valid(edge_json, "src_primary_key_values") &&
+      !check_primary_key_value_valid(edge_json, "src_primary_key_value")) {
+    throw std::runtime_error(
+        "src_primary_key_values/src_primary_key_value is invalid");
   }
-  if (!check_primary_key_value_valid(edge_json, "dst_primary_key_values")) {
+  if (!check_primary_key_value_valid(edge_json, "dst_primary_key_values") &&
+      !check_primary_key_value_valid(edge_json, "dst_primary_key_value")) {
     throw std::runtime_error("dst_primary_key_values is invalid");
   }
-  edge.src_pk_value =
-      Any(jsonToString(edge_json["src_primary_key_values"][0]["value"]));
-  edge.dst_pk_value =
-      Any(jsonToString(edge_json["dst_primary_key_values"][0]["value"]));
+  if (edge_json.HasMember("src_primary_key_values")) {
+    edge.src_pk_value =
+        Any(jsonToString(edge_json["src_primary_key_values"][0]["value"]));
+  } else {
+    edge.src_pk_value = Any(jsonToString(edge_json["src_primary_key_value"]));
+  }
+  if (edge_json.HasMember("dst_primary_key_values")) {
+    edge.dst_pk_value =
+        Any(jsonToString(edge_json["dst_primary_key_values"][0]["value"]));
+  } else {
+    edge.dst_pk_value = Any(jsonToString(edge_json["dst_primary_key_value"]));
+  }
   // Check that all parameters in the parameter
   if (edge_json["properties"].Size() > 1) {
     throw std::runtime_error(

flex/tests/hqps/hqps_cypher_write_test.sh

@@ -0,0 +1,152 @@
+#!/bin/bash
+# Copyright 2020 Alibaba Group Holding Limited.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+set -e
+SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
+FLEX_HOME=${SCRIPT_DIR}/../../
+BULK_LOADER=${FLEX_HOME}/build/bin/bulk_loader
+SERVER_BIN=${FLEX_HOME}/build/bin/interactive_server
+GIE_HOME=${FLEX_HOME}/../interactive_engine/
+
+#
+if [ $# -ne 3 ]; then
+  echo "Receives: $# args, need 3 args"
+  echo "Usage: $0 <INTERACTIVE_WORKSPACE> <GRAPH_NAME> <ENGINE_CONFIG>"
+  exit 1
+fi
+
+INTERACTIVE_WORKSPACE=$1
+GRAPH_NAME=$2
+ENGINE_CONFIG_PATH=$3
+# if ENGINE_CONFIG_PATH is not absolute path, convert it to absolute path
+if [[ ! ${ENGINE_CONFIG_PATH} == /* ]]; then
+  ENGINE_CONFIG_PATH=$(cd $(dirname ${ENGINE_CONFIG_PATH}) && pwd)/$(basename ${ENGINE_CONFIG_PATH})
+fi
+
+
+if [ ! -d ${INTERACTIVE_WORKSPACE} ]; then
+  echo "INTERACTIVE_WORKSPACE: ${INTERACTIVE_WORKSPACE} not exists"
+  exit 1
+fi
+
+if [ ! -d ${INTERACTIVE_WORKSPACE}/data/${GRAPH_NAME} ]; then
+  echo "GRAPH: ${GRAPH_NAME} not exists"
+  exit 1
+fi
+if [ ! -f ${INTERACTIVE_WORKSPACE}/data/${GRAPH_NAME}/graph.yaml ]; then
+  echo "GRAPH_SCHEMA_FILE: ${BULK_LOAD_FILE} not exists"
+  exit 1
+fi
+if [ ! -f ${ENGINE_CONFIG_PATH} ]; then
+  echo "ENGINE_CONFIG: ${ENGINE_CONFIG_PATH} not exists"
+  exit 1
+fi
+
+GRAPH_SCHEMA_YAML=${INTERACTIVE_WORKSPACE}/data/${GRAPH_NAME}/graph.yaml
+GRAPH_CSR_DATA_DIR=${INTERACTIVE_WORKSPACE}/data/${GRAPH_NAME}/indices
+rm -rf ${GRAPH_CSR_DATA_DIR}/wal
+
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+NC='\033[0m' # No Color
+err() {
+  echo -e "${RED}[$(date +'%Y-%m-%d %H:%M:%S')] -ERROR- $* ${NC}" >&2
+}
+
+info() {
+  echo -e "${GREEN}[$(date +'%Y-%m-%d %H:%M:%S')] -INFO- $* ${NC}"
+}
+
+
+kill_service(){
+    info "Kill Service first"
+    ps -ef | grep "com.alibaba.graphscope.GraphServer" | awk '{print $2}' | xargs kill -9 || true
+    ps -ef | grep "interactive_server" |  awk '{print $2}' | xargs kill -9  || true
+    sleep 3
+    sed -i "s/default_graph: .*/default_graph: modern_graph/g" ${ENGINE_CONFIG_PATH}
+    rm -rf ${INTERACTIVE_WORKSPACE}/METADATA || (err "rm METADATA failed")
+    rm ${INTERACTIVE_WORKSPACE}/data/1 || (err "rm builtin graph failed")
+    info "Kill Service success"
+    rm -rf /tmp/neo4j-* || true
+    # clean the wal
+    rm -rf ${GRAPH_CSR_DATA_DIR}/wal || true
+    rm -rf ${GRAPH_CSR_DATA_DIR}/runtime || true
+}
+
+# kill service when exit
+trap kill_service EXIT
+
+
+# start engine service and load ldbc graph
+start_engine_service(){
+    # suppose graph has been loaded, check ${GRAPH_CSR_DATA_DIR} exists
+    rm -rf ${INTERACTIVE_WORKSPACE}/metadata/
+    rm ${INTERACTIVE_WORKSPACE}/data/1 || true
+    if [ ! -d ${GRAPH_CSR_DATA_DIR} ]; then
+        err "GRAPH_CSR_DATA_DIR not found"
+        exit 1
+    fi
+
+    #check SERVER_BIN exists
+    if [ ! -f ${SERVER_BIN} ]; then
+        err "SERVER_BIN not found"
+        exit 1
+    fi
+    sed -i "s/default_graph: .*/default_graph: ${GRAPH_NAME}/g" ${ENGINE_CONFIG_PATH}
+
+    cmd="${SERVER_BIN} -c ${ENGINE_CONFIG_PATH} "
+    cmd="${cmd} -w ${INTERACTIVE_WORKSPACE} --enable-admin-service true --start-compiler true"
+
+    info "Start engine service with command: ${cmd}"
+    ${cmd} &
+    sleep 5
+    #check interactive_server is running, if not, exit
+    ps -ef | grep "interactive_server" | grep -v grep
+
+    info "Start engine service success"
+}
+
+
+start_compiler_service(){
+  echo "try to start compiler service"
+  pushd ${GIE_HOME}/compiler
+  cmd="make run graph.schema=${GRAPH_SCHEMA_YAML} config.path=${ENGINE_CONFIG_PATH}"
+  echo "Start compiler service with command: ${cmd}"
+  ${cmd} &
+  sleep 5
+  # check if Graph Server is running, if not exist
+  ps -ef | grep "com.alibaba.graphscope.GraphServer" | grep -v grep
+  info "Start compiler service success"
+  popd
+}
+
+run_cypher_write_test(){
+    echo "run cypher write test"
+    pushd ${GIE_HOME}/compiler
+    cmd="mvn test -Dskip.ir.core=true -Dtest=com.alibaba.graphscope.cypher.integration.modern.ModernGraphWriteTest"
+    echo "Run cypher write test with command: ${cmd}"
+    ${cmd}
+    info "Run cypher write test success"
+    popd
+}
+
+
+kill_service
+start_engine_service
+start_compiler_service
+run_cypher_write_test
+
+
+
+kill_service

interactive_engine/compiler/pom.xml

@@ -362,6 +362,7 @@
                         <exclude>**/IrPatternTest.java</exclude>
                         <exclude>**/MovieTest.java</exclude>
                         <exclude>**/GraphAlgoTest.java</exclude>
+                        <exclude>**/ModernGraphWriteTest.java</exclude>
                     </excludes>
                 </configuration>
             </plugin>

interactive_engine/compiler/src/main/antlr4/CypherGS.g4

@@ -52,9 +52,14 @@ CALL : ( 'C' | 'c' ) ( 'A' | 'a' ) ( 'L' | 'l' ) ( 'L' | 'l' ) ;
 YIELD : ( 'Y' | 'y' ) ( 'I' | 'i' ) ( 'E' | 'e' ) ( 'L' | 'l' ) ( 'D' | 'd' ) ;
 
 oC_RegularQuery
-     : ( ( oC_ReadingClause ) SP? )+  ( SP oC_Return )
+     : ( ( oC_ReadingClause | oC_UpdatingClause ) SP? )* SP? oC_ReadingClause ( SP oC_Return )
+     | ( ( oC_ReadingClause | oC_UpdatingClause ) SP? )* SP? oC_UpdatingClause ( SP oC_Return )?
      ;
 
+oC_UpdatingClause
+      : oC_Create
+      ;
+
 oC_ReadingClause
      :    oC_Match
      |    oC_With
@@ -63,6 +68,11 @@ oC_ReadingClause
      |    oC_UnionCallSubQuery
      ;
 
+CREATE : ( 'C' | 'c' ) ( 'R' | 'r' ) ( 'E' | 'e' ) ( 'A' | 'a' ) ( 'T' | 't' ) ( 'E' | 'e' ) ;
+
+oC_Create
+      :  CREATE SP? oC_Pattern ;
+
 oC_SubQuery
      : ( ( oC_Match | oC_With | oC_Unwind ) SP? )* ( SP? oC_Return ) ;
 

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/client/HttpExecutionClient.java

@@ -53,6 +53,10 @@ public HttpExecutionClient(Configs graphConfig, ChannelFetcher<URI> channelFetch
                         HiactorConfig.INTERACTIVE_QUERY_ENDPOINT.get(graphConfig));
     }
 
+    public Session getSession() {
+        return session;
+    }
+
     @Override
     public void submit(
             ExecutionRequest request,

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/client/type/ExecutionResponseListener.java

@@ -16,13 +16,11 @@
 
 package com.alibaba.graphscope.common.client.type;
 
-import com.alibaba.graphscope.gaia.proto.IrResult;
-
 /**
  * listener to handle response from remote engine service
  */
-public interface ExecutionResponseListener {
-    void onNext(IrResult.Record record);
+public interface ExecutionResponseListener<T> {
+    void onNext(T record);
 
     void onCompleted();