fix(interactive): Implement HttpIrMetaReader to Get Meta Data From Remote Http Service

.github/workflows/interactive.yml

@@ -180,6 +180,18 @@ jobs:
         ./tests/hqps/query_test ${GS_TEST_DIR}/flex/ldbc-sf01-long-date/audit_graph_schema.yaml \
         /tmp/csr-data-dir/
 
+    - name: Test get graph meta from admin service
+      env:
+        GS_TEST_DIR: ${{ github.workspace }}/gstest
+        INTERACTIVE_WORKSPACE: /tmp/interactive_workspace
+      run: |
+        cd ${GITHUB_WORKSPACE}/flex/tests/hqps
+        sed -i 's/default_graph: ldbc/default_graph: modern_graph/g' ./engine_config_test.yaml
+        pip3 install argparse
+        pip3 install neo4j
+        bash hqps_compiler_get_meta_test.sh ${INTERACTIVE_WORKSPACE} ./engine_config_test.yaml
+        sed -i 's/default_graph: modern_graph/default_graph: ldbc/g' ./engine_config_test.yaml
+
     - name: Run codegen test.
       env:
         GS_TEST_DIR: ${{ github.workspace }}/gstest

flex/tests/hqps/engine_config_test.yaml

@@ -23,6 +23,14 @@ compiler:
       - FilterIntoJoinRule
       - FilterMatchRule
       - NotMatchToAntiJoinRule
+  meta:
+    reader:
+      schema:
+        uri: http://localhost:7777/v1/service/status
+        interval: 1000 # ms
+      statistics:
+        uri: http://localhost:7777/v1/graph/%s/statistics
+        interval: 86400000 # ms
   endpoint:
     default_listen_address: localhost
     bolt_connector:

flex/tests/hqps/hqps_compiler_get_meta_test.sh

@@ -0,0 +1,99 @@
+#!/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}/../../
+SERVER_BIN=${FLEX_HOME}/build/bin/interactive_server
+GIE_HOME=${FLEX_HOME}/../interactive_engine/
+ADMIN_PORT=7777
+QUERY_PORT=10000
+
+# 
+if [ ! $# -eq 2 ]; then
+  echo "only receives: $# args, need 2"
+  echo "Usage: $0 <INTERACTIVE_WORKSPACE> <ENGINE_CONFIG>"
+  exit 1
+fi
+
+INTERACTIVE_WORKSPACE=$1
+ENGINE_CONFIG_PATH=$2
+if [ ! -d ${INTERACTIVE_WORKSPACE} ]; then
+  echo "INTERACTIVE_WORKSPACE: ${INTERACTIVE_WORKSPACE} not exists"
+  exit 1
+fi
+if [ ! -f ${ENGINE_CONFIG_PATH} ]; then
+  echo "ENGINE_CONFIG: ${ENGINE_CONFIG_PATH} not exists"
+  exit 1
+fi
+
+
+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 "interactive_server" |  awk '{print $2}' | xargs kill -9  || true
+    ps -ef | grep "GraphServer" |  awk '{print $2}' | xargs kill -9  || true
+    sleep 3
+    # check if service is killed
+    info "Kill Service success"
+}
+
+# kill service when exit
+trap kill_service EXIT
+
+# start engine service and load ldbc graph
+start_engine_service(){
+    #check SERVER_BIN exists
+    if [ ! -f ${SERVER_BIN} ]; then
+        err "SERVER_BIN not found"
+        exit 1
+    fi
+
+    cmd="${SERVER_BIN} -c ${ENGINE_CONFIG_PATH} --enable-admin-service true "
+    cmd="${cmd}  -w ${INTERACTIVE_WORKSPACE} --start-compiler true"
+
+    echo "Start engine service with command: ${cmd}"
+    ${cmd} &
+    sleep 10
+    #check interactive_server is running, if not, exit
+    ps -ef | grep "interactive_server" | grep -v grep
+
+    info "Start engine service success"
+}
+
+run_cypher_test() {
+  # run a simple cypher query: MATCH (n) RETURN count(n)
+  python3 ./test_count_vertices.py --endpoint neo4j://localhost:7687
+}
+
+kill_service
+start_engine_service
+# comiper service will fail to start, if the graph meta can not be retrieved
+run_cypher_test
+kill_service
+
+
+
+

flex/tests/hqps/test_count_vertices.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+#
+# Copyright 2020 Alibaba Group Holding Limited. All Rights Reserved.
+#
+# 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.
+#
+
+from neo4j import GraphDatabase
+from neo4j import Session as Neo4jSession
+
+import argparse
+
+def count_vertices(sess: Neo4jSession):
+    query = "MATCH (n) RETURN COUNT(n);"
+    result = sess.run(query)
+    for record in result:
+        print(record[0])
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Count the number of vertices in a graph.")
+    parser.add_argument("--endpoint", type=str, required=True, help="The endpoint to connect.")
+    args = parser.parse_args()
+
+    driver = GraphDatabase.driver(args.endpoint, auth=None)
+    with driver.session() as session:
+        count_vertices(session)
+    driver.close()

interactive_engine/compiler/conf/ir.compiler.properties

@@ -13,16 +13,19 @@ pegasus.hosts: localhost:1234
 # set timeout in milliseconds to connect to hiactor service
 # hiactor.timeout: 6000000
 
-# graph.schema
+# set schema access uri, two formats are supported:
+# 1. local file path: file://<absolute path to the local schema file>, i.e. file:///path/to/your/schema.json
+# 2. remote http path: http://<host>:<port>/<query>, i.e. http://localhost:8080/schema
+# we specifically support reading from a relative path if it is a local file.
 graph.schema: ../executor/ir/core/resource/modern_schema.json
 graph.store: exp
 
 graph.planner.is.on: true
 graph.planner.opt: RBO
 graph.planner.rules: FilterIntoJoinRule, FilterMatchRule, ExtendIntersectRule, ExpandGetVFusionRule
 
-# set file path of glogue input statistics
-# graph.statistics: src/test/resources/statistics/modern_statistics.json
+# set statistics access uri
+graph.meta.statistics.uri: src/test/resources/statistics/modern_statistics.json
 
 # set stored procedures directory path
 # graph.stored.procedures: <your stored procedures directory path>
@@ -61,3 +64,9 @@ calcite.default.charset: UTF-8
 
 # set the max capacity of the result streaming buffer for each query
 # per.query.stream.buffer.max.capacity: 256
+
+# 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
+# graph.meta.statistics.fetch.interval.ms: 86400000l

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/GraphServer.java

@@ -25,8 +25,10 @@
 import com.alibaba.graphscope.common.config.GraphConfig;
 import com.alibaba.graphscope.common.ir.meta.IrMeta;
 import com.alibaba.graphscope.common.ir.meta.IrMetaTracker;
+import com.alibaba.graphscope.common.ir.meta.fetcher.DynamicIrMetaFetcher;
 import com.alibaba.graphscope.common.ir.meta.fetcher.IrMetaFetcher;
 import com.alibaba.graphscope.common.ir.meta.fetcher.StaticIrMetaFetcher;
+import com.alibaba.graphscope.common.ir.meta.reader.HttpIrMetaReader;
 import com.alibaba.graphscope.common.ir.meta.reader.LocalIrMetaReader;
 import com.alibaba.graphscope.common.ir.planner.GraphRelOptimizer;
 import com.alibaba.graphscope.common.ir.tools.*;
@@ -48,6 +50,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.net.URI;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
@@ -187,7 +190,14 @@ public static void main(String[] args) throws Exception {
 
     private static IrMetaFetcher createIrMetaFetcher(Configs configs, IrMetaTracker tracker)
             throws IOException {
-        return new StaticIrMetaFetcher(new LocalIrMetaReader(configs), tracker);
+        URI schemaUri = URI.create(GraphConfig.GRAPH_META_SCHEMA_URI.get(configs));
+        if (schemaUri.getScheme() == null || schemaUri.getScheme().equals("file")) {
+            return new StaticIrMetaFetcher(new LocalIrMetaReader(configs), tracker);
+        } else if (schemaUri.getScheme().equals("http")) {
+            return new DynamicIrMetaFetcher(configs, new HttpIrMetaReader(configs), tracker);
+        }
+        throw new IllegalArgumentException(
+                "unknown graph meta reader mode: " + schemaUri.getScheme());
     }
 
     private static GraphProperties getTestGraph(Configs configs) {

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

@@ -17,8 +17,17 @@
 package com.alibaba.graphscope.common.config;
 
 public class GraphConfig {
-    public static final Config<String> GRAPH_SCHEMA = Config.stringConfig("graph.schema", ".");
-    public static final Config<String> GRAPH_STATISTICS =
-            Config.stringConfig("graph.statistics", "");
+    public static final Config<String> GRAPH_META_SCHEMA_URI =
+            Config.stringConfig("graph.schema", ".");
+
+    public static final Config<String> GRAPH_META_STATISTICS_URI =
+            Config.stringConfig("graph.meta.statistics.uri", "");
+
     public static final Config<String> GRAPH_STORE = Config.stringConfig("graph.store", "exp");
+
+    public static final Config<Long> GRAPH_META_SCHEMA_FETCH_INTERVAL_MS =
+            Config.longConfig("graph.meta.schema.fetch.interval.ms", 1000);
+
+    public static final Config<Long> GRAPH_META_STATISTICS_FETCH_INTERVAL_MS =
+            Config.longConfig("graph.meta.statistics.fetch.interval.ms", 24 * 3600 * 1000l);
 }

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

@@ -9,7 +9,9 @@ public class PlannerConfig {
     public static final Config<String> GRAPH_PLANNER_OPT =
             Config.stringConfig("graph.planner.opt", "RBO");
     public static final Config<String> GRAPH_PLANNER_RULES =
-            Config.stringConfig("graph.planner.rules", "");
+            Config.stringConfig(
+                    "graph.planner.rules",
+                    "FilterIntoJoinRule,FilterMatchRule,ExtendIntersectRule,ExpandGetVFusionRule");
     public static final Config<Integer> GRAPH_PLANNER_CBO_GLOGUE_SIZE =
             Config.intConfig("graph.planner.cbo.glogue.size", 3);
     public static final Config<Integer> JOIN_MIN_PATTERN_SIZE =

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

@@ -25,143 +25,150 @@
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.nio.file.Path;
 import java.util.Map;
 
 public class YamlConfigs extends Configs {
     private static ImmutableMap<String, ValueGetter> valueGetterMap;
 
     static {
         ImmutableMap.Builder<String, ValueGetter> mapBuilder = ImmutableMap.builder();
         mapBuilder
                 .put(
                         "graph.planner.is.on",
                         (Configs configs) -> configs.get("compiler.planner.is_on"))
                 .put("graph.planner.opt", (Configs configs) -> configs.get("compiler.planner.opt"))
                 .put(
                         "graph.planner.rules",
                         (Configs configs) -> {
                             String rules = configs.get("compiler.planner.rules");
                             if (rules != null) {
                                 rules = rules.replace("[", "").replace("]", "");
                             }
                             return rules;
                         })
                 .put(
                         "graph.schema",
                         (Configs configs) -> {
                             // if System.properties contains graph.schema, use it
                             String schema = System.getProperty("graph.schema");
                             if (schema != null) {
                                 return schema;
                             }
-                            String workspace = configs.get("directories.workspace");
-                            String subdir = configs.get("directories.subdirs.data");
-                            String graphName = configs.get("default_graph");
-                            if (workspace != null && subdir != null && graphName != null) {
-                                return Path.of(workspace, subdir, graphName, "graph.yaml")
-                                        .toString();
-                            } else {
-                                return null;
+                            return configs.get("compiler.meta.reader.schema.uri");
+                        })
+                .put(
+                        "graph.meta.statistics.uri",
+                        (Configs configs) -> {
+                            String statistics = System.getProperty("graph.meta.statistics.uri");
+                            if (statistics != null) {
+                                return statistics;
                             }
+                            return configs.get("compiler.meta.reader.statistics.uri");
                         })
+                .put(
+                        "graph.meta.schema.fetch.interval.ms",
+                        (Configs configs) -> configs.get("compiler.meta.reader.schema.interval"))
+                .put(
+                        "graph.meta.statistics.fetch.interval.ms",
+                        (Configs configs) ->
+                                configs.get("compiler.meta.reader.statistics.interval"))
                 .put(
                         "graph.store",
                         (Configs configs) -> {
                             if (configs.get("compute_engine.store.type") != null) {
                                 return configs.get("compute_engine.store.type");
                             } else {
                                 return "cpp-mcsr";
                             }
                         })
                 .put(
                         "pegasus.worker.num",
                         (Configs configs) -> {
                             String type = configs.get("compute_engine.type");
                             if (type == null || !type.equals("pegasus")) {
                                 return null;
                             } else {
                                 return configs.get("compute_engine.thread_num_per_worker");
                             }
                         })
                 .put(
                         "pegasus.batch.size",
                         (Configs configs) -> {
                             String type = configs.get("compute_engine.type");
                             if (type == null || !type.equals("pegasus")) {
                                 return null;
                             } else {
                                 return configs.get("compute_engine.batch_size");
                             }
                         })
                 .put(
                         "pegasus.output.capacity",
                         (Configs configs) -> {
                             String type = configs.get("compute_engine.type");
                             if (type == null || !type.equals("pegasus")) {
                                 return null;
                             } else {
                                 return configs.get("compute_engine.output_capacity");
                             }
                         })
                 .put(
                         "pegasus.hosts",
                         (Configs configs) -> {
                             String type = configs.get("compute_engine.type");
                             if (type == null || !type.equals("pegasus")) {
                                 return null;
                             } else {
                                 String hosts = configs.get("compute_engine.workers");
                                 if (hosts != null) {
                                     return hosts.replace("[", "").replace("]", "");
                                 }
                                 return hosts;
                             }
                         })
                 .put(
                         "hiactor.hosts",
                         (Configs configs) -> {
                             String type = configs.get("compute_engine.type");
                             if (type == null || !type.equals("hiactor")) {
                                 return null;
                             } else {
                                 String port = configs.get("http_service.query_port");
                                 String address = configs.get("http_service.default_listen_address");
                                 if (port != null && address != null) {
                                     String hosts = address + ":" + port;
                                     return hosts.replace("[", "").replace("]", "");
                                 }
                                 return null;
                             }
                         })
                 .put(
                         "neo4j.bolt.server.disabled",
                         (Configs configs) ->
                                 configs.get("compiler.endpoint.bolt_connector.disabled"))
                 .put(
                         "neo4j.bolt.server.port",
                         (Configs configs) -> configs.get("compiler.endpoint.bolt_connector.port"))
                 .put(
                         "gremlin.server.disabled",
                         (Configs configs) ->
                                 configs.get("compiler.endpoint.gremlin_connector.disabled"))
                 .put(
                         "gremlin.server.port",
                         (Configs configs) ->
                                 configs.get("compiler.endpoint.gremlin_connector.port"))
                 .put(
                         "query.execution.timeout.ms",
                         (Configs configs) -> configs.get("compiler.query_timeout"))
                 .put("engine.type", (Configs configs) -> configs.get("compute_engine.type"))
                 .put(
                         "calcite.default.charset",
                         (Configs configs) -> configs.get("compiler.calcite_default_charset"))
                 .put(
                         "gremlin.script.language.name",
                         (Configs configs) -> configs.get("compiler.gremlin_script_language_name"));
         valueGetterMap = mapBuilder.build();
     }
 
     public YamlConfigs(String file) throws IOException {
         this(file, FileLoadType.RELATIVE_PATH);
     }

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

@@ -18,6 +18,8 @@
 
 package com.alibaba.graphscope.common.ir.meta;
 
+import com.google.common.base.Objects;
+
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
@@ -36,4 +38,17 @@ public GraphId(T id) {
     public @Nullable T getId() {
         return id;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        GraphId<?> graphId = (GraphId<?>) o;
+        return Objects.equal(id, graphId.id);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(id);
+    }
 }