Merge branch 'master' into fix_session_pool_del_before_lock

Author: BeautyyuYanli

.github/workflows/deploy_release.yaml

@@ -7,13 +7,12 @@ on:
 
 jobs:
   ci:
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v3
-    - uses: pdm-project/setup-pdm@v3
+    - uses: actions/checkout@v5
+    - uses: pdm-project/setup-pdm@v4
       with:
-        python-version: '3.x'
         cache: true
     - name: Install dependencies
       run: pdm install

.github/workflows/run_test.yaml

@@ -21,15 +21,15 @@ on:
 jobs:
   ci-pip-install-from-source:
     # This is to verify the setup.py as a mitigation for remain python 3.6.2+ capability
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
     strategy:
       max-parallel: 2
       matrix:
-        python-version: [3.6, 3.7]
+        python-version: [3.8, 3.11]
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v5
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
       - name: Install nebulagraph-python from source and test dependencies
@@ -52,7 +52,7 @@ jobs:
     strategy:
       max-parallel: 2
       matrix:
-        python-version: [3.8, 3.9, '3.10', 3.11]
+        python-version: ['3.10', 3.11, 3.12, 3.13]
 
     steps:
       - name: Maximize runner space
@@ -63,9 +63,9 @@ jobs:
           remove-android: 'true'
           remove-haskell: 'true'
 
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v5
       - name: Set up Python ${{ matrix.python-version }}
-        uses: pdm-project/setup-pdm@v3
+        uses: pdm-project/setup-pdm@v4
         with:
           python-version: ${{ matrix.python-version }}
           cache: true
@@ -94,14 +94,14 @@ jobs:
           files: coverage.xml
 
   example-test:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
     strategy:
       matrix:
-        python-version: [3.11, 3.12]
+        python-version: ['3.10', 3.11, 3.12, 3.13]
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v5
       - name: Set up Python ${{ matrix.python-version }}
-        uses: pdm-project/setup-pdm@v3
+        uses: pdm-project/setup-pdm@v4
         with:
           python-version: ${{ matrix.python-version }}
           cache: true

nebula3/gclient/net/base.py

@@ -84,17 +84,6 @@ def _cast_value(value: Any) -> Value:
         casted_value.set_sVal(value)
     elif isinstance(value, float):
         casted_value.set_fVal(value)
-    elif isinstance(value, datetime.date):
-        date_value = Date(year=value.year, month=value.month, day=value.day)
-        casted_value.set_dVal(date_value)
-    elif isinstance(value, datetime.time):
-        time_value = Time(
-            hour=value.hour,
-            minute=value.minute,
-            sec=value.second,
-            microsec=value.microsecond,
-        )
-        casted_value.set_tVal(time_value)
     elif isinstance(value, datetime.datetime):
         datetime_value = DateTime(
             year=value.year,
@@ -106,6 +95,17 @@ def _cast_value(value: Any) -> Value:
             microsec=value.microsecond,
         )
         casted_value.set_dtVal(datetime_value)
+    elif isinstance(value, datetime.date):
+        date_value = Date(year=value.year, month=value.month, day=value.day)
+        casted_value.set_dVal(date_value)
+    elif isinstance(value, datetime.time):
+        time_value = Time(
+            hour=value.hour,
+            minute=value.minute,
+            sec=value.second,
+            microsec=value.microsecond,
+        )
+        casted_value.set_tVal(time_value)
     # TODO: add support for GeoSpatial
     elif isinstance(value, list):
         byte_list = []

nebula3/sclient/GraphStorageClient.py

@@ -10,7 +10,9 @@
 The client to scan vertex and edge from storage,
 the return data is from thr graph database
 """
+
 import sys
+import concurrent.futures
 
 from nebula3.sclient.ScanResult import ScanResult
 from nebula3.sclient.net import GraphStorageConnection
@@ -192,6 +194,69 @@ def scan_vertex_with_part(
             partial_success,
         )
 
+    # TODO: 1.Native async or PyO3
+    #       2.Error Handling
+    #       3.Statistical indicators
+    def scan_vertex_async(
+        self,
+        space_name,
+        tag_name,
+        prop_names=[],
+        start_time=DEFAULT_START_TIME,
+        end_time=DEFAULT_END_TIME,
+        where=None,
+        only_latest_version=False,
+        enable_read_from_follower=True,
+        partial_success=False,
+        batch_size=1000,
+        max_workers=8,
+    ):
+        """
+        scan_vertex_async：Multi-partition concurrency and streaming batch yield
+
+        :param space_name: the space name
+        :param tag_name: the tag name
+        :param prop_names: if given empty, return all property
+        :param start_time: the min version of vertex
+        :param end_time: the max version of vertex
+        :param where: now is unsupported
+        :param only_latest_version: when storage enable multi versions and only_latest_version is true,
+        only return latest version.
+        when storage disable multi versions, just use the default value.
+        :param enable_read_from_follower: if set to false, forbid follower read
+        :param partial_success: if set true, when partial success, it will continue until finish
+        :param batch_size: The number of points in each batch (passed to scan_vertex_with_part)
+        :param max_workers: Number of concurrent threads
+        :yield: part_id, VertexResult  # Each batch of data
+
+        """
+        part_leaders = self._meta_cache.get_part_leaders(space_name)
+        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+            future_to_part = {}
+            for part, leader in part_leaders.items():
+                future = executor.submit(
+                    self.scan_vertex_with_part,
+                    space_name,
+                    part,
+                    tag_name,
+                    prop_names,
+                    batch_size,  # The limit passed to scan_vertex_with_part
+                    start_time,
+                    end_time,
+                    where,
+                    only_latest_version,
+                    enable_read_from_follower,
+                    partial_success,
+                )
+                future_to_part[future] = part
+
+            for future in concurrent.futures.as_completed(future_to_part):
+                part = future_to_part[future]
+                scan_result = future.result()  # ScanResult
+                while scan_result is not None and scan_result.has_next():
+                    batch = scan_result.next()
+                    yield part, batch
+
     def _scan_vertex(
         self,
         space_name,
@@ -337,6 +402,65 @@ def scan_edge_with_part(
             partial_success,
         )
 
+    def scan_edge_async(
+        self,
+        space_name,
+        edge_name,
+        prop_names=[],
+        start_time=DEFAULT_START_TIME,
+        end_time=DEFAULT_END_TIME,
+        where=None,
+        only_latest_version=False,
+        enable_read_from_follower=True,
+        partial_success=False,
+        batch_size=1000,
+        max_workers=8,
+    ):
+        """
+        scan_edge_async：Multi-partition concurrency and streaming batch yield
+
+        :param space_name: the space name
+        :param prop_names: if given empty, return all property
+        :param edge_name: the edge name
+        :param start_time: the min version of edge
+        :param end_time: the max version of edge
+        :param where: now is unsupported
+        :param only_latest_version: when storage enable multi versions and only_latest_version is true,
+        only return latest version.
+        when storage disable multi versions, just use the default value.
+        :param enable_read_from_follower: if set to false, forbid follower read
+        :param partial_success: if set true, when partial success, it will continue until finish
+        :param batch_size: The number of edges per batch (passed to scan_edge_with_part)
+        :param max_workers: Number of concurrent threads
+        :yield: part_id, EdgeResult  # Each batch of data
+        """
+        part_leaders = self._meta_cache.get_part_leaders(space_name)
+        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+            future_to_part = {}
+            for part, leader in part_leaders.items():
+                future = executor.submit(
+                    self.scan_edge_with_part,
+                    space_name,
+                    part,
+                    edge_name,
+                    prop_names,
+                    batch_size,
+                    start_time,
+                    end_time,
+                    where,
+                    only_latest_version,
+                    enable_read_from_follower,
+                    partial_success,
+                )
+                future_to_part[future] = part
+
+            for future in concurrent.futures.as_completed(future_to_part):
+                part = future_to_part[future]
+                scan_result = future.result()
+                while scan_result is not None and scan_result.has_next():
+                    batch = scan_result.next()
+                    yield part, batch
+
     def _scan_edge(
         self,
         space_name,

nebula3/utils/__init__.py

@@ -0,0 +1 @@
+from .hash import hash

nebula3/utils/hash.py

@@ -0,0 +1,49 @@
+# nebula3/hash.py
+from __future__ import annotations
+
+_M: int = 0xC6A4A7935BD1E995
+_R: int = 47
+_MASK64: int = (1 << 64) - 1
+
+
+def _read_u64_le(buf: bytes) -> int:
+    """Convert little-endian bytes of up to 8 bytes to an unsigned integer."""
+    return int.from_bytes(buf, byteorder="little", signed=False)
+
+
+def hash(data: bytes | str, seed: int = 0xC70F6907) -> int:
+    """MurmurHash2 64-bit variant:
+    :Param data: supports str (utf-8 encoding), bytes, bytearray
+    :Param seed: defaults to 0xC70F6907
+    :return: Python int, in the range of signed 64-bit
+    """
+    if isinstance(data, str):
+        data_as_bytes = data.encode("utf-8")
+    elif isinstance(data, (bytes, bytearray)):
+        data_as_bytes = bytes(data)
+    else:
+        raise TypeError("Input must be str, bytes, or bytearray")
+
+    h = (seed ^ (_M * len(data_as_bytes) & _MASK64)) & _MASK64
+    off = len(data_as_bytes) // 8 * 8
+    for i in range(0, off, 8):
+        k = _read_u64_le(data_as_bytes[i : i + 8])
+        k = (k * _M) & _MASK64
+        k ^= k >> _R
+        k = (k * _M) & _MASK64
+        h ^= k
+        h = (h * _M) & _MASK64
+
+    tail = data_as_bytes[off:]
+    if tail:
+        t = _read_u64_le(tail)
+        h ^= t
+        h = (h * _M) & _MASK64
+
+    h ^= h >> _R
+    h = (h * _M) & _MASK64
+    h ^= h >> _R
+
+    if h & (1 << 63):
+        h -= 1 << 64
+    return h