From cb678b553f749c467cd82e2e8c3c35aa01fff499 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 18:53:19 +0900 Subject: [PATCH 1/8] Native insert_columns(parallel=True): take the columnar entry (closes #325) A parallel call skipped the columnar fast path and went through one dict per row and one JSON text per chunk of 10,000 rows. The columnar entry arcadedb_n_insert_colp now takes "parallel": true in its header: the image builds the records from the column buffers and hands them to the async writers (db.async().createRecord with a failure token), as DocumentBatcher.insertColumnsParallel does on the JVM wheel. The Python side waits for the writers and reads the failures exactly as insert_many(parallel=True) does (_finish_parallel, shared). Rows the buffers cannot carry still take the row path. The native library is rebuilt from the unchanged engine commit. Tests (native only): a spy proves one colp crossing and no insert_parallel chunk; the stored records equal those of insert_many(parallel=True); nulls, vertex types, rejected records, an edge type, a missing type, the fall back and the off switch. Co-Authored-By: Claude Sonnet 5.5 --- .../org/humemai/nativelib/ArcadeNative.java | 5 +- .../java/org/humemai/nativelib/BulkOps.java | 117 ++++++---- .../org/humemai/nativelib/NativeExtras.java | 37 +++- src/arcadedb/_native/bulk.py | 27 ++- src/arcadedb/_native/columnar.py | 21 +- src/arcadedb/_native/database.py | 24 +- tests/conftest.py | 2 + tests/test_native_columnar_pack.py | 5 + tests/test_native_columnar_parallel.py | 208 ++++++++++++++++++ 9 files changed, 378 insertions(+), 68 deletions(-) create mode 100644 tests/test_native_columnar_parallel.py diff --git a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java index 9b5acb91..1703b965 100644 --- a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java +++ b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java @@ -561,7 +561,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. - * The caller keeps every buffer alive and unchanged until the call returns. Returns {"ok":true,"n":rows}. + * The caller keeps every buffer alive and unchanged until the call returns. Returns {"ok":true,"n":rows}; with "parallel":true in the + * header the records go to the async executor and the reply also carries "fid" (see BulkOps.insertColumnsReply). */ @CEntryPoint(name = "arcadedb_n_insert_colp") public static CCharPointer insertColp(final IsolateThread thread, final long h, final CCharPointer type, final CCharPointer header, @@ -572,7 +573,7 @@ public static CCharPointer insertColp(final IsolateThread thread, final long h, 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) + "}"); + return out(BulkOps.insertColumnsReply(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 37cedf90..40f789e0 100644 --- a/native/src/main/java/org/humemai/nativelib/BulkOps.java +++ b/native/src/main/java/org/humemai/nativelib/BulkOps.java @@ -10,6 +10,9 @@ import com.arcadedb.database.Database; import com.arcadedb.database.MutableDocument; +import com.arcadedb.database.async.DatabaseAsyncExecutor; +import com.arcadedb.database.async.ErrorCallback; +import com.arcadedb.schema.EdgeType; import com.arcadedb.schema.VertexType; import org.graalvm.nativeimage.c.type.CTypeConversion; @@ -55,13 +58,27 @@ static long insertColumns(final Database db, final String type, final String hea spec[SLOTS * c + 1] = bytes; pos += bytes; } - return insertColumns(db, type, h, spec); + return insertColumns(db, type, h, spec, null); } - /** The pointer form (arcadedb_n_insert_colp): the table holds SLOTS longs per column, see {@link #insertColumns(Database, String, Map, long[])}. */ + /** + * The pointer form (arcadedb_n_insert_colp): the table holds SLOTS longs per column, see {@link #insertColumns(Database, String, Map, long[], ErrorCallback)}. + * Returns the reply text: {"ok":true,"n":rows}, and for a header with "parallel":true also "fid", the token that the "failures" operation of + * arcadedb_n_async reads (see there). In that mode the records are handed to the database's async executor, the call does not wait for + * the writers, and it neither opens, commits nor rolls back a transaction ("commit" does not apply), exactly as insert_parallel does for rows. + */ @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); + static String insertColumnsReply(final Database db, final String type, final String header, final long[] spec) { + final Map h = (Map) Json.parse(header); + if (!Boolean.TRUE.equals(h.get("parallel"))) + return "{\"ok\":true,\"n\":" + insertColumns(db, type, h, spec, null) + "}"; + final NativeExtras.FailureToken token = new NativeExtras.FailureToken(); + try { + return "{\"ok\":true,\"n\":" + insertColumns(db, type, h, spec, token.onError) + ",\"fid\":" + token.fid + "}"; + } catch (final Throwable t) { + token.drop(); // the caller gets the error, not a token to read failures from + throw t; + } } /** @@ -72,9 +89,13 @@ static long insertColumns(final Database db, final String type, final String hea * 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. + * onError null: the synchronous mode above. Not null (header "parallel":true): every record goes to db.async().createRecord with that + * callback and no transaction is touched, as DocumentBatcher.insertColumnsParallel does; an edge type is refused there, as it is by + * insert_parallel. The columns are validated before the first record is handed over in both modes. */ @SuppressWarnings("unchecked") - private static long insertColumns(final Database db, final String type, final Map h, final long[] spec) { + private static long insertColumns(final Database db, final String type, final Map h, final long[] spec, + final ErrorCallback onError) { final long nRows = ((Number) h.get("n")).longValue(); if (nRows < 0 || nRows > Integer.MAX_VALUE) throw new IllegalArgumentException("row count out of range: " + nRows); @@ -137,45 +158,27 @@ private static long insertColumns(final Database db, final String type, final Ma } } final byte[] scratch = new byte[longest]; - final boolean vertex = db.getSchema().getType(type) instanceof VertexType; + final Object schemaType = db.getSchema().getType(type); + final boolean vertex = schemaType instanceof VertexType; + if (onError != null) { + if (schemaType instanceof EdgeType) + throw new IllegalArgumentException("Cannot bulk-insert into '" + type + + "' because it is an edge type; load the vertices, then create the edges (graph_batch().new_edges() or CREATE EDGE)"); + final DatabaseAsyncExecutor async = db.async(); + for (int i = 0; i < n; i++) { + final MutableDocument doc = vertex ? db.newVertex(type) : db.newDocument(type); + setRow(doc, i, names, kinds, dims, data, nulls, blobs, scratch); + async.createRecord(doc, null, onError); + } + return n; + } final boolean wasActive = db.isTransactionActive(); if (!wasActive) db.begin(); try { 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], 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 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); - } - } - } + setRow(doc, i, names, kinds, dims, data, nulls, blobs, scratch); doc.save(); if (!wasActive && commitEvery > 0 && (i + 1) % commitEvery == 0) { db.commit(); @@ -197,6 +200,44 @@ private static long insertColumns(final Database db, final String type, final Ma return n; } + /** Sets the properties of row i on doc from the columns: the one place that reads a value out of a buffer. */ + private static void setRow(final MutableDocument doc, final int i, final String[] names, final char[] kinds, final int[] dims, + final long[] data, final long[] nulls, final ByteBuffer[] blobs, final byte[] scratch) { + final int k = names.length; + 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], 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 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); + } + } + } + } + /** * 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/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java index a62574ad..c0e87e87 100644 --- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java +++ b/native/src/main/java/org/humemai/nativelib/NativeExtras.java @@ -527,6 +527,28 @@ else if (opName.equals("create_vertices")) private static final Map ASYNC_FAILS = new ConcurrentHashMap<>(); private static final Map> ASYNC_FIRST = new ConcurrentHashMap<>(); + /** A failure token for one parallel load: the id the caller reads with the "failures" operation, and the callback that counts. */ + static final class FailureToken { + final long fid = ArcadeNative.IDS.getAndIncrement(); + final AtomicLong count = new AtomicLong(); + final AtomicReference first = new AtomicReference<>(); + final ErrorCallback onError = t -> { + count.incrementAndGet(); + first.compareAndSet(null, t); + }; + + FailureToken() { + ASYNC_FAILS.put(fid, count); + ASYNC_FIRST.put(fid, first); + } + + /** The load failed before it returned: the caller gets the error, not a token to read failures from. */ + void drop() { + ASYNC_FAILS.remove(fid); + ASYNC_FIRST.remove(fid); + } + } + private static String replyRows(final ResultSet rs) { final StringBuilder sb = new StringBuilder("{\"ok\":true,\"rs\":0,"); NativeCodec.appendBatch(sb, rs, Integer.MAX_VALUE); @@ -824,15 +846,9 @@ else if (a.get("slot") instanceof Number slot) ex.appendSamples((String) a.get("type"), stamps, columns); } case "insert_parallel" -> { - final long fid = ArcadeNative.IDS.getAndIncrement(); - final AtomicLong count = new AtomicLong(); - final AtomicReference first = new AtomicReference<>(); - ASYNC_FAILS.put(fid, count); - ASYNC_FIRST.put(fid, first); - final ErrorCallback onError = t -> { - count.incrementAndGet(); - first.compareAndSet(null, t); - }; + final FailureToken token = new FailureToken(); + final long fid = token.fid; + final ErrorCallback onError = token.onError; final String type = (String) a.get("type"); try { final List rows = (List) a.get("rows"); @@ -846,8 +862,7 @@ else if (a.get("slot") instanceof Number slot) ex.createRecord(doc, null, onError); } } catch (final Throwable t) { - ASYNC_FAILS.remove(fid); // the caller gets the error, not a token to read failures from - ASYNC_FIRST.remove(fid); + token.drop(); // the caller gets the error, not a token to read failures from throw t; } return ArcadeNative.out("{\"ok\":true,\"fid\":" + fid + "}"); diff --git a/src/arcadedb/_native/bulk.py b/src/arcadedb/_native/bulk.py index d244c48c..dcaca257 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; with ``parallel`` they are handed to the async writers instead of a transaction +(``DocumentBatcher.insertColumnsParallel``). ``ARCADEDB_NATIVE_BULK=0`` turns the columnar path off. """ import os @@ -21,20 +22,19 @@ def _usable(): ) -def _send(db, type_name, packed, rows, commit_every, prefix): +def _send(db, type_name, packed, rows, commit_every, prefix, parallel=False): from .database import _call spec = columnar.spec_of(packed) - reply = _call( + return _call( prefix, "arcadedb_n_insert_colp", db._h, lib.b(type_name), - lib.b(columnar.header_of(rows, commit_every, packed)), + lib.b(columnar.header_of(rows, commit_every, packed, parallel)), spec, len(spec), ) # `packed` holds the arrays the addresses point into until the call returns - return int(reply["n"]) def insert_columns_fast( @@ -51,7 +51,22 @@ def insert_columns_fast( if packed is None: return None rows, cols = packed - return _send(db, type_name, cols, rows, commit_every, prefix) + return int(_send(db, type_name, cols, rows, commit_every, prefix)["n"]) + + +def insert_columns_parallel(db, type_name, columns, declared=None): + """Hand ``columns`` (name -> column) to the async writers in one crossing: ``(rows, failure token)``, or None when a column cannot be + sent this way. The call does not wait for the writers: the caller waits for them and reads the failures for the token + (``Database._finish_parallel``), as for ``insert_many(parallel=True)``. + """ + if not _usable(): + return None + packed = columnar.pack(columns, columnar.Declared(db, type_name, declared)) + if packed is None: + return None + rows, cols = packed + reply = _send(db, type_name, cols, rows, 0, "Failed to bulk-insert", parallel=True) + return int(reply["n"]), reply["fid"] def insert_rows_columnar(db, type_name, chunk): diff --git a/src/arcadedb/_native/columnar.py b/src/arcadedb/_native/columnar.py index f288043c..fe62837a 100644 --- a/src/arcadedb/_native/columnar.py +++ b/src/arcadedb/_native/columnar.py @@ -343,17 +343,20 @@ def spec_of(packed): 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.""" +def header_of(rows, commit_every, packed, parallel=False): + """The JSON header text: row count, commit interval and ``[name, kind, dim]`` per column; ``"parallel": true`` hands the records to the + async executor (the image then ignores the commit interval and the reply carries the failure token). + """ 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], - } - ) + header = { + "n": rows, + "commit": min(int(commit_every), 2**31 - 1), + "cols": [[c.name, c.kind, c.dim] for c in packed], + } + if parallel: + header["parallel"] = True + return json.dumps(header) def rows_to_columns(rows): diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index f30dc526..b0c07932 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -1040,6 +1040,12 @@ def _insert_many_parallel(self, type_name, rows) -> int: except Exception: # nosec B110 - the original error is the one to report pass raise + return self._finish_parallel(type_name, tokens, n) + + def _finish_parallel(self, type_name, tokens, n) -> int: + """The end of a parallel load: wait for the async writers, then read the failure count of every token and raise when a record failed.""" + from . import extras + try: self._async().wait_completion() except Exception as e: @@ -1314,7 +1320,11 @@ def export_to_csv( def insert_columns( self, type_name, columns, commit_every: int = 10_000, parallel: bool = False ) -> int: - """Bulk insert from whole columns (dict of name -> numpy array or list), as rows through :meth:`insert_many`.""" + """Bulk insert from whole columns (dict of name -> numpy array or list): columns the buffers can carry cross in one call + (``arcadedb_n_insert_colp``), any other column set goes as rows through :meth:`insert_many`. With ``parallel=True`` the + same one call hands the records to the async writers (no transaction, ``commit_every`` does not apply, the call waits for the + writers and raises when a record failed), as ``insert_many(parallel=True)`` does for rows. + """ self._check_not_closed() items = list(columns.items()) if hasattr(columns, "items") else None if not items: @@ -1330,7 +1340,17 @@ def insert_columns( known = {n: len(v) for n, v in items if hasattr(v, "__len__")} if len(set(known.values())) > 1: raise ValueError(f"columns differ in length: {known}") - if not parallel: + if parallel: + from .bulk import insert_columns_parallel + + try: + sent = insert_columns_parallel(self, type_name, columns) + except ArcadeDBError as e: + # as insert_many(parallel=True): rows handed to the writers before the failure may be stored + raise self._parallel_load_failed(type_name, e) from e + if sent is not None: + return self._finish_parallel(type_name, [sent[1]], sent[0]) + else: from .bulk import insert_columns_fast done = insert_columns_fast(self, type_name, columns, commit_every) diff --git a/tests/conftest.py b/tests/conftest.py index ec3147c7..be5fa583 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -47,6 +47,8 @@ collect_ignore.append("test_native_vector_hit_lazy.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") + # insert_columns(parallel=True) takes the columnar entry and stores what the row path stores (arcadedb_n_insert_colp with "parallel") + collect_ignore.append("test_native_columnar_parallel.py") # the rows of an insert_many chunk with date-times, Decimals, bytes and numpy scalars are told apart by the entry points they cross collect_ignore.append("test_native_mixed_chunk.py") # the native build's ingest defaults (insert_many and graph_batch commit size), observed on the native library diff --git a/tests/test_native_columnar_pack.py b/tests/test_native_columnar_pack.py index cda0a1f7..56b283e1 100644 --- a/tests/test_native_columnar_pack.py +++ b/tests/test_native_columnar_pack.py @@ -233,6 +233,11 @@ def test_the_address_table_and_header_describe_the_columns(): "cols": [["id", "q", 0], ["s", "s", 0]], } assert json.loads(columnar.header_of(rows, 2**40, packed))["commit"] == 2**31 - 1 + # "parallel" is in the header only when asked for (the image hands the records to the async writers then) + assert json.loads(columnar.header_of(rows, 10_000, packed, parallel=True))[ + "parallel" + ] + assert "parallel" not in json.loads(columnar.header_of(rows, 10_000, packed)) def test_rows_become_columns_only_when_every_row_has_the_same_keys(): diff --git a/tests/test_native_columnar_parallel.py b/tests/test_native_columnar_parallel.py new file mode 100644 index 00000000..f51fd7c7 --- /dev/null +++ b/tests/test_native_columnar_parallel.py @@ -0,0 +1,208 @@ +"""Native ``insert_columns(parallel=True)`` takes the columnar entry and stores what the row path of ``insert_many(parallel=True)`` stores. + +Before, a parallel call skipped the columnar fast path: the columns became one dict per row, then one JSON text per chunk of 10,000 +rows for ``arcadedb_n_async insert_parallel``. Now the same ``arcadedb_n_insert_colp`` crossing that the synchronous call uses carries +``"parallel": true`` and the image hands the records to the async writers, as ``DocumentBatcher.insertColumnsParallel`` does on the +JVM wheel. Needs the native library, so conftest.py collects this file on the native backend only. Each test loads the same data +through the row path (the private reference: ``insert_many(parallel=True)`` with dict rows) and through the columns, compares the +stored records value by value and type by type, and spies on the entry points so a silent fall back cannot pass. +""" + +import numpy as np +import pytest + +from arcadedb._native import bulk, extras, lib # noqa: E402 +from arcadedb.exceptions import ArcadeDBError # noqa: E402 + +COLP, ROWS = "arcadedb_n_insert_colp", "arcadedb_n_insert_rows" + + +class Spy: + """Counts the crossings by name and the calls that hand a chunk of dict rows to the parallel writers.""" + + def __init__(self, monkeypatch): + self.calls, self.row_chunks = [], 0 + real_call, real_rows = lib.call, extras.insert_parallel + + def call(name, *args): + self.calls.append(name) + return real_call(name, *args) + + def insert_parallel(database, type_name, rows): + self.row_chunks += 1 + return real_rows(database, type_name, rows) + + monkeypatch.setattr(lib, "call", call) + monkeypatch.setattr(extras, "insert_parallel", insert_parallel) + + def count(self, name): + return self.calls.count(name) + + +@pytest.fixture +def spy(monkeypatch): + return Spy(monkeypatch) + + +def _count(db, name): + return db.query("sql", f"SELECT count(*) AS n FROM {name}").to_list()[0][ + "n" + ] # nosec B608 - test-owned type name + + +def _types(db, *names, kind="DOCUMENT", buckets=3, unique=False): + for name in names: + db.command("sql", f"CREATE {kind} TYPE {name} BUCKETS {buckets}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + db.command("sql", f"CREATE PROPERTY {name}.val DOUBLE") + db.command("sql", f"CREATE PROPERTY {name}.flag BOOLEAN") + 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 (the writers store in no fixed order).""" + sql = f"SELECT FROM {name} ORDER BY id" # nosec B608 - test-owned type name + return [ + {k: (v, type(v).__name__) for k, v in row.items() if not k.startswith("@")} + for row in db.query("sql", sql).to_list() + ] + + +def _columns(n, dim=4, seed=11): + rng = np.random.default_rng(seed) + ids = np.arange(n, dtype=np.int64) + return { + "id": ids, + "val": (ids * 37 % 10007) / 7.0, + "flag": ids % 3 == 0, + "txt": ["row-%d-%s" % (k, "é" * (k % 5)) 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 test_a_parallel_call_uses_the_columnar_entry_and_stores_the_row_paths_records( + temp_db, spy +): + _types(temp_db, "Ref", "Col") + columns = _columns(3_100) + assert temp_db.insert_many("Ref", _rows(columns), parallel=True) == 3_100 + assert spy.row_chunks == 1 and spy.count(COLP) == 0 # the reference is the row path + + spy.calls.clear() + spy.row_chunks = 0 + assert temp_db.insert_columns("Col", columns, parallel=True) == 3_100 + assert spy.count(COLP) == 1, spy.calls # one crossing for the whole call + assert spy.row_chunks == 0 and spy.count(ROWS) == 0 # no dict rows, no JSON text + + expected = _stored(temp_db, "Ref") + assert len(expected) == 3_100 and _count(temp_db, "Col") == 3_100 + assert expected[7]["id"][1] == "int" and expected[7]["vec"][1] == "list" + assert _stored(temp_db, "Col") == expected + + +def test_nulls_and_awkward_strings_are_stored_as_the_row_path_stores_them(temp_db, spy): + _types(temp_db, "Ref", "Col") + columns = { + "id": list(range(8)), + "txt": ["", "a", None, "snow☃man", "x" * 300, None, "\n\t", "z"], + "val": [1.5, None, 2.5, 1e300, -1e-300, 0.0, 3.0, None], + } + temp_db.insert_many("Ref", _rows(columns), parallel=True) + spy.calls.clear() + temp_db.insert_columns("Col", columns, parallel=True) + assert spy.count(COLP) == 1 + got, want = _stored(temp_db, "Col"), _stored(temp_db, "Ref") + assert got == want and got[2]["txt"][0] is None + + +def test_a_vertex_type_gets_vertices_in_parallel(temp_db, spy): + _types(temp_db, "Ref", "Col", kind="VERTEX") + columns = _columns(1_500) + temp_db.insert_many("Ref", _rows(columns), parallel=True) + spy.calls.clear() + temp_db.insert_columns("Col", columns, parallel=True) + assert spy.count(COLP) == 1 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + kinds = temp_db.query( + "sql", "SELECT FROM schema:types WHERE name = 'Col'" + ).to_list() + assert kinds[0]["type"] == "vertex" + + +def test_a_record_the_writers_reject_is_reported_after_the_wait(temp_db, spy): + _types(temp_db, "Col", unique=True) + ids = np.arange(400, dtype=np.int64) + ids[300] = ids[7] # a duplicate key + with pytest.raises(ArcadeDBError, match="failed record"): + temp_db.insert_columns("Col", {"id": ids, "txt": ["t"] * 400}, parallel=True) + assert spy.count(COLP) == 1 and spy.row_chunks == 0 + stored = _count(temp_db, "Col") + assert 0 < stored <= 400 # the call had waited for the writers: nothing lands later + + +def test_a_value_the_declared_type_refuses_raises_and_the_call_is_final(temp_db, spy): + _types(temp_db, "Col") + values = [str(i) for i in range(1_000)] + values[700] = "not a number" + with pytest.raises( + ArcadeDBError, match="may have been stored|failed record" + ) as err: + temp_db.insert_columns("Col", {"id": values}, parallel=True) + assert "rolled back" not in str(err.value) + assert spy.count(COLP) == 1 + stored = _count(temp_db, "Col") + temp_db.async_executor().wait_completion() + assert _count(temp_db, "Col") == stored <= 1_000 + + +def test_an_edge_type_is_refused_as_before(temp_db): + temp_db.command("sql", "CREATE EDGE TYPE Link") + with pytest.raises(ArcadeDBError, match="edge type"): + temp_db.insert_columns("Link", {"id": [1, 2, 3]}, parallel=True) + + +def test_a_missing_type_is_the_schemas_error(temp_db): + with pytest.raises(ArcadeDBError, match="NoSuchType"): + temp_db.insert_columns("NoSuchType", {"id": [1, 2, 3]}, parallel=True) + + +def test_a_column_the_buffers_cannot_carry_still_goes_through_the_row_path( + temp_db, spy +): + _types(temp_db, "Col") + columns = {"id": [1, 2, 3, 4], "txt": ["a", 2, "c", 4.5]} # a mixed-type column + assert temp_db.insert_columns("Col", columns, parallel=True) == 4 + assert spy.count(COLP) == 0 and spy.row_chunks == 1 + assert _count(temp_db, "Col") == 4 + + +def test_the_switch_that_turns_the_columnar_path_off_applies_to_parallel( + temp_db, spy, monkeypatch +): + _types(temp_db, "Col") + monkeypatch.setattr(bulk, "_ENABLED", False) + assert temp_db.insert_columns("Col", _columns(50), parallel=True) == 50 + assert spy.count(COLP) == 0 and spy.row_chunks == 1 + assert _count(temp_db, "Col") == 50 + + +def test_no_rows_is_zero_and_writes_nothing(temp_db, spy): + _types(temp_db, "Col") + assert temp_db.insert_columns("Col", {"id": []}, parallel=True) == 0 + assert spy.count(COLP) == 0 and _count(temp_db, "Col") == 0 + + +def test_the_synchronous_call_is_unchanged(temp_db, spy): + _types(temp_db, "Col", unique=True) + assert temp_db.insert_columns("Col", _columns(500), commit_every=100) == 500 + assert spy.count(COLP) == 1 and spy.row_chunks == 0 + assert _count(temp_db, "Col") == 500 From 25bceaa8d7dce93cf57d438a010031b0344df502 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 18:56:41 +0900 Subject: [PATCH 2/8] Native insert_arrow and insert_dataframe with parallel=True: the same columnar entry (refs #325) A document load with parallel=True left the Arrow columns as Python values and went to insert_many(parallel=True). It now keeps the columns as they are for the columnar entry, which hands the records to the async writers; a batch the buffers cannot carry still takes the row path. A test spies on the entry points and compares the stored records with the row path of the old code. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/dataframe.py | 19 +++++++++++--- tests/test_native_columnar_parallel.py | 35 ++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/src/arcadedb/dataframe.py b/src/arcadedb/dataframe.py index 9d3ee0b7..d65e51b8 100644 --- a/src/arcadedb/dataframe.py +++ b/src/arcadedb/dataframe.py @@ -434,7 +434,7 @@ def insert_table( batch.column(i), pa, native, - arrow=native and not is_vertex and not parallel, + arrow=native and not is_vertex, ) for i, name in enumerate(names) } @@ -497,9 +497,22 @@ def _insert_documents_native( """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. + the batch is one transaction, or joins the caller's. With ``parallel`` the same columnar entry hands the records to the async + writers (no transaction), and the call waits for them and raises when a record failed, as ``insert_many(parallel=True)`` does. """ - if not parallel: + if parallel: + from ._native.bulk import insert_columns_parallel + + try: + sent = insert_columns_parallel( + db, type_name, columns, {} if declared is None else declared + ) + except ArcadeDBError as e: + raise db._parallel_load_failed(type_name, e) from e + if sent is not None: + db._finish_parallel(type_name, [sent[1]], sent[0]) + return + else: from ._native.bulk import insert_columns_fast if ( diff --git a/tests/test_native_columnar_parallel.py b/tests/test_native_columnar_parallel.py index f51fd7c7..0f5d80d9 100644 --- a/tests/test_native_columnar_parallel.py +++ b/tests/test_native_columnar_parallel.py @@ -206,3 +206,38 @@ def test_the_synchronous_call_is_unchanged(temp_db, spy): assert temp_db.insert_columns("Col", _columns(500), commit_every=100) == 500 assert spy.count(COLP) == 1 and spy.row_chunks == 0 assert _count(temp_db, "Col") == 500 + + +def test_insert_arrow_parallel_takes_the_columnar_entry(temp_db, spy, monkeypatch): + import pyarrow as pa + + _types(temp_db, "Ref", "ByTable") + columns = _columns(2_000) + table = pa.table( + { + "id": columns["id"], + "val": pa.array( + [None if i % 7 == 0 else v for i, v in enumerate(columns["val"])] + ), + "txt": pa.array(columns["txt"]), + } + ) + with ( + monkeypatch.context() as m + ): # the private reference: the row path of the old code + m.setattr(bulk, "_ENABLED", False) + assert ( + temp_db.insert_arrow("Ref", table, batch_size=700, parallel=True) == 2_000 + ) + assert spy.count(COLP) == 0 and spy.row_chunks >= 1 + + spy.calls.clear() + spy.row_chunks = 0 + assert ( + temp_db.insert_arrow("ByTable", table, batch_size=700, parallel=True) == 2_000 + ) + assert spy.count(COLP) == 3 and spy.row_chunks == 0 # one crossing per batch + + expected = _stored(temp_db, "Ref") + assert len(expected) == 2_000 and expected[0]["val"][0] is None + assert _stored(temp_db, "ByTable") == expected From 8d4ec4a4e5f8e0f49efed8faee8db31c1a53b838 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 19:07:55 +0900 Subject: [PATCH 3/8] tests: nosec B608 on the test-owned count query, which bandit flagged after black wrapped it Co-Authored-By: Claude Sonnet 5.5 --- tests/test_native_columnar_parallel.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/test_native_columnar_parallel.py b/tests/test_native_columnar_parallel.py index 0f5d80d9..90d6f79a 100644 --- a/tests/test_native_columnar_parallel.py +++ b/tests/test_native_columnar_parallel.py @@ -45,9 +45,8 @@ def spy(monkeypatch): def _count(db, name): - return db.query("sql", f"SELECT count(*) AS n FROM {name}").to_list()[0][ - "n" - ] # nosec B608 - test-owned type name + sql = f"SELECT count(*) AS n FROM {name}" # nosec B608 - test-owned type name + return db.query("sql", sql).to_list()[0]["n"] def _types(db, *names, kind="DOCUMENT", buckets=3, unique=False): From c24f154cfb3619dec5e130fb670ea160626082e1 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 21:21:37 +0900 Subject: [PATCH 4/8] Native parallel columnar load: interrupt-safe, token-safe, no silent loss (review of #328) The failure token is now made by the caller (arcadedb_n_async "token") and its id travels in the colp header, so the caller holds the id whatever happens to the crossing. An interrupt in it (or between the chunks of the row path) waits for the writers and reads the tokens before it propagates; a wait that raises still reads every token (the entries of an unread token lived forever); the error of a failed handover says how many rows were handed over and that some may have been stored. insert_table(parallel=True) no longer says the earlier batches were written in the caller's transaction (the writers commit them, a rollback does not remove them). Tests: duplicate keys (stored + reported >= n, stored == distinct), the writers' batch boundary, nothing lands after the call, an interrupt in the crossing and in the row path, a failing wait, the in-transaction message. Co-Authored-By: Claude Sonnet 5.5 --- .../java/org/humemai/nativelib/BulkOps.java | 32 +-- .../org/humemai/nativelib/NativeExtras.java | 50 +++-- src/arcadedb/_native/bulk.py | 25 ++- src/arcadedb/_native/columnar.py | 10 +- src/arcadedb/_native/database.py | 80 ++++--- src/arcadedb/_native/extras.py | 13 ++ src/arcadedb/dataframe.py | 25 ++- tests/conftest.py | 2 + tests/test_native_columnar_pack.py | 10 +- ...t_native_columnar_parallel_review_fixes.py | 200 ++++++++++++++++++ 10 files changed, 368 insertions(+), 79 deletions(-) create mode 100644 tests/test_native_columnar_parallel_review_fixes.py diff --git a/native/src/main/java/org/humemai/nativelib/BulkOps.java b/native/src/main/java/org/humemai/nativelib/BulkOps.java index 40f789e0..cdf3cd98 100644 --- a/native/src/main/java/org/humemai/nativelib/BulkOps.java +++ b/native/src/main/java/org/humemai/nativelib/BulkOps.java @@ -63,22 +63,20 @@ static long insertColumns(final Database db, final String type, final String hea /** * The pointer form (arcadedb_n_insert_colp): the table holds SLOTS longs per column, see {@link #insertColumns(Database, String, Map, long[], ErrorCallback)}. - * Returns the reply text: {"ok":true,"n":rows}, and for a header with "parallel":true also "fid", the token that the "failures" operation of - * arcadedb_n_async reads (see there). In that mode the records are handed to the database's async executor, the call does not wait for - * the writers, and it neither opens, commits nor rolls back a transaction ("commit" does not apply), exactly as insert_parallel does for rows. + * Returns the reply text {"ok":true,"n":rows}. A header with "parallel":true must also carry "fid", a failure token that the caller made + * with the "token" operation of arcadedb_n_async: the records go to the database's async executor, the call does not wait for the writers + * and neither opens, commits nor rolls back a transaction ("commit" does not apply), exactly as insert_parallel does for rows. The + * caller waits for the writers and reads (and so removes) the token with the "failures" operation, also when this call throws or never + * returns to it: the token is the caller's, so this call never drops it. */ @SuppressWarnings("unchecked") static String insertColumnsReply(final Database db, final String type, final String header, final long[] spec) { final Map h = (Map) Json.parse(header); if (!Boolean.TRUE.equals(h.get("parallel"))) return "{\"ok\":true,\"n\":" + insertColumns(db, type, h, spec, null) + "}"; - final NativeExtras.FailureToken token = new NativeExtras.FailureToken(); - try { - return "{\"ok\":true,\"n\":" + insertColumns(db, type, h, spec, token.onError) + ",\"fid\":" + token.fid + "}"; - } catch (final Throwable t) { - token.drop(); // the caller gets the error, not a token to read failures from - throw t; - } + if (!(h.get("fid") instanceof Number fid)) + throw new IllegalArgumentException("a parallel columnar insert needs a failure token (header \"fid\", from the \"token\" operation)"); + return "{\"ok\":true,\"n\":" + insertColumns(db, type, h, spec, NativeExtras.FailureToken.of(fid.longValue()).onError) + "}"; } /** @@ -165,10 +163,16 @@ private static long insertColumns(final Database db, final String type, final Ma throw new IllegalArgumentException("Cannot bulk-insert into '" + type + "' because it is an edge type; load the vertices, then create the edges (graph_batch().new_edges() or CREATE EDGE)"); final DatabaseAsyncExecutor async = db.async(); - for (int i = 0; i < n; i++) { - final MutableDocument doc = vertex ? db.newVertex(type) : db.newDocument(type); - setRow(doc, i, names, kinds, dims, data, nulls, blobs, scratch); - async.createRecord(doc, null, onError); + int handed = 0; + try { + for (; handed < n; handed++) { + final MutableDocument doc = vertex ? db.newVertex(type) : db.newDocument(type); + setRow(doc, handed, names, kinds, dims, data, nulls, blobs, scratch); + async.createRecord(doc, null, onError); + } + } catch (final Throwable t) { + // the records handed over before the failure are the writers' now, whatever this call does + throw new IllegalStateException("after handing " + handed + " of " + n + " rows to the async writers (those may have been stored): " + t, t); } return n; } diff --git a/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java index c0e87e87..404181b5 100644 --- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java +++ b/native/src/main/java/org/humemai/nativelib/NativeExtras.java @@ -527,19 +527,42 @@ else if (opName.equals("create_vertices")) private static final Map ASYNC_FAILS = new ConcurrentHashMap<>(); private static final Map> ASYNC_FIRST = new ConcurrentHashMap<>(); - /** A failure token for one parallel load: the id the caller reads with the "failures" operation, and the callback that counts. */ + /** + * A failure token for one parallel load: the id the caller reads (and so removes) with the "failures" operation, and the callback that + * counts. The caller creates it before the load ("token" operation) and passes the id in the load's header, so that it holds the id + * whatever happens to the call, an interrupt included, and always has the means to wait for the writers and remove the entries. + */ static final class FailureToken { - final long fid = ArcadeNative.IDS.getAndIncrement(); - final AtomicLong count = new AtomicLong(); - final AtomicReference first = new AtomicReference<>(); - final ErrorCallback onError = t -> { - count.incrementAndGet(); - first.compareAndSet(null, t); - }; + final long fid; + final AtomicLong count; + final AtomicReference first; + final ErrorCallback onError; + + private FailureToken(final long fid, final AtomicLong count, final AtomicReference first) { + this.fid = fid; + this.count = count; + this.first = first; + this.onError = t -> { + count.incrementAndGet(); + first.compareAndSet(null, t); + }; + } - FailureToken() { - ASYNC_FAILS.put(fid, count); - ASYNC_FIRST.put(fid, first); + /** A new token, registered for the "failures" operation. */ + static FailureToken create() { + final FailureToken token = new FailureToken(ArcadeNative.IDS.getAndIncrement(), new AtomicLong(), new AtomicReference<>()); + ASYNC_FAILS.put(token.fid, token.count); + ASYNC_FIRST.put(token.fid, token.first); + return token; + } + + /** The token that "token" created, for the load that its id was passed to. */ + static FailureToken of(final long fid) { + final AtomicLong count = ASYNC_FAILS.get(fid); + final AtomicReference first = ASYNC_FIRST.get(fid); + if (count == null || first == null) + throw new IllegalArgumentException("unknown failure token " + fid + " (already read, or never created)"); + return new FailureToken(fid, count, first); } /** The load failed before it returned: the caller gets the error, not a token to read failures from. */ @@ -846,7 +869,7 @@ else if (a.get("slot") instanceof Number slot) ex.appendSamples((String) a.get("type"), stamps, columns); } case "insert_parallel" -> { - final FailureToken token = new FailureToken(); + final FailureToken token = FailureToken.create(); final long fid = token.fid; final ErrorCallback onError = token.onError; final String type = (String) a.get("type"); @@ -867,6 +890,9 @@ else if (a.get("slot") instanceof Number slot) } return ArcadeNative.out("{\"ok\":true,\"fid\":" + fid + "}"); } + case "token" -> { + return ArcadeNative.out("{\"ok\":true,\"fid\":" + FailureToken.create().fid + "}"); + } case "failures" -> { final long fid = ((Number) a.get("fid")).longValue(); final AtomicLong count = ASYNC_FAILS.remove(fid); diff --git a/src/arcadedb/_native/bulk.py b/src/arcadedb/_native/bulk.py index dcaca257..9851848f 100644 --- a/src/arcadedb/_native/bulk.py +++ b/src/arcadedb/_native/bulk.py @@ -8,6 +8,7 @@ import os +from ..exceptions import ArcadeDBError from . import columnar, lib _ENABLED = os.environ.get("ARCADEDB_NATIVE_BULK", "1") != "0" @@ -22,7 +23,7 @@ def _usable(): ) -def _send(db, type_name, packed, rows, commit_every, prefix, parallel=False): +def _send(db, type_name, packed, rows, commit_every, prefix, fid=None): from .database import _call spec = columnar.spec_of(packed) @@ -31,7 +32,7 @@ def _send(db, type_name, packed, rows, commit_every, prefix, parallel=False): "arcadedb_n_insert_colp", db._h, lib.b(type_name), - lib.b(columnar.header_of(rows, commit_every, packed, parallel)), + lib.b(columnar.header_of(rows, commit_every, packed, fid)), spec, len(spec), ) # `packed` holds the arrays the addresses point into until the call returns @@ -55,9 +56,10 @@ def insert_columns_fast( def insert_columns_parallel(db, type_name, columns, declared=None): - """Hand ``columns`` (name -> column) to the async writers in one crossing: ``(rows, failure token)``, or None when a column cannot be - sent this way. The call does not wait for the writers: the caller waits for them and reads the failures for the token - (``Database._finish_parallel``), as for ``insert_many(parallel=True)``. + """Hand ``columns`` (name -> column) to the async writers in one crossing, wait for them and return the row count, or None when a + column cannot be sent this way (nothing was handed over then). A record that a writer rejects raises + (``Database._finish_parallel``), as for ``insert_many(parallel=True)``, and so does an interrupt, after the writers have finished: + the failure token is made before the crossing, so the call always has the id to wait for and to read. """ if not _usable(): return None @@ -65,8 +67,17 @@ def insert_columns_parallel(db, type_name, columns, declared=None): if packed is None: return None rows, cols = packed - reply = _send(db, type_name, cols, rows, 0, "Failed to bulk-insert", parallel=True) - return int(reply["n"]), reply["fid"] + from . import extras + + fid = extras.failure_token(db) + try: + _send(db, type_name, cols, rows, 0, "Failed to bulk-insert", fid=fid) + except ArcadeDBError as e: + raise db._parallel_load_failed(type_name, e, [fid]) from e + except BaseException: + db._abandon_parallel([fid]) # a ^C: the writers still run, so wait for them + raise + return db._finish_parallel(type_name, [fid], rows) def insert_rows_columnar(db, type_name, chunk): diff --git a/src/arcadedb/_native/columnar.py b/src/arcadedb/_native/columnar.py index fe62837a..27d261a4 100644 --- a/src/arcadedb/_native/columnar.py +++ b/src/arcadedb/_native/columnar.py @@ -343,9 +343,10 @@ def spec_of(packed): return (ctypes.c_int64 * len(flat))(*flat) -def header_of(rows, commit_every, packed, parallel=False): - """The JSON header text: row count, commit interval and ``[name, kind, dim]`` per column; ``"parallel": true`` hands the records to the - async executor (the image then ignores the commit interval and the reply carries the failure token). +def header_of(rows, commit_every, packed, fid=None): + """The JSON header text: row count, commit interval and ``[name, kind, dim]`` per column; with a failure token ``fid`` it also says + ``"parallel": true``, which hands the records to the async executor (the image then ignores the commit interval and counts the + failed records on that token). """ import json @@ -354,8 +355,9 @@ def header_of(rows, commit_every, packed, parallel=False): "commit": min(int(commit_every), 2**31 - 1), "cols": [[c.name, c.kind, c.dim] for c in packed], } - if parallel: + if fid is not None: header["parallel"] = True + header["fid"] = fid return json.dumps(header) diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index b0c07932..6fab6c58 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -995,20 +995,50 @@ def _command_many(self, language: str, command: str, rows) -> int: raise _error("Command failed", reply) return reply["n"] - def _parallel_load_failed(self, type_name, error) -> ArcadeDBError: - """The error for a parallel load that failed while handing rows to the writers: waits for the queue to drain first, so - the caller sees a final state, and says that the rows handed over before the failure may have been stored. + def _drain_parallel(self, tokens): + """Waits for the async writers, then reads, and so removes, the failure token of every parallel load handed in. Returns + ``(problem, failed, first)``: the error of the wait or of a read (or None), the failed records the writers reported on all the + tokens and the first failure's text. The tokens are read whatever the wait did: an unread token keeps its entries for good. """ + from . import extras + + problem = None try: self._async().wait_completion() - except Exception: # nosec B110 - the original error is the one to report - pass + except Exception as e: + problem = e + failed, first = 0, None + for token in tokens: + try: + result = extras.failures(self, token) + except Exception as e: + problem = problem or e + continue + if result["count"] and first is None: + first = result["first"] + failed += result["count"] + return problem, failed, first + + def _abandon_parallel(self, tokens): + """A parallel load that is being interrupted: waits for the writers (they run on) and removes the tokens; the error that is on its + way out is the one to report. + """ + self._drain_parallel(tokens) + + def _parallel_load_failed(self, type_name, error, tokens=()) -> ArcadeDBError: + """The error for a parallel load that failed while handing rows to the writers: waits for the queue to drain first, so + the caller sees a final state, and says that the rows handed over before the failure may have been stored. + """ + _, failed, _ = self._drain_parallel(tokens) text = str(error) prefix = "Failed to bulk-insert: " if text.startswith(prefix): text = text[len(prefix) :] + reported = ( + f"; the writers had reported {failed} failed record(s)" if failed else "" + ) return ArcadeDBError( - f"Failed to bulk-insert into '{type_name}': {text}; the rows handed to the " + f"Failed to bulk-insert into '{type_name}': {text}{reported}; the rows handed to the " f"parallel writers before the failure may have been stored (the call waited " f"for them before raising)" ) @@ -1030,32 +1060,28 @@ def _insert_many_parallel(self, type_name, rows) -> int: tokens.append(extras.insert_parallel(self, type_name, chunk)["fid"]) except ArcadeDBError as e: # as the JPype path: the rows handed to the writers before the failure are written whatever this call does - raise self._parallel_load_failed(type_name, e) from e + raise self._parallel_load_failed(type_name, e, tokens) from e n += len(chunk) except ArcadeDBError: raise except BaseException: - try: - self._async().wait_completion() - except Exception: # nosec B110 - the original error is the one to report - pass + self._abandon_parallel(tokens) raise return self._finish_parallel(type_name, tokens, n) def _finish_parallel(self, type_name, tokens, n) -> int: """The end of a parallel load: wait for the async writers, then read the failure count of every token and raise when a record failed.""" - from . import extras - try: - self._async().wait_completion() - except Exception as e: - raise ArcadeDBError(f"Failed to bulk-insert into '{type_name}': {e}") from e - n_failed, first = 0, None - for token in tokens: - result = extras.failures(self, token) - if result["count"] and first is None: - first = result["first"] - n_failed += result["count"] + problem, n_failed, first = self._drain_parallel(tokens) + except BaseException: + self._abandon_parallel( + tokens + ) # a ^C in the wait: finish it, and remove the tokens + raise + if problem is not None: + raise ArcadeDBError( + f"Failed to bulk-insert into '{type_name}': {problem}" + ) from problem if n_failed: raise ArcadeDBError( f"Failed to bulk-insert into '{type_name}': the parallel writers reported {n_failed} failed record(s) " @@ -1343,13 +1369,9 @@ def insert_columns( if parallel: from .bulk import insert_columns_parallel - try: - sent = insert_columns_parallel(self, type_name, columns) - except ArcadeDBError as e: - # as insert_many(parallel=True): rows handed to the writers before the failure may be stored - raise self._parallel_load_failed(type_name, e) from e - if sent is not None: - return self._finish_parallel(type_name, [sent[1]], sent[0]) + done = insert_columns_parallel(self, type_name, columns) + if done is not None: + return done else: from .bulk import insert_columns_fast diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index 5f8a4702..439fb9ef 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -915,6 +915,19 @@ def insert_parallel(database, type_name: str, rows) -> dict: ) +def failure_token(database) -> int: + """A new failure token for one parallel load: the caller makes it before the load, hands its id to the load and reads it with + :func:`failures` after the final wait, so it holds the id whatever happens to the load (an interrupt included). + """ + return _call( + "Failed to make a failure token", + "arcadedb_n_async", + database._h, + b"token", + lib.b(_enc({})), + )["fid"] + + def failures(database, token: int) -> dict: return _call( "Failed to read the parallel writers", diff --git a/src/arcadedb/dataframe.py b/src/arcadedb/dataframe.py index d65e51b8..46362e53 100644 --- a/src/arcadedb/dataframe.py +++ b/src/arcadedb/dataframe.py @@ -421,8 +421,12 @@ def insert_table( native = _is_native(db) written = 0 # a document load inside the caller's own transaction joins it (insert_columns commits - # nothing then); a vertex load refuses to run inside one - in_caller_transaction = (not is_vertex) and bool(db.is_transaction_active()) + # nothing then); a vertex load refuses to run inside one. parallel=True never joins it: the + # async writers commit the records themselves, so a rollback of the caller's transaction does + # not remove them + in_caller_transaction = ( + (not is_vertex) and (not parallel) and bool(db.is_transaction_active()) + ) graph_batch = db.graph_batch() if is_vertex else None try: for batch in table.to_batches(max_chunksize=batch_size): @@ -459,6 +463,13 @@ def insert_table( f"{exc}; the {written} row(s) of the batches before this one " f"were written in your transaction, not committed" ) from exc + if written and parallel: + # the async writers commit on their own, inside the caller's transaction or not + raise ArcadeDBError( + f"{exc}; the {written} row(s) of the batches before this one " + f"were handed to the async writers, which commit on their own: " + f"a rollback of your transaction does not remove them" + ) from exc if written: raise ArcadeDBError( f"{exc}; the {written} row(s) of the batches before this one " @@ -503,14 +514,12 @@ def _insert_documents_native( if parallel: from ._native.bulk import insert_columns_parallel - try: - sent = insert_columns_parallel( + if ( + insert_columns_parallel( db, type_name, columns, {} if declared is None else declared ) - except ArcadeDBError as e: - raise db._parallel_load_failed(type_name, e) from e - if sent is not None: - db._finish_parallel(type_name, [sent[1]], sent[0]) + is not None + ): return else: from ._native.bulk import insert_columns_fast diff --git a/tests/conftest.py b/tests/conftest.py index be5fa583..707833d1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -49,6 +49,8 @@ collect_ignore.append("test_native_columnar_insert.py") # insert_columns(parallel=True) takes the columnar entry and stores what the row path stores (arcadedb_n_insert_colp with "parallel") collect_ignore.append("test_native_columnar_parallel.py") + # the review of the native parallel loads: no silent loss, an interrupt waits for the writers and reads the failure token + collect_ignore.append("test_native_columnar_parallel_review_fixes.py") # the rows of an insert_many chunk with date-times, Decimals, bytes and numpy scalars are told apart by the entry points they cross collect_ignore.append("test_native_mixed_chunk.py") # the native build's ingest defaults (insert_many and graph_batch commit size), observed on the native library diff --git a/tests/test_native_columnar_pack.py b/tests/test_native_columnar_pack.py index 56b283e1..af26f211 100644 --- a/tests/test_native_columnar_pack.py +++ b/tests/test_native_columnar_pack.py @@ -233,11 +233,11 @@ def test_the_address_table_and_header_describe_the_columns(): "cols": [["id", "q", 0], ["s", "s", 0]], } assert json.loads(columnar.header_of(rows, 2**40, packed))["commit"] == 2**31 - 1 - # "parallel" is in the header only when asked for (the image hands the records to the async writers then) - assert json.loads(columnar.header_of(rows, 10_000, packed, parallel=True))[ - "parallel" - ] - assert "parallel" not in json.loads(columnar.header_of(rows, 10_000, packed)) + # "parallel" and the failure token are in the header only when a token is given (the image hands the records to the async writers then) + parallel = json.loads(columnar.header_of(rows, 10_000, packed, fid=41)) + assert parallel["parallel"] is True and parallel["fid"] == 41 + plain = json.loads(columnar.header_of(rows, 10_000, packed)) + assert "parallel" not in plain and "fid" not in plain def test_rows_become_columns_only_when_every_row_has_the_same_keys(): diff --git a/tests/test_native_columnar_parallel_review_fixes.py b/tests/test_native_columnar_parallel_review_fixes.py new file mode 100644 index 00000000..72313ee6 --- /dev/null +++ b/tests/test_native_columnar_parallel_review_fixes.py @@ -0,0 +1,200 @@ +"""Review of the native parallel loads (#328): nothing is lost silently, an interrupt leaves no writer running and no token behind. + +``insert_columns(parallel=True)`` hands the records to the async writers in one crossing and waits for them before it returns. These +tests pin what a caller relies on: a record that a writer rejects is counted (``stored + reported failed >= rows``) and the stored +count is exactly the distinct keys; a load at the writers' batch boundary stores every row and nothing lands after the call returns; +a ^C during the crossing, or between the chunks of the row path, waits for the writers and reads (so removes) the failure token; a wait +that fails still reads the tokens; and the message of a failed parallel batch inside a caller's transaction does not claim that a +rollback removes rows the writers committed. Needs the native library: conftest.py collects this file on the native backend only. +""" + +import re +import time + +import numpy as np +import pytest + +from arcadedb._native import database as native_database +from arcadedb._native import extras +from arcadedb.exceptions import ArcadeDBError + +COLP = "arcadedb_n_insert_colp" + + +def _count(db, name): + sql = f"SELECT count(*) AS n FROM {name}" # nosec B608 - test-owned type name + return db.query("sql", sql).to_list()[0]["n"] + + +def _type(db, name="Col", unique=False): + db.command("sql", f"CREATE DOCUMENT TYPE {name} BUCKETS 3") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + db.command("sql", f"CREATE PROPERTY {name}.txt STRING") + if unique: + db.command("sql", f"CREATE INDEX ON {name} (id) UNIQUE") + + +class TokenSpy: + """The failure tokens that were made and the ones that were read (a read removes the token's entries in the image).""" + + def __init__(self, monkeypatch): + self.made, self.read = [], [] + real_make, real_read = extras.failure_token, extras.failures + real_rows = extras.insert_parallel + + def make(database): + fid = real_make(database) + self.made.append(fid) + return fid + + def failures(database, token): + self.read.append(token) + return real_read(database, token) + + def rows(database, type_name, chunk): + fid = real_rows(database, type_name, chunk) + self.made.append(fid["fid"]) + return fid + + monkeypatch.setattr(extras, "failure_token", make) + monkeypatch.setattr(extras, "failures", failures) + monkeypatch.setattr(extras, "insert_parallel", rows) + + +@pytest.fixture +def tokens(monkeypatch): + return TokenSpy(monkeypatch) + + +def _failed_count(error): + return int(re.search(r"reported (\d+) failed record", str(error)).group(1)) + + +def test_a_duplicate_key_is_stored_once_and_the_rest_are_reported(temp_db): + _type(temp_db, unique=True) + n, dupes = 3_000, 50 + ids = np.arange(n, dtype=np.int64) + ids[n - dupes :] = np.arange(dupes) # the last 50 repeat the first 50 keys + with pytest.raises(ArcadeDBError, match="failed record") as err: + temp_db.insert_columns("Col", {"id": ids, "txt": ["t"] * n}, parallel=True) + stored = _count(temp_db, "Col") + assert ( + stored + _failed_count(err.value) >= n + ) # every row is a stored row or a reported one + assert stored == len(set(ids.tolist())) == n - dupes + keys = {r["id"] for r in temp_db.query("sql", "SELECT id FROM Col").to_list()} + assert keys == set(ids.tolist()) # each key once, no key lost + + +def test_a_load_at_the_writers_batch_boundary_stores_every_row_and_nothing_later( + temp_db, +): + _type(temp_db) + threads = temp_db.async_executor().get_parallel_level() + n = threads * 10_240 + 1 # one row past a full batch for every writer + ids = np.arange(n, dtype=np.int64) + assert ( + temp_db.insert_columns("Col", {"id": ids, "txt": ["t"] * n}, parallel=True) == n + ) + stored = _count(temp_db, "Col") + assert stored == n + time.sleep(0.3) + temp_db.async_executor().wait_completion() + assert _count(temp_db, "Col") == n # nothing landed after the call returned + distinct = temp_db.query("sql", "SELECT count(DISTINCT id) AS c FROM Col") + assert distinct.to_list()[0]["c"] == n + + +def test_an_interrupt_in_the_crossing_waits_for_the_writers_and_reads_the_token( + temp_db, tokens, monkeypatch +): + _type(temp_db) + real_call = native_database._call + + def interrupted(prefix, name, *args): + reply = real_call(prefix, name, *args) + if name == COLP: + raise KeyboardInterrupt # the records are handed over; the reply never reaches the caller + return reply + + monkeypatch.setattr(native_database, "_call", interrupted) + n = 30_000 + with pytest.raises(KeyboardInterrupt): + temp_db.insert_columns( + "Col", {"id": np.arange(n, dtype=np.int64), "txt": ["t"] * n}, parallel=True + ) + assert ( + temp_db.async_executor().is_processing() is False + ) # waited: no writer runs on + assert len(tokens.made) == 1 and tokens.read == tokens.made # and the token is gone + assert ( + _count(temp_db, "Col") == n + ) # the rows handed over were all written, none half + + +def test_an_interrupt_between_the_chunks_of_the_row_path_waits_and_reads_the_tokens( + temp_db, tokens, monkeypatch +): + _type(temp_db) + real_rows = extras.insert_parallel + seen = [] + + def second_chunk_interrupted(database, type_name, chunk): + if seen: + raise KeyboardInterrupt + seen.append(1) + return real_rows(database, type_name, chunk) + + monkeypatch.setattr(extras, "insert_parallel", second_chunk_interrupted) + rows = [{"id": i, "txt": "t"} for i in range(25_000)] + with pytest.raises(KeyboardInterrupt): + temp_db.insert_many("Col", rows, parallel=True) + assert temp_db.async_executor().is_processing() is False + assert _count(temp_db, "Col") == 10_000 # the first chunk, whole + assert len(tokens.made) == 1 and tokens.read == tokens.made + + +def test_a_wait_that_fails_still_reads_the_tokens(temp_db, tokens, monkeypatch): + _type(temp_db) + real_wait = extras.NativeAsyncExecutor.wait_completion + state = {"raised": False} + + def failing_wait(self, timeout_ms=None): + if not state["raised"]: + state["raised"] = True + raise RuntimeError("the wait failed") + return real_wait(self, timeout_ms) + + monkeypatch.setattr(extras.NativeAsyncExecutor, "wait_completion", failing_wait) + n = 2_000 + with pytest.raises(ArcadeDBError, match="the wait failed"): + temp_db.insert_columns( + "Col", {"id": np.arange(n, dtype=np.int64), "txt": ["t"] * n}, parallel=True + ) + assert len(tokens.made) == 1 and tokens.read == tokens.made + temp_db.async_executor().wait_completion() + assert _count(temp_db, "Col") == n + + +def test_a_failed_parallel_batch_in_a_caller_transaction_does_not_promise_a_rollback( + temp_db, +): + import pyarrow as pa + + _type(temp_db, unique=True) + ids = list(range(300)) + ids[250] = 7 # a duplicate key in the third batch of 100 + table = pa.table({"id": pa.array(ids, pa.int64()), "txt": ["t"] * 300}) + temp_db.begin() + try: + with pytest.raises(ArcadeDBError) as err: + temp_db.insert_arrow("Col", table, batch_size=100, parallel=True) + message = str(err.value) + assert "not committed" not in message and "your transaction" in message + assert "async writers" in message and "does not remove them" in message + finally: + temp_db.rollback() + temp_db.async_executor().wait_completion() + assert ( + _count(temp_db, "Col") >= 200 + ) # the writers committed the first two batches: the rollback did not remove them From dc55dec8d4b62cda52d4c34aedf7877a1bd6ad22 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 22:22:23 +0900 Subject: [PATCH 5/8] tests: a duplicate key in a parallel load abandons the failing writer's open batch (CI of the first candidate) The first candidate run showed stored 1000 of 2950 distinct keys: a writer that meets a duplicate key rolls back its whole open batch (up to commit_every rows), so the valid rows of that batch are abandoned and reported too. stored == distinct was the wrong invariant. Assert stored + reported >= n, stored <= distinct, only sent ids, none stored twice. Co-Authored-By: Claude Sonnet 5.5 --- ...est_native_columnar_parallel_review_fixes.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/tests/test_native_columnar_parallel_review_fixes.py b/tests/test_native_columnar_parallel_review_fixes.py index 72313ee6..70f17bc3 100644 --- a/tests/test_native_columnar_parallel_review_fixes.py +++ b/tests/test_native_columnar_parallel_review_fixes.py @@ -1,8 +1,8 @@ """Review of the native parallel loads (#328): nothing is lost silently, an interrupt leaves no writer running and no token behind. ``insert_columns(parallel=True)`` hands the records to the async writers in one crossing and waits for them before it returns. These -tests pin what a caller relies on: a record that a writer rejects is counted (``stored + reported failed >= rows``) and the stored -count is exactly the distinct keys; a load at the writers' batch boundary stores every row and nothing lands after the call returns; +tests pin what a caller relies on: a record that a writer rejects is counted (``stored + reported failed >= rows``) and no key is stored twice +(a writer that meets a duplicate abandons its whole open batch, and reports it); a load at the writers' batch boundary stores every row and nothing lands after the call returns; a ^C during the crossing, or between the chunks of the row path, waits for the writers and reads (so removes) the failure token; a wait that fails still reads the tokens; and the message of a failed parallel batch inside a caller's transaction does not claim that a rollback removes rows the writers committed. Needs the native library: conftest.py collects this file on the native backend only. @@ -70,20 +70,23 @@ def _failed_count(error): return int(re.search(r"reported (\d+) failed record", str(error)).group(1)) -def test_a_duplicate_key_is_stored_once_and_the_rest_are_reported(temp_db): +def test_a_duplicate_key_is_reported_and_no_row_is_stored_twice(temp_db): _type(temp_db, unique=True) n, dupes = 3_000, 50 ids = np.arange(n, dtype=np.int64) ids[n - dupes :] = np.arange(dupes) # the last 50 repeat the first 50 keys with pytest.raises(ArcadeDBError, match="failed record") as err: temp_db.insert_columns("Col", {"id": ids, "txt": ["t"] * n}, parallel=True) - stored = _count(temp_db, "Col") + stored_ids = [r["id"] for r in temp_db.query("sql", "SELECT id FROM Col").to_list()] + stored = len(stored_ids) assert ( stored + _failed_count(err.value) >= n ) # every row is a stored row or a reported one - assert stored == len(set(ids.tolist())) == n - dupes - keys = {r["id"] for r in temp_db.query("sql", "SELECT id FROM Col").to_list()} - assert keys == set(ids.tolist()) # each key once, no key lost + # A writer that meets a duplicate key rolls back its WHOLE open batch (up to commit_every rows), so the valid rows of that batch are + # abandoned and reported too: `stored` can be below the number of distinct keys, never above it. + assert stored <= len(set(ids.tolist())) + assert set(stored_ids) <= set(ids.tolist()) # only ids that were sent + assert len(stored_ids) == len(set(stored_ids)) # none stored twice def test_a_load_at_the_writers_batch_boundary_stores_every_row_and_nothing_later( From 79bb5193271d5554f560296f5a82959a8afae77f Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 19:02:40 +0900 Subject: [PATCH 6/8] Native columnar insert: date-time, date, decimal and bytes columns stay columnar (refs #316) One column of a date-time, date, Decimal or bytes value used to make columnar.pack return None, so every column of the batch went through one dict per row and a JSON text with tagged values. BulkOps.insertColumns gets four kinds that read the same objects Json.decodeTag builds for those tagged values, so the stored value is the same: T (int64 ticks of the UTC wall clock since the epoch, dim = ticks per second so an Arrow timestamp keeps its unit, LocalDateTime), D (int32 epoch days, LocalDate), n (decimal text in the string layout, BigDecimal) and x (bytes in the string layout, byte[]). columnar.py lays out lists of datetime (also pandas.Timestamp), date, Decimal, bytes and bytearray and Arrow timestamp, date32, date64, decimal and binary arrays; a time-zone-aware value is its UTC wall clock and a naive one is taken as it is, the rule of _tag_datetime. The DataFrame and Arrow loader keeps those Arrow columns for the entry and turns them into tagged values only when a batch falls back to the row path. The native library is rebuilt from the unchanged engine commit. Co-Authored-By: Claude Sonnet 5.5 --- .../java/org/humemai/nativelib/BulkOps.java | 39 ++- src/arcadedb/_native/bulk.py | 4 +- src/arcadedb/_native/columnar.py | 114 ++++++- src/arcadedb/_native/database.py | 3 +- src/arcadedb/dataframe.py | 24 +- tests/conftest.py | 2 + tests/test_columnar_temporal_lists.py | 113 ++++++ tests/test_native_columnar_insert.py | 23 +- tests/test_native_columnar_pack.py | 97 +++++- tests/test_native_columnar_temporal.py | 322 ++++++++++++++++++ tests/test_native_mixed_chunk.py | 3 +- 11 files changed, 712 insertions(+), 32 deletions(-) create mode 100644 tests/test_columnar_temporal_lists.py create mode 100644 tests/test_native_columnar_temporal.py diff --git a/native/src/main/java/org/humemai/nativelib/BulkOps.java b/native/src/main/java/org/humemai/nativelib/BulkOps.java index cdf3cd98..00798947 100644 --- a/native/src/main/java/org/humemai/nativelib/BulkOps.java +++ b/native/src/main/java/org/humemai/nativelib/BulkOps.java @@ -19,8 +19,12 @@ import org.graalvm.word.Pointer; import org.graalvm.word.WordFactory; +import java.math.BigDecimal; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.List; import java.util.Map; @@ -82,7 +86,10 @@ static String insertColumnsReply(final Database db, final String type, final Str /** * 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). The kinds that the JSON rows carry as tagged values (Json.decodeTag), read as the same + * objects: T (int64 ticks since 1970-01-01T00:00:00 of the UTC wall clock, dim = ticks per second: 1, 1000, 10^6 or 10^9, read as a + * LocalDateTime), D (int32 days since the epoch, a LocalDate), n (a decimal as text in the string layout, a BigDecimal) and x (bytes in + * the string layout, a byte[]), so a column of these is stored as the row path stores it. 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 @@ -121,13 +128,18 @@ private static long insertColumns(final Database db, final String type, final Ma final long need = switch (kinds[c]) { case 'q', 'd' -> 8L * n; case 'b' -> n; - case 'i', 'e' -> 4L * n; + case 'i', 'e', 'D' -> 4L * n; + case 'T' -> { + if (dims[c] != 1 && dims[c] != 1_000 && dims[c] != 1_000_000 && dims[c] != 1_000_000_000) + throw new IllegalArgumentException("column '" + names[c] + "': a date-time column has 1, 1000, 10^6 or 10^9 ticks per second, not " + dims[c]); + yield 8L * 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); + case 's', 'n', 'x' -> 4L * (n + 1L); default -> throw new IllegalArgumentException("unknown column kind " + kinds[c]); }; if (dataBytes < need || (need > 0 && dataAddr == 0)) @@ -138,7 +150,7 @@ 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 (kinds[c] == 's') { + if (kinds[c] == 's' || kinds[c] == 'n' || kinds[c] == 'x') { 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); @@ -220,6 +232,25 @@ private static void setRow(final MutableDocument doc, final int i, final String[ 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 'T' -> { + final long ticks = base.readLong(WordFactory.signed(8L * i)); + final long perSecond = dims[c]; + doc.set(names[c], LocalDateTime.ofEpochSecond(Math.floorDiv(ticks, perSecond), + (int) (Math.floorMod(ticks, perSecond) * (1_000_000_000L / perSecond)), ZoneOffset.UTC)); + } + case 'D' -> doc.set(names[c], LocalDate.ofEpochDay(base.readInt(WordFactory.signed(4L * i)))); + case 'n' -> { + final int from = base.readInt(WordFactory.signed(4L * i)); + final int len = base.readInt(WordFactory.signed(4L * (i + 1L))) - from; + blobs[c].get(from, scratch, 0, len); + doc.set(names[c], new BigDecimal(new String(scratch, 0, len, StandardCharsets.US_ASCII))); + } + case 'x' -> { + final int from = base.readInt(WordFactory.signed(4L * i)); + final byte[] bytes = new byte[base.readInt(WordFactory.signed(4L * (i + 1L))) - from]; + blobs[c].get(from, bytes, 0, bytes.length); + doc.set(names[c], bytes); + } case 's' -> { final int from = base.readInt(WordFactory.signed(4L * i)); final int len = base.readInt(WordFactory.signed(4L * (i + 1L))) - from; diff --git a/src/arcadedb/_native/bulk.py b/src/arcadedb/_native/bulk.py index 9851848f..c16d225b 100644 --- a/src/arcadedb/_native/bulk.py +++ b/src/arcadedb/_native/bulk.py @@ -1,7 +1,7 @@ """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 +vector columns, date-times, dates, decimals, bytes and Arrow arrays are laid out by :mod:`.columnar`; anything else (nested values, +mixed-type columns, 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; with ``parallel`` they are handed to the async writers instead of a transaction (``DocumentBatcher.insertColumnsParallel``). ``ARCADEDB_NATIVE_BULK=0`` turns the columnar path off. """ diff --git a/src/arcadedb/_native/columnar.py b/src/arcadedb/_native/columnar.py index 27d261a4..46537e38 100644 --- a/src/arcadedb/_native/columnar.py +++ b/src/arcadedb/_native/columnar.py @@ -2,19 +2,29 @@ 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. +without one. A column that cannot be laid out (a nested value, a mixed-type column, a numpy datetime64, 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). +Temporal, decimal and binary kinds (the values the JSON path sends as tagged values, so the stored value is the same object): ``T`` a +date-time, int64 ticks since 1970-01-01T00:00:00 of the UTC wall clock, ``dim`` ticks per second (1, 1000, 10**6 or 10**9: an Arrow +timestamp keeps its unit, a Python ``datetime`` is microseconds), read as ``LocalDateTime``; ``D`` a date, int32 days since the epoch, +read as ``LocalDate``; ``n`` a decimal, the string layout holding ``str(Decimal)``, read as ``BigDecimal``; ``x`` bytes, the string +layout, read as ``byte[]``. A time-zone-aware ``datetime`` or Arrow timestamp is its UTC wall clock and a naive one is taken as it is, +the rule of ``_tag_datetime``; nothing here decides a time zone. + ``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. """ +from datetime import date, datetime, timezone +from decimal import Decimal + from .schema_objects import ( # the schema-entry reader shared with NativeType (#195) _name_type, ) @@ -27,6 +37,12 @@ _INT64_MAX = 2**63 - 1 _INT32_MAX = 2**31 - 1 _NoneType = type(None) +_EPOCH_ORDINAL = date(1970, 1, 1).toordinal() +_UTC = timezone.utc +_BLOB_KINDS = ( + "snx" # the kinds laid out as int32 offsets[n + 1] plus the bytes they index +) +_TICKS = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000} class Packed: @@ -39,8 +55,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 _BLOB_KINDS else 0 + ) # a string, decimal or bytes 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) @@ -166,6 +182,35 @@ def _string_column(name, texts, nulls=None): return Packed(name, "s", 0, offsets, aux=aux, nulls=nulls, keep=(blob_array, data)) +def _blob_column(name, kind, items, nulls=None): + """A list of bytes (a null already replaced by b'') as an offsets-plus-bytes column of ``kind``, or None past 2 GiB.""" + n = len(items) + data = b"".join(items) + if len(data) > _INT32_MAX: + return None + lengths = _np.fromiter(map(len, items), dtype=_np.int64, count=n) + 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, kind, 0, offsets, aux=aux, nulls=nulls, keep=(blob_array, data)) + + +def _epoch_micros(value): + """Microseconds since 1970-01-01T00:00:00 of the UTC wall clock of a ``datetime`` (a naive one is taken as it is, as ``_tag_datetime``).""" + if value.utcoffset() is not None: + value = value.astimezone(_UTC) + seconds = ( + (value.toordinal() - _EPOCH_ORDINAL) * 86400 + + value.hour * 3600 + + value.minute * 60 + + value.second + ) + return seconds * 1_000_000 + value.microsecond + + 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) @@ -203,6 +248,33 @@ def _list_column(name, values, declared): 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 issubclass(only, datetime): + if only.__name__ == "NaTType": + return None # the JSON path stores it as a null + try: + ticks = [0 if v is None else _epoch_micros(v) for v in values] + except (OverflowError, ValueError): + return None # the JSON path raises what it raised before + return Packed( + name, "T", 1_000_000, _np.array(ticks, dtype=_np.int64), nulls=nulls + ) + if only is date: + days = [0 if v is None else v.toordinal() - _EPOCH_ORDINAL for v in values] + return Packed(name, "D", 0, _np.array(days, dtype=_np.int32), nulls=nulls) + if only is Decimal: + texts = [] + for v in values: + if v is None: + texts.append(b"") + elif v.is_finite(): + texts.append(str(v).encode("ascii")) + else: + return None # the JSON path sends a non-finite Decimal down the per-row path + return _blob_column(name, "n", texts, nulls) + if only is bytes or only is bytearray: + return _blob_column( + name, "x", values if present is None else [v or b"" for v in values], nulls + ) if only is list and declared.get(name) == "ARRAY_OF_FLOATS" and present is None: try: arr = _np.array(values, dtype=_np.float32) @@ -248,10 +320,11 @@ def _arrow_column(name, arr, declared): if arr.null_count else None ) - if types.is_string(t) or types.is_large_string(t): + is_text = types.is_string(t) or types.is_large_string(t) + if is_text or types.is_binary(t) or types.is_large_binary(t): bufs = arr.buffers() off_buf, data_buf = bufs[1], bufs[2] - if types.is_string(t): + if types.is_string(t) or types.is_binary(t): offsets = _np.frombuffer(off_buf, dtype=_np.int32)[ arr.offset : arr.offset + n + 1 ] @@ -266,8 +339,35 @@ def _arrow_column(name, arr, declared): 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) + name, + "s" if is_text else "x", + 0, + offsets, + aux=aux, + nulls=nulls, + keep=(arr, off_buf, data_buf), + ) + if types.is_timestamp(t): + values = arr.cast(pa.int64()).fill_null(0).to_numpy(zero_copy_only=False) + return Packed( + name, + "T", + _TICKS[t.unit], + _np.ascontiguousarray(values, dtype=_np.int64), + nulls=nulls, + ) + if types.is_date(t): + if types.is_date64(t): # milliseconds since the epoch, floor-divided to days + raw = arr.cast(pa.int64()).fill_null(0).to_numpy(zero_copy_only=False) + days = raw // 86_400_000 + else: + days = arr.cast(pa.int32()).fill_null(0).to_numpy(zero_copy_only=False) + return Packed( + name, "D", 0, _np.ascontiguousarray(days, dtype=_np.int32), nulls=nulls ) + if types.is_decimal(t): + texts = [b"" if v is None else str(v).encode("ascii") for v in arr.to_pylist()] + return _blob_column(name, "n", texts, nulls) 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) diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 6fab6c58..3ee0f51b 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -877,7 +877,8 @@ def insert_many( a newDocument/set/save loop inside the image, as ``DocumentBatcher.insertManyJson`` on the JPype path). A ``datetime`` (also ``pandas.Timestamp`` and ``numpy.datetime64``), ``date``, ``Decimal``, ``bytes``/``bytearray``, numpy scalar, ``pandas.NA`` or ``pandas.NaT`` sitting directly in a row crosses in that text as a tagged value (``{"\\u0001": [kind, payload]}``, decoded by - ``Json.java``), so it does not move the chunk off this path; such a chunk loses the columnar entry. A row with a value the JSON + ``Json.java``), so it does not move the chunk off this path; but when a column holds only date-times, dates, Decimals or + bytes (None is a null) the chunk goes as columns, the same stored objects. A row with a value the JSON text cannot carry unchanged (NaN or Infinity, an integer beyond 64 bits, a date-time or decimal inside a list or map, a non-str key, a lone surrogate, a ``datetime.time``) goes through ``INSERT INTO t CONTENT :row`` on its own, in one crossing for a run of such rows; the rows around it keep the bulk path and every row is inserted in the order given. Inside the caller's diff --git a/src/arcadedb/dataframe.py b/src/arcadedb/dataframe.py index 46362e53..37c67a29 100644 --- a/src/arcadedb/dataframe.py +++ b/src/arcadedb/dataframe.py @@ -325,8 +325,10 @@ def _column_values(name: str, chunk, pa, native: bool = False, arrow: bool = Fal """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. + ``arrow`` (the native backend's document load): a string, binary, timestamp, date or decimal column, and a numeric one with + nulls, stay Arrow arrays, which the columnar entry reads in place (the offsets, the bytes, the integer ticks and the validity + bitmap) instead of as Python values; ``_insert_documents_native`` turns them into tagged values for the row path when a batch + cannot go columnar. """ types = pa.types if types.is_dictionary(chunk.type): # a pandas categorical or a Polars Categorical @@ -361,6 +363,8 @@ def _column_values(name: str, chunk, pa, native: bool = False, arrow: bool = Fal values = values.astype("float64") # float16/32 widen exactly return values return chunk.to_pylist() + if arrow and kind in ("timestamp", "date", "decimal", "binary"): + return chunk if kind in ("timestamp", "date"): return ( _temporal_to_tagged(chunk, pa) if native else _temporal_to_java(chunk, pa) @@ -496,6 +500,20 @@ def _insert_vertices( ) +def _arrow_row_values(col) -> list: + """A column kept as an Arrow array, as the values the row path takes: Python values, or tagged values for a timestamp, date, + decimal or binary column (the form it had before the columnar entry took them, so the stored value is unchanged). + """ + import pyarrow as pa + + kind = _kind(pa, col.type) + if kind in ("timestamp", "date"): + return _temporal_to_tagged(col, pa) + if kind in ("decimal", "binary"): + return _scalars_to_tagged(kind, col) + return col.to_pylist() + + def _insert_documents_native( db, type_name: str, @@ -542,7 +560,7 @@ def _insert_documents_native( if hasattr( col, "to_pylist" ): # a column kept as an Arrow array for the columnar entry - lists.append(col.to_pylist()) + lists.append(_arrow_row_values(col)) 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) diff --git a/tests/conftest.py b/tests/conftest.py index 707833d1..dc34241e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -51,6 +51,8 @@ collect_ignore.append("test_native_columnar_parallel.py") # the review of the native parallel loads: no silent loss, an interrupt waits for the writers and reads the failure token collect_ignore.append("test_native_columnar_parallel_review_fixes.py") + # a column of date-times, dates, Decimals or bytes crosses as a column (kinds T D n x), observed on the native library + collect_ignore.append("test_native_columnar_temporal.py") # the rows of an insert_many chunk with date-times, Decimals, bytes and numpy scalars are told apart by the entry points they cross collect_ignore.append("test_native_mixed_chunk.py") # the native build's ingest defaults (insert_many and graph_batch commit size), observed on the native library diff --git a/tests/test_columnar_temporal_lists.py b/tests/test_columnar_temporal_lists.py new file mode 100644 index 00000000..35aacabb --- /dev/null +++ b/tests/test_columnar_temporal_lists.py @@ -0,0 +1,113 @@ +"""A column of date-times, dates, Decimals or bytes stores the same values on both wheels (ArcadeData/arcadedb-python #316). + +On the native wheel such a column used to move the whole batch off the columnar entry; now ``insert_columns`` and ``insert_many`` send +it as a column (``tests/test_native_columnar_temporal.py`` proves the entry and compares with the row path). Here the expected values +are literals, so the same file on the JVM wheel and on the native wheel says that both store them: a naive date-time as it is, a +time-zone-aware one as its UTC wall clock, a date-time before 1970, a null, a date, a Decimal and bytes. +""" + +import datetime as dt +import decimal + +import pytest + +UTC9 = dt.timezone(dt.timedelta(hours=9)) +TS = [ + dt.datetime(2020, 1, 2, 3, 4, 5, 123456), + dt.datetime(1969, 12, 31, 23, 59, 59, 999999), + dt.datetime(1900, 1, 1, 0, 0, 0, 1), + dt.datetime(2400, 2, 29, 23, 59, 59), + None, +] +TS_STORED = TS # a naive date-time is the wall clock +AWARE = [ + dt.datetime(2020, 1, 2, 12, 0, 0, 5, tzinfo=UTC9), + dt.datetime(1969, 12, 31, 23, 0, 0, tzinfo=dt.timezone.utc), + dt.datetime(2020, 6, 1, 0, 0, 0, tzinfo=dt.timezone(dt.timedelta(hours=-7))), + None, + dt.datetime(2000, 1, 1, tzinfo=dt.timezone.utc), +] +AWARE_STORED = [ + dt.datetime(2020, 1, 2, 3, 0, 0, 5), + dt.datetime(1969, 12, 31, 23, 0, 0), + dt.datetime(2020, 6, 1, 7, 0, 0), + None, + dt.datetime(2000, 1, 1), +] +DAYS = [ + dt.date(2020, 2, 29), + dt.date(1960, 1, 1), + None, + dt.date(1970, 1, 1), + dt.date(1969, 12, 31), +] +AMOUNTS = [ + decimal.Decimal("12345.6789"), + decimal.Decimal("-0.50"), + decimal.Decimal("1E+3"), + None, + decimal.Decimal("0"), +] +BLOBS = [b"\x00\xff abc\x7f", b"", None, bytearray(b"ab\x80"), b"z"] + + +def _type(db, name): + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + db.command("sql", f"CREATE PROPERTY {name}.ts DATETIME_MICROS") + db.command("sql", f"CREATE PROPERTY {name}.day DATE") + db.command("sql", f"CREATE PROPERTY {name}.amount DECIMAL") + db.command("sql", f"CREATE PROPERTY {name}.blob BINARY") + + +def _stored(db, name, column): + sql = f"SELECT id, {column} AS v FROM {name} ORDER BY id" # nosec B608 - test-owned names + return [r.get("v") for r in db.query("sql", sql).to_list()] + + +def _bytes(values): + return [None if v is None else bytes(v) for v in values] + + +@pytest.mark.parametrize("path", ["insert_columns", "insert_many"]) +def test_naive_and_aware_datetimes_dates_decimals_and_bytes_are_stored_the_same( + temp_db, path +): + _type(temp_db, "Ev") + n = len(TS) + columns = { + "id": list(range(n)), + "ts": TS, + "day": DAYS, + "amount": AMOUNTS, + "blob": BLOBS, + } + if path == "insert_columns": + assert temp_db.insert_columns("Ev", columns) == n + else: + rows = [dict(zip(columns, vals)) for vals in zip(*columns.values())] + assert temp_db.insert_many("Ev", rows) == n + assert _stored(temp_db, "Ev", "ts") == TS_STORED + assert _stored(temp_db, "Ev", "day") == DAYS + assert _stored(temp_db, "Ev", "amount") == AMOUNTS + assert _bytes(_stored(temp_db, "Ev", "blob")) == _bytes(BLOBS) + + +def test_time_zone_aware_datetimes_are_stored_as_their_utc_wall_clock(temp_db): + _type(temp_db, "Zoned") + n = len(AWARE) + assert temp_db.insert_columns("Zoned", {"id": list(range(n)), "ts": AWARE}) == n + assert _stored(temp_db, "Zoned", "ts") == AWARE_STORED + + +def test_a_parallel_load_of_datetimes_stores_the_same_values(temp_db): + _type(temp_db, "Par") + n = len(TS) + assert ( + temp_db.insert_columns( + "Par", {"id": list(range(n)), "ts": TS, "day": DAYS}, parallel=True + ) + == n + ) + assert _stored(temp_db, "Par", "ts") == TS_STORED + assert _stored(temp_db, "Par", "day") == DAYS diff --git a/tests/test_native_columnar_insert.py b/tests/test_native_columnar_insert.py index 1ec2984c..e7296960 100644 --- a/tests/test_native_columnar_insert.py +++ b/tests/test_native_columnar_insert.py @@ -315,15 +315,18 @@ def test_rows_with_different_keys_fall_back_to_json(temp_db, spy, monkeypatch): _fallback_case(temp_db, spy, monkeypatch, rows, ROWS) -def test_a_datetime_leaves_the_columnar_entry_but_stays_on_the_json_entry( - temp_db, spy, monkeypatch -): - """columnar.pack refuses a date-time, so the chunk takes the JSON entry, which carries it as a tagged value - (test_native_mixed_chunk.py compares what it stores with INSERT ... CONTENT).""" +def test_a_datetime_column_stays_on_the_columnar_entry(temp_db, spy, monkeypatch): + """A column of date-times is the kind T of the columnar entry (#316): the stored value is the one the JSON entry stores as a tagged + value (tests/test_native_columnar_temporal.py has the full comparison, test_native_mixed_chunk.py the rows that stay on JSON). + """ 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, ROWS) - assert spy.count(MANY) == 0 + _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) == 1 and spy.count(ROWS) == 0 and spy.count(MANY) == 0 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") def test_values_json_cannot_carry_still_go_row_by_row_in_one_crossing( @@ -344,9 +347,7 @@ def test_an_integer_past_64_bits_is_refused_on_both_paths(temp_db, spy): 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 -): +def test_a_dataframe_with_a_timestamp_column_stays_columnar(temp_db, spy, monkeypatch): import pandas as pd import pyarrow as pa @@ -362,7 +363,7 @@ def test_a_dataframe_with_a_column_the_buffers_cannot_carry_falls_back( } ) temp_db.insert_arrow("Col", table) - assert spy.count(COLP) == 0 + assert spy.count(COLP) == 1 and spy.count(ROWS) == 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")] diff --git a/tests/test_native_columnar_pack.py b/tests/test_native_columnar_pack.py index af26f211..92a0eb89 100644 --- a/tests/test_native_columnar_pack.py +++ b/tests/test_native_columnar_pack.py @@ -198,16 +198,107 @@ 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(): +def test_numpy_datetime64_arrays_and_mixed_temporal_columns_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 ) + assert ( + columnar.pack( + {"d": [datetime.datetime(2020, 1, 1), datetime.date(2020, 1, 1)]}, NO_TYPES + ) + is None + ) + assert ( + columnar.pack({"d": [decimal.Decimal("1"), decimal.Decimal("NaN")]}, NO_TYPES) + is None + ) + assert columnar.pack({"d": [b"a", "b"]}, NO_TYPES) is None + + +def _ticks(col, dtype): + return np.frombuffer(_bytes(col.data), dtype=dtype).tolist() + + +def test_python_datetimes_are_utc_wall_clock_microseconds(): + import datetime + + utc9 = datetime.timezone(datetime.timedelta(hours=9)) + values = [ + datetime.datetime(1970, 1, 1), + datetime.datetime(1969, 12, 31, 23, 59, 59, 999999), + datetime.datetime(2020, 1, 2, 12, 0, 0, 5, tzinfo=utc9), # 03:00:00.000005 UTC + None, + ] + rows, cols = columnar.pack({"t": values}, NO_TYPES) + col = cols[0] + assert rows == 4 and col.kind == "T" and col.dim == 1_000_000 + assert _ticks(col, np.int64) == [0, -1, 1_577_934_000_000_005, 0] + assert _present(col) == [True, True, True, False] + + +def test_python_dates_are_epoch_days(): + import datetime + + rows, cols = columnar.pack( + {"d": [datetime.date(1970, 1, 2), datetime.date(1969, 12, 31), None]}, NO_TYPES + ) + assert cols[0].kind == "D" and _ticks(cols[0], np.int32) == [1, -1, 0] + assert _present(cols[0]) == [True, True, False] + + +def test_decimals_and_bytes_use_the_string_layout(): + import decimal + + rows, cols = columnar.pack( + { + "a": [decimal.Decimal("1E+3"), None, decimal.Decimal("-0.50")], + "b": [b"\x00\xff", b"", None], + "c": [bytearray(b"xy"), bytearray(b"z"), bytearray(b"")], + }, + NO_TYPES, + ) + a, b, c = cols + assert [x.kind for x in cols] == ["n", "x", "x"] + offsets = np.frombuffer(_bytes(a.data), dtype=np.int32).tolist() + assert offsets == [0, 4, 4, 9] and _bytes(a.aux) == b"1E+3-0.50" + offsets = np.frombuffer(_bytes(b.data), dtype=np.int32).tolist() + assert offsets == [0, 2, 2, 2] and _bytes(b.aux) == b"\x00\xff" + assert _bytes(c.aux) == b"xyz" and _present(a) == [True, False, True] + + +def test_arrow_timestamps_dates_decimals_and_binary(): + import decimal + + import pyarrow as pa + + stamps = pa.array([1, None, -2], pa.int64()).cast( + pa.timestamp("ms", tz="Asia/Seoul") + ) + days = pa.array([3, -4, None], pa.int32()).cast(pa.date32()) + day_ms = pa.array([86_400_000 * 2, -1, None], pa.int64()).cast(pa.date64()) + amounts = pa.array([decimal.Decimal("1.50"), None], pa.decimal128(10, 2)) + blobs = pa.array([b"ab", None, b"c"]) + for name, col, expected in ( + ("t", stamps, ("T", 1_000)), + ("d", days, ("D", 0)), + ("m", day_ms, ("D", 0)), + ): + packed = columnar.pack({name: col}, NO_TYPES)[1][0] + assert (packed.kind, packed.dim) == expected + assert _ticks(columnar.pack({"t": stamps}, NO_TYPES)[1][0], np.int64) == [1, 0, -2] + assert _ticks(columnar.pack({"d": days}, NO_TYPES)[1][0], np.int32) == [3, -4, 0] + # a date64 is floor-divided to days: 2 days, then -1 ms is the day before the epoch + assert _ticks(columnar.pack({"m": day_ms}, NO_TYPES)[1][0], np.int32) == [2, -1, 0] + n = columnar.pack({"n": amounts}, NO_TYPES)[1][0] + assert n.kind == "n" and _bytes(n.aux) == b"1.50" + x = columnar.pack({"x": blobs}, NO_TYPES)[1][0] + assert ( + x.kind == "x" and _bytes(x.aux) == b"abc" and _present(x) == [True, False, True] + ) def test_ragged_columns_and_empty_input_are_left_to_the_row_path(): diff --git a/tests/test_native_columnar_temporal.py b/tests/test_native_columnar_temporal.py new file mode 100644 index 00000000..a97d0b61 --- /dev/null +++ b/tests/test_native_columnar_temporal.py @@ -0,0 +1,322 @@ +"""Native: a column of date-times, dates, Decimals or bytes stays on the columnar entry and stores what the row path stores (#316). + +Before, one such column made ``columnar.pack`` return None, so every column of the batch went through one dict per row and a JSON text +with tagged values. Now the kinds T, D, n and x of ``arcadedb_n_insert_colp`` carry them and the image builds the same objects +(``LocalDateTime``, ``LocalDate``, ``BigDecimal``, ``byte[]``) that ``Json.decodeTag`` builds for the tagged values. Each test loads the +same data through the row path (the private reference: the columnar path switched off, so the chunk goes as tagged JSON) and through +the columns, compares every stored value and its type, and spies on the entry points so a silent fall back cannot pass. Needs the +native library, so conftest.py collects this file on the native backend only. +""" + +import datetime as dt +import decimal + +import pytest + +from arcadedb._native import bulk, columnar, lib # noqa: E402 + +COLP, ROWS, MANY = ( + "arcadedb_n_insert_colp", + "arcadedb_n_insert_rows", + "arcadedb_n_exec_many", +) +UTC9 = dt.timezone(dt.timedelta(hours=9)) + + +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 _type(db, name, declared=True): + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + if declared: + for prop, ty in ( + ("ts", "DATETIME_MICROS"), + ("ns", "DATETIME_NANOS"), + ("day", "DATE"), + ("amount", "DECIMAL"), + ("blob", "BINARY"), + ): + db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}") + + +def _stored(db, name): + """Every record by id as {property: (python type name, repr)}, so a changed type is as visible as a changed value.""" + sql = f"SELECT FROM {name} ORDER BY id" # nosec B608 - test-owned type name + return [ + { + k: (type(v).__name__, repr(v)) + for k, v in sorted(rec.items()) + if not k.startswith("@") and k != "rid" + } + for rec in db.query("sql", sql).to_list() + ] + + +def _rows(columns): + return [dict(zip(columns, vals)) for vals in zip(*columns.values())] + + +def _reference(db, name, rows, monkeypatch): + with monkeypatch.context() as m: + m.setattr(bulk, "_ENABLED", False) + db.insert_many(name, rows) + + +def _columns(n=60): + base = dt.datetime(1969, 12, 31, 23, 59, 59, 999999) + return { + "id": list(range(n)), + "ts": [ + ( + None + if i % 11 == 0 + else (base + dt.timedelta(seconds=i * 86_399, microseconds=i * 7)) + ) + for i in range(n) + ], + "day": [ + None if i % 13 == 0 else dt.date(1960, 1, 1) + dt.timedelta(days=i * 401) + for i in range(n) + ], + "amount": [ + ( + None + if i % 9 == 0 + else decimal.Decimal(f"{i * 1234567}.{i % 100:02d}") + * (-1 if i % 2 else 1) + ) + for i in range(n) + ], + "blob": [ + None if i % 8 == 0 else bytes(range(i % 256)) + b"\x00\xff" + for i in range(n) + ], + } + + +@pytest.mark.parametrize("declared", [True, False], ids=["declared", "schemaless"]) +def test_a_batch_with_temporal_decimal_and_bytes_columns_stays_columnar( + temp_db, spy, monkeypatch, declared +): + for name in ("Ref", "ByRows", "ByColumns"): + _type(temp_db, name, declared) + columns = _columns() + _reference(temp_db, "Ref", _rows(columns), monkeypatch) + assert ( + spy.count(COLP) == 0 and spy.count(ROWS) == 1 + ) # the reference is the JSON path + spy.calls.clear() + assert temp_db.insert_many("ByRows", _rows(columns)) == 60 + assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 and spy.count(MANY) == 0 + spy.calls.clear() + assert temp_db.insert_columns("ByColumns", columns) == 60 + assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 + expected = _stored(temp_db, "Ref") + assert len(expected) == 60 and expected[1]["ts"][0] == "datetime" + assert _stored(temp_db, "ByRows") == expected + assert _stored(temp_db, "ByColumns") == expected + + +def test_time_zone_aware_and_naive_datetimes_follow_the_json_rule( + temp_db, spy, monkeypatch +): + """An aware date-time is its UTC wall clock, a naive one is taken as it is (_tag_datetime): the column does not decide a zone.""" + _type(temp_db, "Ref") + _type(temp_db, "Col") + values = [ + dt.datetime(2020, 1, 2, 12, 0, 0, 5, tzinfo=UTC9), + dt.datetime(1969, 12, 31, 23, 0, 0, tzinfo=dt.timezone.utc), + dt.datetime(2020, 6, 1, 0, 0, 0, tzinfo=dt.timezone(dt.timedelta(hours=-7))), + ] + rows = [{"id": i, "ts": v} for i, v in enumerate(values)] + _reference(temp_db, "Ref", rows, monkeypatch) + spy.calls.clear() + temp_db.insert_many("Col", rows) + assert spy.count(COLP) == 1 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + + +def test_pandas_timestamps_keep_microseconds_as_the_json_path_does( + temp_db, spy, monkeypatch +): + import pandas as pd + + _type(temp_db, "Ref") + _type(temp_db, "Col") + stamps = [ + pd.Timestamp("2026-05-06 07:08:09.123456789"), + pd.Timestamp("1960-05-06 07:08:09.123456", tz="Asia/Seoul"), + ] + rows = [{"id": i, "ts": v} for i, v in enumerate(stamps)] + _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 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + + +@pytest.mark.parametrize( + "column", + [ + [dt.datetime(2020, 1, 1), dt.date(2020, 1, 1)], # a date-time and a date + [decimal.Decimal("1.5"), decimal.Decimal("NaN")], # a non-finite Decimal + [b"a", "b"], # bytes and text + ], + ids=["datetime-and-date", "nan-decimal", "bytes-and-text"], +) +def test_a_column_the_buffers_cannot_carry_unchanged_still_falls_back( + temp_db, spy, monkeypatch, column +): + _type(temp_db, "Ref", declared=False) + _type(temp_db, "Col", declared=False) + rows = [{"id": i, "x": v} for i, v in enumerate(column)] + try: + _reference(temp_db, "Ref", rows, monkeypatch) + except ( + Exception + ): # the statement path may refuse it; then the new path must refuse it the same way + with pytest.raises(Exception): + temp_db.insert_many("Col", rows) + assert spy.count(COLP) == 0 + return + spy.calls.clear() + temp_db.insert_many("Col", rows) + assert spy.count(COLP) == 0 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + + +def _arrow_table(pa): + import numpy as np + + n = 40 + ids = np.arange(n, dtype=np.int64) + epoch_us = (ids - 20) * 1_000_003_000 + 7 # microseconds, before and after 1970 + mask = ids % 6 == 0 + + def stamps(unit, tz=None): + raw = { + "s": epoch_us // 1_000_000, + "ms": epoch_us // 1_000, + "us": epoch_us, + "ns": epoch_us * 1_000, + }[unit] + return pa.array(raw.astype(np.int64), mask=mask).cast(pa.timestamp(unit, tz=tz)) + + return pa.table( + { + "id": ids, + "ts": stamps("us"), + "ns": stamps("ns"), + "sec": stamps("s"), + "zoned": stamps("ms", tz="Asia/Seoul"), + "day": pa.array( + ((ids - 10) * 397).astype(np.int32), mask=ids % 5 == 0 + ).cast(pa.date32()), + "amount": pa.array( + [ + None if i % 7 == 0 else decimal.Decimal(f"{i}.{i % 100:02d}") + for i in range(n) + ], + type=pa.decimal128(20, 2), + ), + "blob": pa.array( + [None if i % 4 == 0 else bytes(range(i)) for i in range(n)] + ), + } + ) + + +def test_arrow_timestamps_dates_decimals_and_binary_stay_columnar( + temp_db, spy, monkeypatch +): + import pyarrow as pa + + for name in ("Ref", "Col"): + _type(temp_db, name) + for prop, ty in (("sec", "DATETIME_SECOND"), ("zoned", "DATETIME")): + temp_db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}") + table = _arrow_table(pa) + with ( + monkeypatch.context() as m + ): # the reference: tagged values through the JSON path + m.setattr(bulk, "_ENABLED", False) + assert temp_db.insert_arrow("Ref", table) == 40 + assert spy.count(COLP) == 0 + spy.calls.clear() + assert temp_db.insert_arrow("Col", table) == 40 + assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 and spy.count(MANY) == 0 + expected = _stored(temp_db, "Ref") + assert len(expected) == 40 and expected[1]["ns"][0] == "datetime" + assert _stored(temp_db, "Col") == expected + + +def test_an_arrow_batch_with_one_list_column_falls_back_with_the_same_values( + temp_db, spy, monkeypatch +): + """A list column cannot cross as a buffer: the whole batch takes the row path, and the temporal columns reach it as tagged values.""" + import pyarrow as pa + + for name in ("Ref", "Col"): + _type(temp_db, name) + for prop, ty in (("sec", "DATETIME_SECOND"), ("zoned", "DATETIME")): + temp_db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}") + table = _arrow_table(pa).append_column( + "tags", pa.array([["a", "b"]] * 40, type=pa.list_(pa.string())) + ) + with monkeypatch.context() as m: + m.setattr(bulk, "_ENABLED", False) + temp_db.insert_arrow("Ref", table) + spy.calls.clear() + temp_db.insert_arrow("Col", table) + assert spy.count(COLP) == 0 + assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") + + +def test_a_parallel_load_of_datetimes_is_one_columnar_crossing(temp_db, spy): + _type(temp_db, "Par") + columns = _columns(300) + assert temp_db.insert_columns("Par", columns, parallel=True) == 300 + assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 + assert temp_db.query("sql", "SELECT count(*) AS n FROM Par").first().get("n") == 300 + + +def test_the_image_refuses_a_wrong_ticks_per_second_and_writes_nothing(temp_db): + import json + + _type(temp_db, "Col") + rows, packed = columnar.pack( + {"id": [1, 2], "ts": [dt.datetime(2020, 1, 1)] * 2}, columnar.Declared(known={}) + ) + assert packed[1].kind == "T" and packed[1].dim == 1_000_000 + packed[1].dim = 7 + spec = columnar.spec_of(packed) + reply = json.loads( + lib.call( + COLP, + temp_db._h, + lib.b("Col"), + lib.b(columnar.header_of(rows, 10_000, packed)), + spec, + len(spec), + ) + ) + assert not reply["ok"] and "ticks per second" in reply["error"] + assert temp_db.query("sql", "SELECT count(*) AS n FROM Col").first().get("n") == 0 diff --git a/tests/test_native_mixed_chunk.py b/tests/test_native_mixed_chunk.py index db4162cf..0c746bdf 100644 --- a/tests/test_native_mixed_chunk.py +++ b/tests/test_native_mixed_chunk.py @@ -169,7 +169,8 @@ def test_a_pandas_timestamp_is_stored_as_the_statement_path_stores_it( _statement_path(db, "Ref", rows) spy.clear() db.insert_many("Bulk", rows) - assert spy.count(MANY) == 0 and spy.count(ROWS) == 1 + # one column of Timestamps is the kind T of the columnar entry (#316): microseconds, as the tagged path keeps them + assert spy.count(MANY) == 0 and spy.count(ROWS) == 0 and spy.count(COLP) == 1 assert _stored(db, "Bulk") == _stored(db, "Ref") From ca91fd836334aa587e5fb4c6feb1ad39dbbd7327 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 21:25:41 +0900 Subject: [PATCH 7/8] Temporal columnar kinds: refuse out-of-range values in Python, before the crossing (review of #333) A date64 past int32 days wrapped in the int32 cast (the row path raised) and a timestamp[s] past the years of java.time threw DateTimeException in the middle of a batch (BulkOps T kind, LocalDateTime.ofEpochSecond). Both now make pack() return None, so the call takes the row path and raises what it raised before, with nothing stored. Only a timestamp[s] can pass the java.time range: an int64 of milliseconds is at most 9.2e15 seconds against 3.15e16. Tests: the pack-level limits, the sync call (nothing stored, no transaction left open), the parallel call (may have been stored), and a date-time chunk with NaN stores the NaN that INSERT ... CONTENT :row stores (both end in MutableDocument.putProperty). Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/_native/columnar.py | 11 ++ tests/conftest.py | 2 + ...t_native_columnar_temporal_review_fixes.py | 176 ++++++++++++++++++ 3 files changed, 189 insertions(+) create mode 100644 tests/test_native_columnar_temporal_review_fixes.py diff --git a/src/arcadedb/_native/columnar.py b/src/arcadedb/_native/columnar.py index 46537e38..d001b3da 100644 --- a/src/arcadedb/_native/columnar.py +++ b/src/arcadedb/_native/columnar.py @@ -43,6 +43,10 @@ "snx" # the kinds laid out as int32 offsets[n + 1] plus the bytes they index ) _TICKS = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000} +# the seconds that java.time.LocalDateTime.ofEpochSecond takes (years -999999999 to 999999999): beyond them it throws in the image, in the +# middle of a batch. Only a timestamp[s] can pass them: an int64 of milliseconds is at most 9.2e15 seconds +_LDT_MIN_SECONDS = -31_557_014_167_219_200 +_LDT_MAX_SECONDS = 31_556_889_864_403_199 class Packed: @@ -349,6 +353,9 @@ def _arrow_column(name, arr, declared): ) if types.is_timestamp(t): values = arr.cast(pa.int64()).fill_null(0).to_numpy(zero_copy_only=False) + if t.unit == "s" and values.size: + if values.min() < _LDT_MIN_SECONDS or values.max() > _LDT_MAX_SECONDS: + return None # the row path raises its usual error for it, before anything is stored return Packed( name, "T", @@ -360,6 +367,10 @@ def _arrow_column(name, arr, declared): if types.is_date64(t): # milliseconds since the epoch, floor-divided to days raw = arr.cast(pa.int64()).fill_null(0).to_numpy(zero_copy_only=False) days = raw // 86_400_000 + if days.size and ( + days.min() < -(_INT32_MAX + 1) or days.max() > _INT32_MAX + ): + return None # int32 days would wrap: the row path raises its usual error for it else: days = arr.cast(pa.int32()).fill_null(0).to_numpy(zero_copy_only=False) return Packed( diff --git a/tests/conftest.py b/tests/conftest.py index dc34241e..d7d95ce1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -53,6 +53,8 @@ collect_ignore.append("test_native_columnar_parallel_review_fixes.py") # a column of date-times, dates, Decimals or bytes crosses as a column (kinds T D n x), observed on the native library collect_ignore.append("test_native_columnar_temporal.py") + # the review of the temporal kinds: a value the image cannot take is refused in Python, never wrapped or thrown mid-batch + collect_ignore.append("test_native_columnar_temporal_review_fixes.py") # the rows of an insert_many chunk with date-times, Decimals, bytes and numpy scalars are told apart by the entry points they cross collect_ignore.append("test_native_mixed_chunk.py") # the native build's ingest defaults (insert_many and graph_batch commit size), observed on the native library diff --git a/tests/test_native_columnar_temporal_review_fixes.py b/tests/test_native_columnar_temporal_review_fixes.py new file mode 100644 index 00000000..31c8430e --- /dev/null +++ b/tests/test_native_columnar_temporal_review_fixes.py @@ -0,0 +1,176 @@ +"""Review of the temporal columnar kinds (#333): a value the image cannot take is refused in Python, before the crossing, and never wraps. + +``T`` and ``D`` columns carry integers that the image turns into ``LocalDateTime`` and ``LocalDate``. A ``date64`` past int32 days used +to wrap in the int32 cast and a ``timestamp[s]`` past the years of ``java.time`` threw ``DateTimeException`` in the middle of a batch. Now +``columnar.pack`` returns None for both, so the call takes the row path, which raises what it raised before (nothing stored, no +transaction left open on the synchronous call; ``may have been stored`` for a parallel call, which has handed rows over by then). Also +pinned: a chunk with a date-time column and a float column holding NaN goes columnar now and stores the same NaN that ``INSERT INTO t +CONTENT :row`` stores (both end in ``MutableDocument.putProperty``). Needs the native library: conftest.py collects this file on the +native backend only. +""" + +import datetime as dt +import math + +import pytest + +from arcadedb._native import columnar, lib # noqa: E402 +from arcadedb.exceptions import ArcadeDBError + +pa = pytest.importorskip("pyarrow") + +COLP = "arcadedb_n_insert_colp" +DAY_MS = 86_400_000 +LDT_MIN_SECONDS, LDT_MAX_SECONDS = -31_557_014_167_219_200, 31_556_889_864_403_199 +NO_TYPES = columnar.Declared(known={}) + + +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(db, name): + sql = f"SELECT count(*) AS n FROM {name}" # nosec B608 - test-owned type name + return db.query("sql", sql).first().get("n") + + +def _kinds(columns): + packed = columnar.pack(columns, NO_TYPES) + return None if packed is None else [c.kind for c in packed[1]] + + +def test_a_date64_past_int32_days_is_not_laid_out_and_never_wraps(): + wraps = pa.array([DAY_MS * (2**31 + 5)], pa.int64()).cast(pa.date64()) + assert columnar.pack({"d": wraps}, NO_TYPES) is None + below = pa.array([DAY_MS * (-(2**31) - 1)], pa.int64()).cast(pa.date64()) + assert columnar.pack({"d": below}, NO_TYPES) is None + one_bad = pa.array([0, DAY_MS, DAY_MS * 2**40], pa.int64()).cast(pa.date64()) + assert columnar.pack({"d": one_bad}, NO_TYPES) is None # one value is enough + + +def test_a_date64_at_the_int32_limits_stays_columnar_and_exact(): + edge = pa.array( + [DAY_MS * (2**31 - 1), DAY_MS * -(2**31), None], pa.int64() + ).cast(pa.date64()) + rows, packed = columnar.pack({"d": edge}, NO_TYPES) + assert rows == 3 and packed[0].kind == "D" + import ctypes + + import numpy as np + + addr, length = packed[0].data + days = np.frombuffer(ctypes.string_at(addr, length), dtype=np.int32) + assert days.tolist() == [2**31 - 1, -(2**31), 0] + + +def test_a_timestamp_past_the_years_of_java_time_is_not_laid_out(): + assert ( + columnar.pack({"t": pa.array([0, 2**62], pa.timestamp("s"))}, NO_TYPES) is None + ) + over = pa.array([LDT_MAX_SECONDS + 1], pa.timestamp("s")) + assert columnar.pack({"t": over}, NO_TYPES) is None + under = pa.array([LDT_MIN_SECONDS - 1], pa.timestamp("s")) + assert columnar.pack({"t": under}, NO_TYPES) is None + + +def test_a_timestamp_at_the_limits_stays_columnar(): + edge = pa.array([LDT_MAX_SECONDS, LDT_MIN_SECONDS, None], pa.timestamp("s")) + assert _kinds({"t": edge}) == ["T"] + # an int64 of milliseconds, microseconds or nanoseconds is always inside the years of java.time + for unit in ("ms", "us", "ns"): + big = pa.array([2**63 - 1, -(2**63)], pa.timestamp(unit)) + assert _kinds({"t": big}) == ["T"], unit + + +def _type(db, name): + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + + +def test_a_timestamp_the_image_cannot_take_stores_nothing_and_leaves_no_transaction( + temp_db, +): + _type(temp_db, "Col") + table = pa.table( + { + "id": pa.array([1, 2], pa.int64()), + "t": pa.array([0, 2**62], pa.timestamp("s")), + } + ) + with pytest.raises(ArcadeDBError): + temp_db.insert_arrow("Col", table) + assert not temp_db.is_transaction_active() + assert _count(temp_db, "Col") == 0 + + +def test_a_date_the_image_cannot_take_stores_nothing_and_leaves_no_transaction(temp_db): + _type(temp_db, "Col") + wraps = pa.array([0, DAY_MS * (2**31 + 5)], pa.int64()).cast(pa.date64()) + table = pa.table({"id": pa.array([1, 2], pa.int64()), "d": wraps}) + with pytest.raises(ArcadeDBError): + temp_db.insert_arrow("Col", table) + assert not temp_db.is_transaction_active() + assert _count(temp_db, "Col") == 0 + + +def test_a_parallel_load_of_such_a_timestamp_says_that_rows_may_have_been_stored( + temp_db, +): + _type(temp_db, "Col") + table = pa.table( + { + "id": pa.array([1, 2], pa.int64()), + "t": pa.array([0, 2**62], pa.timestamp("s")), + } + ) + with pytest.raises(ArcadeDBError, match="may have been stored"): + temp_db.insert_arrow("Col", table, parallel=True) + temp_db.async_executor().wait_completion() + assert ( + _count(temp_db, "Col") <= 1 + ) # the row before the bad one may have been handed over + + +def test_a_datetime_chunk_with_nan_stores_the_nan_the_statement_path_stores( + temp_db, monkeypatch +): + for name in ("Col", "Ref"): + _type(temp_db, name) + temp_db.command("sql", f"CREATE PROPERTY {name}.ts DATETIME_MICROS") + temp_db.command("sql", f"CREATE PROPERTY {name}.val DOUBLE") + rows = [ + { + "id": i, + "ts": dt.datetime(2020, 1, 1, 0, 0, i), + "val": float("nan") if i % 2 else 1.5 * i, + } + for i in range(10) + ] + spy = Spy(monkeypatch) + temp_db.insert_many("Col", rows) + assert ( + spy.calls.count(COLP) == 1 + ) # the date-time column no longer forces the JSON path + sql = "INSERT INTO `Ref` CONTENT :row" + temp_db.begin() + temp_db._command_many("sql", sql, [{"row": row} for row in rows]) + temp_db.commit() + + def stored(name): + select = f"SELECT FROM {name} ORDER BY id" # nosec B608 - test-owned type name + return [ + (r["id"], type(r["val"]).__name__, repr(r["val"]), repr(r["ts"])) + for r in temp_db.query("sql", select).to_list() + ] + + got, want = stored("Col"), stored("Ref") + assert got == want + assert all(math.isnan(float(v[2])) for v in got if v[0] % 2) From c4d41e5e004a69272514c4d3795b457fe81ee96f Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 22:23:25 +0900 Subject: [PATCH 8/8] tests: fix the cases of the first candidate run that the temporal kinds changed or that were mis-built - review fixes: the date64 case built an int64 beyond its own range (DAY_MS * 2**40); use 2**33, still past int32 days. A non-finite Decimal (NaN, -NaN, sNaN, Infinity, -Infinity) is refused in lists, with None, and in object columns. - nan-decimal fall back: the NaN row goes to the statement as before; the finite row around it is a run of carriable rows and crosses on the columnar entry now (a finite Decimal is carried), so one colp call is expected, not none. - pack refusals: bytes alone are carried (kind x) now; refuse bytes with text and bytes with bytearray instead. - temporal lists: a BINARY property reads back as the signed values of a Java byte[] or as bytes; compare after & 0xFF. - mixed chunk: the rows around an uncarriable row cross on the columnar entry (the date-time column is carried), one colp call per run. Co-Authored-By: Claude Sonnet 5.5 --- tests/test_columnar_temporal_lists.py | 3 ++- tests/test_native_columnar_pack.py | 6 +++++- tests/test_native_columnar_temporal.py | 18 +++++++++++------- ...st_native_columnar_temporal_review_fixes.py | 16 +++++++++++++++- tests/test_native_mixed_chunk.py | 4 +++- 5 files changed, 36 insertions(+), 11 deletions(-) diff --git a/tests/test_columnar_temporal_lists.py b/tests/test_columnar_temporal_lists.py index 35aacabb..9a9d62b7 100644 --- a/tests/test_columnar_temporal_lists.py +++ b/tests/test_columnar_temporal_lists.py @@ -66,7 +66,8 @@ def _stored(db, name, column): def _bytes(values): - return [None if v is None else bytes(v) for v in values] + # a BINARY property reads back as the values of a Java byte[] (signed, -128..127) or as bytes; & 0xFF takes both to the same bytes + return [None if v is None else bytes(b & 0xFF for b in v) for v in values] @pytest.mark.parametrize("path", ["insert_columns", "insert_many"]) diff --git a/tests/test_native_columnar_pack.py b/tests/test_native_columnar_pack.py index 92a0eb89..e8925661 100644 --- a/tests/test_native_columnar_pack.py +++ b/tests/test_native_columnar_pack.py @@ -190,7 +190,11 @@ def test_a_bitmap_longer_than_a_byte(): [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"], + [ + b"ab", + "cd", + ], # bytes and text (a column of bytes alone crosses as kind x since #316) + [b"ab", bytearray(b"cd")], # two bytes-like types [np.int64(1), np.int64(2)], # numpy scalars in a list are not Python numbers ], ) diff --git a/tests/test_native_columnar_temporal.py b/tests/test_native_columnar_temporal.py index a97d0b61..7ae3b545 100644 --- a/tests/test_native_columnar_temporal.py +++ b/tests/test_native_columnar_temporal.py @@ -174,17 +174,21 @@ def test_pandas_timestamps_keep_microseconds_as_the_json_path_does( @pytest.mark.parametrize( - "column", + "column, colp_runs", [ - [dt.datetime(2020, 1, 1), dt.date(2020, 1, 1)], # a date-time and a date - [decimal.Decimal("1.5"), decimal.Decimal("NaN")], # a non-finite Decimal - [b"a", "b"], # bytes and text + ([dt.datetime(2020, 1, 1), dt.date(2020, 1, 1)], 0), # a date-time and a date + # A non-finite Decimal: the column is refused (BigDecimal("NaN") would throw mid-batch in the image), so the NaN row goes to the + # statement on its own, as before. The row around it is a run of carriable rows: it goes in bulk, and the columnar entry + # carries a finite Decimal now, so that one run crosses on arcadedb_n_insert_colp (before, it was the JSON entry). + ([decimal.Decimal("1.5"), decimal.Decimal("NaN")], 1), + ([b"a", "b"], 0), # bytes and text ], ids=["datetime-and-date", "nan-decimal", "bytes-and-text"], ) def test_a_column_the_buffers_cannot_carry_unchanged_still_falls_back( - temp_db, spy, monkeypatch, column + temp_db, spy, monkeypatch, column, colp_runs ): + assert columnar.pack({"x": column}, columnar.Declared(known={})) is None _type(temp_db, "Ref", declared=False) _type(temp_db, "Col", declared=False) rows = [{"id": i, "x": v} for i, v in enumerate(column)] @@ -195,11 +199,11 @@ def test_a_column_the_buffers_cannot_carry_unchanged_still_falls_back( ): # the statement path may refuse it; then the new path must refuse it the same way with pytest.raises(Exception): temp_db.insert_many("Col", rows) - assert spy.count(COLP) == 0 + assert spy.count(COLP) == colp_runs return spy.calls.clear() temp_db.insert_many("Col", rows) - assert spy.count(COLP) == 0 + assert spy.count(COLP) == colp_runs assert _stored(temp_db, "Col") == _stored(temp_db, "Ref") diff --git a/tests/test_native_columnar_temporal_review_fixes.py b/tests/test_native_columnar_temporal_review_fixes.py index 31c8430e..34898f9c 100644 --- a/tests/test_native_columnar_temporal_review_fixes.py +++ b/tests/test_native_columnar_temporal_review_fixes.py @@ -10,8 +10,10 @@ """ import datetime as dt +import decimal import math +import numpy as np import pytest from arcadedb._native import columnar, lib # noqa: E402 @@ -52,7 +54,7 @@ def test_a_date64_past_int32_days_is_not_laid_out_and_never_wraps(): assert columnar.pack({"d": wraps}, NO_TYPES) is None below = pa.array([DAY_MS * (-(2**31) - 1)], pa.int64()).cast(pa.date64()) assert columnar.pack({"d": below}, NO_TYPES) is None - one_bad = pa.array([0, DAY_MS, DAY_MS * 2**40], pa.int64()).cast(pa.date64()) + one_bad = pa.array([0, DAY_MS, DAY_MS * 2**33], pa.int64()).cast(pa.date64()) assert columnar.pack({"d": one_bad}, NO_TYPES) is None # one value is enough @@ -90,6 +92,18 @@ def test_a_timestamp_at_the_limits_stays_columnar(): assert _kinds({"t": big}) == ["T"], unit +@pytest.mark.parametrize("text", ["NaN", "-NaN", "sNaN", "Infinity", "-Infinity"]) +def test_a_non_finite_decimal_is_never_laid_out(text): + # new BigDecimal("NaN") throws in the image, in the middle of a batch: the column must stay in Python, in every container + bad = decimal.Decimal(text) + assert columnar.pack({"x": [decimal.Decimal("1.5"), bad]}, NO_TYPES) is None + assert columnar.pack({"x": [None, bad, None]}, NO_TYPES) is None + as_object = np.array([decimal.Decimal("1"), bad], dtype=object) + assert columnar.pack({"x": as_object}, NO_TYPES) is None + finite = columnar.pack({"x": [decimal.Decimal("1.5"), None]}, NO_TYPES) + assert finite is not None and finite[1][0].kind == "n" # the finite column crosses + + def _type(db, name): db.command("sql", f"CREATE DOCUMENT TYPE {name}") db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") diff --git a/tests/test_native_mixed_chunk.py b/tests/test_native_mixed_chunk.py index 0c746bdf..cf409cb6 100644 --- a/tests/test_native_mixed_chunk.py +++ b/tests/test_native_mixed_chunk.py @@ -250,8 +250,10 @@ def test_an_uncarriable_value_sends_its_own_row_to_the_statement_not_the_chunk( spy.clear() db.insert_many("Bulk", rows) assert spy.count(MANY) == 1, "one statement run for the one uncarriable row" + # the rows before and after it stay on the bulk entry; its columnar form carries the date-time column now, so each run + # crosses on arcadedb_n_insert_colp (before it was the JSON entry, arcadedb_n_insert_rows) assert ( - spy.count(ROWS) == 2 + spy.count(COLP) == 2 and spy.count(ROWS) == 0 ), "the rows before and after it stay on the bulk entry" assert _stored(db, "Bulk") == _stored(db, "Ref")