add point store and candidate set

example/random/small_schema.ngql

@@ -0,0 +1,6 @@
+CREATE SPACE `random_small`( partition_num=8, replica_factor=1, vid_type=INT64);
+
+create tag point(vector string);
+create edge e();
+
+insert vertex point(vector) VALUES 0:("")) 

merak/candidate_set.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+
+from asyncio import futures
+from concurrent.futures.thread import _worker
+from merak.point import Point
+from typing import List, Tuple
+from merak.point_store import PointStore
+import heapq
+from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
+import concurrent
+
+executor = ThreadPoolExecutor(max_workers=40)
+
+class CandidateSet(object):
+    def __init__(self, target: Point, points: List[Point], max_size: int, point_store: PointStore) -> None:
+        global executor
+        self.target = target
+        self.max_size = max_size
+        self.point_store = point_store
+        self.executor = executor
+        self.points: List[Tuple(float, Point)] = [(
+            p.distance(self.target), p) for p in points]
+        self.points.sort(key=lambda x: x[0])
+        self.visited = set([p.id for p in points])
+        self.furthest = self.points[-1]
+
+    def pop(self) -> Point:
+
+        # if len(self.futures) != 0:
+        #     ret = concurrent.futures.wait(
+        #         self.futures, return_when=concurrent.futures.FIRST_COMPLETED)
+        #     done = ret.done
+        #     self.futures = ret.not_done
+        #     for f in done:
+        #         point: Point = f.result()
+        #         if point.id in self.visited:
+        #             continue
+        #         if len(self.points) < self.max_size:
+        #             heapq.heappush(
+        #                 self.points, (point.distance(self.target), point))
+        #         else:
+        #             heapq.heappushpop(
+        #                 self.points, point.distance(self.target), point)
+        if len(self.points) == 0:
+            return None
+        ret = heapq.heappop(self.points)
+        return ret[1]
+
+    def add(self, id_list: List[int]):
+        points = self.executor.map(self.worker,id_list)
+        for point in points:
+            if point.id in self.visited:
+                return
+            self.visited.add(point.id)
+            x = (point.distance(self.target), point)
+            self.points.append(x)
+        self.points.sort(key=lambda x:x[0])
+        self.points = self.points[0:self.max_size]
+
+
+
+    def worker(self, id: int) -> Point:
+        point = self.point_store.get_point(id, True)
+        return point

merak/client.py

@@ -43,7 +43,8 @@ def get_neighbors(self, vid) -> Tuple[str, Dict]:
         '''
         # todo: replace t1 as tag, col1 as property
         # todo: use int id?
-        query = "FETCH PROP ON t1 \'{}\' YIELD properties(vertex).col1".format(vid)
+        query = "FETCH PROP ON t1 \'{}\' YIELD properties(vertex).col1".format(
+            vid)
         result = self.session.execute(query)
         if not result.is_succeeded():
             raise RuntimeError("fetch failed")
@@ -54,7 +55,8 @@ def get_neighbors(self, vid) -> Tuple[str, Dict]:
             raise RuntimeError("fetch no result")
 
         # todo: replace e1 as edge
-        query = "GO FROM \'{}\' OVER e1 YIELD rank(edge) as rank, dst(edge) as dst".format(vid)
+        query = "GO FROM \'{}\' OVER e1 YIELD rank(edge) as rank, dst(edge) as dst".format(
+            vid)
         result = self.session.execute(query)
         if not result.is_succeeded():
             raise RuntimeError("go failed")
@@ -76,7 +78,8 @@ def get_neighbors(self, vid) -> Tuple[str, Dict]:
         return (vec, neighbors)
 
     def insert_vertex(self, vid, vector):
-        query = "INSERT VERTEX t1(col1) VALUES \'{}\': (\'{}\')".format(vid, vector)
+        query = "INSERT VERTEX t1(col1) VALUES \'{}\': (\'{}\')".format(
+            vid, vector)
         result = self.session.execute(query)
         if not result.is_succeeded():
             raise RuntimeError("insert vertex failed")
@@ -97,7 +100,10 @@ def insert(self, batch: InsertBatch):
             raise RuntimeError(f"insert batch {batchStr} failed")
 
     def execute(self, query):
-        return self.session.execute(query)
+        session = self.pool.get_session('root', 'nebula')
+        result = session.execute(query)
+        session.release()
+        return result
 
     def close(self):
         self.session.release()