From 395f2f9af9d15eb052fdf1d95d9e361b4afd5af2 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Wed, 4 Nov 2026 05:47:46 +0900 Subject: [PATCH 1/3] Native insert_many: ragged int and float lists go through the columnar entry (kinds I Q E D) A chunk with a list-of-ints column (the sparse-vector tokens) or a ragged list-of-floats column was refused by columnar.pack and sent as one JSON text. The columnar entry now carries such columns as int32 element offsets plus one buffer of elements, like a string column: I (int32, stored int[]), E (float32, float[]), Q (int64, list of Long), D (float64, list of Double), chosen from the declared property type so the engine stores what the JSON text stores. A library without the new probe entry arcadedb_n_colp_lists keeps lists on the JSON path (lib.COLP_LISTS, ARCADEDB_NATIVE_COLP_LISTS=0 turns it off). Closes #231. Co-Authored-By: Claude Opus 5.5 --- .../org/humemai/nativelib/ArcadeNative.java | 12 +- .../java/org/humemai/nativelib/BulkOps.java | 68 +++++- src/arcadedb/_native/bulk.py | 11 +- src/arcadedb/_native/columnar.py | 108 ++++++++- src/arcadedb/_native/database.py | 2 +- src/arcadedb/_native/lib.py | 9 +- tests/test_native_columnar_insert.py | 222 ++++++++++++++++++ tests/test_native_columnar_pack.py | 133 +++++++++-- 8 files changed, 530 insertions(+), 35 deletions(-) diff --git a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java index 9b5acb9..e7f33d1 100644 --- a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java +++ b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java @@ -560,7 +560,8 @@ public static CCharPointer insertCols(final IsolateThread thread, final long h, /** * Columnar insert by pointer: no copy of the columns into one buffer. spec points at 6 longs per column (data address and bytes, aux * address and bytes, null-bitmap address and bytes, see BulkOps.insertColumns) and nspec is their count. Kinds q d b f as - * arcadedb_n_insert_cols, plus i (int32), e (float32) and s (Arrow UTF-8 string: int32 offsets and the bytes they index), and null bitmaps. + * arcadedb_n_insert_cols, plus i (int32), e (float32) and s (Arrow UTF-8 string: int32 offsets and the bytes they index), the ragged + * list kinds I Q E D (int32 element offsets and the elements they index, see BulkOps.insertColumns), and null bitmaps. * The caller keeps every buffer alive and unchanged until the call returns. Returns {"ok":true,"n":rows}. */ @CEntryPoint(name = "arcadedb_n_insert_colp") @@ -578,6 +579,15 @@ public static CCharPointer insertColp(final IsolateThread thread, final long h, } } + /** + * Probe: a library that has this entry decodes the ragged list kinds I Q E D of arcadedb_n_insert_colp (the Python side asks for + * the symbol, it does not call it). An older library lacks the symbol and the Python side keeps such columns on the JSON path. + */ + @CEntryPoint(name = "arcadedb_n_colp_lists") + public static CCharPointer colpLists(final IsolateThread thread) { + return out("{\"ok\":true,\"kinds\":\"IQED\"}"); + } + // ---- phase 8: fewer crossings and no reply text for a command whose rows are read late or never; batches of statements and rows per crossing private static final ThreadLocal LAST_ERROR = new ThreadLocal<>(); diff --git a/native/src/main/java/org/humemai/nativelib/BulkOps.java b/native/src/main/java/org/humemai/nativelib/BulkOps.java index 37cedf9..5e4a56b 100644 --- a/native/src/main/java/org/humemai/nativelib/BulkOps.java +++ b/native/src/main/java/org/humemai/nativelib/BulkOps.java @@ -4,7 +4,9 @@ * Phase 7: columnar bulk insert through one crossing. The documents are built here from whole columns that Python lays out * in native-endian buffers (int64, float64, bool bytes, int32, float32, float32 vector matrix, UTF-8 strings as Arrow offsets plus * bytes, optional Arrow null bitmaps), as DocumentBatcher.insertColumns does on the JPype path from long[], double[], boolean[] and - * Object[] columns. Values are read straight from the caller's buffers; no JSON text, no parser, no intermediate arrays. + * Object[] columns. Values are read straight from the caller's buffers; no JSON text, no parser, no intermediate arrays. Ragged lists of + * numbers (the sparse-vector token and weight columns) cross as int32 element offsets plus one buffer of their elements, see + * {@link #insertColumns(Database, String, Map, long[])}. */ package org.humemai.nativelib; @@ -18,6 +20,7 @@ import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -67,7 +70,12 @@ static long insertColumns(final Database db, final String type, final String hea /** * header: {"n":rows,"commit":every,"cols":[[name,kind,dim],..]}. Kinds: q (int64), d (float64), b (bool byte), i (int32), e (float32), * f (float32 matrix, dim floats per row), s (UTF-8 string: int32 offsets[n+1] in the data slot, the bytes they index in the aux slot, as - * an Arrow string array; offsets[0] need not be 0). spec: per column six longs, [data address, data bytes, aux address, aux bytes, null + * an Arrow string array; offsets[0] need not be 0), and the ragged list kinds I Q E D: the data slot holds int32 offsets[n+1] counted in + * ELEMENTS (offsets[0] need not be 0, a null or empty row has offsets[i] == offsets[i+1]) and the aux slot the elements they index, + * native-endian: I = int32 elements, stored as int[] (a property declared ARRAY_OF_INTEGERS); Q = int64 elements, stored as a + * List of Long (what the JSON text of a list of integers parses to); E = float32 elements, stored as float[] (ARRAY_OF_FLOATS); D = + * float64 elements, stored as a List of Double. The Python side chooses the kind from the declared property type so that the engine + * stores what the JSON path stores. spec: per column six longs, [data address, data bytes, aux address, aux bytes, null * bitmap address, null bitmap bytes]; the bitmap is Arrow's (bit i of byte i/8, least significant first; 1 = present), an address of 0 * means no nulls. The memory belongs to the caller for the duration of the call; nothing is kept. Every length is checked here before * the first document is built, so a wrong table is an exception and nothing is written. Same transaction contract as @@ -90,6 +98,7 @@ private static long insertColumns(final Database db, final String type, final Ma final long[] data = new long[k]; // addresses: a Word value cannot live in an array or be compared with null final long[] nulls = new long[k]; // 0 = no nulls final ByteBuffer[] blobs = new ByteBuffer[k]; + final long[] elems = new long[k]; // aux address of a ragged list column int longest = 0; for (int c = 0; c < k; c++) { final List col = (List) cols.get(c); @@ -108,7 +117,7 @@ private static long insertColumns(final Database db, final String type, final Ma throw new IllegalArgumentException("column '" + names[c] + "': a vector column needs a dimension above 0"); yield 4L * n * dims[c]; } - case 's' -> 4L * (n + 1L); + case 's', 'I', 'Q', 'E', 'D' -> 4L * (n + 1L); default -> throw new IllegalArgumentException("unknown column kind " + kinds[c]); }; if (dataBytes < need || (need > 0 && dataAddr == 0)) @@ -119,6 +128,22 @@ private static long insertColumns(final Database db, final String type, final Ma throw new IllegalArgumentException("column '" + names[c] + "' has a null bitmap of " + nullBytes + " bytes, " + (n + 7L) / 8 + " needed"); nulls[c] = nullAddr; } + if (isList(kinds[c])) { + final long width = kinds[c] == 'Q' || kinds[c] == 'D' ? 8L : 4L; + final Pointer offsets = WordFactory.pointer(data[c]); + int prev = offsets.readInt(WordFactory.signed(0)); + if (prev < 0) + throw new IllegalArgumentException("column '" + names[c] + "': negative list offset"); + for (int i = 1; i <= n; i++) { + final int off = offsets.readInt(WordFactory.signed(4L * i)); + if (off < prev || width * off > auxBytes) + throw new IllegalArgumentException("column '" + names[c] + "': list offsets are not increasing within the " + auxBytes + " bytes of elements (row " + (i - 1) + ")"); + prev = off; + } + if (auxBytes > 0 && auxAddr == 0) + throw new IllegalArgumentException("column '" + names[c] + "': the list elements are missing"); + elems[c] = auxAddr; + } if (kinds[c] == 's') { if (auxBytes < 0 || auxBytes > Integer.MAX_VALUE || (auxBytes > 0 && auxAddr == 0)) throw new IllegalArgumentException("column '" + names[c] + "': the string bytes are missing or beyond 2 GiB"); @@ -166,6 +191,7 @@ private static long insertColumns(final Database db, final String type, final Ma doc.set(names[c], new String(scratch, 0, len, StandardCharsets.UTF_8)); } } + case 'I', 'Q', 'E', 'D' -> doc.set(names[c], listValue(kinds[c], base, WordFactory.pointer(elems[c]), i)); default -> { final int dim = dims[c]; final float[] v = new float[dim]; @@ -197,6 +223,42 @@ private static long insertColumns(final Database db, final String type, final Ma return n; } + private static boolean isList(final char kind) { + return kind == 'I' || kind == 'Q' || kind == 'E' || kind == 'D'; + } + + /** Row i of a ragged list column: offsets (int32, in elements) at {@code offsets}, the elements at {@code elems}. */ + private static Object listValue(final char kind, final Pointer offsets, final Pointer elems, final int i) { + final int from = offsets.readInt(WordFactory.signed(4L * i)); + final int len = offsets.readInt(WordFactory.signed(4L * (i + 1L))) - from; + switch (kind) { + case 'I': { + final int[] v = new int[len]; + for (int j = 0; j < len; j++) + v[j] = elems.readInt(WordFactory.signed(4L * (from + (long) j))); + return v; + } + case 'E': { + final float[] v = new float[len]; + for (int j = 0; j < len; j++) + v[j] = elems.readFloat(WordFactory.signed(4L * (from + (long) j))); + return v; + } + case 'Q': { + final List v = new ArrayList<>(len); + for (int j = 0; j < len; j++) + v.add(elems.readLong(WordFactory.signed(8L * (from + (long) j)))); + return v; + } + default: { + final List v = new ArrayList<>(len); + for (int j = 0; j < len; j++) + v.add(elems.readDouble(WordFactory.signed(8L * (from + (long) j)))); + return v; + } + } + } + /** * Documents (or vertices of a vertex type) from rows [{property: value},..] in the UTF-8 JSON text b[0, n): one record per row through * newDocument/newVertex, set per property and save, as DocumentBatcher.insertManyJson does on the JPype path; no transaction is opened diff --git a/src/arcadedb/_native/bulk.py b/src/arcadedb/_native/bulk.py index d244c48..90ea177 100644 --- a/src/arcadedb/_native/bulk.py +++ b/src/arcadedb/_native/bulk.py @@ -2,7 +2,8 @@ ints, floats and bools, lists of one value type (None is a null), strings (UTF-8 bytes plus offsets, as Arrow lays them out), float32 vector columns and Arrow arrays are laid out by :mod:`.columnar`; anything else (nested values, mixed-type columns, date-times, decimals, integers past 64 bits) returns None and the caller uses the per-row JSON path. The documents are built image-side as -``DocumentBatcher.insertColumns`` builds them on JPype. ``ARCADEDB_NATIVE_BULK=0`` turns the columnar path off. +``DocumentBatcher.insertColumns`` builds them on JPype. Lists of ints and floats (ragged or not, None allowed) cross as int32 offsets plus a buffer of their elements when the library has +the decoder (``lib.COLP_LISTS``). ``ARCADEDB_NATIVE_BULK=0`` turns the columnar path off. """ import os @@ -21,6 +22,10 @@ def _usable(): ) +def _declared(db, type_name, known=None): + return columnar.Declared(db, type_name, known, lists=lib.COLP_LISTS) + + def _send(db, type_name, packed, rows, commit_every, prefix): from .database import _call @@ -47,7 +52,7 @@ def insert_columns_fast( """ if not _usable(): return None - packed = columnar.pack(columns, columnar.Declared(db, type_name, declared)) + packed = columnar.pack(columns, _declared(db, type_name, declared)) if packed is None: return None rows, cols = packed @@ -63,7 +68,7 @@ def insert_rows_columnar(db, type_name, chunk): columns = columnar.rows_to_columns(chunk) if columns is None: return False - packed = columnar.pack(columns, columnar.Declared(db, type_name)) + packed = columnar.pack(columns, _declared(db, type_name)) if packed is None: return False rows, cols = packed diff --git a/src/arcadedb/_native/columnar.py b/src/arcadedb/_native/columnar.py index f288043..07ef33e 100644 --- a/src/arcadedb/_native/columnar.py +++ b/src/arcadedb/_native/columnar.py @@ -9,12 +9,20 @@ matrix (``dim`` per row), ``s`` UTF-8 string (``int32 offsets[n+1]`` and the bytes they index: Arrow's string layout). A column may carry a null bitmap (Arrow's: bit ``i`` of byte ``i // 8``, least significant first, 1 = present). +Ragged lists of numbers (``I`` int32, ``Q`` int64, ``E`` float32, ``D`` float64 elements) lay out like a string column: ``int32 +offsets[n+1]`` counted in elements as the data buffer and the elements they index as the aux buffer; a null or empty row has equal +neighbouring offsets. The image stores ``I`` as ``int[]`` (a property declared ``ARRAY_OF_INTEGERS``), ``E`` as ``float[]`` (declared +``ARRAY_OF_FLOATS``), ``Q`` and ``D`` as a list of Long or Double (undeclared or declared ``LIST``): what the JSON path stores for the same +rows. A library without the decoder (``Declared(lists=False)``) gets none of them. + ``i`` and ``e`` and the float matrix from a list of lists are chosen only for a property declared ``INTEGER``, ``FLOAT`` and ``ARRAY_OF_FLOATS``: the engine would convert to those types anyway, so the stored value is the one the JSON path stores. For an undeclared property an ``int32`` column crosses as ``q`` (a Long) and a ``float32`` column as ``d`` (a Double), as the JSON path stores them. """ +import itertools + from .schema_objects import ( # the schema-entry reader shared with NativeType (#195) _name_type, ) @@ -26,7 +34,14 @@ _INT64_MAX = 2**63 - 1 _INT32_MAX = 2**31 - 1 +_INT32_MIN = -(2**31) _NoneType = type(None) +_OFFSET_KINDS = frozenset( + "sIQED" +) # the kinds whose data buffer is int32 offsets[n + 1] +_EMPTY = ( + [] +) # stands in for a null row of a list column (the null bitmap says it is absent) class Packed: @@ -39,8 +54,8 @@ def __init__(self, name, kind, dim, array, aux=None, nulls=None, keep=()): self.kind = kind self.dim = dim self.n = int(array.shape[0]) - ( - 1 if kind == "s" else 0 - ) # a string column's array is its n + 1 offsets + 1 if kind in _OFFSET_KINDS else 0 + ) # a string or list column's array is its n + 1 offsets self.data = (array.ctypes.data, array.nbytes) self.aux = aux or (0, 0) self.nulls = (nulls.ctypes.data, nulls.nbytes) if nulls is not None else (0, 0) @@ -50,11 +65,13 @@ def __init__(self, name, kind, dim, array, aux=None, nulls=None, keep=()): class Declared: """The declared property types of a type, read once and only when a column needs them (``INTEGER``, ``FLOAT``, ``ARRAY_OF_FLOATS``). - ``known`` (a dict) is used as it is; otherwise ``db`` is asked for ``schema:types`` the first time :meth:`get` is called. + ``known`` (a dict) is used as it is; otherwise ``db`` is asked for ``schema:types`` the first time :meth:`get` is called. ``lists``: + whether the library decodes the ragged list kinds (``lib.COLP_LISTS``); False keeps list columns on the JSON path. """ - def __init__(self, db=None, type_name=None, known=None): + def __init__(self, db=None, type_name=None, known=None, lists=True): self._db, self._type, self._known = db, type_name, known + self.lists = lists def get(self, name): if self._known is None: @@ -207,10 +224,11 @@ def _list_column(name, values, declared): try: arr = _np.array(values, dtype=_np.float32) except (ValueError, TypeError): - return None # ragged, or not numbers - if arr.ndim != 2 or arr.shape[1] == 0: - return None - return Packed(name, "f", int(arr.shape[1]), arr) + arr = None # ragged, or not numbers + if arr is not None and arr.ndim == 2 and arr.shape[1] > 0: + return Packed(name, "f", int(arr.shape[1]), arr) + if only is list: + return _ragged_column(name, values, declared, present, nulls) if ( only is _np.ndarray and present is None @@ -229,6 +247,80 @@ def _list_column(name, values, declared): return None +def _ragged_column(name, values, declared, present, nulls): + """A list of lists of ints or of floats (ragged, empty rows and None rows allowed) as an offsets-plus-elements column, or None. + + The kind follows the declared property type so that the engine stores what the JSON text stores: ``ARRAY_OF_INTEGERS`` takes ints + in int32 range as ``I``, ``ARRAY_OF_FLOATS`` takes Python floats as ``E`` (float32, as the engine converts a list of doubles), an + undeclared or ``LIST`` property takes ``Q`` (ints) or ``D`` (floats), which the image stores as a list of Long or Double. Anything + else (another declared type, bools, strings, nested lists, mixed ints and floats, an int past 64 bits, NaN or an infinity, more than + 2**31 - 1 elements) returns None and the chunk goes the JSON way. + """ + if not declared.lists: + return None + rows = ( + values if present is None else [v if v is not None else _EMPTY for v in values] + ) + n = len(rows) + decl = declared.get(name) + if decl not in (None, "LIST", "ARRAY_OF_INTEGERS", "ARRAY_OF_FLOATS"): + return None + counts = _np.fromiter(map(len, rows), dtype=_np.int64, count=n) + total = int(counts.sum()) + if total > _INT32_MAX: + return None + flat = list(itertools.chain.from_iterable(rows)) + if ( + total == 0 + ): # only empty rows: the engine stores an empty list or array whatever the element type + kind, elements = {"ARRAY_OF_INTEGERS": "I", "ARRAY_OF_FLOATS": "E"}.get( + decl, "Q" + ), None + else: + element_types = set(map(type, flat)) + if len(element_types) != 1: + return None + (element,) = element_types + if element is int and decl in (None, "LIST", "ARRAY_OF_INTEGERS"): + try: + elements = _np.array(flat, dtype=_np.int64) + except OverflowError: + return None # an integer beyond 64 bits + if decl == "ARRAY_OF_INTEGERS": + if int(elements.min()) < _INT32_MIN or int(elements.max()) > _INT32_MAX: + return None # the engine refuses such a value for the property: leave the error to the JSON path + kind, elements = "I", elements.astype(_np.int32) + else: + kind = "Q" + elif element is float and decl in (None, "LIST", "ARRAY_OF_FLOATS"): + elements = _np.array(flat, dtype=_np.float64) + if not _np.isfinite(elements).all(): + return None + if decl == "ARRAY_OF_FLOATS": + with _np.errstate( + all="ignore" + ): # a double beyond float32 becomes Infinity, as in the engine + kind, elements = "E", elements.astype(_np.float32) + else: + kind = "D" + else: + return None + offsets = _np.zeros(n + 1, dtype=_np.int32) + _np.cumsum(counts, out=offsets[1:], dtype=_np.int32) + if elements is None or not elements.size: + return Packed(name, kind, 0, offsets, nulls=nulls) + elements = _np.ascontiguousarray(elements) + return Packed( + name, + kind, + 0, + offsets, + aux=(elements.ctypes.data, elements.nbytes), + nulls=nulls, + keep=(elements,), + ) + + def _arrow_column(name, arr, declared): """A ``pyarrow`` array as a column (strings in place), or None.""" import pyarrow as pa diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 8fb321f..7c31ce9 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -752,7 +752,7 @@ def insert_many( parallel: bool = False, ) -> int: """Bulk insert. Rows go in chunks of at most 10,000 (a chunk never crosses a commit), one crossing per chunk. A chunk of rows with the same keys whose columns each hold one - value type (ints, floats, bools, strings, equal-length float lists for an ``ARRAY_OF_FLOATS`` property; None is a null) goes as + value type (ints, floats, bools, strings, lists of ints or of floats, ragged or not; None is a null) goes as columns by address (``arcadedb_n_insert_colp``, no JSON text); any other chunk goes as one JSON text (``arcadedb_n_insert_rows``: a newDocument/set/save loop inside the image, as ``DocumentBatcher.insertManyJson`` on the JPype path); a chunk with a value the JSON text cannot carry unchanged goes through ``INSERT INTO t CONTENT :row`` for each row, in one crossing. Inside the caller's diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index 00d20b7..27660ec 100644 --- a/src/arcadedb/_native/lib.py +++ b/src/arcadedb/_native/lib.py @@ -31,6 +31,8 @@ EXEC_MANY = os.environ.get("ARCADEDB_NATIVE_EXEC_MANY", "1") != "0" # ARCADEDB_NATIVE_NEXT_BIN: the batches after the first read through arcadedb_n_next_bin, float arrays of 8 or more elements in a float32 tail (as flag 4 of arcadedb_n_exec does for the first batch), not as number text NEXT_BIN = os.environ.get("ARCADEDB_NATIVE_NEXT_BIN", "1") != "0" +# ARCADEDB_NATIVE_COLP_LISTS: ragged lists of ints and floats (the sparse-vector token and weight columns) cross in arcadedb_n_insert_colp as int32 offsets plus one buffer of elements (kinds I Q E D), not inside the JSON text of the chunk; 0 keeps them on the JSON path, as does a library without arcadedb_n_colp_lists +COLP_LISTS = os.environ.get("ARCADEDB_NATIVE_COLP_LISTS", "1") != "0" # the directory the engine log, logging.properties and the crash files go to (None until the library is loaded) LOG_DIR_USED = None @@ -225,6 +227,7 @@ class _Params(ctypes.Structure): "arcadedb_n_exec_bin": [P, L, I, S, S, S, I, S, L], "arcadedb_n_insert_cols": [P, L, S, S, S, L], "arcadedb_n_insert_colp": [P, L, S, S, P, I], + "arcadedb_n_colp_lists": [P], "arcadedb_n_next": [P, L, I], "arcadedb_n_next_bin": [P, L, I], "arcadedb_n_rs_close": [P, L], @@ -266,7 +269,7 @@ class _Params(ctypes.Structure): ) else ctypes.c_longlong if name == "arcadedb_n_cmd" else ctypes.c_void_p ) - global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY, NEXT_BIN + global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY, NEXT_BIN, COLP_LISTS if not hasattr( lib, "arcadedb_n_exec_bin" ): # a library built before the binary path @@ -283,6 +286,10 @@ class _Params(ctypes.Structure): lib, "arcadedb_n_next_bin" ): # a library built before the binary tail of later batches NEXT_BIN = False + if not hasattr( + lib, "arcadedb_n_colp_lists" + ): # a library built before the ragged list kinds of arcadedb_n_insert_colp + COLP_LISTS = False _isolate = isolate _creator = threading.get_ident() _local.thread = _Thread(thread, owned=False) diff --git a/tests/test_native_columnar_insert.py b/tests/test_native_columnar_insert.py index a3591d5..463618a 100644 --- a/tests/test_native_columnar_insert.py +++ b/tests/test_native_columnar_insert.py @@ -502,3 +502,225 @@ def test_an_unknown_kind_and_a_wrong_table_size_are_errors(temp_db): ) assert not reply["ok"] # 5 entries are not a multiple of 6 assert _count(temp_db) == 0 + + +# ---- ragged lists of ints and floats (the sparse-vector token and weight columns): offsets plus one buffer of elements + +SPARSE_DIMS = 400 + + +def _sparse_types( + db, *names, tokens="ARRAY_OF_INTEGERS", weights="ARRAY_OF_FLOATS", index=False +): + for name in names: + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + if tokens: + db.command("sql", f"CREATE PROPERTY {name}.tokens {tokens}") + if weights: + db.command("sql", f"CREATE PROPERTY {name}.weights {weights}") + if index: + db.command( + "sql", + f"CREATE INDEX ON {name} (tokens, weights) LSM_SPARSE_VECTOR " + f'METADATA {{"dimensions": {SPARSE_DIMS}}}', + ) + + +def _sparse_rows(n, ragged, seed=11): + """Rows of the lanes' shape: an id, the token ids of a sparse vector and their weights (30 nonzeros, or 1 to 40 when ragged).""" + import random + + rng = random.Random(seed) + rows = [] + for i in range(n): + nnz = rng.randint(1, 40) if ragged else 30 + tokens = rng.sample(range(SPARSE_DIMS), nnz) + rows.append( + {"id": i, "tokens": tokens, "weights": [0.1 + rng.random() for _ in tokens]} + ) + return rows + + +def _tagged(value): + if isinstance(value, list): + return ("list", [(type(e).__name__, e) for e in value]) + return (type(value).__name__, value) + + +def _stored_deep(db, name): + """Every record of a type, a list property as its elements with their Python types.""" + sql = f"SELECT FROM {name} ORDER BY id" # nosec B608 - test-owned type name, no user input + return [ + {k: _tagged(v) for k, v in row.items() if not k.startswith("@")} + for row in db.query("sql", sql).to_list() + ] + + +def _stored_kinds(db, name, props): + """What the engine says each stored property value is (its type() per record), to compare between the two paths.""" + select = ", ".join(f"{p}.type() AS {p}" for p in props) + sql = f"SELECT id, {select} FROM {name} ORDER BY id" # nosec B608 - test-owned names, no user input + return db.query("sql", sql).to_list() + + +def _neighbours(db, name, tokens, weights, k=10): + sql = ( + "SELECT id, score FROM " # nosec B608 - test-owned type name + f"(SELECT expand(`vector.sparseNeighbors`('{name}[tokens,weights]', ?, ?, ?))) ORDER BY score DESC, id" + ) + rows = db.query( + "sql", + sql, + np.asarray(tokens, dtype=np.int32), + np.asarray(weights, dtype=np.float32), + k, + ).to_list() + return [(int(r["id"]), r["score"]) for r in rows] + + +@pytest.mark.parametrize("ragged", [False, True]) +def test_sparse_rows_cross_as_columns_and_the_sparse_index_accepts_them( + temp_db, spy, monkeypatch, ragged +): + _sparse_types(temp_db, "Ref", "Col", index=True) + rows = _sparse_rows(300, ragged) + _json_reference(temp_db, "Ref", rows, monkeypatch) + assert spy.count(COLP) == 0 and spy.count(ROWS) == 1 + + spy.calls.clear() + assert temp_db.insert_many("Col", rows) == 300 + assert spy.count(COLP) == 1, spy.calls # the columnar entry ... + assert ( + spy.count(ROWS) == 0 and spy.count(MANY) == 0 + ), spy.calls # ... and no JSON text + + expected = _stored_deep(temp_db, "Ref") + assert len(expected) == 300 + assert ( + expected[3]["tokens"][0] == "list" and expected[3]["tokens"][1][0][0] == "int" + ) + assert expected[3]["weights"][1][0][0] == "float" + assert _stored_deep(temp_db, "Col") == expected + assert _stored_kinds(temp_db, "Col", ("tokens", "weights")) == _stored_kinds( + temp_db, "Ref", ("tokens", "weights") + ) + for row in rows[ + :5 + ]: # the LSM_SPARSE_VECTOR index of each type finds the rows it was loaded with + found = _neighbours(temp_db, "Col", row["tokens"], row["weights"]) + assert found == _neighbours(temp_db, "Ref", row["tokens"], row["weights"]) + assert found + + +def test_every_list_shape_stores_what_the_json_path_stores(temp_db, spy, monkeypatch): + for name in ("Ref", "Col"): + temp_db.command("sql", f"CREATE DOCUMENT TYPE {name}") + temp_db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + temp_db.command("sql", f"CREATE PROPERTY {name}.ai ARRAY_OF_INTEGERS") # int[] + temp_db.command("sql", f"CREATE PROPERTY {name}.af ARRAY_OF_FLOATS") # float[] + temp_db.command("sql", f"CREATE PROPERTY {name}.li LIST") # a list of Long + temp_db.command("sql", f"CREATE PROPERTY {name}.lf LIST") # a list of Double + n = 60 + rows = [] + for i in range(n): + rows.append( + { + "id": i, + "ai": None if i % 7 == 0 else [j * 1000 - 5 for j in range(i % 5)], + "af": None if i % 6 == 0 else [j / 8.0 + 0.1 for j in range(i % 4)], + "li": None if i % 5 == 0 else [j - 2**40 for j in range(i % 6)], + "lf": None if i % 4 == 0 else [j * 0.1 for j in range(i % 3)], + "ui": [-(2**63), 2**63 - 1, 0] if i % 2 else [], # undeclared + "uf": [1e300, 1e-300, -0.0] if i % 3 else [], # undeclared + } + ) + rows[1]["ai"] = [-(2**31), 2**31 - 1] + rows[2]["af"] = [3.4e38, 1e-45] # float32 edges + props = ("ai", "af", "li", "lf", "ui", "uf") + _json_reference(temp_db, "Ref", rows, monkeypatch) + spy.calls.clear() + temp_db.insert_many("Col", rows) + assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 and spy.count(MANY) == 0 + expected = _stored_deep(temp_db, "Ref") + assert expected[3]["ai"] == ("list", [("int", -5), ("int", 995), ("int", 1995)]) + assert expected[1]["ai"][1] == [("int", -(2**31)), ("int", 2**31 - 1)] + assert expected[0]["ai"] == ("NoneType", None) + assert expected[4]["li"] == ("list", [("int", j - 2**40) for j in range(4)]) + assert _stored_deep(temp_db, "Col") == expected + assert _stored_kinds(temp_db, "Col", props) == _stored_kinds(temp_db, "Ref", props) + + empty = [{"id": 1000 + k, "ai": [], "af": []} for k in range(2)] # only empty rows + _json_reference(temp_db, "Ref", empty, monkeypatch) + temp_db.insert_many("Col", empty) + assert spy.count(COLP) == 2 and spy.count(ROWS) == 0 + assert _stored_deep(temp_db, "Col") == _stored_deep(temp_db, "Ref") + assert _stored_deep(temp_db, "Col")[-1]["ai"] == ("list", []) + + +def test_lists_in_a_library_without_the_decoder_go_as_json(temp_db, spy, monkeypatch): + _sparse_types(temp_db, "Ref", "Col") + rows = _sparse_rows(80, ragged=True) + _json_reference(temp_db, "Ref", rows, monkeypatch) + spy.calls.clear() + monkeypatch.setattr(lib, "COLP_LISTS", False) + temp_db.insert_many("Col", rows) + assert spy.count(COLP) == 0 and spy.count(ROWS) == 1, spy.calls + assert _stored_deep(temp_db, "Col") == _stored_deep(temp_db, "Ref") + + +def test_lists_json_cannot_carry_unchanged_still_go_the_json_way( + temp_db, spy, monkeypatch +): + _sparse_types(temp_db, "Ref", "Col", tokens=None, weights=None) + rows = [{"id": i, "w": [1.0, math.inf] if i % 2 else [2.5]} for i in range(20)] + _json_reference(temp_db, "Ref", rows, monkeypatch) + spy.calls.clear() + temp_db.insert_many("Col", rows) + assert spy.count(COLP) == 0 and spy.count(MANY) == 1, spy.calls + assert _stored_deep(temp_db, "Col") == _stored_deep(temp_db, "Ref") + + +def test_more_ragged_rows_than_one_chunk(temp_db, spy, monkeypatch): + _sparse_types(temp_db, "Ref", "Col") + rows = [ + {"id": i, "tokens": list(range(i % 9)), "weights": [0.5] * (i % 9)} + for i in range(12_000) + ] + _json_reference(temp_db, "Ref", rows, monkeypatch) + spy.calls.clear() + temp_db.insert_many("Col", rows) + assert spy.count(COLP) == 2 and spy.count(ROWS) == 0, spy.calls # 10,000 + 2,000 + assert _stored_deep(temp_db, "Col") == _stored_deep(temp_db, "Ref") + + +def test_a_list_table_that_does_not_fit_its_buffers_is_an_error_and_writes_nothing( + temp_db, +): + _types(temp_db, "Col") + for elements, kind in ( + ([[1, 2], [3], [], [4, 5, 6], [7]], "Q"), + ([[0.5], [1.5, 2.5], [], [3.5], [4.5]], "D"), + ): + columns = {"id": np.arange(5), "t": elements} + + def decreasing(spec, packed): + offsets = (ctypes.c_int32 * 6).from_address(spec[6]) + offsets[2], offsets[3] = 9, 4 # not increasing + + reply, packed = _send(temp_db, columns, decreasing) + assert packed[1].kind == kind + assert not reply["ok"] and "offsets" in reply["error"], reply + + reply, _ = _send( + temp_db, columns, lambda spec, p: spec.__setitem__(9, 3) + ) # three bytes of elements for seven or more + assert not reply["ok"] and "offsets" in reply["error"], reply + + def negative(spec, packed): + (ctypes.c_int32 * 1).from_address(spec[6])[0] = -1 + + reply, _ = _send(temp_db, columns, negative) + assert not reply["ok"] and "offset" in reply["error"], reply + assert _count(temp_db) == 0 + assert temp_db.insert_columns("Col", columns) == 5 and _count(temp_db) == 5 diff --git a/tests/test_native_columnar_pack.py b/tests/test_native_columnar_pack.py index cda0a1f..e3f4843 100644 --- a/tests/test_native_columnar_pack.py +++ b/tests/test_native_columnar_pack.py @@ -105,26 +105,124 @@ def test_float64_arrays_are_not_vector_columns(): assert columnar.pack({"v": [row for row in m]}, NO_TYPES) is None -def test_a_list_of_float_lists_is_a_vector_only_for_an_array_of_floats_property(): +def test_a_list_of_equal_float_lists_is_a_vector_matrix_for_an_array_of_floats_property(): lists = [[0.5, 1.5], [2.5, 3.5]] - assert ( - columnar.pack({"v": lists}, NO_TYPES) is None - ) # undeclared: the JSON path keeps the doubles _, cols = _pack({"v": lists}, {"v": "ARRAY_OF_FLOATS"}) - assert (cols["v"].kind, cols["v"].dim) == ("f", 2) - assert ( - columnar.pack( - {"v": [[1.0], [1.0, 2.0]]}, - columnar.Declared(known={"v": "ARRAY_OF_FLOATS"}), - ) - is None - ) # ragged + assert (cols["v"].kind, cols["v"].dim) == ("f", 2) # a matrix, not a ragged list + _, cols = _pack({"v": lists}) # undeclared: the JSON path keeps the doubles + assert (cols["v"].kind, cols["v"].dim) == ("D", 0) + + +def _lists(col): + """(offsets, elements, present) of a ragged list column as read back from the addresses.""" + offsets = np.frombuffer(_bytes(col.data), dtype=np.int32) + dtype = {"I": np.int32, "Q": np.int64, "E": np.float32, "D": np.float64}[col.kind] + elements = np.frombuffer(_bytes(col.aux), dtype=dtype) + return ( + offsets.tolist(), + elements.tolist(), + _present(col) if col.nulls[1] else None, + ) + + +def test_ragged_int_and_float_lists_are_offsets_plus_elements(): + tokens = [[5, 7, 9], [1], [], [2, 4]] + weights = [[0.5, 1.25, 2.0], [3.5], [], [0.1, 0.2]] + rows, cols = _pack( + {"t": tokens, "w": weights, "t2": tokens, "w2": weights}, + {"t": "ARRAY_OF_INTEGERS", "w": "ARRAY_OF_FLOATS", "t2": "LIST", "w2": "LIST"}, + ) + assert rows == 4 + assert [cols[k].kind for k in ("t", "w", "t2", "w2")] == ["I", "E", "Q", "D"] + assert all(cols[k].n == 4 and cols[k].dim == 0 for k in cols) + assert _lists(cols["t"]) == ([0, 3, 4, 4, 6], [5, 7, 9, 1, 2, 4], None) + assert _lists(cols["t2"]) == ([0, 3, 4, 4, 6], [5, 7, 9, 1, 2, 4], None) + assert _lists(cols["w"])[0] == [0, 3, 4, 4, 6] + assert _lists(cols["w"])[1] == np.array(sum(weights, []), dtype=np.float32).tolist() + assert _lists(cols["w2"])[1] == sum( + weights, [] + ) # float64: exactly the Python floats + + +def test_undeclared_lists_cross_as_the_64_bit_kinds_the_json_text_parses_to(): + _, cols = _pack({"t": [[1, 2], [3]], "w": [[0.5], [1.5, 2.5]]}) + assert (cols["t"].kind, cols["w"].kind) == ("Q", "D") + big = [[-(2**63), 2**63 - 1], [2**31, -(2**31) - 1]] + _, cols = _pack({"t": big}) + assert _lists(cols["t"])[1] == [-(2**63), 2**63 - 1, 2**31, -(2**31) - 1] + + +def test_a_declared_float_array_narrows_to_float32_and_a_huge_double_becomes_infinity(): + _, cols = _pack({"w": [[1e300, 0.1], [-1e300]]}, {"w": "ARRAY_OF_FLOATS"}) + got = _lists(cols["w"])[1] + assert got[0] == float("inf") and got[2] == float("-inf") + assert got[1] == float(np.float32(0.1)) + + +def test_null_and_empty_rows_of_a_list_column(): + _, cols = _pack({"t": [[1, 2], None, [], [3]], "w": [None, None, [0.5], None]}) + assert _lists(cols["t"]) == ([0, 2, 2, 2, 3], [1, 2, 3], [True, False, True, True]) + assert _lists(cols["w"]) == ([0, 0, 0, 1, 1], [0.5], [False, False, True, False]) + _, cols = _pack({"t": [[], []]}, {"t": "ARRAY_OF_INTEGERS"}) + assert (cols["t"].kind, cols["t"].n, cols["t"].aux) == ("I", 2, (0, 0)) + assert _lists(cols["t"])[0] == [0, 0, 0] + for declared, kind in ( + ({}, "Q"), + ({"t": "LIST"}, "Q"), + ({"t": "ARRAY_OF_FLOATS"}, "E"), + ): + _, cols = _pack({"t": [[], []]}, declared) + assert cols["t"].kind == kind + _, cols = _pack({"t": [None, [1.5]]}, {"t": "ARRAY_OF_FLOATS"}) + assert cols["t"].kind == "E" # a null row keeps a float list off the fixed matrix + assert _lists(cols["t"]) == ([0, 0, 1], [1.5], [False, True]) + + +def test_the_offsets_of_a_long_ragged_list_column_index_its_elements(): + lists = [list(range(k % 7)) for k in range(25_000)] + rows, cols = _pack({"t": lists}) + offsets, elements, _ = _lists(cols["t"]) + assert rows == 25_000 and offsets[-1] == len(elements) == sum(map(len, lists)) + assert elements[offsets[1234] : offsets[1235]] == lists[1234] + + +@pytest.mark.parametrize( + "column, declared", + [ + ([[1, 2.5]], {}), # mixed int and float elements + ([[True, False]], {}), # bool is not an int here + ([[1], [2.5]], {}), # an int row and a float row + ([["a"]], {}), + ([[[1]], [[2]]], {}), # nested + ([[1], (2, 3)], {}), # a tuple row + ([(1, 2), (3,)], {}), + ([[np.int64(1)]], {}), + ([[1, 2**63]], {}), # past 64 bits + ([[float("nan")]], {}), + ([[1.0], [2.0, float("inf")]], {"t": "ARRAY_OF_FLOATS"}), # ragged: no matrix + ([[2**31]], {"t": "ARRAY_OF_INTEGERS"}), # the engine refuses it: JSON path + ([[-(2**31) - 1]], {"t": "ARRAY_OF_INTEGERS"}), + ([[1.5]], {"t": "ARRAY_OF_INTEGERS"}), + ([[1]], {"t": "ARRAY_OF_LONGS"}), # a declared type the kinds do not cover + ([[1.0]], {"t": "ARRAY_OF_DOUBLES"}), + ([[1]], {"t": "STRING"}), + ], +) +def test_list_columns_the_buffers_cannot_carry_unchanged_are_refused(column, declared): + named = {"t": column} + assert columnar.pack(named, columnar.Declared(known=declared)) is None + + +def test_a_library_without_the_list_decoder_keeps_lists_on_the_json_path(): + off = columnar.Declared( + known={"t": "ARRAY_OF_INTEGERS", "w": "ARRAY_OF_FLOATS"}, lists=False + ) + assert columnar.pack({"t": [[1], [2, 3]]}, off) is None + assert columnar.pack({"w": [[1.0], [2.0, 3.0]]}, off) is None assert ( - columnar.pack( - {"v": [[], []]}, columnar.Declared(known={"v": "ARRAY_OF_FLOATS"}) - ) - is None - ) # no dimension + columnar.pack({"id": [1, 2], "w": [[1.0, 2.0], [3.0, 4.0]]}, off) is not None + ) # a float matrix needs no decoder + assert columnar.pack({"s": ["a", "b"]}, off) is not None def test_strings_are_arrow_offsets_plus_utf8_bytes(): @@ -189,7 +287,6 @@ def test_a_bitmap_longer_than_a_byte(): [True, 1], # bool is not an int here [1, 2**63], # past 64 bits [{"a": 1}, {"a": 2}], # nested - [[1], [2]], # a list without a declared ARRAY_OF_FLOATS [b"ab", b"cd"], [np.int64(1), np.int64(2)], # numpy scalars in a list are not Python numbers ], From 5691c4775cac2a0eaa8303ca909e0e21643fe6c3 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Wed, 4 Nov 2026 05:48:19 +0900 Subject: [PATCH 2/3] test: numpy generator for the sparse rows (bandit B311) Co-Authored-By: Claude Opus 5.5 --- tests/test_native_columnar_insert.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/tests/test_native_columnar_insert.py b/tests/test_native_columnar_insert.py index 463618a..dcee0a1 100644 --- a/tests/test_native_columnar_insert.py +++ b/tests/test_native_columnar_insert.py @@ -529,15 +529,13 @@ def _sparse_types( def _sparse_rows(n, ragged, seed=11): """Rows of the lanes' shape: an id, the token ids of a sparse vector and their weights (30 nonzeros, or 1 to 40 when ragged).""" - import random - - rng = random.Random(seed) + rng = np.random.default_rng(seed) rows = [] for i in range(n): - nnz = rng.randint(1, 40) if ragged else 30 - tokens = rng.sample(range(SPARSE_DIMS), nnz) + nnz = int(rng.integers(1, 41)) if ragged else 30 + tokens = rng.choice(SPARSE_DIMS, size=nnz, replace=False).tolist() rows.append( - {"id": i, "tokens": tokens, "weights": [0.1 + rng.random() for _ in tokens]} + {"id": i, "tokens": tokens, "weights": (0.1 + rng.random(nnz)).tolist()} ) return rows From f91cf142aa2b9c21f9c28611b18c0ad2560e8f06 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Wed, 4 Nov 2026 06:16:39 +0900 Subject: [PATCH 3/3] test: clear the spy before the empty-rows insert (the JSON reference crossed rows once) Co-Authored-By: Claude Opus 5.5 --- tests/test_native_columnar_insert.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_native_columnar_insert.py b/tests/test_native_columnar_insert.py index dcee0a1..f53555f 100644 --- a/tests/test_native_columnar_insert.py +++ b/tests/test_native_columnar_insert.py @@ -650,8 +650,9 @@ def test_every_list_shape_stores_what_the_json_path_stores(temp_db, spy, monkeyp empty = [{"id": 1000 + k, "ai": [], "af": []} for k in range(2)] # only empty rows _json_reference(temp_db, "Ref", empty, monkeypatch) + spy.calls.clear() temp_db.insert_many("Col", empty) - assert spy.count(COLP) == 2 and spy.count(ROWS) == 0 + assert spy.count(COLP) == 1 and spy.count(ROWS) == 0, spy.calls assert _stored_deep(temp_db, "Col") == _stored_deep(temp_db, "Ref") assert _stored_deep(temp_db, "Col")[-1]["ai"] == ("list", [])