fix(interactive): Fix yaml node parsing (#4377)

As titled.

Author: zhanglei1949

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,16 +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 urllib3.exceptions import ProtocolError
-
-from gs_interactive.client.generated.interactive_pb2 import Code as StatusCode
 from gs_interactive.models.api_response_with_code import APIResponseWithCode
 
 

flex/tests/hqps/interactive_config_test.yaml

@@ -5,7 +5,7 @@ compute_engine:
   type: hiactor
   workers:
     - localhost:10000
-  thread_num_per_worker: 1
+  thread_num_per_worker: 4
   store:
     type: cpp-mcsr
   metadata_store:
@@ -26,6 +26,7 @@ compiler:
       statistics:
         uri: http://localhost:7777/v1/graph/%s/statistics
         interval: 86400000 # ms
+      timeout: 1000 # ms
   endpoint:
     default_listen_address: localhost
     bolt_connector:

flex/utils/yaml_utils.cc

@@ -52,12 +52,20 @@ void convert_yaml_node_to_json(const YAML::Node& node,
         json.SetInt(node.as<int>());
       } catch (const YAML::BadConversion& e) {
         try {
-          json.SetDouble(node.as<double>());
+          json.SetInt64(node.as<int64_t>());
         } catch (const YAML::BadConversion& e) {
           try {
-            json.SetBool(node.as<bool>());
+            json.SetUint64(node.as<uint64_t>());
           } catch (const YAML::BadConversion& e) {
-            json.SetString(node.as<std::string>().c_str(), allocator);
+            try {
+              json.SetDouble(node.as<double>());
+            } catch (const YAML::BadConversion& e) {
+              try {
+                json.SetBool(node.as<bool>());
+              } catch (const YAML::BadConversion& e) {
+                json.SetString(node.as<std::string>().c_str(), allocator);
+              }
+            }
           }
         }
       }

interactive_engine/compiler/conf/ir.compiler.properties

@@ -68,5 +68,8 @@ calcite.default.charset: UTF-8
 # set the interval in milliseconds to fetch graph schema
 # graph.meta.schema.fetch.interval.ms: 1000
 
-# set the timeout in milliseconds to fetch graph statistics
+# set the interval in milliseconds to fetch graph statistics
 # graph.meta.statistics.fetch.interval.ms: 86400000l
+
+# set the timeout in milliseconds to fetch graph statistics
+# graph.meta.fetch.timeout.ms: 1000

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/config/GraphConfig.java

@@ -31,6 +31,9 @@ public class GraphConfig {
     public static final Config<Long> GRAPH_META_STATISTICS_FETCH_INTERVAL_MS =
             Config.longConfig("graph.meta.statistics.fetch.interval.ms", 24 * 3600 * 1000l);
 
+    public static final Config<Long> GRAPH_META_FETCH_TIMEOUT_MS =
+            Config.longConfig("graph.meta.fetch.timeout.ms", 1000);
+
     // an intermediate solution to support foreign key, will be integrated into schema
     public static final Config<String> GRAPH_FOREIGN_KEY_URI =
             Config.stringConfig("graph.foreign.key", "");

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/config/YamlConfigs.java

@@ -72,6 +72,9 @@ public class YamlConfigs extends Configs {
                         "graph.meta.statistics.fetch.interval.ms",
                         (Configs configs) ->
                                 configs.get("compiler.meta.reader.statistics.interval"))
+                .put(
+                        "graph.meta.fetch.timeout.ms",
+                        (Configs configs) -> configs.get("compiler.meta.reader.timeout"))
                 .put(
                         "graph.store",
                         (Configs configs) -> {

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/meta/reader/HttpIrMetaReader.java

@@ -57,7 +57,9 @@ public HttpIrMetaReader(Configs configs) {
     public IrMeta readMeta() throws IOException {
         try {
             HttpResponse<String> response =
-                    sendRequest(GraphConfig.GRAPH_META_SCHEMA_URI.get(configs));
+                    sendRequest(
+                            GraphConfig.GRAPH_META_SCHEMA_URI.get(configs),
+                            GraphConfig.GRAPH_META_FETCH_TIMEOUT_MS.get(configs));
             String res = response.body();
             Preconditions.checkArgument(
                     response.statusCode() == 200,
@@ -91,7 +93,8 @@ public IrGraphStatistics readStats(GraphId graphId) throws IOException {
                     sendRequest(
                             String.format(
                                     GraphConfig.GRAPH_META_STATISTICS_URI.get(configs),
-                                    graphId.getId()));
+                                    graphId.getId()),
+                            GraphConfig.GRAPH_META_FETCH_TIMEOUT_MS.get(configs));
             String res = response.body();
             Preconditions.checkArgument(
                     response.statusCode() == 200,
@@ -109,7 +112,9 @@ public IrGraphStatistics readStats(GraphId graphId) throws IOException {
     public boolean syncStatsEnabled(GraphId graphId) throws IOException {
         try {
             HttpResponse<String> response =
-                    sendRequest(GraphConfig.GRAPH_META_SCHEMA_URI.get(configs));
+                    sendRequest(
+                            GraphConfig.GRAPH_META_SCHEMA_URI.get(configs),
+                            GraphConfig.GRAPH_META_FETCH_TIMEOUT_MS.get(configs));
             String res = response.body();
             Preconditions.checkArgument(
                     response.statusCode() == 200,
@@ -122,13 +127,14 @@ public boolean syncStatsEnabled(GraphId graphId) throws IOException {
         }
     }
 
-    private HttpResponse<String> sendRequest(String requestUri)
+    private HttpResponse<String> sendRequest(String requestUri, long timeOut)
             throws IOException, InterruptedException {
         HttpRequest request =
                 HttpRequest.newBuilder()
                         .uri(URI.create(requestUri))
                         .headers(CONTENT_TYPE, APPLICATION_JSON)
                         .GET()
+                        .timeout(java.time.Duration.ofMillis(timeOut))
                         .build();
         return httpClient.send(request, HttpResponse.BodyHandlers.ofString());
     }