GG-49932 Vector query fast paths: one-pass response decode, shared ctypes classes, sane paging default

pygridgain/aio_cache.py

@@ -494,7 +494,7 @@ def scan(self, page_size: int = 1, partitions: int = -1, local: bool = False) ->
         return AioScanCursor(self.client, self.cache_info, page_size, partitions, local)
 
     def vector(self, type_name: str, field: str, clause_vector: List[float],
-               k: int, threshold: float, page_size: int = 1, ef_search: int = 0,
+               k: int, threshold: float, page_size: int = None, ef_search: int = 0,
                with_scores: bool = False, no_content: bool = False) -> AioVectorCursor:
         """
         Ignite supports vector queries based on Apache Lucene engine.
@@ -506,10 +506,12 @@ def vector(self, type_name: str, field: str, clause_vector: List[float],
          per-element conversion and is by far the cheapest to send, which matters because embedding
          dimensions are typically in the hundreds or thousands. Converting such an array to a Python
          list before passing it costs more than the query itself.
-        :param k: [K]NN, how many vectors to return.
+        :param k: [K]NN, how many vectors to return. Must be positive; the server also bounds it
+         (GRIDGAIN_VECTOR_MAX_K, 10000 by default) and rejects a query above that bound.
         :param threshold: similarity threshold, non-positive values disable it.
-        :param page_size: (optional) page size. Default size is 1 (slowest
-         and safest),
+        :param page_size: (optional) page size. Defaults to k: a vector query returns at most k
+         rows, so the whole result arrives in one page. Set it lower only to cap the size of a
+         single response,
         :param ef_search: (optional) search beam width: how many candidate vectors the engine
          keeps while traversing the index graph. The engine returns the k best of those
          candidates, so k controls the result size while the beam controls the search quality:
@@ -525,6 +527,12 @@ def vector(self, type_name: str, field: str, clause_vector: List[float],
          `(key, value, score)` with `with_scores`, bare `key` with `no_content`, and
          `(key, score)` with both.
         """
+        if k < 1:
+            raise ValueError(f'k must be positive, got {k}')
+
+        if page_size is None:
+            page_size = k
+
         query_flags = ((VECTOR_FLAG_WITH_SCORES if with_scores else 0)
                        | (VECTOR_FLAG_NOCONTENT if no_content else 0))
 

pygridgain/api/sql.py

@@ -22,6 +22,7 @@
 from pygridgain.datatypes.sql import StatementType
 from pygridgain.exceptions import NotSupportedByClusterError
 from pygridgain.queries import Query, query_perform
+from pygridgain.queries.response import VectorResponse
 from pygridgain.queries.op_codes import (
     OP_QUERY_SCAN, OP_QUERY_SCAN_CURSOR_GET_PAGE, OP_QUERY_SQL, OP_QUERY_SQL_CURSOR_GET_PAGE, OP_QUERY_SQL_FIELDS,
     OP_QUERY_SQL_FIELDS_CURSOR_GET_PAGE, OP_RESOURCE_CLOSE, OP_QUERY_VECTOR, OP_QUERY_VECTOR_CURSOR_GET_PAGE
@@ -480,9 +481,9 @@ def vector(conn: 'Connection', cache_info: CacheInfo, page_size: int,
      Value dict is of following format:
 
      * `cursor`: int, cursor ID,
-     * `data`: result rows - a dict of key-value pairs when `query_flags` is 0, otherwise
-       a list of per-row dicts with `key`, optionally `value` (no VECTOR_FLAG_NOCONTENT) and
-       optionally `score` (VECTOR_FLAG_WITH_SCORES) entries,
+     * `data`: result rows as final Python values, shaped by the flags: `(key, value)` tuples
+       when `query_flags` is 0, otherwise `key`, `(key, score)`, `(key, value)` or
+       `(key, value, score)` per VECTOR_FLAG_NOCONTENT / VECTOR_FLAG_WITH_SCORES,
      * `more`: bool, True if more data is available for subsequent
        ‘vector_cursor_get_page’ calls.
     """
@@ -500,25 +501,6 @@ async def vector_async(conn: 'AioConnection', cache_info: CacheInfo, page_size:
                           ef_search, query_flags)
 
 
-def __vector_rows_type(query_flags):
-    """
-    Response rows encoding: a plain key-value sequence for legacy queries, a row struct shaped
-    by the flags otherwise.
-    """
-    if not query_flags:
-        return Map
-
-    row = [('key', AnyDataObject)]
-
-    if not query_flags & VECTOR_FLAG_NOCONTENT:
-        row.append(('value', AnyDataObject))
-
-    if query_flags & VECTOR_FLAG_WITH_SCORES:
-        row.append(('score', PyFloat))
-
-    return StructArray(row)
-
-
 def __vector(conn, cache_info, page_size, type_name, field, clause_vector, k, threshold, ef_search, query_flags):
     fields = [
         ('cache_info', CacheInfo),
@@ -553,16 +535,17 @@ def __vector(conn, cache_info, page_size, type_name, field, clause_vector, k, th
         raise NotSupportedByClusterError('The cluster does not support extended vector queries '
                                          '(efSearch, scores, NOCONTENT) - QUERY_VECTOR_EXTENDED feature is absent.')
 
-    query_struct = Query(OP_QUERY_VECTOR, fields)
+    query_struct = Query(OP_QUERY_VECTOR, fields, response_type=VectorResponse)
 
+    # The response decodes in one pass (VectorResponse): rows leave it as final Python values,
+    # already shaped the way the cursor yields them.
     return query_perform(
         query_struct, conn,
         query_params=query_params,
-        response_config=[
-            ('cursor', Long),
-            ('data', __vector_rows_type(query_flags)),
-            ('more', Bool),
-        ],
+        with_scores=bool(query_flags & VECTOR_FLAG_WITH_SCORES),
+        no_content=bool(query_flags & VECTOR_FLAG_NOCONTENT),
+        legacy=not query_flags,
+        has_cursor=True,
         post_process_fun=__query_result_post_process
     )
 
@@ -597,17 +580,18 @@ def __vector_cursor_get_page(conn, cursor, query_flags):
         OP_QUERY_VECTOR_CURSOR_GET_PAGE,
         [
             ('cursor', Long),
-        ]
+        ],
+        response_type=VectorResponse,
     )
     return query_perform(
         query_struct, conn,
         query_params={
             'cursor': cursor,
         },
-        response_config=[
-            ('data', __vector_rows_type(query_flags)),
-            ('more', Bool),
-        ],
+        with_scores=bool(query_flags & VECTOR_FLAG_WITH_SCORES),
+        no_content=bool(query_flags & VECTOR_FLAG_NOCONTENT),
+        legacy=not query_flags,
+        has_cursor=False,
         post_process_fun=__query_result_post_process
     )
 

pygridgain/cache.py

@@ -660,7 +660,7 @@ def select_row(
                          distributed_joins, replicated_only, local, timeout)
 
     def vector(self, type_name: str, field: str, clause_vector: List[float],
-               k: int, threshold: float, page_size: int = 1, ef_search: int = 0,
+               k: int, threshold: float, page_size: int = None, ef_search: int = 0,
                with_scores: bool = False, no_content: bool = False) -> VectorCursor:
         """
         Ignite supports vector queries based on Apache Lucene engine.
@@ -672,10 +672,12 @@ def vector(self, type_name: str, field: str, clause_vector: List[float],
          per-element conversion and is by far the cheapest to send, which matters because embedding
          dimensions are typically in the hundreds or thousands. Converting such an array to a Python
          list before passing it costs more than the query itself.
-        :param k: [K]NN, how many vectors to return.
+        :param k: [K]NN, how many vectors to return. Must be positive; the server also bounds it
+         (GRIDGAIN_VECTOR_MAX_K, 10000 by default) and rejects a query above that bound.
         :param threshold: similarity threshold, non-positive values disable it.
-        :param page_size: (optional) page size. Default size is 1 (slowest
-         and safest),
+        :param page_size: (optional) page size. Defaults to k: a vector query returns at most k
+         rows, so the whole result arrives in one page. Set it lower only to cap the size of a
+         single response,
         :param ef_search: (optional) search beam width: how many candidate vectors the engine
          keeps while traversing the index graph. The engine returns the k best of those
          candidates, so k controls the result size while the beam controls the search quality:
@@ -691,6 +693,12 @@ def vector(self, type_name: str, field: str, clause_vector: List[float],
          `(key, value, score)` with `with_scores`, bare `key` with `no_content`, and
          `(key, score)` with both.
         """
+        if k < 1:
+            raise ValueError(f'k must be positive, got {k}')
+
+        if page_size is None:
+            page_size = k
+
         query_flags = ((VECTOR_FLAG_WITH_SCORES if with_scores else 0)
                        | (VECTOR_FLAG_NOCONTENT if no_content else 0))
 

pygridgain/cursors.py

@@ -25,7 +25,7 @@
     sql_cursor_get_page, sql_fields, sql_fields_cursor_get_page, sql_fields_cursor_get_page_async, sql_fields_async
 )
 from pygridgain.api.sql import (
-    VECTOR_FLAG_NOCONTENT, VECTOR_FLAG_WITH_SCORES, vector, vector_cursor_get_page, vector_async,
+    vector, vector_cursor_get_page, vector_async,
     vector_cursor_get_page_async
 )
 from pygridgain.exceptions import CacheError, SQLError
@@ -417,8 +417,8 @@ def _process_page_response(self, result):
         self.data, self.more = self._rows_iter(result.value['data']), result.value['more']
 
     def _rows_iter(self, data):
-        # Legacy responses are key-value maps; flagged responses are lists of per-row dicts.
-        return iter(data if self._query_flags else data.items())
+        # Rows arrive from VectorResponse as final Python values, already shaped by the flags.
+        return iter(data)
 
 
 class VectorCursor(AbstractVectorCursor, CursorMixin):
@@ -452,30 +452,14 @@ def __next__(self):
             raise StopIteration
 
         try:
-            row = next(self.data)
+            return next(self.data)
         except StopIteration:
             if self.more:
                 self._process_page_response(
                     vector_cursor_get_page(self.connection, self.cursor_id, self._query_flags))
-                row = next(self.data)
-            else:
-                raise StopIteration
-
-        if not self._query_flags:
-            k, v = row
-            return self.client.unwrap_binary(k), self.client.unwrap_binary(v)
+                return next(self.data)
 
-        key = self.client.unwrap_binary(row['key'])
-
-        if self._query_flags & VECTOR_FLAG_NOCONTENT:
-            if self._query_flags & VECTOR_FLAG_WITH_SCORES:
-                return key, row['score']
-
-            return key
-
-        value = self.client.unwrap_binary(row['value'])
-
-        return key, value, row['score']
+            raise StopIteration
 
 
 class AioVectorCursor(AbstractVectorCursor, AioCursorMixin):
@@ -515,32 +499,14 @@ async def __anext__(self):
             raise StopAsyncIteration
 
         try:
-            row = next(self.data)
+            return next(self.data)
         except StopIteration:
             if self.more:
                 self._process_page_response(
                     await vector_cursor_get_page_async(self.connection, self.cursor_id, self._query_flags))
                 try:
-                    row = next(self.data)
+                    return next(self.data)
                 except StopIteration:
                     raise StopAsyncIteration
-            else:
-                raise StopAsyncIteration
-
-        if not self._query_flags:
-            k, v = row
-            return await asyncio.gather(
-                *[self.client.unwrap_binary(k), self.client.unwrap_binary(v)]
-            )
 
-        key = await self.client.unwrap_binary(row['key'])
-
-        if self._query_flags & VECTOR_FLAG_NOCONTENT:
-            if self._query_flags & VECTOR_FLAG_WITH_SCORES:
-                return key, row['score']
-
-            return key
-
-        value = await self.client.unwrap_binary(row['value'])
-
-        return key, value, row['score']
+            raise StopAsyncIteration

pygridgain/datatypes/complex.py

@@ -22,7 +22,7 @@
 from pygridgain.constants import *
 from pygridgain.exceptions import ParseError
 from .base import GridGainDataType
-from .internal import AnyDataObject, Struct, infer_from_python, infer_from_python_async
+from .internal import AnyDataObject, Struct, cached_c_type, infer_from_python, infer_from_python_async
 from .type_codes import *
 from .type_ids import *
 from .type_names import *
@@ -157,19 +157,14 @@ def parse_not_null(cls, stream):
             byteorder=PROTOCOL_BYTE_ORDER
         )
 
-        final_class = type(
-            cls.__name__,
-            (ctypes.LittleEndianStructure,),
-            {
-                '_pack_': 1,
-                '_fields_': [
-                    ('type_code', ctypes.c_byte),
-                    ('length', ctypes.c_int),
-                    ('payload', ctypes.c_byte * length),
-                    ('offset', ctypes.c_int),
-                ],
-            }
-        )
+        # One shared class per payload length: the rows of one result set carry same-shaped
+        # objects, so the length recurs and a fresh class per row is pure overhead.
+        final_class = cached_c_type(cls.__name__, (ctypes.LittleEndianStructure,), (
+            ('type_code', ctypes.c_byte),
+            ('length', ctypes.c_int),
+            ('payload', ctypes.c_byte * length),
+            ('offset', ctypes.c_int),
+        ))
 
         stream.seek(ctypes.sizeof(final_class), SEEK_CUR)
         return final_class
@@ -257,14 +252,7 @@ def __parse_header(cls, stream):
 
     @classmethod
     def __build_final_class(cls, fields):
-        return type(
-            cls.__name__,
-            (ctypes.LittleEndianStructure,),
-            {
-                '_pack_': 1,
-                '_fields_': fields,
-            }
-        )
+        return cached_c_type(cls.__name__, (ctypes.LittleEndianStructure,), fields)
 
     @classmethod
     def to_python_not_null(cls, ctypes_object, *args, **kwargs):
@@ -349,14 +337,7 @@ async def _parse_async(cls, stream):
 
     @classmethod
     def __build_final_class(cls, fields):
-        return type(
-            cls.__name__,
-            (ctypes.LittleEndianStructure,),
-            {
-                '_pack_': 1,
-                '_fields_': fields,
-            }
-        )
+        return cached_c_type(cls.__name__, (ctypes.LittleEndianStructure,), fields)
 
     @classmethod
     def _to_python(cls, ctypes_object, **kwargs):
@@ -597,17 +578,10 @@ def offset_c_type(cls, flags: int):
     def schema_type(cls, flags: int):
         if flags & cls.COMPACT_FOOTER:
             return cls.offset_c_type(flags)
-        return type(
-            'SchemaElement',
-            (ctypes.LittleEndianStructure,),
-            {
-                '_pack_': 1,
-                '_fields_': [
-                    ('field_id', ctypes.c_int),
-                    ('offset', cls.offset_c_type(flags)),
-                ],
-            },
-        )
+        return cached_c_type('SchemaElement', (ctypes.LittleEndianStructure,), (
+            ('field_id', ctypes.c_int),
+            ('offset', cls.offset_c_type(flags)),
+        ))
 
     @classmethod
     def parse_not_null(cls, stream):
@@ -655,14 +629,9 @@ def __build_final_class(cls, stream, header, header_class, object_fields, fields
             stream.seek(ctypes.sizeof(schema), SEEK_CUR)
             final_class_fields.append(('schema', schema))
 
-        final_class = type(
-            cls.__name__,
-            (header_class,),
-            {
-                '_pack_': 1,
-                '_fields_': final_class_fields,
-            }
-        )
+        # Rows of one result set share the object shape, so this class recurs: object_fields comes
+        # from a shared-per-shape cache and ctypes caches array types, which makes the key stable.
+        final_class = cached_c_type(cls.__name__, (header_class,), final_class_fields)
         # register schema encoding approach
         stream.compact_footer = bool(header.flags & cls.COMPACT_FOOTER)
         return final_class