From 6fd16246271be7cb4be90c1d207c77408cb3c5fc Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 22:03:10 +0900 Subject: [PATCH] Native: columnar bulk insert by address for strings, nulls and vectors (no JSON text) Port of humemai/arcadedb-embedded-python#514 onto the P-04 port fix. - arcadedb_n_insert_colp: the image reads whole columns in place (int64, float64, bool, int32, float32, UTF-8 strings as Arrow offsets plus bytes, float32 vector matrices, an Arrow validity bitmap per column), no JSON text and no per-value crossing. - _native/columnar.py lays the columns out (numpy, lists of one value type with None as null, lists of float32 arrays, pyarrow arrays); _native/bulk.py sends them; insert_many chunks and the DataFrame/Arrow document load use the entry when every column maps to a supported kind, any other batch keeps the JSON path. - Only a float32 matrix (or a list of float32 arrays) is a column of vectors, as on the JPype path (tests/test_float_vector_bulk.py); #514 took any float matrix. The pack test says so. - New tests: test_native_columnar_pack.py (no native library needed, runs on both backends) and test_native_columnar_insert.py (compares the stored records of the JSON and the columnar entry points, native backend only, collect_ignore otherwise). - The two new test files use plain imports of numpy, pandas and pyarrow, not importorskip (the repo's no-skips policy). Needs a rebuilt native library (BulkOps.java, ArcadeNative.java). Co-Authored-By: Claude Sonnet 5.5 --- .../org/humemai/nativelib/ArcadeNative.java | 27 +- .../java/org/humemai/nativelib/BulkOps.java | 162 ++++-- src/arcadedb/_native/bulk.py | 139 ++--- src/arcadedb/_native/columnar.py | 373 +++++++++++++ src/arcadedb/_native/database.py | 13 +- src/arcadedb/_native/lib.py | 1 + src/arcadedb/dataframe.py | 65 ++- tests/conftest.py | 2 + tests/test_native_columnar_insert.py | 488 ++++++++++++++++++ tests/test_native_columnar_pack.py | 301 +++++++++++ 10 files changed, 1433 insertions(+), 138 deletions(-) create mode 100644 src/arcadedb/_native/columnar.py create mode 100644 tests/test_native_columnar_insert.py create mode 100644 tests/test_native_columnar_pack.py diff --git a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java index 49a102f..2cbd978 100644 --- a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java +++ b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java @@ -29,6 +29,7 @@ import org.graalvm.nativeimage.c.type.CCharPointer; import org.graalvm.nativeimage.c.type.CFloatPointer; import org.graalvm.nativeimage.c.type.CLongPointer; +import org.graalvm.word.Pointer; import org.graalvm.word.WordFactory; import org.graalvm.nativeimage.c.type.CTypeConversion; @@ -510,13 +511,33 @@ private static CCharPointer replyOf(final ResultSet rs, final int max, final int return p; } - /** Columnar insert (see BulkOps): header JSON, one buffer holding the columns. Returns {"ok":true,"n":rows}. */ + /** Columnar insert (see BulkOps): header JSON, one buffer holding the columns back to back. Returns {"ok":true,"n":rows}. */ @CEntryPoint(name = "arcadedb_n_insert_cols") public static CCharPointer insertCols(final IsolateThread thread, final long h, final CCharPointer type, final CCharPointer header, final CCharPointer buf, final long bufBytes) { try { - final java.nio.ByteBuffer bb = CTypeConversion.asByteBuffer(buf, (int) bufBytes); - return out("{\"ok\":true,\"n\":" + BulkOps.insertColumns(db(h), str(type), str(header), bb) + "}"); + return out("{\"ok\":true,\"n\":" + BulkOps.insertColumns(db(h), str(type), str(header), (Pointer) buf, bufBytes) + "}"); + } catch (final Throwable t) { + return err(t); + } + } + + /** + * 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. + * The caller keeps every buffer alive and unchanged until the call returns. Returns {"ok":true,"n":rows}. + */ + @CEntryPoint(name = "arcadedb_n_insert_colp") + public static CCharPointer insertColp(final IsolateThread thread, final long h, final CCharPointer type, final CCharPointer header, + final CCharPointer spec, final int nspec) { + try { + if (nspec < 0 || nspec % BulkOps.SLOTS != 0) + throw new IllegalArgumentException("address table of " + nspec + " entries is not a multiple of " + BulkOps.SLOTS); + final long[] table = new long[nspec]; + for (int i = 0; i < nspec; i++) + table[i] = ((Pointer) spec).readLong(WordFactory.signed(8L * i)); + return out("{\"ok\":true,\"n\":" + BulkOps.insertColumns(db(h), str(type), str(header), table) + "}"); } catch (final Throwable t) { return err(t); } diff --git a/native/src/main/java/org/humemai/nativelib/BulkOps.java b/native/src/main/java/org/humemai/nativelib/BulkOps.java index a3042bb..37cedf9 100644 --- a/native/src/main/java/org/humemai/nativelib/BulkOps.java +++ b/native/src/main/java/org/humemai/nativelib/BulkOps.java @@ -2,8 +2,9 @@ * SPDX-License-Identifier: Apache-2.0 * * Phase 7: columnar bulk insert through one crossing. The documents are built here from whole columns that Python lays out - * in one native-endian buffer (int64, float64, bool bytes, float32 vector matrix), as DocumentBatcher.insertColumns does on - * the JPype path from long[], double[], boolean[] and Object[] columns. + * 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. */ package org.humemai.nativelib; @@ -11,8 +12,12 @@ import com.arcadedb.database.MutableDocument; import com.arcadedb.schema.VertexType; +import org.graalvm.nativeimage.c.type.CTypeConversion; +import org.graalvm.word.Pointer; +import org.graalvm.word.WordFactory; + import java.nio.ByteBuffer; -import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; @@ -20,57 +25,118 @@ final class BulkOps { private BulkOps() { } + /** Slots per column in the address table: data (address, bytes), aux (address, bytes), nulls (address, bytes). */ + static final int SLOTS = 6; + /** - * header: {"n":rows,"commit":every,"cols":[[name,kind,dim],..]} with kind q (int64), d (float64), b (bool byte) or f - * (float32 matrix, dim floats per row); the buffer holds the columns in that order. Same transaction contract as - * DocumentBatcher.insertColumns: batch commits and the rollback on failure apply only to a transaction opened here. + * The original single-buffer form (arcadedb_n_insert_cols): the columns of kind q, d, b or f lie back to back in one buffer, in + * header order. The address table of the pointer form is derived from the layout, so one loop builds the documents for both. */ @SuppressWarnings("unchecked") - static long insertColumns(final Database db, final String type, final String header, final ByteBuffer raw) { + static long insertColumns(final Database db, final String type, final String header, final Pointer buf, final long bufBytes) { final Map h = (Map) Json.parse(header); - final int n = ((Number) h.get("n")).intValue(); + final long n = ((Number) h.get("n")).longValue(); + final List cols = (List) h.get("cols"); + final long[] spec = new long[SLOTS * cols.size()]; + long pos = 0; + for (int c = 0; c < cols.size(); c++) { + final List col = (List) cols.get(c); + final char kind = ((String) col.get(1)).charAt(0); + final long dim = ((Number) col.get(2)).longValue(); + final long bytes = switch (kind) { + case 'q', 'd' -> 8 * n; + case 'b' -> n; + case 'f' -> 4 * n * dim; + default -> throw new IllegalArgumentException("unknown column kind " + kind); + }; + if (pos + bytes > bufBytes) + throw new IllegalArgumentException("column '" + col.get(0) + "' ends at byte " + (pos + bytes) + ", the buffer has " + bufBytes); + spec[SLOTS * c] = buf.rawValue() + pos; + spec[SLOTS * c + 1] = bytes; + pos += bytes; + } + return insertColumns(db, type, h, spec); + } + + /** The pointer form (arcadedb_n_insert_colp): the table holds SLOTS longs per column, see {@link #insertColumns(Database, String, Map, long[])}. */ + @SuppressWarnings("unchecked") + static long insertColumns(final Database db, final String type, final String header, final long[] spec) { + return insertColumns(db, type, (Map) Json.parse(header), spec); + } + + /** + * 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 + * 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 + * DocumentBatcher.insertColumns: batch commits and the rollback on failure apply only to a transaction opened here. + */ + @SuppressWarnings("unchecked") + private static long insertColumns(final Database db, final String type, final Map h, final long[] spec) { + final long nRows = ((Number) h.get("n")).longValue(); + if (nRows < 0 || nRows > Integer.MAX_VALUE) + throw new IllegalArgumentException("row count out of range: " + nRows); + final int n = (int) nRows; final int commitEvery = ((Number) h.get("commit")).intValue(); final List cols = (List) h.get("cols"); final int k = cols.size(); + if (spec.length != SLOTS * k) + throw new IllegalArgumentException("address table has " + spec.length + " entries for " + k + " columns, expected " + SLOTS * k); final String[] names = new String[k]; final char[] kinds = new char[k]; - final Object[] data = new Object[k]; final int[] dims = new int[k]; - raw.order(ByteOrder.nativeOrder()); - int pos = 0; + 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]; + int longest = 0; for (int c = 0; c < k; c++) { final List col = (List) cols.get(c); names[c] = (String) col.get(0); kinds[c] = ((String) col.get(1)).charAt(0); dims[c] = ((Number) col.get(2)).intValue(); - final int bytes = switch (kinds[c]) { - case 'q', 'd' -> 8 * n; + final long dataAddr = spec[SLOTS * c], dataBytes = spec[SLOTS * c + 1]; + final long auxAddr = spec[SLOTS * c + 2], auxBytes = spec[SLOTS * c + 3]; + final long nullAddr = spec[SLOTS * c + 4], nullBytes = spec[SLOTS * c + 5]; + final long need = switch (kinds[c]) { + case 'q', 'd' -> 8L * n; case 'b' -> n; - case 'f' -> 4 * n * dims[c]; + case 'i', 'e' -> 4L * n; + case 'f' -> { + if (dims[c] <= 0) + throw new IllegalArgumentException("column '" + names[c] + "': a vector column needs a dimension above 0"); + yield 4L * n * dims[c]; + } + case 's' -> 4L * (n + 1L); default -> throw new IllegalArgumentException("unknown column kind " + kinds[c]); }; - final ByteBuffer part = raw.slice(pos, bytes).order(ByteOrder.nativeOrder()); - pos += bytes; - switch (kinds[c]) { - case 'q' -> { - final long[] a = new long[n]; - part.asLongBuffer().get(a); - data[c] = a; - } - case 'd' -> { - final double[] a = new double[n]; - part.asDoubleBuffer().get(a); - data[c] = a; - } - case 'b' -> { - final boolean[] a = new boolean[n]; - for (int i = 0; i < n; i++) - a[i] = part.get(i) != 0; - data[c] = a; + if (dataBytes < need || (need > 0 && dataAddr == 0)) + throw new IllegalArgumentException("column '" + names[c] + "' has " + dataBytes + " bytes, " + need + " needed"); + data[c] = dataAddr; + if (nullAddr != 0) { + if (nullBytes < (n + 7L) / 8) + throw new IllegalArgumentException("column '" + names[c] + "' has a null bitmap of " + nullBytes + " bytes, " + (n + 7L) / 8 + " needed"); + nulls[c] = nullAddr; + } + 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"); + blobs[c] = auxBytes == 0 ? ByteBuffer.allocate(0) : CTypeConversion.asByteBuffer(WordFactory.pointer(auxAddr), (int) auxBytes); + final Pointer offsets = WordFactory.pointer(data[c]); + int prev = offsets.readInt(WordFactory.signed(0)); + if (prev < 0) + throw new IllegalArgumentException("column '" + names[c] + "': negative string offset"); + for (int i = 1; i <= n; i++) { + final int off = offsets.readInt(WordFactory.signed(4L * i)); + if (off < prev || off > auxBytes) + throw new IllegalArgumentException("column '" + names[c] + "': string offsets are not increasing within the " + auxBytes + " bytes (row " + (i - 1) + ")"); + longest = Math.max(longest, off - prev); + prev = off; } - default -> data[c] = part.asFloatBuffer(); } } + final byte[] scratch = new byte[longest]; final boolean vertex = db.getSchema().getType(type) instanceof VertexType; final boolean wasActive = db.isTransactionActive(); if (!wasActive) @@ -79,13 +145,33 @@ static long insertColumns(final Database db, final String type, final String hea for (int i = 0; i < n; i++) { final MutableDocument doc = vertex ? db.newVertex(type) : db.newDocument(type); for (int c = 0; c < k; c++) { + final Pointer base = WordFactory.pointer(data[c]); + if (nulls[c] != 0 && ((WordFactory.pointer(nulls[c]).readByte(WordFactory.signed(i >> 3)) >> (i & 7)) & 1) == 0) { + doc.set(names[c], null); + continue; + } switch (kinds[c]) { - case 'q' -> doc.set(names[c], ((long[]) data[c])[i]); - case 'd' -> doc.set(names[c], ((double[]) data[c])[i]); - case 'b' -> doc.set(names[c], ((boolean[]) data[c])[i]); + case 'q' -> doc.set(names[c], base.readLong(WordFactory.signed(8L * i))); + case 'd' -> doc.set(names[c], base.readDouble(WordFactory.signed(8L * i))); + case 'b' -> doc.set(names[c], base.readByte(WordFactory.signed(i)) != 0); + case 'i' -> doc.set(names[c], base.readInt(WordFactory.signed(4L * i))); + case 'e' -> doc.set(names[c], base.readFloat(WordFactory.signed(4L * i))); + case 's' -> { + final int from = base.readInt(WordFactory.signed(4L * i)); + final int len = base.readInt(WordFactory.signed(4L * (i + 1L))) - from; + if (len == 0) + doc.set(names[c], ""); + else { + blobs[c].get(from, scratch, 0, len); + doc.set(names[c], new String(scratch, 0, len, StandardCharsets.UTF_8)); + } + } default -> { - final float[] v = new float[dims[c]]; - ((java.nio.FloatBuffer) data[c]).get(i * dims[c], v); + final int dim = dims[c]; + final float[] v = new float[dim]; + final long at = 4L * i * dim; + for (int j = 0; j < dim; j++) + v[j] = base.readFloat(WordFactory.signed(at + 4L * j)); doc.set(names[c], v); } } diff --git a/src/arcadedb/_native/bulk.py b/src/arcadedb/_native/bulk.py index 9cae198..d244c48 100644 --- a/src/arcadedb/_native/bulk.py +++ b/src/arcadedb/_native/bulk.py @@ -1,98 +1,71 @@ -"""Columnar bulk insert through one crossing (``arcadedb_n_insert_cols``): numpy int/float/bool columns and float32 vector -columns (a 2-D array or a list of equal-length 1-D float arrays) go over as one buffer; anything else returns None and the -caller falls back to the per-row path. The documents are built Java-side as ``DocumentBatcher.insertColumns`` builds them on JPype. +"""Columnar bulk insert through one crossing (``arcadedb_n_insert_colp``): whole columns go over by address, with no JSON text. Numpy +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. """ -import json import os -from . import lib - -try: - import numpy as _np -except ImportError: # pragma: no cover - _np = None +from . import columnar, lib _ENABLED = os.environ.get("ARCADEDB_NATIVE_BULK", "1") != "0" -def insert_columns_fast(db, type_name, columns, commit_every): - if ( - not _ENABLED - or _np is None - or lib._lib is None - or not hasattr(lib._lib, "arcadedb_n_insert_cols") - ): - return None - cols, parts, n = [], [], None - for name, values in columns.items(): - if hasattr(values, "to_numpy") and not isinstance(values, (list, tuple)): - return None - kind, dim = None, 0 - if isinstance(values, _np.ndarray): - k = values.dtype.kind - if ( - values.ndim == 1 - and k in "iu" - and ( - k == "i" - or not values.size - or int(values.max()) <= _np.iinfo(_np.int64).max - ) - ): - kind, arr = "q", _np.ascontiguousarray(values, dtype=_np.int64) - elif values.ndim == 1 and k == "f": - kind, arr = "d", _np.ascontiguousarray(values, dtype=_np.float64) - elif values.ndim == 1 and k == "b": - kind, arr = "b", _np.ascontiguousarray(values, dtype=_np.uint8) - elif ( - values.ndim == 2 and values.dtype == _np.float32 and values.shape[1] > 0 - ): # only a float32 matrix is a column of vectors, as on the JPype path - kind, arr, dim = ( - "f", - _np.ascontiguousarray(values, dtype=_np.float32), - int(values.shape[1]), - ) - elif ( - isinstance(values, (list, tuple)) - and values - and all( - isinstance(v, _np.ndarray) - and v.ndim == 1 - and v.dtype.kind == "f" - and v.shape == values[0].shape - for v in values - ) - and values[0].shape[0] > 0 - ): - kind, arr, dim = ( - "f", - _np.ascontiguousarray(_np.stack(values), dtype=_np.float32), - int(values[0].shape[0]), - ) - if kind is None: - return None - rows = int(arr.shape[0]) - if n is not None and rows != n: - return None # the per-row path reports the ragged columns - n = rows - cols.append([name, kind, dim]) - parts.append(arr.tobytes()) - if not n: - return None - header = json.dumps( - {"n": n, "commit": min(int(commit_every), 2**31 - 1), "cols": cols} +def _usable(): + return ( + _ENABLED + and columnar._np is not None + and lib._lib is not None + and hasattr(lib._lib, "arcadedb_n_insert_colp") ) - buf = b"".join(parts) + + +def _send(db, type_name, packed, rows, commit_every, prefix): from .database import _call + spec = columnar.spec_of(packed) reply = _call( - "Failed to bulk-insert", - "arcadedb_n_insert_cols", + prefix, + "arcadedb_n_insert_colp", db._h, lib.b(type_name), - lib.b(header), - buf, - len(buf), - ) + lib.b(columnar.header_of(rows, commit_every, packed)), + spec, + len(spec), + ) # `packed` holds the arrays the addresses point into until the call returns return int(reply["n"]) + + +def insert_columns_fast( + db, type_name, columns, commit_every, declared=None, prefix="Failed to bulk-insert" +): + """Insert ``columns`` (name -> column) in one crossing and return the row count, or None when a column cannot be sent this way. + + ``declared``: a dict of declared property types when the caller already knows them (the DataFrame path), else they are read from the + schema if a column needs them. + """ + if not _usable(): + return None + packed = columnar.pack(columns, columnar.Declared(db, type_name, declared)) + if packed is None: + return None + rows, cols = packed + return _send(db, type_name, cols, rows, commit_every, prefix) + + +def insert_rows_columnar(db, type_name, chunk): + """One ``insert_many`` chunk (a list of dicts with the same keys) through the columnar entry inside the caller's transaction (no + commit here). Returns True when it was sent, False when the chunk needs the JSON path. + """ + if not _usable(): + return False + columns = columnar.rows_to_columns(chunk) + if columns is None: + return False + packed = columnar.pack(columns, columnar.Declared(db, type_name)) + if packed is None: + return False + rows, cols = packed + _send(db, type_name, cols, rows, 0, f"Failed to bulk-insert into '{type_name}'") + return True diff --git a/src/arcadedb/_native/columnar.py b/src/arcadedb/_native/columnar.py new file mode 100644 index 0000000..4bb79f1 --- /dev/null +++ b/src/arcadedb/_native/columnar.py @@ -0,0 +1,373 @@ +"""Packing of whole columns for ``arcadedb_n_insert_colp``: the columnar bulk insert of the native backend. + +A column crosses as raw memory, not as JSON text: the image reads each buffer in place and builds the documents without a parser. This +module only decides how a column is laid out and hands the addresses over; it needs no native library, so the layouts are unit-tested +without one. A column that cannot be laid out (a nested value, a mixed-type column, a date-time, an integer past 64 bits, a string with +a lone surrogate) makes :func:`pack` return ``None`` and the caller uses the JSON path for that call; nothing is half sent. + +Kinds (the image's, ``BulkOps.insertColumns``): ``q`` int64, ``d`` float64, ``b`` bool byte, ``i`` int32, ``e`` float32, ``f`` float32 +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). + +``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 re + +try: + import numpy as _np +except ImportError: # pragma: no cover + _np = None + +_INT64_MAX = 2**63 - 1 +_INT32_MAX = 2**31 - 1 +_NoneType = type(None) +_PROPERTY = re.compile(r"name: (.+?), type: ([A-Z_]+)") + + +class Packed: + """One column laid out for the image: ``data``, ``aux`` and ``nulls`` are ``(address, bytes)``; ``keep`` holds what the addresses point into.""" + + __slots__ = ("name", "kind", "dim", "n", "data", "aux", "nulls", "keep") + + def __init__(self, name, kind, dim, array, aux=None, nulls=None, keep=()): + self.name = name + 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 + 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) + self.keep = (array, nulls, *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. + """ + + def __init__(self, db=None, type_name=None, known=None): + self._db, self._type, self._known = db, type_name, known + + def get(self, name): + if self._known is None: + self._known = _declared_types(self._db, self._type) + return self._known.get(name) + + +def _declared_types(db, type_name): + """``{property: TYPE}`` of ``type_name`` and its parent types (the nearest declaration wins); ``{}`` when it cannot be read.""" + try: + rows = { + r["name"]: r for r in db.query("sql", "SELECT FROM schema:types").to_list() + } + except Exception: # nosec B110 - the insert reports the real error + return {} + out, seen, queue = {}, set(), [type_name] + while queue: + current = queue.pop(0) + if current in seen or current not in rows: + continue + seen.add(current) + for text in rows[current].get("properties") or []: + m = _PROPERTY.search(text) + if m: + out.setdefault(m.group(1), m.group(2)) + queue.extend(rows[current].get("parentTypes") or []) + return out + + +def _bitmap(present): + """Arrow validity bitmap of a bool array (True = present).""" + return _np.packbits(present, bitorder="little") + + +def _numpy_column(name, values, declared, nulls=None): + """A 1-D or 2-D numpy array as a column, or None.""" + k = values.dtype.kind + if ( + values.ndim == 2 + and values.dtype + == _np.float32 # only a float32 matrix is a column of vectors, as on the JPype path + and values.shape[1] > 0 + and nulls is None + ): + arr = _np.ascontiguousarray(values, dtype=_np.float32) + return Packed(name, "f", int(values.shape[1]), arr) + if values.ndim != 1: + return None + if k == "b": + return Packed( + name, "b", 0, _np.ascontiguousarray(values).view(_np.uint8), nulls=nulls + ) + if k in "iu": + if ( + k == "u" + and values.dtype.itemsize == 8 + and values.size + and int(values.max()) > _INT64_MAX + ): + return None + if k == "i" and values.dtype.itemsize <= 4 and declared.get(name) == "INTEGER": + return Packed( + name, + "i", + 0, + _np.ascontiguousarray(values, dtype=_np.int32), + nulls=nulls, + ) + return Packed( + name, "q", 0, _np.ascontiguousarray(values, dtype=_np.int64), nulls=nulls + ) + if k == "f": + if values.dtype.itemsize == 4 and declared.get(name) == "FLOAT": + return Packed( + name, + "e", + 0, + _np.ascontiguousarray(values, dtype=_np.float32), + nulls=nulls, + ) + return Packed( + name, "d", 0, _np.ascontiguousarray(values, dtype=_np.float64), nulls=nulls + ) + return None + + +def _string_column(name, texts, nulls=None): + """A list of str (a null already replaced by '') as a string column, or None (a lone surrogate, or 2 GiB of text).""" + n = len(texts) + blob = "".join(texts) + if blob.isascii(): + data = blob.encode("ascii") + lengths = _np.fromiter(map(len, texts), dtype=_np.int64, count=n) + else: + try: + encoded = [t.encode("utf-8") for t in texts] + except UnicodeEncodeError: + return None # a lone surrogate: the JSON path decides what happens to it + data = b"".join(encoded) + lengths = _np.fromiter(map(len, encoded), dtype=_np.int64, count=n) + if len(data) > _INT32_MAX: + return None + offsets = _np.zeros(n + 1, dtype=_np.int32) + _np.cumsum(lengths, out=offsets[1:], dtype=_np.int32) + blob_array = ( + _np.frombuffer(data, dtype=_np.uint8) if data else _np.zeros(0, dtype=_np.uint8) + ) + aux = (blob_array.ctypes.data, blob_array.nbytes) if data else (0, 0) + return Packed(name, "s", 0, offsets, aux=aux, nulls=nulls, keep=(blob_array, data)) + + +def _list_column(name, values, declared): + """A list or tuple of Python values as a column: one type (None allowed) or a list of equal-length float arrays; else None.""" + n = len(values) + types = set(map(type, values)) + present = None + if _NoneType in types: + types.discard(_NoneType) + present = _np.fromiter((v is not None for v in values), dtype=bool, count=n) + nulls = _bitmap(present) + else: + nulls = None + if not types: # every value is None + return _string_column(name, [""] * n, nulls) + if len(types) != 1: + return None + (only,) = types + if only is str: + texts = ( + values if present is None else [v if v is not None else "" for v in values] + ) + return _string_column(name, texts, nulls) + if only is int or only is float or only is bool: + filler = 0 if only is int else 0.0 if only is float else False + vals = ( + values + if present is None + else [v if v is not None else filler for v in values] + ) + dtype = ( + _np.int64 if only is int else _np.float64 if only is float else _np.uint8 + ) + try: + arr = _np.array(vals, dtype=dtype) + except OverflowError: + return None # an integer beyond 64 bits + kind = "q" if only is int else "d" if only is float else "b" + return Packed(name, kind, 0, arr, nulls=nulls) + if only is list and declared.get(name) == "ARRAY_OF_FLOATS" and present is None: + 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) + if ( + only is _np.ndarray + and present is None + and all( + v.ndim == 1 and v.dtype == _np.float32 and v.shape == values[0].shape + for v in values + ) + and values[0].shape[0] > 0 + ): + return Packed( + name, + "f", + int(values[0].shape[0]), + _np.ascontiguousarray(_np.stack(values), dtype=_np.float32), + ) + return None + + +def _arrow_column(name, arr, declared): + """A ``pyarrow`` array as a column (strings in place), or None.""" + import pyarrow as pa + + types = pa.types + if isinstance(arr, pa.ChunkedArray): + arr = arr.combine_chunks() + if isinstance(arr, pa.ChunkedArray): + arr = arr.chunk(0) if arr.num_chunks == 1 else None + if arr is None: + return None + if types.is_dictionary(arr.type): + arr = arr.cast(arr.type.value_type) + t, n = arr.type, len(arr) + nulls = ( + _bitmap(arr.is_valid().to_numpy(zero_copy_only=False)) + if arr.null_count + else None + ) + if types.is_string(t) or types.is_large_string(t): + bufs = arr.buffers() + off_buf, data_buf = bufs[1], bufs[2] + if types.is_string(t): + offsets = _np.frombuffer(off_buf, dtype=_np.int32)[ + arr.offset : arr.offset + n + 1 + ] + else: + wide = _np.frombuffer(off_buf, dtype=_np.int64)[ + arr.offset : arr.offset + n + 1 + ] + if len(wide) and int(wide[-1]) > _INT32_MAX: + return None + offsets = wide.astype(_np.int32) + offsets = _np.ascontiguousarray(offsets) + size = 0 if data_buf is None else min(data_buf.size, _INT32_MAX) + aux = (data_buf.address, size) if size else (0, 0) + return Packed( + name, "s", 0, offsets, aux=aux, nulls=nulls, keep=(arr, off_buf, data_buf) + ) + if types.is_integer(t) or types.is_floating(t) or types.is_boolean(t): + if arr.null_count: + arr = arr.fill_null(False if types.is_boolean(t) else 0) + return _numpy_column(name, arr.to_numpy(zero_copy_only=False), declared, nulls) + if ( + (types.is_list(t) or types.is_large_list(t) or types.is_fixed_size_list(t)) + and not arr.null_count + and n + and declared.get(name) == "ARRAY_OF_FLOATS" + and (types.is_floating(t.value_type) or types.is_integer(t.value_type)) + ): + flat = arr.flatten() + if flat.null_count or len(flat) % n: + return None + dim = len(flat) // n + if dim == 0: + return None + if not types.is_fixed_size_list(t): + import pyarrow.compute as pc + + lengths = pc.list_value_length(arr) + if pc.min(lengths).as_py() != dim or pc.max(lengths).as_py() != dim: + return None + values = flat.to_numpy(zero_copy_only=False).reshape(n, dim) + return Packed(name, "f", dim, _np.ascontiguousarray(values, dtype=_np.float32)) + return None + + +def pack(columns, declared): + """``(rows, [Packed])`` for a dict of name -> column, or None when any column cannot be laid out or the columns differ in length. + + A column is a numpy array (1-D ints, floats and bools, or a 2-D float32 matrix), a list or tuple of Python values of one type (None + allowed), a list of equal-length 1-D float32 arrays, or a ``pyarrow`` array. + """ + items = list( + columns.items() + ) # a dict, or a DataFrame (its columns are pandas Series: the row path takes those) + if _np is None or not items: + return None + rows, packed = None, [] + for name, values in items: + if type(values).__module__.split(".")[0] == "pyarrow": + col = _arrow_column(name, values, declared) + elif isinstance(values, _np.ndarray): + if values.dtype.kind in "OUS": + col = ( + _list_column(name, values.tolist(), declared) + if values.ndim == 1 + else None + ) + else: + col = _numpy_column(name, values, declared) + elif isinstance(values, (list, tuple)) and values: + col = _list_column(name, values, declared) + else: + col = None # a pandas Series, a generator, an empty list: the row path handles them + if col is None: + return None + if rows is not None and col.n != rows: + return None # the row path reports the ragged columns + rows = col.n + packed.append(col) + return (rows, packed) if rows else None + + +def spec_of(packed): + """The address table the image reads: six int64 per column (data, aux and null-bitmap address and byte count).""" + import ctypes + + flat = [] + for col in packed: + flat.extend((*col.data, *col.aux, *col.nulls)) + return (ctypes.c_int64 * len(flat))(*flat) + + +def header_of(rows, commit_every, packed): + """The JSON header text: row count, commit interval and ``[name, kind, dim]`` per column.""" + import json + + return json.dumps( + { + "n": rows, + "commit": min(int(commit_every), 2**31 - 1), + "cols": [[c.name, c.kind, c.dim] for c in packed], + } + ) + + +def rows_to_columns(rows): + """A list of dicts with the same keys as a dict of name -> list of values, or None (no rows, different keys, a non-str key).""" + if not rows: + return None + try: + keys = list(rows[0]) + if any(type(key) is not str for key in keys): + return None + if set(map(len, rows)) != {len(keys)}: + return None + return {key: [row[key] for row in rows] for key in keys} + except ( + KeyError, + TypeError, + ): # same size but different keys, or a row that is not a dict + return None diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index d640e8a..1b87e20 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -14,6 +14,7 @@ from ..transactions import TransactionContext from ..type_conversion import json_bulk_dumps from . import lib, not_yet +from .bulk import insert_rows_columnar from .codec import _prep as _prep_row from .codec import encode_params, json_native, loads, row_default @@ -677,10 +678,12 @@ def lookup_by_key(self, type_name: str, keys: List[str], values: List[Any]): def insert_many( self, type_name: str, rows, commit_every: int = 10_000, parallel: bool = False ) -> int: - """Bulk insert. Rows go in chunks of 10,000, one JSON text and one crossing per chunk (``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 row by row through ``INSERT INTO t CONTENT :row``. Inside the caller's transaction nothing is - committed; otherwise every ``commit_every`` rows are one transaction (0: all rows in one). + """Bulk insert. Rows go in chunks of 10,000, 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 + 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 + transaction nothing is committed; otherwise every ``commit_every`` rows are one transaction (0: all rows in one). """ self._check_not_closed() if parallel: @@ -719,6 +722,8 @@ def insert_many( def _insert_chunk(self, type_name: str, sql: str, chunk) -> None: if lib.INSERT_ROWS: + if insert_rows_columnar(self, type_name, chunk): + return try: payload = json_bulk_dumps(chunk) except (TypeError, ValueError): diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index b2f16ac..0309ee5 100644 --- a/src/arcadedb/_native/lib.py +++ b/src/arcadedb/_native/lib.py @@ -158,6 +158,7 @@ class _Params(ctypes.Structure): "arcadedb_n_exec": [P, L, I, S, S, S, I], "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_next": [P, L, I], "arcadedb_n_rs_close": [P, L], "arcadedb_n_cmd": [P, L, I, S, S, S, S, L, I], diff --git a/src/arcadedb/dataframe.py b/src/arcadedb/dataframe.py index 634ff9c..0e4e9b9 100644 --- a/src/arcadedb/dataframe.py +++ b/src/arcadedb/dataframe.py @@ -311,14 +311,30 @@ def _scalars_to_tagged(kind: str, chunk): ] -def _column_values(name: str, chunk, pa, native: bool = False): +def _column_values(name: str, chunk, pa, native: bool = False, arrow: bool = False): """One Arrow array as what ``Database.insert_columns`` takes: a numpy array when the column crosses as one buffer (integers, floats and booleans without nulls), else a list of values (``None`` is a null). + + ``arrow`` (the native backend's document load): a string column, and a numeric one with nulls, stay Arrow arrays, which + the columnar entry reads in place (the offsets, the bytes and the validity bitmap) instead of as Python values. """ types = pa.types if types.is_dictionary(chunk.type): # a pandas categorical or a Polars Categorical chunk = chunk.cast(chunk.type.value_type) kind = _kind(pa, chunk.type) + if arrow and ( + ( + kind == "string" + and (types.is_string(chunk.type) or types.is_large_string(chunk.type)) + ) + or (kind in ("integer", "floating", "bool") and chunk.null_count) + ): + if not ( + kind == "integer" + and types.is_uint64(chunk.type) + and len(chunk) > chunk.null_count + ): # uint64 keeps the range check below + return chunk if kind in ("integer", "floating", "bool"): if native and types.is_uint64(chunk.type) and len(chunk) > chunk.null_count: import pyarrow.compute as pc @@ -403,7 +419,13 @@ def insert_table( if batch.num_rows == 0: continue columns = { - name: _column_values(name, batch.column(i), pa, native) + name: _column_values( + name, + batch.column(i), + pa, + native, + arrow=native and not is_vertex and not parallel, + ) for i, name in enumerate(names) } try: @@ -413,7 +435,7 @@ def insert_table( ) elif native: _insert_documents_native( - db, type_name, names, columns, batch_size, parallel + db, type_name, names, columns, batch_size, parallel, declared ) else: db.insert_columns( @@ -460,15 +482,38 @@ def _insert_documents_native( columns: Dict[str, Any], batch_size: int, parallel: bool, + declared: Optional[Dict[str, str]] = None, ): - """One batch of documents on the native backend: the columns as one dict per row, handed to ``insert_many`` - (one crossing per 10,000 rows; the batch is one transaction, or joins the caller's). + """One batch of documents on the native backend. When every column can cross as memory (numbers, bools, strings, nulls, + float vectors for an ``ARRAY_OF_FLOATS`` property) the batch goes through the columnar entry, no Python values and no JSON + text; otherwise the columns become one dict per row and go to ``insert_many`` (one crossing per 10,000 rows). Either way + the batch is one transaction, or joins the caller's. """ - lists = [ - # jvm-payload: not-a-crossing -- no JVM here: insert_many serialises the rows to JSON text - columns[name].tolist() if hasattr(columns[name], "tolist") else columns[name] - for name in names - ] + if not parallel: + from ._native.bulk import insert_columns_fast + + if ( + insert_columns_fast( + db, + type_name, + columns, + batch_size, + declared={} if declared is None else declared, + prefix=f"Failed to bulk-insert into '{type_name}'", + ) + is not None + ): + return + lists = [] + for name in names: + col = columns[name] + if hasattr( + col, "to_pylist" + ): # a column kept as an Arrow array for the columnar entry + lists.append(col.to_pylist()) + else: + # jvm-payload: not-a-crossing -- no JVM here: insert_many serialises the rows to JSON text + lists.append(col.tolist() if hasattr(col, "tolist") else col) db.insert_many( type_name, [dict(zip(names, row)) for row in zip(*lists)], diff --git a/tests/conftest.py b/tests/conftest.py index 365d268..86b601d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -39,6 +39,8 @@ if BACKEND != "native": # Java -> Python upcalls (Python functions in SQL) exist on the native build only collect_ignore.append("test_native_udf.py") + # the columnar insert's stored records are compared between the JSON and the columnar entry points of the native library + collect_ignore.append("test_native_columnar_insert.py") # the native library's start-up output (#477) can only be checked with the native library loaded collect_ignore.append("test_native_startup_runtime.py") # the XML entity limits of an import, lifted by the native glue (the JVM build lifts them with JVM system properties) diff --git a/tests/test_native_columnar_insert.py b/tests/test_native_columnar_insert.py new file mode 100644 index 0000000..3aac3df --- /dev/null +++ b/tests/test_native_columnar_insert.py @@ -0,0 +1,488 @@ +"""The native backend's columnar bulk insert stores what the JSON path stores (``arcadedb_n_insert_colp`` against ``arcadedb_n_insert_rows``). + +Needs the native library, so conftest.py collects this file on the native backend only (a skip elsewhere would fail +scripts/check_test_skips.py). Each test loads the same data through the JSON path (``bulk._ENABLED`` off) and through the columnar entry and +compares the stored records value by value and type by type; a spy on the crossing proves which entry point ran, so a silent fall back +to JSON cannot pass for the columnar path. +""" + +import ctypes +import datetime +import math + +import numpy as np +import pytest + +from arcadedb._native import bulk, columnar, lib # noqa: E402 +from arcadedb.exceptions import ArcadeDBError # noqa: E402 + +COLP, ROWS, MANY = ( + "arcadedb_n_insert_colp", + "arcadedb_n_insert_rows", + "arcadedb_n_exec_many", +) + + +class Spy: + def __init__(self, monkeypatch): + self.calls = [] + real = lib.call + + def call(name, *args): + self.calls.append(name) + return real(name, *args) + + monkeypatch.setattr(lib, "call", call) + + def count(self, name): + return self.calls.count(name) + + +@pytest.fixture +def spy(monkeypatch): + return Spy(monkeypatch) + + +def _count_sql(name): + return f"SELECT count(*) AS n FROM {name}" # nosec B608 - test-owned type name, no user input + + +def _types(db, *names, extra="", unique=False): + for name in names: + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + db.command("sql", f"CREATE PROPERTY {name}.grp INTEGER") + db.command("sql", f"CREATE PROPERTY {name}.val DOUBLE") + db.command("sql", f"CREATE PROPERTY {name}.cat STRING") + db.command("sql", f"CREATE PROPERTY {name}.txt STRING") + db.command("sql", f"CREATE PROPERTY {name}.vec ARRAY_OF_FLOATS") + if unique: + db.command("sql", f"CREATE INDEX ON {name} (id) UNIQUE") + + +def _stored(db, name): + """Every record of a type as (value, python type) pairs per property, ordered by id.""" + out = [] + sql = f"SELECT FROM {name} ORDER BY id" # nosec B608 - test-owned type name, no user input + for row in db.query("sql", sql).to_list(): + out.append( + {k: (v, type(v).__name__) for k, v in row.items() if not k.startswith("@")} + ) + return out + + +def _probe(n, dim=8, seed=7): + """The probe workload's column types: INTEGER ids, a DOUBLE, a 20-value category, a variable-length text, a float32 vector.""" + rng = np.random.default_rng(seed) + ids = np.arange(n, dtype=np.int64) + return { + "id": ids, + "grp": ids % 1000, + "val": (ids * 37 % 10007) / 7.0, + "cat": ["c%d" % (k % 20) for k in range(n)], + "txt": ["row-%d-%s" % (k, "x" * (k % 24)) for k in range(n)], + "vec": rng.standard_normal((n, dim), dtype=np.float32), + } + + +def _rows(columns): + names = list(columns) + lists = [c.tolist() if hasattr(c, "tolist") else c for c in columns.values()] + return [dict(zip(names, vals)) for vals in zip(*lists)] + + +def _json_reference(db, name, rows, monkeypatch, **kw): + with monkeypatch.context() as m: + m.setattr(bulk, "_ENABLED", False) + db.insert_many(name, rows, **kw) + + +def test_probe_workload_columns_store_the_same_records_on_every_columnar_entry( + temp_db, spy, monkeypatch +): + _types(temp_db, "Ref", "ByRows", "ByCols", "ByFrame", "ByTable") + columns = _probe(2_500) + rows = _rows(columns) + _json_reference(temp_db, "Ref", rows, monkeypatch) + assert spy.count(COLP) == 0 and spy.count(ROWS) == 1 + + spy.calls.clear() + temp_db.insert_many("ByRows", rows) # dict rows, transposed to columns + assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 + temp_db.insert_columns("ByCols", columns) # numpy columns and lists of str + assert spy.count(COLP) == 2 and spy.count(ROWS) == 0 + + import pandas as pd + import pyarrow as pa + + frame = pd.DataFrame( + { + **{k: v for k, v in columns.items() if k != "vec"}, + "vec": list(columns["vec"]), + } + ) + temp_db.insert_dataframe("ByFrame", frame) + temp_db.insert_arrow("ByTable", pa.table({k: v for k, v in frame.items()})) + assert spy.count(COLP) == 4 and spy.count(ROWS) == 0 and spy.count(MANY) == 0 + + expected = _stored(temp_db, "Ref") + assert len(expected) == 2_500 + assert expected[7]["id"][1] == "int" and expected[7]["vec"][1] == "list" + for name in ("ByRows", "ByCols", "ByFrame", "ByTable"): + assert _stored(temp_db, name) == expected, name + + +def test_unicode_empty_and_awkward_strings_survive_the_columnar_path( + temp_db, spy, monkeypatch +): + _types(temp_db, "Ref", "Col") + texts = [ + "", + "plain", + "héllo wörld", + "日本語のテキスト", + "한국어", + "😀 emoji, outside the BMP", + 'tab\tnewline\nquote" backslash\\', + "embedded\x00nul", + "x" * 70_000, # larger than the text scratch of the JSON path + "é́ composed and decomposed", + ] + n = len(texts) + columns = { + "id": np.arange(n), + "grp": np.zeros(n, dtype=np.int64), + "txt": texts, + "cat": list(reversed(texts)), + } + _json_reference(temp_db, "Ref", _rows(columns), monkeypatch) + temp_db.insert_columns("Col", columns) + assert spy.count(COLP) == 1 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + assert [ + r["txt"] + for r in temp_db.query("sql", "SELECT txt FROM Col ORDER BY id").to_list() + ] == texts + + +def test_nulls_are_stored_as_the_json_path_stores_them(temp_db, spy, monkeypatch): + _types(temp_db, "Ref", "Col", "ByFrame") + n = 40 + columns = { + "id": np.arange(n), + "grp": [None if i % 3 == 0 else i for i in range(n)], + "val": [None if i % 4 == 0 else i / 4 for i in range(n)], + "cat": [None if i % 5 == 0 else "c%d" % i for i in range(n)], + "txt": [None] * n, + } + _json_reference(temp_db, "Ref", _rows(columns), monkeypatch) + temp_db.insert_columns("Col", columns) + assert spy.count(COLP) == 1 + expected = _stored(temp_db, "Ref") + assert _stored(temp_db, "Col") == expected + assert expected[0]["grp"][0] is None and expected[1]["grp"][0] == 1 + + import pandas as pd + + spy.calls.clear() + frame = pd.DataFrame( + { + "id": np.arange(n), + "grp": pd.array(columns["grp"], dtype="Int64"), + "val": pd.array(columns["val"], dtype="Float64"), + "cat": pd.array(columns["cat"], dtype="string"), + "txt": pd.array(columns["txt"], dtype="string"), + } + ) + temp_db.insert_dataframe("ByFrame", frame) + assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 + assert _stored(temp_db, "ByFrame") == expected + + +def test_int64_edges_nan_and_infinity(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}.k INTEGER") + temp_db.command("sql", f"CREATE PROPERTY {name}.big LONG") + ints = [-(2**63), -(2**63) + 1, -1, 0, 1, 2**63 - 2, 2**63 - 1] + floats = [0.0, -0.0, 1.5, math.inf, -math.inf, 5e-324, 1.7976931348623157e308] + columns = {"k": list(range(len(ints))), "big": ints, "x": floats} + _json_reference(temp_db, "Ref", _rows(columns), monkeypatch) + temp_db.insert_columns("Col", columns) + assert spy.count(COLP) == 1 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + + nan = {"k": [0, 1], "x": [float("nan"), 1.0]} + temp_db.insert_columns("Col", nan) + stored = temp_db.query( + "sql", "SELECT x FROM Col WHERE k = 0 AND x IS NOT NULL ORDER BY x" + ).to_list() + assert any(isinstance(r["x"], float) and math.isnan(r["x"]) for r in stored) + + +def test_booleans_and_narrow_numpy_dtypes_keep_the_json_paths_types( + 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}.k INTEGER") + temp_db.command("sql", f"CREATE PROPERTY {name}.flag BOOLEAN") + temp_db.command("sql", f"CREATE PROPERTY {name}.small INTEGER") + temp_db.command("sql", f"CREATE PROPERTY {name}.single FLOAT") + n = 300 + ids = np.arange(n) + columns = { + "k": ids, + "flag": ids % 3 == 0, + "small": (ids - 150).astype(np.int32), # declared INTEGER: crosses as int32 + "single": (ids / 8).astype(np.float32), # declared FLOAT: crosses as float32 + "narrow": ids.astype(np.int16), # undeclared: a Long, as from JSON + "half": (ids / 4).astype(np.float32), # undeclared: a Double, as from JSON + } + _json_reference(temp_db, "Ref", _rows(columns), monkeypatch) + temp_db.insert_columns("Col", columns) + assert spy.count(COLP) == 1 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + + +def test_vertex_types_load_like_documents(temp_db, spy, monkeypatch): + for name in ("Ref", "Col"): + temp_db.command("sql", f"CREATE VERTEX TYPE {name}") + temp_db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + columns = {"id": np.arange(50), "label": ["v%d" % i for i in range(50)]} + _json_reference(temp_db, "Ref", _rows(columns), monkeypatch) + temp_db.insert_columns("Col", columns) + assert spy.count(COLP) == 1 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + assert temp_db.query("sql", "SELECT count(*) AS n FROM Col").to_list()[0]["n"] == 50 + + +# ---- the fallback: anything the buffers cannot carry unchanged goes through the JSON path, with the same result + + +def _fallback_case(temp_db, spy, monkeypatch, rows, expect_entry): + _types(temp_db, "Ref", "Col") + _json_reference(temp_db, "Ref", rows, monkeypatch) + spy.calls.clear() + temp_db.insert_many("Col", rows) + assert spy.count(COLP) == 0 + assert spy.count(expect_entry) == 1, spy.calls + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + + +def test_nested_maps_and_lists_fall_back_to_json(temp_db, spy, monkeypatch): + rows = [ + {"id": i, "txt": "t%d" % i, "meta": {"k": i, "tags": ["a", "b"]}} + for i in range(20) + ] + _fallback_case(temp_db, spy, monkeypatch, rows, ROWS) + + +def test_a_mixed_type_column_falls_back_to_json(temp_db, spy, monkeypatch): + rows = [ + {"id": i, "txt": "t%d" % i, "mixed": i if i % 2 else "s%d" % i} + for i in range(20) + ] + _fallback_case(temp_db, spy, monkeypatch, rows, ROWS) + rows = [ + {"id": i, "txt": "t%d" % i, "n": i if i % 2 else i + 0.5} for i in range(20) + ] # int and float + spy.calls.clear() + temp_db.insert_many("Col", rows) + assert spy.count(COLP) == 0 and spy.count(ROWS) == 1 + + +def test_rows_with_different_keys_fall_back_to_json(temp_db, spy, monkeypatch): + rows = [ + {"id": i, "txt": "t"} if i % 2 else {"id": i, "cat": "c"} for i in range(20) + ] + _fallback_case(temp_db, spy, monkeypatch, rows, ROWS) + + +def test_values_json_cannot_carry_still_go_row_by_row_in_one_crossing( + temp_db, spy, monkeypatch +): + when = datetime.datetime(2024, 5, 6, 7, 8, 9) + rows = [{"id": i, "txt": "t%d" % i, "when": when} for i in range(10)] + _fallback_case(temp_db, spy, monkeypatch, rows, MANY) + + +def test_an_integer_past_64_bits_is_refused_on_both_paths(temp_db, spy): + _types(temp_db, "Col") + with pytest.raises(ArcadeDBError): + temp_db.insert_many("Col", [{"id": 1, "txt": "a", "huge": 2**70}]) + assert spy.count(COLP) == 0 + with pytest.raises(ValueError): + temp_db.insert_columns("Col", {"id": np.array([2**63], dtype=np.uint64)}) + + +def test_a_dataframe_with_a_column_the_buffers_cannot_carry_falls_back( + temp_db, spy, monkeypatch +): + import pandas as pd + import pyarrow as pa + + _types(temp_db, "Col") + table = pa.table( + { + "id": pa.array([1, 2, 3], type=pa.int32()), + "txt": pa.array(["a", "b", "c"]), + "when": pa.array( + [datetime.datetime(2024, 1, d) for d in (1, 2, 3)], + type=pa.timestamp("us"), + ), + } + ) + temp_db.insert_arrow("Col", table) + assert spy.count(COLP) == 0 + assert pd is not None + rows = temp_db.query("sql", "SELECT id, txt FROM Col ORDER BY id").to_list() + assert [(r["id"], r["txt"]) for r in rows] == [(1, "a"), (2, "b"), (3, "c")] + + +# ---- transactions, chunking and failures + + +def test_more_rows_than_one_chunk_and_the_commit_interval(temp_db, spy): + _types(temp_db, "Col") + n = 25_000 + columns = _probe(n) + assert temp_db.insert_many("Col", _rows(columns), commit_every=4_000) == n + assert spy.count(COLP) == 7 # chunks of 4,000 rows: six full ones and one of 1,000 + assert temp_db.query("sql", "SELECT count(*) AS n FROM Col").to_list()[0]["n"] == n + assert temp_db.insert_columns("Col", columns, commit_every=1_000) == n + assert spy.count(COLP) == 8 # one crossing for the whole call + assert ( + temp_db.query("sql", "SELECT count(*) AS n FROM Col").to_list()[0]["n"] == 2 * n + ) + + +def test_a_duplicate_key_fails_the_same_way_on_both_paths(temp_db, monkeypatch): + _types(temp_db, "Ref", "Col", unique=True) + ids = list(range(250)) + ids[150] = 7 # a duplicate of an earlier row + columns = {"id": ids, "txt": ["t%d" % i for i in ids]} + with monkeypatch.context() as m: + m.setattr(bulk, "_ENABLED", False) + with pytest.raises(ArcadeDBError) as ref: + temp_db.insert_many("Ref", _rows(columns), commit_every=100) + with pytest.raises(ArcadeDBError) as col: + temp_db.insert_many("Col", _rows(columns), commit_every=100) + for error in ( + ref.value, + col.value, + ): # the unique index is checked when the batch commits + assert "DuplicatedKeyException" in str(error) and "Duplicated key [7]" in str( + error + ) + # the rows of the failed batch are rolled back, the batches before it stay + assert [ + temp_db.query("sql", _count_sql(name)).to_list()[0]["n"] + for name in ("Ref", "Col") + ] == [100, 100] + + with pytest.raises(ArcadeDBError) as one_call: + temp_db.insert_columns( + "Col", {"id": ids, "txt": columns["txt"]}, commit_every=100 + ) + assert "Failed to bulk-insert" in str(one_call.value) + + +def test_a_callers_transaction_is_not_committed_by_the_call(temp_db, spy): + _types(temp_db, "Col", unique=True) + temp_db.begin() + temp_db.insert_columns("Col", {"id": [1, 2, 3], "txt": ["a", "b", "c"]}) + assert spy.count(COLP) == 1 + assert temp_db.is_transaction_active() + temp_db.rollback() + assert temp_db.query("sql", "SELECT count(*) AS n FROM Col").to_list()[0]["n"] == 0 + + +# ---- the image checks the table it is given: a wrong table is an error, never a fault, and writes nothing + + +def _send(db, columns, mutate): + rows, packed = columnar.pack(columns, columnar.Declared(known={})) + spec = columnar.spec_of(packed) + mutate(spec, packed) + reply = lib.call( + COLP, + db._h, + lib.b("Col"), + lib.b(columnar.header_of(rows, 10_000, packed)), + spec, + len(spec), + ) + import json + + return json.loads(reply), packed + + +def _count(db): + return db.query("sql", "SELECT count(*) AS n FROM Col").to_list()[0]["n"] + + +def test_a_table_that_does_not_fit_its_buffers_is_an_error_and_writes_nothing(temp_db): + _types(temp_db, "Col") + columns = {"id": np.arange(5), "txt": ["a", "bb", "ccc", "dddd", "eeeee"]} + + reply, _ = _send( + temp_db, columns, lambda spec, p: spec.__setitem__(1, 8) + ) # id: 8 bytes for 5 rows + assert not reply["ok"] and "needed" in reply["error"] + + def decreasing(spec, packed): + offsets = (ctypes.c_int32 * 6).from_address(spec[6]) + offsets[2], offsets[3] = 9, 4 # not increasing + + reply, _ = _send(temp_db, columns, decreasing) + assert not reply["ok"] and "offsets" in reply["error"] + + def past_the_bytes(spec, packed): + spec[9] = 3 # three bytes for 15 characters of text + + reply, _ = _send(temp_db, columns, past_the_bytes) + assert not reply["ok"] and "offsets" in reply["error"] + + nulls = {"id": np.arange(16), "txt": ["a"] * 8 + [None] * 8} + reply, _ = _send( + temp_db, nulls, lambda spec, p: spec.__setitem__(11, 1) + ) # bitmap of 1 byte for 16 rows + assert not reply["ok"] and "bitmap" in reply["error"] + assert _count(temp_db) == 0 + # the connection is fine afterwards + assert temp_db.insert_columns("Col", columns) == 5 and _count(temp_db) == 5 + + +def test_an_unknown_kind_and_a_wrong_table_size_are_errors(temp_db): + _types(temp_db, "Col") + import json + + arr = np.arange(3, dtype=np.int64) + spec = (ctypes.c_int64 * 6)(arr.ctypes.data, arr.nbytes, 0, 0, 0, 0) + for header, entries in ( + ({"n": 3, "commit": 0, "cols": [["id", "z", 0]]}, 6), + ( + {"n": 3, "commit": 0, "cols": [["id", "q", 0], ["grp", "q", 0]]}, + 6, + ), # two columns, one entry + ({"n": -1, "commit": 0, "cols": [["id", "q", 0]]}, 6), + ): + reply = json.loads( + lib.call( + COLP, temp_db._h, lib.b("Col"), lib.b(json.dumps(header)), spec, entries + ) + ) + assert not reply["ok"], header + reply = json.loads( + lib.call( + COLP, + temp_db._h, + lib.b("Col"), + lib.b(json.dumps({"n": 3, "commit": 0, "cols": []})), + spec, + 5, + ) + ) + assert not reply["ok"] # 5 entries are not a multiple of 6 + assert _count(temp_db) == 0 diff --git a/tests/test_native_columnar_pack.py b/tests/test_native_columnar_pack.py new file mode 100644 index 0000000..cda0a1f --- /dev/null +++ b/tests/test_native_columnar_pack.py @@ -0,0 +1,301 @@ +"""Packing of columns for the native backend's columnar insert (``arcadedb/_native/columnar.py``). + +No native library is needed: the layouts (kinds, offsets, null bitmaps, which columns are refused) are read back from the addresses the +packer would hand to ``arcadedb_n_insert_colp``. The run-time half, which compares the stored records of the JSON and the columnar paths, +is in ``test_native_columnar_insert.py``, collected on the native backend only. +""" + +import ctypes + +import numpy as np +import pytest + +from arcadedb._native import columnar # noqa: E402 + + +def _bytes(addr_len): + addr, length = addr_len + return ctypes.string_at(addr, length) if length else b"" + + +def _strings(col): + offsets = np.frombuffer(_bytes(col.data), dtype=np.int32) + blob = _bytes(col.aux) + return [blob[offsets[i] : offsets[i + 1]].decode("utf-8") for i in range(col.n)] + + +def _present(col): + bits = np.frombuffer(_bytes(col.nulls), dtype=np.uint8) + return np.unpackbits(bits, bitorder="little")[: col.n].astype(bool).tolist() + + +NO_TYPES = columnar.Declared(known={}) + + +def _pack(columns, declared=None): + got = columnar.pack(columns, columnar.Declared(known=declared or {})) + assert got is not None + rows, cols = got + return rows, {c.name: c for c in cols} + + +def test_numpy_columns_cross_as_their_64_bit_kinds_without_a_copy(): + ids = np.arange(5, dtype=np.int64) + vals = np.linspace(0, 1, 5) + rows, cols = _pack({"id": ids, "val": vals}) + assert rows == 5 + assert (cols["id"].kind, cols["val"].kind) == ("q", "d") + assert cols["id"].data == (ids.ctypes.data, 40) # the array itself, not a copy + assert cols["val"].data == (vals.ctypes.data, 40) + assert cols["id"].nulls == (0, 0) and cols["id"].aux == (0, 0) + + +def test_narrow_numbers_widen_unless_the_property_is_declared_that_narrow(): + small = np.array([1, -2, 3], dtype=np.int32) + single = np.array([0.5, 1.5, 2.5], dtype=np.float32) + _, cols = _pack( + {"n": small, "x": single} + ) # undeclared: a Long and a Double, as the JSON path stores them + assert (cols["n"].kind, cols["x"].kind) == ("q", "d") + assert np.frombuffer(_bytes(cols["n"].data), dtype=np.int64).tolist() == [1, -2, 3] + _, cols = _pack({"n": small, "x": single}, {"n": "INTEGER", "x": "FLOAT"}) + assert (cols["n"].kind, cols["x"].kind) == ("i", "e") + assert cols["n"].data == (small.ctypes.data, 12) and cols["x"].data == ( + single.ctypes.data, + 12, + ) + _, cols = _pack({"n": small.astype(np.int64)}, {"n": "INTEGER"}) + assert ( + cols["n"].kind == "q" + ) # a 64-bit source is range-checked by the engine, as before + + +def test_unsigned_past_the_signed_range_is_refused_and_below_it_is_not(): + assert columnar.pack({"u": np.array([2**63], dtype=np.uint64)}, NO_TYPES) is None + _, cols = _pack({"u": np.array([2**63 - 1, 1], dtype=np.uint64)}) + assert np.frombuffer(_bytes(cols["u"].data), dtype=np.int64).tolist() == [ + 2**63 - 1, + 1, + ] + + +def test_bool_and_nan_and_infinity_keep_their_values(): + _, cols = _pack( + {"f": np.array([True, False, True]), "x": np.array([np.nan, np.inf, -np.inf])} + ) + assert cols["f"].kind == "b" and _bytes(cols["f"].data) == b"\x01\x00\x01" + x = np.frombuffer(_bytes(cols["x"].data), dtype=np.float64) + assert np.isnan(x[0]) and x[1] == np.inf and x[2] == -np.inf + + +def test_a_float32_matrix_and_a_list_of_float32_arrays_are_vector_columns(): + m = np.arange(6, dtype=np.float32).reshape(3, 2) + _, cols = _pack({"v": m, "w": [row for row in m]}) + for name in ("v", "w"): + assert (cols[name].kind, cols[name].dim, cols[name].n) == ("f", 2, 3) + assert np.frombuffer( + _bytes(cols[name].data), dtype=np.float32 + ).tolist() == list(range(6)) + + +def test_float64_arrays_are_not_vector_columns(): + # only float32 is a column of vectors, as on the JPype path (tests/test_float_vector_bulk.py): float64 keeps the row path + m = np.arange(6, dtype=np.float64).reshape(3, 2) + assert columnar.pack({"v": m}, NO_TYPES) is None + 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(): + 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 ( + columnar.pack( + {"v": [[], []]}, columnar.Declared(known={"v": "ARRAY_OF_FLOATS"}) + ) + is None + ) # no dimension + + +def test_strings_are_arrow_offsets_plus_utf8_bytes(): + texts = ["", "abc", "héllo", "日本語", "😀", "x\x00y"] + rows, cols = _pack({"s": texts}) + assert rows == 6 and cols["s"].kind == "s" + assert _strings(cols["s"]) == texts + offsets = np.frombuffer(_bytes(cols["s"].data), dtype=np.int32).tolist() + assert ( + offsets[0] == 0 + and offsets == sorted(offsets) + and offsets[-1] == len(_bytes(cols["s"].aux)) + ) + + +def test_ascii_and_all_empty_strings(): + _, cols = _pack({"a": ["a", "bb", "ccc"], "e": ["", "", ""]}) + assert _strings(cols["a"]) == ["a", "bb", "ccc"] + assert _strings(cols["e"]) == ["", "", ""] and cols["e"].aux == (0, 0) + + +def test_a_lone_surrogate_is_refused_so_the_json_path_decides(): + assert columnar.pack({"s": ["ok", "\ud800"]}, NO_TYPES) is None + + +def test_nulls_become_a_bitmap_and_a_filler_value(): + rows, cols = _pack( + { + "n": [1, None, 3, None], + "x": [None, 2.5, None, 4.5], + "s": ["a", None, "c", None], + "b": [True, None, False, True], + "z": [None, None, None, None], + } + ) + assert rows == 4 + assert _present(cols["n"]) == [True, False, True, False] + assert _present(cols["s"]) == [True, False, True, False] + assert _present(cols["x"]) == [False, True, False, True] + assert _present(cols["b"]) == [True, False, True, True] + assert _present(cols["z"]) == [False] * 4 and cols["z"].kind == "s" + assert _strings(cols["s"])[0] == "a" and _strings(cols["s"])[2] == "c" + assert np.frombuffer(_bytes(cols["n"].data), dtype=np.int64)[[0, 2]].tolist() == [ + 1, + 3, + ] + assert len(_bytes(cols["n"].nulls)) == 1 + + +def test_a_bitmap_longer_than_a_byte(): + values = [None if i % 3 == 0 else i for i in range(20)] + _, cols = _pack({"n": values}) + assert _present(cols["n"]) == [v is not None for v in values] + assert len(_bytes(cols["n"].nulls)) == 3 + + +@pytest.mark.parametrize( + "column", + [ + [1, 2.5], # mixed int and float + [1, "a"], + [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 + ], +) +def test_columns_the_buffer_cannot_carry_unchanged_are_refused(column): + assert columnar.pack({"c": column}, NO_TYPES) is None + + +def test_datetimes_decimals_and_object_arrays_of_them_are_refused(): + import datetime + import decimal + + assert columnar.pack({"d": [datetime.datetime(2020, 1, 1)]}, NO_TYPES) is None + assert columnar.pack({"d": [decimal.Decimal("1.5")]}, NO_TYPES) is None + assert ( + columnar.pack({"d": np.array(["2020-01-01"], dtype="datetime64[D]")}, NO_TYPES) + is None + ) + + +def test_ragged_columns_and_empty_input_are_left_to_the_row_path(): + assert columnar.pack({"a": [1, 2, 3], "b": [1, 2]}, NO_TYPES) is None + assert columnar.pack({"a": []}, NO_TYPES) is None + assert columnar.pack({}, NO_TYPES) is None + assert columnar.pack({"a": (x for x in range(3))}, NO_TYPES) is None + + +def test_the_address_table_and_header_describe_the_columns(): + rows, packed = columnar.pack({"id": np.arange(3), "s": ["a", "bc", None]}, NO_TYPES) + spec = columnar.spec_of(packed) + assert len(spec) == 12 # six int64 per column + assert spec[0] == packed[0].data[0] and spec[1] == 24 + assert spec[6] == packed[1].data[0] and spec[7] == 16 # n + 1 offsets of 4 bytes + assert spec[8] == packed[1].aux[0] and spec[9] == 3 + assert spec[10] == packed[1].nulls[0] and spec[11] == 1 + import json + + assert json.loads(columnar.header_of(rows, 10_000, packed)) == { + "n": 3, + "commit": 10_000, + "cols": [["id", "q", 0], ["s", "s", 0]], + } + assert json.loads(columnar.header_of(rows, 2**40, packed))["commit"] == 2**31 - 1 + + +def test_rows_become_columns_only_when_every_row_has_the_same_keys(): + rows = [{"a": 1, "b": "x"}, {"a": 2, "b": "y"}] + assert columnar.rows_to_columns(rows) == {"a": [1, 2], "b": ["x", "y"]} + assert columnar.rows_to_columns([{"a": 1}, {"a": 2, "b": 3}]) is None + assert ( + columnar.rows_to_columns([{"a": 1, "b": 2}, {"a": 1, "c": 2}]) is None + ) # same size, other key + assert columnar.rows_to_columns([{1: "x"}]) is None + assert columnar.rows_to_columns([]) is None + assert columnar.rows_to_columns([[1, 2]]) is None + + +def test_arrow_strings_cross_in_place_with_their_offset_and_validity(): + import pyarrow as pa + + arr = pa.array(["a", None, "héllo", "", "xyz", "tail"]) + rows, cols = _pack({"s": arr}) + assert rows == 6 and cols["s"].kind == "s" + assert _present(cols["s"]) == [True, False, True, True, True, True] + assert [t for t, ok in zip(_strings(cols["s"]), _present(cols["s"])) if ok] == [ + "a", + "héllo", + "", + "xyz", + "tail", + ] + # a slice keeps the buffers and moves the offsets: the offsets no longer start at 0 + part = arr.slice(2, 3) + _, cols = _pack({"s": part}) + assert _strings(cols["s"]) == ["héllo", "", "xyz"] + assert int(np.frombuffer(_bytes(cols["s"].data), dtype=np.int32)[0]) > 0 + big = pa.array(["a", "bc"], type=pa.large_string()) + _, cols = _pack({"s": big}) + assert _strings(cols["s"]) == ["a", "bc"] + + +def test_arrow_numbers_with_nulls_and_arrow_vectors(): + import pyarrow as pa + + _, cols = _pack( + { + "n": pa.array([1, None, 3], type=pa.int64()), + "b": pa.array([True, None, False]), + } + ) + assert _present(cols["n"]) == [True, False, True] and cols["n"].kind == "q" + assert cols["b"].kind == "b" and _present(cols["b"]) == [True, False, True] + vec = pa.array([[1.0, 2.0], [3.0, 4.0]], type=pa.list_(pa.float32())) + assert columnar.pack({"v": vec}, NO_TYPES) is None + _, cols = _pack({"v": vec}, {"v": "ARRAY_OF_FLOATS"}) + assert (cols["v"].kind, cols["v"].dim) == ("f", 2) + fixed = pa.array([[1.0, 2.0], [3.0, 4.0]], type=pa.list_(pa.float32(), 2)) + _, cols = _pack({"v": fixed}, {"v": "ARRAY_OF_FLOATS"}) + assert np.frombuffer(_bytes(cols["v"].data), dtype=np.float32).tolist() == [ + 1.0, + 2.0, + 3.0, + 4.0, + ] + ragged = pa.array([[1.0], [1.0, 2.0]], type=pa.list_(pa.float32())) + assert ( + columnar.pack({"v": ragged}, columnar.Declared(known={"v": "ARRAY_OF_FLOATS"})) + is None + )