Merge branch 'main' into upgrade_rust_version

.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

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/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")

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

@@ -27,6 +27,7 @@
 from gs_interactive.tests.conftest import call_procedure  # noqa: E402
 from gs_interactive.tests.conftest import create_procedure
 from gs_interactive.tests.conftest import delete_procedure
+from gs_interactive.tests.conftest import ensure_compiler_schema_ready
 from gs_interactive.tests.conftest import import_data_to_full_modern_graph
 from gs_interactive.tests.conftest import import_data_to_partial_modern_graph
 from gs_interactive.tests.conftest import import_data_to_vertex_only_modern_graph
@@ -61,6 +62,9 @@ def test_query_on_vertex_only_graph(
     """
     print("[Query on vertex only graph]")
     start_service_on_graph(interactive_session, create_vertex_only_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_vertex_only_modern_graph
+    )
     run_cypher_test_suite(
         neo4j_session, create_vertex_only_modern_graph, vertex_only_cypher_queries
     )
@@ -69,6 +73,10 @@ def test_query_on_vertex_only_graph(
     import_data_to_vertex_only_modern_graph(
         interactive_session, create_vertex_only_modern_graph
     )
+    start_service_on_graph(interactive_session, create_vertex_only_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_vertex_only_modern_graph
+    )
     run_cypher_test_suite(
         neo4j_session, create_vertex_only_modern_graph, vertex_only_cypher_queries
     )
@@ -83,12 +91,19 @@ def test_query_on_partial_graph(
     print("[Query on partial graph]")
     # start service on new graph
     start_service_on_graph(interactive_session, create_partial_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_partial_modern_graph
+    )
     # try to query on the graph
     run_cypher_test_suite(neo4j_session, create_partial_modern_graph, cypher_queries)
     start_service_on_graph(interactive_session, "1")
     import_data_to_partial_modern_graph(
         interactive_session, create_partial_modern_graph
     )
+    start_service_on_graph(interactive_session, create_partial_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_partial_modern_graph
+    )
     run_cypher_test_suite(neo4j_session, create_partial_modern_graph, cypher_queries)
 
 
@@ -100,10 +115,17 @@ def test_query_on_full_modern_graph(
     """
     print("[Query on full modern graph]")
     start_service_on_graph(interactive_session, create_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_modern_graph
+    )
     # try to query on the graph
     run_cypher_test_suite(neo4j_session, create_modern_graph, cypher_queries)
     start_service_on_graph(interactive_session, "1")
     import_data_to_full_modern_graph(interactive_session, create_modern_graph)
+    start_service_on_graph(interactive_session, create_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_modern_graph
+    )
     run_cypher_test_suite(neo4j_session, create_modern_graph, cypher_queries)
 
 
@@ -129,6 +151,9 @@ def test_service_switching(
     )
     print("Procedure id: ", a_proc_id)
     start_service_on_graph(interactive_session, create_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_modern_graph
+    )
     call_procedure(neo4j_session, create_modern_graph, a_proc_id)
 
     # create procedure on graph_b_id
@@ -139,6 +164,9 @@ def test_service_switching(
         "MATCH(n: person) return count(n);",
     )
     start_service_on_graph(interactive_session, create_vertex_only_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_vertex_only_modern_graph
+    )
     call_procedure(neo4j_session, create_vertex_only_modern_graph, b_proc_id)
 
 
@@ -156,6 +184,9 @@ def test_procedure_creation(interactive_session, neo4j_session, create_modern_gr
     )
     print("Procedure id: ", a_proc_id)
     start_service_on_graph(interactive_session, create_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_modern_graph
+    )
     call_procedure(neo4j_session, create_modern_graph, a_proc_id)
 
     # create procedure with name containing space,
@@ -202,6 +233,9 @@ def test_builtin_procedure(interactive_session, neo4j_session, create_modern_gra
         )
     # Call the builtin procedure
     start_service_on_graph(interactive_session, create_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_modern_graph
+    )
     call_procedure(
         neo4j_session,
         create_modern_graph,
@@ -259,6 +293,10 @@ def test_list_jobs(interactive_session, create_vertex_only_modern_graph):
 def test_call_proc_in_cypher(interactive_session, neo4j_session, create_modern_graph):
     print("[Test call procedure in cypher]")
     import_data_to_full_modern_graph(interactive_session, create_modern_graph)
+    start_service_on_graph(interactive_session, create_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_modern_graph
+    )
     result = neo4j_session.run(
         'MATCH(p: person) with p.id as oid CALL k_neighbors("person", oid, 1) return label_name, vertex_oid;'
     )
@@ -276,6 +314,9 @@ def test_custom_pk_name(
         interactive_session, create_graph_with_custom_pk_name
     )
     start_service_on_graph(interactive_session, create_graph_with_custom_pk_name)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_graph_with_custom_pk_name
+    )
     result = neo4j_session.run(
         "MATCH (n: person) where n.custom_id = 4 return n.custom_id;"
     )
@@ -289,4 +330,3 @@ def test_custom_pk_name(
     )
     records = result.fetch(1)
     assert len(records) == 1 and records[0]["$f0"] == 2
-    start_service_on_graph(interactive_session, "1")

flex/interactive/sdk/python/setup.cfg

@@ -3,7 +3,7 @@ profile = black
 ensure_newline_before_comments = True
 line_length = 88
 force_single_line = True
-skip = build/,dist/,gs_interactive/api/,gs_interactive/api_response.py,gs_interactive/configuration.py,gs_interactive/exceptions.py,gs_interactive/models/,gs_interactiverest.py,
+skip = build/,dist/,gs_interactive/api/,gs_interactive/api_response.py,gs_interactive/configuration.py,gs_interactive/exceptions.py,gs_interactive/models/,gs_interactiverest.py,gs_interactive/api_client.py,gs_interactive/__init__.py,gs_interactive/rest.py
 skip_glob = *_pb2.py,*_pb2_grpc.py,build/*
 
 [flake8]