From 4cb904dbdbf6b058d981e12b2fab4dff6499d2dd Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Wed, 4 Nov 2026 05:48:50 +0900 Subject: [PATCH 1/3] Native append_samples: numeric columns cross by address, not as .tolist() and JSON text (#228) NativeAsyncExecutor.append_samples called .tolist() on every column, walked the timestamps again with int(), and sent the chunk as one JSON text to arcadedb_n_async. A one-dimensional numpy int, float or bool column and an int or datetime64[ns] timestamp array now go through the new arcadedb_n_async_bin: the JSON holds a kind marker per column ("q" int64, "d" float64) and the sample count, the data crosses in an address table of (address, count) longs, the convention of arcadedb_n_insert_colp. The host tag column and anything the text path converts differently (lists, float timestamps, other datetime64 units, 2-D, empty, uint64 past int64, a column of another length) keep the text. Java builds the same long[] and boxed Object[] columns and calls the same appendSamples, so the stored samples are bit-identical (NaN is the canonical NaN of the text). A library without the entry, or ARCADEDB_NATIVE_TS_BIN=0, keeps the text path; primitive=True is the same length check. Tests (native only, run in CI): the lane's shape takes the entry and reads back by count, sums and a window; every dtype is stored as the text path stores it; the entry choice per shape; the Java checks of a wrong table. Co-Authored-By: Claude Opus 5.5 --- .../org/humemai/nativelib/NativeExtras.java | 85 +++- src/arcadedb/_native/codec.py | 41 ++ src/arcadedb/_native/extras.py | 71 ++- src/arcadedb/_native/lib.py | 9 +- tests/conftest.py | 2 + tests/test_native_timeseries_binary.py | 477 ++++++++++++++++++ 6 files changed, 666 insertions(+), 19 deletions(-) create mode 100644 tests/test_native_timeseries_binary.py diff --git a/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java index 7ee05e9..6c04e7b 100644 --- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java +++ b/native/src/main/java/org/humemai/nativelib/NativeExtras.java @@ -574,8 +574,67 @@ private static boolean isVertexType(final Database db, final String typeName) { * cb, err} | on_ok {cb} | on_error {cb} | append_samples {type, ts, cols} | insert_parallel {type, rows} | failures {fid}. */ @CEntryPoint(name = "arcadedb_n_async") - @SuppressWarnings("unchecked") public static CCharPointer async(final IsolateThread thread, final long dbh, final CCharPointer op, final CCharPointer argsJson) { + return asyncImpl(dbh, op, argsJson, 0L, 0); + } + + /** + * Same as arcadedb_n_async plus an address table (pointer and count of longs, TS_SLOTS per slot: data address, element count) that the + * append_samples arguments point into, as arcadedb_n_insert_colp reads its columns: "ts" and every entry of "cols" is either a JSON array + * of values, as for arcadedb_n_async, or the kind of a column in the table, "q" (int64) or "d" (float64), slot 0 being the timestamps and + * slot 1 + i the column i. "n" is the number of samples. The caller keeps every buffer alive and unchanged until the call returns. + */ + @CEntryPoint(name = "arcadedb_n_async_bin") + public static CCharPointer asyncBin(final IsolateThread thread, final long dbh, final CCharPointer op, final CCharPointer argsJson, + final CCharPointer table, final int nTable) { + return asyncImpl(dbh, op, argsJson, table.rawValue(), nTable); + } + + /** Longs per slot of the address table of arcadedb_n_async_bin: the data address and the element count. */ + private static final int TS_SLOTS = 2; + + /** The data address of a table slot, checked to hold n elements of the kind: "q" for the timestamps, "q" or "d" for a column. */ + private static long sampleAddress(final long table, final int nTable, final int slot, final Object kind, final long n, final boolean timestamps) { + final String what = timestamps ? "the timestamps" : "column " + (slot - 1); + if (!"q".equals(kind) && (timestamps || !"d".equals(kind))) + throw new IllegalArgumentException(what + " has the unknown kind " + kind); + if (n < 0 || n > Integer.MAX_VALUE) + throw new IllegalArgumentException("sample count out of range: " + n); + if (table == 0 || nTable < TS_SLOTS * (slot + 1)) + throw new IllegalArgumentException(what + " is not in the address table of " + nTable + " entries"); + final Pointer t = WordFactory.pointer(table); + final long address = t.readLong(WordFactory.signed(8L * TS_SLOTS * slot)); + final long count = t.readLong(WordFactory.signed(8L * (TS_SLOTS * slot + 1))); + if (count != n || (n > 0 && address == 0)) + throw new IllegalArgumentException(what + " has " + count + " values in the address table, " + n + " samples"); + return address; + } + + /** A column of the table as the boxed values the text path gives: a Long for "q", a Double for "d" (a NaN is the canonical NaN of the text). */ + private static Object[] sampleColumn(final long address, final boolean doubles, final int n) { + final Pointer p = WordFactory.pointer(address); + final Object[] values = new Object[n]; + for (int i = 0; i < n; i++) { + final long bits = p.readLong(WordFactory.signed(8L * i)); + if (doubles) { + final double d = Double.longBitsToDouble(bits); + values[i] = d != d ? Double.NaN : d; + } else + values[i] = bits; + } + return values; + } + + private static long[] sampleLongs(final long address, final int n) { + final Pointer p = WordFactory.pointer(address); + final long[] values = new long[n]; + for (int i = 0; i < n; i++) + values[i] = p.readLong(WordFactory.signed(8L * i)); + return values; + } + + @SuppressWarnings("unchecked") + private static CCharPointer asyncImpl(final long dbh, final CCharPointer op, final CCharPointer argsJson, final long table, final int nTable) { try { final DatabaseInternal db = (DatabaseInternal) ArcadeNative.db(dbh); final DatabaseAsyncExecutor ex = db.async(); @@ -726,14 +785,26 @@ else if (a.get("slot") instanceof Number slot) event("scan_error", "\"rid\":\"" + rid + "\"," + errorBody(t)))); } case "append_samples" -> { - final List ts = (List) a.get("ts"); - final long[] stamps = new long[ts.size()]; - for (int i = 0; i < stamps.length; i++) - stamps[i] = ((Number) ts.get(i)).longValue(); + final Object tsArg = a.get("ts"); + final long n = a.get("n") instanceof Number num ? num.longValue() : -1L; + final long[] stamps; + if (tsArg instanceof String kind) + stamps = sampleLongs(sampleAddress(table, nTable, 0, kind, n, true), (int) n); + else { + final List ts = (List) tsArg; + stamps = new long[ts.size()]; + for (int i = 0; i < stamps.length; i++) + stamps[i] = ((Number) ts.get(i)).longValue(); + } final List cols = (List) a.get("cols"); final Object[][] columns = new Object[cols.size()][]; - for (int i = 0; i < columns.length; i++) - columns[i] = ((List) cols.get(i)).toArray(); + for (int i = 0; i < columns.length; i++) { + final Object col = cols.get(i); + if (col instanceof String kind) + columns[i] = sampleColumn(sampleAddress(table, nTable, i + 1, kind, n, false), "d".equals(kind), (int) n); + else + columns[i] = ((List) col).toArray(); + } ex.appendSamples((String) a.get("type"), stamps, columns); } case "insert_parallel" -> { diff --git a/src/arcadedb/_native/codec.py b/src/arcadedb/_native/codec.py index 929f548..b157b61 100644 --- a/src/arcadedb/_native/codec.py +++ b/src/arcadedb/_native/codec.py @@ -222,6 +222,47 @@ def _dump(obj, binary): return text, buf, state[1] +_INT64_MAX = 2**63 - 1 +_NS_DATETIME = None if _np is None else _np.dtype("datetime64[ns]") + + +def sample_array(values, timestamps=False): + """The int64 or float64 array that stands for a column of ``append_samples`` in the address table of ``arcadedb_n_async_bin``, or + None when the column keeps the JSON text it has always had. It is the array whose ``.tolist()`` the text path writes, value for + value: an int, uint (uint64 only inside the int64 range) or bool array is int64, a float array narrower than long double is float64 + (a float32 widens exactly), and, for the timestamps only, a ``datetime64[ns]`` array without NaT is its int64 count (what ``.tolist()`` + gives for that unit). Anything else, a list, a subclass of ndarray, a 2-D or empty array, a float array as timestamps, other + ``datetime64`` units (``int()`` refuses their datetimes on the text path), a ``datetime64[ns]`` of non-native byte order, returns None. + """ + if ( + _np is None + or type(values) is not _np.ndarray + or values.ndim != 1 + or not values.size + ): + return None + dtype = values.dtype + kind = dtype.kind + if kind in "ib": + return _np.ascontiguousarray(values, dtype=_np.int64) + if kind == "u": + if dtype.itemsize == 8 and int(values.max()) > _INT64_MAX: + return None + return _np.ascontiguousarray(values, dtype=_np.int64) + if kind == "f": + if timestamps or dtype.itemsize > 8: + return None + return _np.ascontiguousarray(values, dtype=_np.float64) + if kind == "M" and timestamps and dtype == _NS_DATETIME: + counts = _np.ascontiguousarray(values.view(_np.int64)) + if ( + counts == _np.iinfo(_np.int64).min + ).any(): # NaT: the text path raises on its None + return None + return counts + return None + + def row_default(value): """json.dumps hook for the rows of a bulk load: a NumPy float array that is not float32 is the list it would have been (a float64 array stored on an undeclared property keeps its doubles, as on the JPype path); the rest as :func:`_default`. diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index 03a68d7..ea9583b 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -20,7 +20,7 @@ from ..exceptions import ArcadeDBError from ..graph_batch import GraphBatch as _BaseGraphBatch from . import interrupts, lib, not_yet -from .codec import _PLAIN, _default, _prep, loads +from .codec import _PLAIN, _default, _np, _prep, loads, sample_array # -------------------------------------------------------------------------------------------------------- upcall @@ -605,20 +605,69 @@ def plain(values): values = values.astype("int64") return values.tolist() if hasattr(values, "tolist") else list(values) - stamps = [int(t) for t in plain(timestamps)] + # numpy int and float columns (and an integer timestamp array) cross by address, the rest as the JSON text they always had + ts_array = sample_array(timestamps, True) if lib.TS_BIN else None + if ts_array is not None: + stamps, count = None, len(ts_array) + else: + stamps = [int(t) for t in plain(timestamps)] + count = len(stamps) if primitive: - # the engine's primitive TimeSeriesBatch is a JVM-boundary optimisation (no boxing per value); here the rows cross - # as one JSON text either way, so the columns take the same path, with the primitive path's length check + # the engine's primitive TimeSeriesBatch is a JVM-boundary optimisation (no boxing per value); here the columns + # take the same path as without it, with the primitive path's length check for index, values in enumerate(column_values): - if len(values) != len(stamps): + if len(values) != count: raise ValueError( - f"column {index} has {len(values)} values for {len(stamps)} timestamps" + f"column {index} has {len(values)} values for {count} timestamps" ) - self._op( - "append_samples", - type=type_name, - ts=stamps, - cols=[plain(c) for c in column_values], + arrays = [] + if lib.TS_BIN: + for values in column_values: + array = sample_array(values) + # a column of another length stays text: the engine pads or cuts it as it always did + arrays.append( + array if array is not None and len(array) == count else None + ) + if ts_array is None and not any(a is not None for a in arrays): + self._op( + "append_samples", + type=type_name, + ts=stamps, + cols=[plain(c) for c in column_values], + ) + return + # the address table: per slot (0 the timestamps, 1 + i the column i) the data address and the element count of an array that + # stays alive until the call returns; a column that is text has no entry (zeros) and its values are in the JSON + table = _np.zeros(2 * (1 + len(column_values)), dtype=_np.int64) + if ts_array is not None: + table[0], table[1] = ts_array.ctypes.data, count + cols = [] + for index, (values, array) in enumerate(zip(column_values, arrays)): + if array is None: + cols.append(plain(values)) + else: + table[2 * (index + 1)], table[2 * (index + 1) + 1] = ( + array.ctypes.data, + count, + ) + cols.append("d" if array.dtype.kind == "f" else "q") + _call( + "Async executor failed", + "arcadedb_n_async_bin", + self._db._h, + lib.b("append_samples"), + lib.b( + _enc( + { + "type": type_name, + "n": count, + "ts": "q" if stamps is None else stamps, + "cols": cols, + } + ) + ), + table.ctypes.data, + len(table), ) def _edge_callback(self, callback): diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index 00d20b7..041f0ed 100644 --- a/src/arcadedb/_native/lib.py +++ b/src/arcadedb/_native/lib.py @@ -31,6 +31,8 @@ EXEC_MANY = os.environ.get("ARCADEDB_NATIVE_EXEC_MANY", "1") != "0" # ARCADEDB_NATIVE_NEXT_BIN: the batches after the first read through arcadedb_n_next_bin, float arrays of 8 or more elements in a float32 tail (as flag 4 of arcadedb_n_exec does for the first batch), not as number text NEXT_BIN = os.environ.get("ARCADEDB_NATIVE_NEXT_BIN", "1") != "0" +# ARCADEDB_NATIVE_TS_BIN: the numpy int and float columns (and an integer timestamp array) of append_samples cross by address in arcadedb_n_async_bin, not as .tolist() and JSON text +TS_BIN = os.environ.get("ARCADEDB_NATIVE_TS_BIN", "1") != "0" # the directory the engine log, logging.properties and the crash files go to (None until the library is loaded) LOG_DIR_USED = None @@ -243,6 +245,7 @@ class _Params(ctypes.Structure): "arcadedb_n_gb_create": [P, L, S], "arcadedb_n_gb": [P, L, S, S], "arcadedb_n_async": [P, L, S, S], + "arcadedb_n_async_bin": [P, L, S, S, P, I], "arcadedb_n_vindex": [P, L, S, S], "arcadedb_n_import": [P, L, S], "arcadedb_n_export": [P, L, S, S, I, S], @@ -266,7 +269,7 @@ class _Params(ctypes.Structure): ) else ctypes.c_longlong if name == "arcadedb_n_cmd" else ctypes.c_void_p ) - global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY, NEXT_BIN + global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY, NEXT_BIN, TS_BIN if not hasattr( lib, "arcadedb_n_exec_bin" ): # a library built before the binary path @@ -283,6 +286,10 @@ class _Params(ctypes.Structure): lib, "arcadedb_n_next_bin" ): # a library built before the binary tail of later batches NEXT_BIN = False + if not hasattr( + lib, "arcadedb_n_async_bin" + ): # a library built before the address table of append_samples + TS_BIN = False _isolate = isolate _creator = threading.get_ident() _local.thread = _Thread(thread, owned=False) diff --git a/tests/conftest.py b/tests/conftest.py index 01ab332..445be73 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -63,6 +63,8 @@ collect_ignore.append("test_native_tx_crossings.py") # counts the statement texts the native backend hands to the engine (the JPype backend has no such layer) collect_ignore.append("test_native_statement_texts.py") + # the address table of append_samples (arcadedb_n_async_bin), observed on the native library + collect_ignore.append("test_native_timeseries_binary.py") if not os.path.isfile( os.path.join(os.path.dirname(os.path.dirname(__file__)), "docs", "index.md") ): diff --git a/tests/test_native_timeseries_binary.py b/tests/test_native_timeseries_binary.py new file mode 100644 index 0000000..3601724 --- /dev/null +++ b/tests/test_native_timeseries_binary.py @@ -0,0 +1,477 @@ +"""``append_samples`` on the native build sends numpy columns by address, not as ``.tolist()`` and one JSON text per chunk. + +A one-dimensional numpy int, float or bool column, and an int or ``datetime64[ns]`` timestamp array, go through ``arcadedb_n_async_bin``: +the JSON of the call holds a kind marker per such column ("q" int64, "d" float64) and the sample count, and the data crosses in an address +table, as ``arcadedb_n_insert_colp`` does for a bulk insert. Everything else (the host tag column of strings, lists, a float array as +timestamps, a column of another length) keeps the text of ``arcadedb_n_async``. The stored samples do not change: the Java side builds the +same ``long[]`` and boxed ``Object[]`` columns the text gave and calls the same ``appendSamples``. A library without the entry +(``lib.TS_BIN`` false) keeps the text path; ``primitive=True`` is the same length check as before. + +The calls are observed through ``lib.call`` (the ctypes door of the native library, as in test_native_tx_crossings.py). Needs the native +library, so conftest.py collects this file on the native backend only. +""" + +import json +import struct + +import numpy as np +import pytest + +from arcadedb import ArcadeDBError +from arcadedb._native import codec +from arcadedb._native import extras as native_extras +from arcadedb._native import lib + +BASE = 1_700_000_000_000 # epoch ms + + +class _Calls: + """The ``(entry point, operation, JSON text, rest of the arguments)`` of every async call made while it is installed. + + ``stub``: the calls are recorded and answered with ``{"ok":true}`` without reaching the library, for the tests that only look at + which entry a call takes (a call that the engine would refuse, or that is not worth a database). + """ + + def __init__(self, monkeypatch, stub=False): + self.calls = [] + real = lib.call + + def spy(name, *args): + if name in ("arcadedb_n_async", "arcadedb_n_async_bin"): + self.calls.append((name, args[1].decode(), args[2].decode(), args[3:])) + if stub and args[1] == b"append_samples": + return '{"ok":true}' + return real(name, *args) + + monkeypatch.setattr(lib, "call", spy) + + def entries(self, op="append_samples"): + return [c[0] for c in self.calls if c[1] == op] + + def texts(self, op="append_samples"): + return [json.loads(c[2]) for c in self.calls if c[1] == op] + + +def _type(db, name, fields="d1 DOUBLE, d2 DOUBLE, d3 DOUBLE"): + db.command( + "sql", + f"CREATE TIMESERIES TYPE {name} TIMESTAMP ts TAGS (host STRING) FIELDS ({fields}) SHARDS 2", + ) + + +def _canon(value): + # a float by its bits (-0.0, NaN and the last digit all count), anything else by its repr + if isinstance(value, float): + return struct.pack("i8") + ex = temp_db.async_executor() + _type(temp_db, "Odd", "d1 DOUBLE, l1 LONG") + ex.append_samples("Odd", ts, ["a"] * n, strided, big) + ex.wait_completion() + assert calls.entries() == ["arcadedb_n_async_bin"] + got = temp_db.query("sql", "SELECT ts, d1, l1 FROM Odd ORDER BY ts").to_list() + assert [float(r["d1"]) for r in got] == [float(x) for x in strided] + assert [int(r["l1"]) for r in got] == [5 * i for i in range(n)] + + +# ----------------------------------------------------------------------------------------------------------- what keeps the text path + + +def test_lists_keep_the_text_entry(temp_db, monkeypatch): + calls = _Calls(monkeypatch) + _type(temp_db, "Lists") + ex = temp_db.async_executor() + ex.append_samples( + "Lists", [BASE, BASE + 1], ["a", "b"], [1.0, 2.0], [3.0, 4.0], [5.0, 6.0] + ) + ex.wait_completion() + assert calls.entries() == ["arcadedb_n_async"] + assert _count_sums(temp_db, "Lists", ["d1", "d2", "d3"]) == [2, 3.0, 7.0, 11.0] + + +def test_a_timestamp_list_with_numpy_columns_sends_the_timestamps_as_text( + temp_db, monkeypatch +): + calls = _Calls(monkeypatch) + _type(temp_db, "MixedTs") + ts = [BASE + i for i in range(20)] + d = np.arange(20, dtype=np.float64) + ex = temp_db.async_executor() + ex.append_samples("MixedTs", ts, ["a"] * 20, d, d, d) + ex.wait_completion() + assert calls.entries() == ["arcadedb_n_async_bin"] + text = calls.texts()[0] + assert text["ts"] == ts and text["cols"][1:] == ["d", "d", "d"] + assert _count_sums(temp_db, "MixedTs", ["d1"]) == [20, float(d.sum())] + + +def test_float_timestamps_keep_the_text_path_and_the_columns_still_go_by_address( + temp_db, monkeypatch +): + calls = _Calls(monkeypatch, stub=True) + ex = temp_db.async_executor() + ts = ( + np.arange(5, dtype=np.float64) + BASE + ) # int() truncates float timestamps on the text path + ex.append_samples("T", ts, ["a"] * 5, [1.0] * 5) + assert calls.entries() == ["arcadedb_n_async"] + assert calls.texts()[0]["ts"] == [BASE + i for i in range(5)] + ex.append_samples("T", ts + 0.7, ["a"] * 5, np.ones(5)) + assert calls.entries() == ["arcadedb_n_async", "arcadedb_n_async_bin"] + text = calls.texts()[1] + assert text["ts"] == [BASE + i for i in range(5)] and text["cols"][1] == "d" + + +def test_datetime64_milliseconds_are_refused_as_before(temp_db, monkeypatch): + calls = _Calls(monkeypatch, stub=True) + ts = np.arange(5).astype("datetime64[ms]") + np.datetime64(BASE, "ms") + with pytest.raises(TypeError): # int() of a datetime, on the text path + temp_db.async_executor().append_samples("T", ts, ["a"] * 5, np.zeros(5)) + assert calls.entries() == [] + + +def test_a_column_of_another_length_keeps_the_text_path(temp_db, monkeypatch): + calls = _Calls(monkeypatch, stub=True) + ex = temp_db.async_executor() + ex.append_samples( + "T", np.arange(10, dtype=np.int64), ["a"] * 10, np.zeros(9), np.zeros(10) + ) + (text,) = calls.texts() + assert calls.entries() == [ + "arcadedb_n_async_bin" + ] # the timestamps and the column of 10 go by address, the column of 9 is text + assert ( + text["cols"][0] == ["a"] * 10 + and text["cols"][1] == [0.0] * 9 + and text["cols"][2] == "d" + ) + calls.calls.clear() + ex.append_samples( + "T", [1, 2, 3], ["a"] * 3, np.zeros(2) + ) # no column of the call's length and the timestamps a list + assert calls.entries() == ["arcadedb_n_async"] + + +def test_primitive_true_keeps_its_length_check_and_takes_the_same_entry( + temp_db, monkeypatch +): + calls = _Calls(monkeypatch, stub=True) + ex = temp_db.async_executor() + with pytest.raises(ValueError, match="column 1 has 9 values for 10 timestamps"): + ex.append_samples( + "T", np.arange(10, dtype=np.int64), ["a"] * 10, np.zeros(9), primitive=True + ) + with pytest.raises(ValueError, match="column 0 has 3 values for 10 timestamps"): + ex.append_samples("T", np.arange(10, dtype=np.int64), ["a"] * 3, primitive=True) + assert calls.entries() == [] # the check comes before the crossing + ex.append_samples( + "T", np.arange(10, dtype=np.int64), ["a"] * 10, np.zeros(10), primitive=True + ) + ex.append_samples("T", np.arange(10, dtype=np.int64), ["a"] * 10, np.zeros(10)) + assert calls.entries() == ["arcadedb_n_async_bin"] * 2 + assert calls.texts()[0] == calls.texts()[1] + + +def test_a_library_without_the_entry_keeps_the_text_path(temp_db, monkeypatch): + # the probe flag as the loader leaves it for a library built before arcadedb_n_async_bin + monkeypatch.setattr(lib, "TS_BIN", False) + calls = _Calls(monkeypatch) + _type(temp_db, "Old") + n = 1_000 + d = np.arange(n, dtype=np.float64) + ex = temp_db.async_executor() + ex.append_samples( + "Old", BASE + np.arange(n, dtype=np.int64), ["a"] * n, d, d, d, primitive=True + ) + ex.wait_completion() + assert calls.entries() == ["arcadedb_n_async"] + assert _count_sums(temp_db, "Old", ["d1", "d2", "d3"]) == [n] + [float(d.sum())] * 3 + + +# ----------------------------------------------------------------------------------------------------------- sample_array + + +_INT64_MAX = 2**63 - 1 + + +@pytest.mark.parametrize( + "values, timestamps, want", + [ + (np.arange(3, dtype=np.int8), False, np.int64), + (np.arange(3, dtype=np.int64), True, np.int64), + (np.arange(3, dtype=np.uint32), False, np.int64), + (np.array([1, 2], dtype=np.uint64), False, np.int64), + ( + np.array([1, _INT64_MAX + 1], dtype=np.uint64), + False, + None, + ), # past the int64 range: the text keeps its Python ints + (np.array([True, False]), False, np.int64), + (np.array([True, False]), True, np.int64), + (np.ones(3, dtype=np.float16), False, np.float64), + (np.ones(3, dtype=np.float32), False, np.float64), + (np.ones(3, dtype=np.float64), False, np.float64), + ( + np.ones(3, dtype=np.float64), + True, + None, + ), # int() truncates float timestamps on the text path + (np.ones(3, dtype=np.longdouble), False, None), + (np.array([1, 2], dtype="datetime64[ns]"), True, np.int64), + (np.array([1, 2], dtype="datetime64[ns]"), False, None), + ( + np.array(["1970-01-01T00:00:00.000000001", "NaT"], dtype="datetime64[ns]"), + True, + None, + ), + (np.array([1, 2], dtype="datetime64[ms]"), True, None), + (np.array([1, 2], dtype=">M8[ns]"), True, None), + (np.zeros(0, dtype=np.float64), False, None), + (np.zeros((2, 2)), False, None), + ( + np.ma.masked_array([1.0, 2.0], mask=[False, True]), + False, + None, + ), # tolist() gives None for the masked value + (np.array(["a", "b"]), False, None), + (np.array([1, 2], dtype=object), False, None), + ([1.0, 2.0], False, None), + (None, False, None), + ], +) +def test_sample_array_picks_the_columns_whose_tolist_it_reproduces( + values, timestamps, want +): + got = codec.sample_array(values, timestamps) + if want is None: + assert got is None + return + assert got.dtype == want and got.flags.c_contiguous and got.ndim == 1 + if isinstance(values, np.ndarray) and values.dtype.kind in "iub": + assert got.tolist() == [int(v) for v in values.tolist()] + if isinstance(values, np.ndarray) and values.dtype.kind == "f": + assert ( + got.tolist() == values.tolist() + ) # the same doubles .tolist() writes into the text (a float32 widens exactly) + if isinstance(values, np.ndarray) and values.dtype.kind == "M": + assert ( + got.tolist() == values.tolist() + ) # the ints .tolist() gives for datetime64[ns] + + +# ----------------------------------------------------------------------------------------------------------- the Java side checks the table + + +def _bin(db, text, table, count): + return native_extras._call( + "append_samples", + "arcadedb_n_async_bin", + db._h, + lib.b("append_samples"), + lib.b(json.dumps(text)), + table.ctypes.data if table is not None else None, + count, + ) + + +def test_a_wrong_address_table_is_refused_before_anything_is_written(temp_db): + _type(temp_db, "Bad") + ts = np.arange(4, dtype=np.int64) + BASE + table = np.zeros(4, dtype=np.int64) + table[0], table[1] = ( + ts.ctypes.data, + 3, + ) # the table says 3 timestamps, the call says 4 + args = {"type": "Bad", "n": 4, "ts": "q", "cols": [["a"] * 4]} + with pytest.raises(ArcadeDBError, match="address table"): + _bin(temp_db, args, table, len(table)) + table[1] = 4 + with pytest.raises(ArcadeDBError, match="address table"): + _bin( + temp_db, args, table, 1 + ) # the table is shorter than the slots the call uses + with pytest.raises(ArcadeDBError, match="address table"): + _bin(temp_db, args, None, 0) + with pytest.raises(ArcadeDBError, match="unknown kind"): + _bin(temp_db, {**args, "ts": "x"}, table, len(table)) + with pytest.raises(ArcadeDBError, match="unknown kind"): + _bin(temp_db, {**args, "ts": "d"}, table, len(table)) # a timestamp is int64 + with pytest.raises(ArcadeDBError, match="sample count out of range"): + _bin(temp_db, {**args, "n": -1}, table, len(table)) + temp_db.async_executor().wait_completion() + assert temp_db.query("sql", "SELECT count(*) AS n FROM Bad").to_list()[0]["n"] == 0 From 28b8cd7c1e76b047f6acc294e6615f40cc31a945 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Wed, 4 Nov 2026 06:40:57 +0900 Subject: [PATCH 2/3] Native append_samples tests: fix three wrong expectations of the new test file (#228) CI run 41716062800 failed only these three, all in tests/test_native_timeseries_binary.py, all in the test code: the window of the lane test ended at index 1900 so host_7 had 9 points, not 10 (end now 1999); an int32 array into a DOUBLE column is an int64 column ("q"), the expected kinds listed "d"; datetime64 + datetime64 is not defined in numpy, the datetime64[ms] timestamps are now datetime64 + timedelta64. No source change: every other parity file of the run passed (2036 passed). Co-Authored-By: Claude Opus 5.5 --- tests/test_native_timeseries_binary.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/tests/test_native_timeseries_binary.py b/tests/test_native_timeseries_binary.py index 3601724..01227b3 100644 --- a/tests/test_native_timeseries_binary.py +++ b/tests/test_native_timeseries_binary.py @@ -120,7 +120,10 @@ def test_the_lane_shape_takes_the_binary_entry_and_stores_the_source( float(c.sum()) for c in cols ] # windows: ten instants of one host, as the probe reads them - lo_ts, hi_ts = BASE + 1000 * 1000, BASE + 1000 * 1000 + 9 * hosts * 1000 + lo_ts, hi_ts = ( + BASE + 1000 * 1000, + BASE + 1000 * 1000 + 10 * hosts * 1000 - 1, + ) # indices 1000 to 1999: host_7 at 1007, 1107 .. 1907 got = temp_db.query( "sql", f"SELECT ts, d1 FROM Lane WHERE host = 'host_7' AND ts BETWEEN {lo_ts} AND {hi_ts} ORDER BY ts", # nosec B608 @@ -185,7 +188,16 @@ def test_every_numeric_dtype_is_stored_as_the_text_path_stored_it(temp_db, monke ex.append_samples("Bin", ts, host, *cols) ex.wait_completion() assert calls.entries() == ["arcadedb_n_async_bin"] - assert calls.texts()[0]["cols"][1:] == ["d", "d", "d", "d", "q", "q", "q", "d"] + assert calls.texts()[0]["cols"][1:] == [ + "d", + "d", + "d", + "q", + "q", + "q", + "q", + "d", + ] # d4 is an int32 array: int64 monkeypatch.setattr( lib, "TS_BIN", False ) # the library as a build without the entry leaves the flag @@ -299,7 +311,7 @@ def test_float_timestamps_keep_the_text_path_and_the_columns_still_go_by_address def test_datetime64_milliseconds_are_refused_as_before(temp_db, monkeypatch): calls = _Calls(monkeypatch, stub=True) - ts = np.arange(5).astype("datetime64[ms]") + np.datetime64(BASE, "ms") + ts = np.datetime64(BASE, "ms") + np.arange(5).astype("timedelta64[ms]") with pytest.raises(TypeError): # int() of a datetime, on the text path temp_db.async_executor().append_samples("T", ts, ["a"] * 5, np.zeros(5)) assert calls.entries() == [] From 8c5f10618ae58c1879b23e9bfee24cceba82bef1 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Wed, 4 Nov 2026 08:44:43 +0900 Subject: [PATCH 3/3] Test: longdouble is float64 on Windows and Apple arm64, so sample_array may take it there The parity leg on windows-x64 py3.14 failed one case (values11): the test expected None for a np.longdouble array, which holds only on platforms where longdouble is 80-bit extended. Where it is float64 the picker is right to take it (tolist reproduces the values). Co-Authored-By: Claude Opus 5.5 --- tests/test_native_timeseries_binary.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/test_native_timeseries_binary.py b/tests/test_native_timeseries_binary.py index 01227b3..bd26af1 100644 --- a/tests/test_native_timeseries_binary.py +++ b/tests/test_native_timeseries_binary.py @@ -403,7 +403,11 @@ def test_a_library_without_the_entry_keeps_the_text_path(temp_db, monkeypatch): True, None, ), # int() truncates float timestamps on the text path - (np.ones(3, dtype=np.longdouble), False, None), + ( + np.ones(3, dtype=np.longdouble), + False, + np.float64 if np.dtype(np.longdouble) == np.dtype(np.float64) else None, + ), # longdouble is float64 on Windows (MSVC) and Apple arm64, 80-bit extended on x86-64 Linux (np.array([1, 2], dtype="datetime64[ns]"), True, np.int64), (np.array([1, 2], dtype="datetime64[ns]"), False, None), (