Merge branch 'master' into fix-package

.github/workflows/package.yml

@@ -18,7 +18,7 @@ jobs:
           - ubuntu1604
           - ubuntu1804
           - ubuntu2004
-          - centos6
+          # - centos6
           - centos7
           - centos8
     container:

.github/workflows/pull_request.yml

@@ -92,7 +92,7 @@ jobs:
                   -DENABLE_TESTING=on \
                   -DENABLE_BUILD_STORAGE=on \
                   -B build
-              echo "::set-output name=j::8"
+              echo "::set-output name=j::10"
               ;;
             ubuntu1804)
               # build with Debug type
@@ -103,7 +103,7 @@ jobs:
                   -DENABLE_TESTING=on \
                   -DENABLE_BUILD_STORAGE=on \
                   -B build
-              echo "::set-output name=j::8"
+              echo "::set-output name=j::10"
               ;;
             esac
             ;;
@@ -117,7 +117,7 @@ jobs:
                 -DENABLE_TESTING=on \
                 -DENABLE_BUILD_STORAGE=on \
                 -B build
-            echo "::set-output name=j::4"
+            echo "::set-output name=j::6"
             ;;
           esac
       - name: Make graph

tests/Makefile

@@ -11,7 +11,7 @@ gherkin_fmt = ~/.local/bin/reformat-gherkin
 
 RM_DIR ?= true
 TEST_DIR ?= $(CURR_DIR)
-J ?= 8
+J ?= 10
 
 install-deps:
 	pip3 install --user -U setuptools wheel -i $(PYPI_MIRROR)

tests/common/comparator.py

@@ -6,6 +6,7 @@
 import math
 import re
 
+from enum import Enum
 from typing import Union, Dict, List
 from nebula2.common.ttypes import (
     DataSet,
@@ -20,17 +21,19 @@
 KV = Dict[Union[str, bytes], Value]
 Pattern = type(re.compile(r'/'))
 
+CmpType = Enum('CmpType', ('EQUAL', 'CONTAINS', 'NOT_CONTAINS'))
+
 
 class DataSetComparator:
     def __init__(self,
                  strict=True,
                  order=False,
-                 included=False,
-                 decode_type: str = 'utf-8',
+                 contains=CmpType.EQUAL,
+                 decode_type='utf-8',
                  vid_fn=None):
         self._strict = strict
         self._order = order
-        self._included = included
+        self._contains = contains
         self._decode_type = decode_type
         self._vid_fn = vid_fn
 
@@ -43,12 +46,18 @@ def b(self, v: str) -> bytes:
     def s(self, b: bytes) -> str:
         return b.decode(self._decode_type)
 
+    def _whether_return(self, cmp: bool) -> bool:
+        return ((self._contains == CmpType.EQUAL and not cmp)
+                or (self._contains == CmpType.NOT_CONTAINS and cmp))
+
     def compare(self, resp: DataSet, expect: DataSet):
+        if self._contains == CmpType.NOT_CONTAINS and len(resp.rows) == 0:
+            return True, None
         if all(x is None for x in [expect, resp]):
             return True, None
         if None in [expect, resp]:
             return False, -1
-        if len(resp.rows) < len(expect.rows):
+        if len(resp.rows) < len(expect.rows) and self._contains == CmpType.EQUAL:
             return False, -1
         if len(resp.column_names) != len(expect.column_names):
             return False, -1
@@ -57,13 +66,14 @@ def compare(self, resp: DataSet, expect: DataSet):
                 return False, -2
         if self._order:
             for i in range(0, len(expect.rows)):
-                if not self.compare_row(resp.rows[i], expect.rows[i]):
+                cmp = self.compare_row(resp.rows[i], expect.rows[i])
+                if self._whether_return(cmp):
                     return False, i
-            if self._included:
+            if self._contains == CmpType.CONTAINS:
                 return True, None
             return len(resp.rows) == len(expect.rows), -1
         return self._compare_list(resp.rows, expect.rows, self.compare_row,
-                                  self._included)
+                                  self._contains)
 
     def compare_value(self, lhs: Value, rhs: Union[Value, Pattern]) -> bool:
         """
@@ -327,7 +337,7 @@ def compare_row(self, lhs: Row, rhs: Row):
         return all(
             self.compare_value(l, r) for (l, r) in zip(lhs.values, rhs.values))
 
-    def _compare_list(self, lhs, rhs, cmp_fn, included=False):
+    def _compare_list(self, lhs, rhs, cmp_fn, contains=False):
         visited = []
         for j, rr in enumerate(rhs):
             found = False
@@ -336,9 +346,11 @@ def _compare_list(self, lhs, rhs, cmp_fn, included=False):
                     visited.append(i)
                     found = True
                     break
-            if not found:
+            if self._whether_return(found):
                 return False, j
         size = len(lhs)
-        if included:
+        if contains == CmpType.CONTAINS:
             return len(visited) <= size, -1
+        if contains == CmpType.NOT_CONTAINS:
+            return True, -1
         return len(visited) == size, -1

tests/common/types.py

@@ -48,6 +48,9 @@ def use_stmt(self) -> str:
     def drop_stmt(self) -> str:
         return f"DROP SPACE IF EXISTS `{self.name}`;"
 
+    def is_int_vid(self) -> bool:
+        return self.vid_type == 'int'
+
 
 class Column:
     def __init__(self, index: int):

tests/common/utils.py

@@ -321,37 +321,72 @@ def space_generator(size=6, chars=string.ascii_uppercase + string.digits):
     return ''.join(random.choice(chars) for _ in range(size))
 
 
-def check_resp(resp, stmt):
-    assert resp is not None, "response is None"
+def check_resp(resp, stmt: str):
+    assert resp is not None, f"response is None, stmt: {stmt}"
     msg = f"Fail to exec: {stmt}, error: {resp.error_msg()}"
     assert resp.is_succeeded(), msg
 
 
-def response(sess, stmt):
+def retry(times: int, predicate=lambda x: x and x.is_succeeded()):
+    def _retry(func):
+        def wrapper(*args, **kwargs):
+            resp = None
+            for i in range(times):
+                resp = func(*args, **kwargs)
+                if predicate(resp):
+                    return resp
+                time.sleep(0.5)
+            return resp
+
+        return wrapper
+
+    return _retry
+
+
+@retry(30)
+def try_execute(sess: Session, stmt: str):
+    return sess.execute(stmt)
+
+
+def return_if_not_leader_changed(resp) -> bool:
+    if not resp:
+        return True
+    if resp.is_succeeded():
+        return True
+
+    err_msg = resp.error_msg()
+    return err_msg.find('Storage Error: The leader has changed') < 0
+
+
+@retry(30, return_if_not_leader_changed)
+def process_leader_changed(sess: Session, stmt: str):
+    return sess.execute(stmt)
+
+
+def response(sess: Session, stmt: str, need_try: bool = False):
     try:
-        resp = sess.execute(stmt)
-        check_resp(resp, stmt)
-        return resp
+        if need_try:
+            return try_execute(sess, stmt)
+        return process_leader_changed(sess, stmt)
     except Exception as ex:
         assert not ex, f"Fail to exec: {stmt}, exception: {ex}"
 
 
-def create_space(space_desc: SpaceDesc, sess: Session):
-    def exec(stmt):
-        response(sess, stmt)
+def resp_ok(sess: Session, stmt: str, need_try: bool = False):
+    resp = response(sess, stmt, need_try)
+    check_resp(resp, stmt)
+    return resp
 
-    stmts = [
-        space_desc.drop_stmt(),
-        space_desc.create_stmt(),
-        space_desc.use_stmt(),
-    ]
 
-    exec(";".join(stmts))
+def create_space(space_desc: SpaceDesc, sess: Session):
+    resp_ok(sess, space_desc.drop_stmt(), True)
+    resp_ok(sess, space_desc.create_stmt(), True)
+    resp_ok(sess, space_desc.use_stmt(), True)
 
 
 def _load_data_from_file(sess, data_dir, fd):
     for stmt in CSVImporter(fd, data_dir):
-        response(sess, stmt)
+        response(sess, stmt, True)
 
 
 def load_csv_data(
@@ -374,6 +409,7 @@ def load_csv_data(
         assert space is not None
         if not space_name:
             space_name = space.get('name', "A" + space_generator())
+
         space_desc = SpaceDesc(
             name=space_name,
             vid_type=space.get('vidType', 'FIXED_STRING(32)'),
@@ -386,10 +422,8 @@ def load_csv_data(
         create_space(space_desc, sess)
 
         schemas = config['schema']
-        stmts = ' '.join(map(lambda x: x.strip(), schemas.splitlines()))
-        response(sess, stmts)
-
-        time.sleep(3)
+        for line in schemas.splitlines():
+            resp_ok(sess, line.strip(), True)
 
         for fd in config["files"]:
             _load_data_from_file(sess, data_dir, fd)