Merge remote-tracking branch 'origin/main' into ir_align_type

.github/workflows/gss.yml

@@ -79,6 +79,8 @@ jobs:
         . ${HOME}/.graphscope_env
         export SCCACHE_DIR=~/.cache/sccache
         export RUSTC_WRAPPER=/usr/local/bin/sccache
+        rustup toolchain install 1.81.0
+        rustup default 1.81.0
         cd ${GITHUB_WORKSPACE}/interactive_engine
         mvn clean install -P groot -Drust.compile.mode=debug -DskipTests --quiet
         mvn clean install -Pgroot-data-load --quiet

.github/workflows/k8s-ci.yml

@@ -282,6 +282,12 @@ jobs:
           ~/.cache/sccache
         key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
 
+    - name: Install Rust
+      uses: actions-rs/toolchain@v1
+      with:
+        toolchain: 1.81.0
+        override: true
+
     - name: Build Artifact
       run: |
         . ${HOME}/.graphscope_env
@@ -640,6 +646,12 @@ jobs:
             ~/.cache/sccache
           key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
 
+      - name: Install Rust
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: 1.81.0
+          override: true
+
       - name: Build GIE Experimental Artifacts
         run: |
           . ~/.graphscope_env

.github/workflows/pr-check.yml

@@ -213,11 +213,15 @@ jobs:
         python3 -m black --check --diff .
         python3 -m flake8 .
         popd
-        pushd flex/interactive/sdk/python
+        # we need to generate the code first
+        pushd flex/interactive/sdk
+        bash generate_sdk.sh -g python
+        pushd python
         python3 -m isort --check --diff .
         python3 -m black --check --diff .
         python3 -m flake8 .
         popd
+        popd
 
     - name: Generate Docs
       shell: bash

docs/zh/frequently_asked_questions.rst

@@ -69,7 +69,7 @@
 
     - ``vineyard_shared_mem``: 存储数据集的内存。我们发现将其设置为数据集在磁盘上的大小的 5 倍通常是一个合理的值。 它相当于 graphscope 的 helm chart 中的 ``vineyard.shared_mem``。
 
-    - ``k8s_engine_mem`: ``engine`` 容器的内存大小。一般将其设置为 ``vineyard_shared_mem`` 的同样大小。它相当于 graphscope 的 helm chart 中的 ``engines.resources.memory.requests`` 和 ``engines.resources.memory.requests``。
+    - ``k8s_engine_mem``: ``engine`` 容器的内存大小。一般将其设置为 ``vineyard_shared_mem`` 的同样大小。它相当于 graphscope 的 helm chart 中的 ``engines.resources.memory.requests`` 和 ``engines.resources.memory.requests``。
 
 10. 导致在 Apple M1 python3.8 环境下安装 GraphScope 失败的原因可能有哪些？
 

flex/engines/graph_db/runtime/adhoc/expr_impl.cc

@@ -207,6 +207,9 @@ RTAny UnaryLogicalExpr::eval_edge(const LabelTriplet& label, vid_t src,
   if (logic_ == common::Logical::NOT) {
     return RTAny::from_bool(
         !expr_->eval_edge(label, src, dst, data, idx).as_bool());
+  } else if (logic_ == common::Logical::ISNULL) {
+    return RTAny::from_bool(
+        expr_->eval_edge(label, src, dst, data, idx, 0).is_null());
   }
   LOG(FATAL) << "not support" << static_cast<int>(logic_);
   return RTAny::from_bool(false);

flex/engines/graph_db/runtime/adhoc/expr_impl.h

@@ -85,17 +85,23 @@ class WithInExpr : public ExprBase {
   WithInExpr(const ReadTransaction& txn, const Context& ctx,
              std::unique_ptr<ExprBase>&& key, const common::Value& array)
       : key_(std::move(key)) {
-    if constexpr (std::is_same_v<T, int64_t>) {
-      CHECK(array.item_case() == common::Value::kI64Array);
-      size_t len = array.i64_array().item_size();
-      for (size_t idx = 0; idx < len; ++idx) {
-        container_.push_back(array.i64_array().item(idx));
-      }
-    } else if constexpr (std::is_same_v<T, int32_t>) {
-      CHECK(array.item_case() == common::Value::kI32Array);
-      size_t len = array.i32_array().item_size();
-      for (size_t idx = 0; idx < len; ++idx) {
-        container_.push_back(array.i32_array().item(idx));
+    if constexpr ((std::is_same_v<T, int64_t>) ||
+                  (std::is_same_v<T, int32_t>) ) {
+      // Implicitly convert to T
+      if (array.item_case() == common::Value::kI64Array) {
+        size_t len = array.i64_array().item_size();
+        for (size_t idx = 0; idx < len; ++idx) {
+          container_.push_back(array.i64_array().item(idx));
+        }
+      } else if (array.item_case() == common::Value::kI32Array) {
+        size_t len = array.i32_array().item_size();
+        for (size_t idx = 0; idx < len; ++idx) {
+          container_.push_back(array.i32_array().item(idx));
+        }
+      } else {
+        LOG(FATAL) << "Fail to construct WithInExpr of type "
+                   << typeid(T).name() << " with array of type "
+                   << array.item_case();
       }
     } else if constexpr (std::is_same_v<T, std::string>) {
       CHECK(array.item_case() == common::Value::kStrArray);

flex/engines/graph_db/runtime/adhoc/operators/group_by.cc

@@ -413,6 +413,29 @@ std::shared_ptr<IContextColumn> string_to_list(
   return builder.finish();
 }
 
+template <typename T>
+std::shared_ptr<IContextColumn> scalar_to_list(
+    const Var& var, const std::vector<std::vector<size_t>>& to_aggregate) {
+  ListValueColumnBuilder<T> builder;
+  size_t col_size = to_aggregate.size();
+  builder.reserve(col_size);
+  std::vector<std::shared_ptr<ListImplBase>> impls;
+  for (size_t k = 0; k < col_size; ++k) {
+    auto& vec = to_aggregate[k];
+
+    std::vector<T> elem;
+    for (auto idx : vec) {
+      elem.push_back(TypedConverter<T>::to_typed(var.get(idx)));
+    }
+    auto impl = ListImpl<T>::make_list_impl(std::move(elem));
+    auto list = List::make_list(impl);
+    impls.emplace_back(impl);
+    builder.push_back_opt(list);
+  }
+  builder.set_list_impls(impls);
+  return builder.finish();
+}
+
 bl::result<std::shared_ptr<IContextColumn>> apply_reduce(
     const AggFunc& func, const std::vector<std::vector<size_t>>& to_aggregate) {
   if (func.aggregate == AggrKind::kSum) {
@@ -497,6 +520,14 @@ bl::result<std::shared_ptr<IContextColumn>> apply_reduce(
       return tuple_to_list(var, to_aggregate);
     } else if (var.type() == RTAnyType::kStringValue) {
       return string_to_list(var, to_aggregate);
+    } else if (var.type() == RTAnyType::kI32Value) {
+      return scalar_to_list<int32_t>(var, to_aggregate);
+    } else if (var.type() == RTAnyType::kI64Value) {
+      return scalar_to_list<int64_t>(var, to_aggregate);
+    } else if (var.type() == RTAnyType::kU64Value) {
+      return scalar_to_list<uint64_t>(var, to_aggregate);
+    } else if (var.type() == RTAnyType::kF64Value) {
+      return scalar_to_list<double>(var, to_aggregate);
     } else {
       LOG(FATAL) << "not support" << static_cast<int>(var.type().type_enum_);
     }

flex/engines/graph_db/runtime/common/operators/scan.cc

@@ -47,11 +47,10 @@ bl::result<Context> Scan::find_vertex_with_id(const ReadTransaction& txn,
     }
     if (GlobalId::get_label_id(gid) == label) {
       vid = GlobalId::get_vid(gid);
+      builder.push_back_opt(vid);
     } else {
       LOG(ERROR) << "Global id " << gid << " does not match label " << label;
-      return Context();
     }
-    builder.push_back_opt(vid);
     Context ctx;
     ctx.set(alias, builder.finish());
     return ctx;

flex/engines/graph_db/runtime/common/rt_any.cc

@@ -132,6 +132,8 @@ RTAny::RTAny(const Any& val) {
   } else if (val.type == PropertyType::Bool()) {
     type_ = RTAnyType::kBoolValue;
     value_.b_val = val.AsBool();
+  } else if (val.type == PropertyType::Empty()) {
+    type_ = RTAnyType::kNull;
   } else {
     LOG(FATAL) << "Any value: " << val.to_string()
                << ", type = " << val.type.type_enum;
@@ -368,6 +370,8 @@ std::string_view RTAny::as_string() const {
     return value_.str_val;
   } else if (type_ == RTAnyType::kUnknown) {
     return std::string_view();
+  } else if (type_ == RTAnyType::kNull) {
+    return std::string_view();
   } else {
     LOG(FATAL) << "unexpected type" << static_cast<int>(type_.type_enum_);
     return std::string_view();

flex/engines/graph_db/runtime/common/rt_any.h

@@ -420,6 +420,7 @@ class ListImpl : ListImplBase {
   static std::shared_ptr<ListImplBase> make_list_impl(std::vector<T>&& vals) {
     auto new_list = new ListImpl<T>();
     new_list->list_ = std::move(vals);
+    new_list->is_valid_.resize(new_list->list_.size(), true);
     return std::shared_ptr<ListImplBase>(static_cast<ListImplBase*>(new_list));
   }
 

flex/engines/http_server/handler/graph_db_http_handler.cc

@@ -176,7 +176,7 @@ class stored_proc_handler : public StoppableHandler {
 
   bool start() override {
     if (get_executors()[StoppableHandler::shard_id()].size() > 0) {
-      LOG(ERROR) << "The actors have been already created!";
+      VLOG(10) << "The actors have been already created!";
       return false;
     }
     return StoppableHandler::start_scope(

flex/engines/http_server/handler/graph_db_http_handler.h

@@ -69,7 +69,7 @@ class StoppableHandler : public seastar::httpd::handler_base {
           } catch (const std::exception& e) {
             // In case the scope is already cancelled, we should ignore the
             // exception.
-            LOG(INFO) << "Failed to cancel IC scope: " << e.what();
+            VLOG(1) << "Failed to cancel IC scope: " << e.what();
           }
           func();
           return seastar::make_ready_future<>();

flex/interactive/sdk/python/gs_interactive/client/result.py

@@ -20,9 +20,8 @@
 from typing import TypeVar
 
 from gs_interactive.api_response import ApiResponse
-from gs_interactive.exceptions import ApiException
-
 from gs_interactive.client.status import Status
+from gs_interactive.exceptions import ApiException
 
 # Define a generic type placeholder
 T = TypeVar("T")

flex/interactive/sdk/python/gs_interactive/client/session.py

@@ -23,6 +23,11 @@
 from typing import Optional
 from typing import Union
 
+from pydantic import Field
+from pydantic import StrictBytes
+from pydantic import StrictStr
+from typing_extensions import Annotated
+
 from gs_interactive.api import AdminServiceGraphManagementApi
 from gs_interactive.api import AdminServiceJobManagementApi
 from gs_interactive.api import AdminServiceProcedureManagementApi
@@ -32,18 +37,13 @@
 from gs_interactive.api import QueryServiceApi
 from gs_interactive.api import UtilsApi
 from gs_interactive.api_client import ApiClient
-from gs_interactive.configuration import Configuration
-from pydantic import Field
-from pydantic import StrictBytes
-from pydantic import StrictStr
-from typing_extensions import Annotated
-
 from gs_interactive.client.generated.results_pb2 import CollectiveResults
 from gs_interactive.client.result import Result
 from gs_interactive.client.status import Status
 from gs_interactive.client.status import StatusCode
 from gs_interactive.client.utils import InputFormat
 from gs_interactive.client.utils import append_format_byte
+from gs_interactive.configuration import Configuration
 from gs_interactive.models import CreateGraphRequest
 from gs_interactive.models import CreateGraphResponse
 from gs_interactive.models import CreateProcedureRequest

flex/interactive/sdk/python/gs_interactive/client/status.py

@@ -16,15 +16,16 @@
 # limitations under the License.
 #
 
+from urllib3.exceptions import MaxRetryError
+from urllib3.exceptions import ProtocolError
+
 from gs_interactive.api_response import ApiResponse
+from gs_interactive.client.generated.interactive_pb2 import Code as StatusCode
 from gs_interactive.exceptions import ApiException
 from gs_interactive.exceptions import BadRequestException
 from gs_interactive.exceptions import ForbiddenException
 from gs_interactive.exceptions import NotFoundException
 from gs_interactive.exceptions import ServiceException
-from urllib3.exceptions import MaxRetryError
-
-from gs_interactive.client.generated.interactive_pb2 import Code as StatusCode
 from gs_interactive.models.api_response_with_code import APIResponseWithCode
 
 
@@ -108,6 +109,8 @@ def from_exception(exception: ApiException):
                 return Status(StatusCode.INTERNAL_ERROR, exception.body)
         elif isinstance(exception, MaxRetryError):
             return Status(StatusCode.INTERNAL_ERROR, exception)
+        elif isinstance(exception, ProtocolError):
+            return Status(StatusCode.INTERNAL_ERROR, exception)
         return Status(
             StatusCode.UNKNOWN, "Unknown Error from exception " + exception.body
         )

flex/interactive/sdk/python/gs_interactive/tests/conftest.py

@@ -17,6 +17,7 @@
 #
 
 # get the directory of the current file
+import copy
 import os
 import time
 
@@ -27,6 +28,7 @@
 from gs_interactive.client.session import Session
 from gs_interactive.models import CreateGraphRequest
 from gs_interactive.models import CreateProcedureRequest
+from gs_interactive.models import GetGraphSchemaResponse
 from gs_interactive.models import SchemaMapping
 from gs_interactive.models import StartServiceRequest
 from gs_interactive.models import UpdateProcedureRequest
@@ -39,7 +41,7 @@
 
 
 modern_graph_full = {
-    "name": "modern_graph",
+    "name": "full_graph",
     "description": "This is a test graph",
     "schema": {
         "vertex_types": [
@@ -120,7 +122,7 @@
 }
 
 modern_graph_vertex_only = {
-    "name": "modern_graph",
+    "name": "vertex_only",
     "description": "This is a test graph, only contains vertex",
     "schema": {
         "vertex_types": [
@@ -148,7 +150,7 @@
 }
 
 modern_graph_partial = {
-    "name": "modern_graph",
+    "name": "partial_graph",
     "description": "This is a test graph",
     "schema": {
         "vertex_types": [
@@ -336,7 +338,7 @@ def create_partial_modern_graph(interactive_session):
 
 @pytest.fixture(scope="function")
 def create_graph_with_custom_pk_name(interactive_session):
-    modern_graph_custom_pk_name = modern_graph_full.copy()
+    modern_graph_custom_pk_name = copy.deepcopy(modern_graph_full)
     for vertex_type in modern_graph_custom_pk_name["schema"]["vertex_types"]:
         vertex_type["properties"][0]["property_name"] = "custom_id"
         vertex_type["primary_keys"] = ["custom_id"]
@@ -479,3 +481,24 @@ def start_service_on_graph(interactive_session, graph_id: str):
     assert resp.is_ok()
     # wait three second to let compiler get the new graph
     time.sleep(3)
+
+
+def ensure_compiler_schema_ready(
+    interactive_session, neo4j_session: Neo4jSession, graph_id: str
+):
+    rel_graph_meta = interactive_session.get_graph_schema(graph_id).get_value()
+    max_times = 10
+    while True:
+        if max_times == 0:
+            raise Exception("compiler schema is not ready")
+        res = neo4j_session.run("CALL gs.procedure.meta.schema();")
+        val = res.single().value()
+        compiler_graph_schema = GetGraphSchemaResponse.from_json(val)
+        # print("compiler_graph_schema: ", compiler_graph_schema)
+        # print("rel_graph_meta: ", rel_graph_meta)
+        if compiler_graph_schema == rel_graph_meta:
+            break
+        print("compiler schema is not ready, wait for 1 second")
+        time.sleep(1)
+        max_times -= 1
+    print("compiler schema is ready")