RPC implementation for sending match aggregate, match group, and match group aggregate queries (#148)

## What is the goal of this PR?

We have implemented RPC support for sending match aggregate, match group, and match group aggregate queries to server

## What are the changes implemented in this PR?

1. Update `@graknlabs_protocol` to the version that supports the new queries.
2. Support processing the new queries

Author: Ganeshwara Hananda

dependencies/graknlabs/artifacts.bzl

@@ -27,5 +27,5 @@ def graknlabs_grakn_core_artifacts():
         artifact_name = "grakn-core-server-{platform}-{version}.{ext}",
         tag_source = deployment["artifact.release"],
         commit_source = deployment["artifact.snapshot"],
-        commit = "81653e39fa681d20b37fdba13f158a0d1c1a6c2e",
+        commit = "118eee244a4949c629de0277804155ebd4b316be",
     )

grakn/concept/answer/concept_map.py

@@ -23,11 +23,10 @@
 
 from grakn.common.exception import GraknClientException
 from grakn.concept.proto import concept_proto_reader
-from grakn.concept.answer.answer import Answer
 from grakn.concept.concept import Concept
 
 
-class ConceptMap(Answer):
+class ConceptMap:
 
     _THING = "thing"
 
@@ -63,10 +62,6 @@ def __hash__(self):
 def _of(concept_map_proto: answer_proto.ConceptMap):
     variable_map = {}
     for res_var in concept_map_proto.map:
-        res_concept = concept_map_proto.map[res_var]
-        if res_concept.HasField(ConceptMap._THING):
-            concept = concept_proto_reader.thing(res_concept.thing)
-        else:
-            concept = concept_proto_reader.type_(res_concept.type)
+        concept = concept_proto_reader.concept(concept_map_proto.map[res_var])
         variable_map[res_var] = concept
     return ConceptMap(variable_map)

grakn/concept/answer/concept_map_group.py

@@ -0,0 +1,37 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+#
+
+import graknprotocol.protobuf.answer_pb2 as answer_proto
+from grakn.concept.answer import concept_map
+from grakn.concept.proto.concept_proto_reader import concept
+
+class ConceptMapGroup:
+    def __init__(self, owner, concept_maps):
+        self._owner = owner
+        self._concept_maps = concept_maps
+
+    def owner(self):
+        return self._owner
+
+    def concept_maps(self):
+        return self._concept_maps
+
+
+def _of(concept_map_group_proto: answer_proto.ConceptMapGroup):
+    return ConceptMapGroup(concept(concept_map_group_proto.owner), map(lambda cm: concept_map._of(cm), concept_map_group_proto.concept_maps))

grakn/concept/answer/numeric.py

@@ -0,0 +1,61 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+#
+
+import graknprotocol.protobuf.answer_pb2 as answer_proto
+
+from grakn.common.exception import GraknClientException
+
+
+class Numeric:
+    def __init__(self, int_value, float_value):
+        self._int_value = int_value
+        self._float_value = float_value
+
+    def is_int(self):
+        return self._int_value is not None
+
+    def is_float(self):
+        return self._float_value is not None
+
+    def is_nan(self):
+        return not self.is_int() and not self.is_float()
+
+    def as_float(self):
+        if (self.is_float()):
+            return self._float_value
+        else:
+            raise GraknClientException("Illegal casting operation to 'float'.")
+
+    def as_int(self):
+        if (self.is_int()):
+            return self._int_value
+        else:
+            raise GraknClientException("Illegal casting operation to 'int'.")
+
+
+def _of(numeric_proto: answer_proto.Numeric):
+    numeric_case = numeric_proto.WhichOneof("value")
+    if numeric_case == "long_value":
+        return Numeric(numeric_proto.long_value, None)
+    elif numeric_case == "double_value":
+        return Numeric(None, numeric_proto.double_value)
+    elif numeric_case == "nan":
+        return Numeric(None, None)
+    else:
+        raise GraknClientException("The answer type '" + numeric_case + "' was not recognised.")

grakn/concept/answer/answer.py renamed to grakn/concept/answer/numeric_group.py

@@ -18,18 +18,20 @@
 #
 
 import graknprotocol.protobuf.answer_pb2 as answer_proto
+from grakn.concept.answer import numeric
+from grakn.concept.proto.concept_proto_reader import concept
 
-from grakn.common.exception import GraknClientException
-from grakn.concept.answer import concept_map
+class NumericGroup:
+    def __init__(self, owner, numeric):
+        self._owner = owner
+        self._numeric = numeric
 
+    def owner(self):
+        return self._owner
 
-class Answer(object):
+    def numeric(self):
+        return self._numeric
 
-    _CONCEPT_MAP = "concept_map"
 
-
-def _of(proto_answer: answer_proto.Answer):
-    answer_case = proto_answer.WhichOneof("answer")
-    if answer_case == Answer._CONCEPT_MAP:
-        return concept_map._of(proto_answer.concept_map)
-    raise GraknClientException("The answer type " + answer_case + " was not recognised.")
+def _of(numeric_group_proto: answer_proto.NumericGroup):
+    return NumericGroup(concept(numeric_group_proto.owner), numeric._of(numeric_group_proto.number))

grakn/concept/proto/concept_proto_reader.py

@@ -20,6 +20,7 @@
 import graknprotocol.protobuf.concept_pb2 as concept_proto
 
 from grakn.common.exception import GraknClientException
+from grakn.concept.answer.concept_map import ConceptMap
 from grakn.concept.thing.attribute import BooleanAttribute, LongAttribute, DoubleAttribute, StringAttribute, \
     DateTimeAttribute
 from grakn.concept.thing.entity import Entity
@@ -32,6 +33,14 @@
 from grakn.concept.type.thing_type import ThingType
 
 
+def concept(con_proto: concept_proto.Concept):
+    if con_proto.HasField(ConceptMap._THING):
+        concept = thing(con_proto.thing)
+    else:
+        concept = type_(con_proto.type)
+    return concept
+
+
 def thing(thing_proto: concept_proto.Thing):
     if thing_proto.encoding == concept_proto.Thing.Encoding.Value("ENTITY"):
         return Entity._of(thing_proto)

grakn/query/query_manager.py

@@ -23,7 +23,7 @@
 import graknprotocol.protobuf.transaction_pb2 as transaction_proto
 
 from grakn import grakn_proto_builder
-from grakn.concept.answer import concept_map
+from grakn.concept.answer import concept_map, concept_map_group, numeric, numeric_group
 from grakn.options import GraknOptions
 
 
@@ -34,45 +34,65 @@ def __init__(self, transaction):
 
     def match(self, query: str, options=GraknOptions()):
         request = query_proto.Query.Req()
-        match_req = query_proto.Graql.Match.Req()
+        match_req = query_proto.Query.Match.Req()
         match_req.query = query
         request.match_req.CopyFrom(match_req)
         return map(lambda answer_proto: concept_map._of(answer_proto), self._iterate_query(request, lambda res: res.query_res.match_res.answers, options))
 
+    def match_aggregate(self, query: str, options=GraknOptions()):
+        request = query_proto.Query.Req()
+        match_aggregate_req = query_proto.Query.MatchAggregate.Req()
+        match_aggregate_req.query = query
+        request.match_aggregate_req.CopyFrom(match_aggregate_req)
+        return self._iterate_query(request, lambda res: [numeric._of(res.query_res.match_aggregate_res.answer)], options)
+
+    def match_group(self, query: str, options=GraknOptions()):
+        request = query_proto.Query.Req()
+        match_group_req = query_proto.Query.MatchGroup.Req()
+        match_group_req.query = query
+        request.match_group_req.CopyFrom(match_group_req)
+        return map(
+            lambda cmg_proto: concept_map_group._of(cmg_proto),
+            self._iterate_query(request, lambda res: res.query_res.match_group_res.answers, options)
+        )
+
+    def match_group_aggregate(self, query: str, options=GraknOptions()):
+        request = query_proto.Query.Req()
+        match_group_aggregate_req = query_proto.Query.MatchGroupAggregate.Req()
+        match_group_aggregate_req.query = query
+        request.match_group_aggregate_req.CopyFrom(match_group_aggregate_req)
+        return map(
+            lambda numeric_group_proto: numeric_group._of(numeric_group_proto),
+            self._iterate_query(request, lambda res: res.query_res.match_group_aggregate_res.answers, options)
+        )
+
     def insert(self, query: str, options=GraknOptions()):
         request = query_proto.Query.Req()
-        insert_req = query_proto.Graql.Insert.Req()
+        insert_req = query_proto.Query.Insert.Req()
         insert_req.query = query
         request.insert_req.CopyFrom(insert_req)
         return map(lambda answer_proto: concept_map._of(answer_proto), self._iterate_query(request, lambda res: res.query_res.insert_res.answers, options))
 
     def delete(self, query: str, options=GraknOptions()):
         request = query_proto.Query.Req()
-        delete_req = query_proto.Graql.Delete.Req()
+        delete_req = query_proto.Query.Delete.Req()
         delete_req.query = query
         request.delete_req.CopyFrom(delete_req)
-        return self._run_query(request, options)
+        return self._iterate_query(request, lambda res: [], options)
 
     def define(self, query: str, options=GraknOptions()):
         request = query_proto.Query.Req()
-        define_req = query_proto.Graql.Define.Req()
+        define_req = query_proto.Query.Define.Req()
         define_req.query = query
         request.define_req.CopyFrom(define_req)
-        return self._run_query(request, options)
+        return self._iterate_query(request, lambda res: [], options)
 
     def undefine(self, query: str, options=GraknOptions()):
         request = query_proto.Query.Req()
-        undefine_req = query_proto.Graql.Undefine.Req()
+        undefine_req = query_proto.Query.Undefine.Req()
         undefine_req.query = query
         request.undefine_req.CopyFrom(undefine_req)
-        return self._run_query(request, options)
-
-    def _run_query(self, query_req: query_proto.Query.Req, options: GraknOptions):
-        req = transaction_proto.Transaction.Req()
-        query_req.options.CopyFrom(grakn_proto_builder.options(options))
-        req.query_req.CopyFrom(query_req)
-        # Using stream makes this request asynchronous.
-        return self._transaction._stream(req)
+        return self._iterate_query(request, lambda res: [], options)
 
     def _iterate_query(self, query_req: query_proto.Query.Req, response_reader: Callable[[transaction_proto.Transaction.Res], List], options: GraknOptions):
         req = transaction_proto.Transaction.Req()

requirements.txt

@@ -17,15 +17,14 @@
 # under the License.
 #
 
-
 ## Configuration options
 
 # --extra-index-url https://repo.grakn.ai/repository/pypi-snapshot/simple  # Allow importing of snapshots
 
 
 ## Dependencies
 
-graknprotocol==2.0.0a4
+graknprotocol==2.0.0a5
 grpcio==1.33.2
 protobuf==3.6.1
 six>=1.11.0

tests/integration/test_concept.py

@@ -22,7 +22,7 @@
 
 from grakn.client import GraknClient, SessionType, TransactionType, ValueType
 from grakn.common.exception import GraknClientException
-from test.integration.base import test_base, GraknServer
+from tests.integration.base import test_base, GraknServer
 
 
 class TestConcept(test_base):

tests/integration/test_connection.py

@@ -19,7 +19,7 @@
 
 import unittest
 from grakn.client import GraknClient, SessionType, TransactionType
-from test.integration.base import test_base, GraknServer
+from tests.integration.base import test_base, GraknServer
 
 
 class TestConnection(test_base):