diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc index ddc06105..9febc7f6 100644 --- a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc +++ b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc @@ -1,5 +1,5 @@ = Bulk loads and commit size -:description: The default commit size of insert_many, graph_batch and create_vertices on the arcadedb (native) and arcadedb-jvm packages, what it means for a failed load, what to do after an interrupted graph batch, how to change it, how insert_many treats Python's cyclic garbage collector on arcadedb-jvm, which property shapes a graph batch loads in bulk on the JVM build, and a known limitation: a graph batch does not reuse the space of deleted edges. +:description: The default commit size of insert_many, graph_batch and create_vertices on the arcadedb (native) and arcadedb-jvm packages, what it means for a failed load, what to do after an interrupted graph batch, how to change it, how insert_many treats Python's cyclic garbage collector on arcadedb-jvm, which property shapes a graph batch loads in bulk on the JVM build, how to pass time-series columns to append_samples on arcadedb (native), and a known limitation: a graph batch does not reuse the space of deleted edges. `insert_many`, `graph_batch` and `GraphBatch.create_vertices` load data in transactions. The commit size is the number of rows in one transaction. It decides how much work the engine holds in memory before it commits, and how many rows stay in the database when a load fails. @@ -111,6 +111,22 @@ The table is one laptop study of the `arcadedb-jvm` package: per-row time of thi |=== +== Time-series samples with `append_samples` (`arcadedb`, native) + +`db.async_executor().append_samples(type_name, timestamps, *columns)` loads a columnar batch into a TIMESERIES type: the timestamps, then one column for each tag and field in the order of the type's declaration. Call `wait_completion()` before you read the samples back. How to declare the type is in the main ArcadeDB docs: see {arcadedb-docs-url}/concepts/timeseries[Time series]. + +On the `arcadedb` (native) package, a NumPy column crosses to the engine by its address, with no JSON text, when it is one of these: + +* The timestamps: a one-dimensional, non-empty integer or bool array (`int64` is the plain case), or a `datetime64[ns]` array without `NaT`. +* A numeric column: a one-dimensional, non-empty array of any integer or bool kind, or of a float kind up to `float64` (`float32` widens to `float64` without a change of value). A `uint64` array, as timestamps or as a column, crosses only if its largest value fits in `int64`. +* Each column must have as many values as there are timestamps. + +Everything else keeps the text path, which stores the same values and only takes longer: a list or a tuple, a subclass of `ndarray`, a two-dimensional or empty array, a float array as the timestamps, a `datetime64` array of another unit or with `NaT`, a `long double` array, and a string or tag column. A call can mix both: the numeric columns cross by address and the tag column goes as text. To get the fast path, build the timestamps and the numeric columns as NumPy arrays with these dtypes and pass the arrays as they are, without `.tolist()`. An array that is not already contiguous `int64` or `float64` is converted first, which copies it in NumPy. + +The measurement is a laptop A/B on the `arcadedb` (native) wheel, one fresh process per run, Python 3.12 and NumPy 2.5.4. The workload loaded 2,000,000 points of 100 hosts in chunks of 100,000: an `int64` timestamp array, one list of host names (the tag), and three `float64` arrays, with one `wait_completion()` at the end. The figure is the whole ingest time in seconds, including that wait, as the ratio of the build with this change to the build without it (below 1 means faster): 0.370 (95% interval 0.362 to 0.390, median of 11 clean paired runs). A document insert in the same runs, which does not use this path, took 1.020 (95% interval 1.016 to 1.022) of the time on the build with this change. The figure holds for this shape: your gain depends on how many numeric columns a call carries and how large the chunks are. + +`ARCADEDB_NATIVE_TS_BIN=0` in the environment turns the address path off and returns to the text path. A native library that is older than this change falls back to the text path by itself. + == Graph batches next to deletes (fixed in ArcadeDB 26.11.1) NOTE: *Fixed in ArcadeDB 26.11.1.* Before 26.11.1, a graph batch did not reuse the space of deleted edges (https://github.com/ArcadeData/arcadedb/issues/10746[ArcadeData/arcadedb#10746]). The engine wrote the edge records of each `graph_batch()` flush onto new pages and never looked at the free space that deletes left on older pages, so the edge bucket gained one 64 KiB page per flush while the live data stayed flat. From 26.11.1 (https://github.com/ArcadeData/arcadedb/pull/10836[ArcadeData/arcadedb#10836]) each edge goes where an ordinary insert would put it while an existing page has room, and only the rest of a flush goes onto new pages, so a large import stays as fast as before. The 26.11.1 packages ship this fix. The `26.11.1.devN` pre-releases embed engine 26.10.2 and still show the growth described below. diff --git a/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java index b1a518a3..a62574ad 100644 --- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java +++ b/native/src/main/java/org/humemai/nativelib/NativeExtras.java @@ -590,8 +590,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(); @@ -742,14 +801,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 9f8816b9..6f0503ba 100644 --- a/src/arcadedb/_native/codec.py +++ b/src/arcadedb/_native/codec.py @@ -276,6 +276,47 @@ def encode_gb_args(args): 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 42e3ef3a..5f8a4702 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, encode_gb_args, loads +from .codec import _PLAIN, _default, _np, _prep, encode_gb_args, loads, sample_array # -------------------------------------------------------------------------------------------------------- upcall @@ -614,6 +614,23 @@ def command( def append_samples( self, type_name, timestamps, *column_values, primitive: bool = False ): + """Columnar bulk append into a TIMESERIES type, the same call as on the JVM build. + + Columns follow the type's declaration order (tags first, then fields); call wait_completion() before reading the samples back. + + What to pass to get the fast path: NumPy arrays, not lists. A one-dimensional, non-empty int, uint (uint64 only inside the int64 + range) or bool array, or a datetime64[ns] array without NaT, as the timestamps, and an int, uint, bool or float array (float32 and + float64; not long double) as a numeric column, each of the length of the timestamps, cross by address, with no JSON text (an array + that is not already contiguous int64 or float64 is converted first, which copies it). A list, a tuple, an ndarray subclass, a 2-D + or empty array, a float array as timestamps, another datetime64 unit, a string or tag column and a column of another length keep + the text path, which stores the same values. One call can mix both. + ARCADEDB_NATIVE_TS_BIN=0 turns the address path off. + + Measured on this package (laptop A/B, 2,000,000 points of 100 hosts in chunks of 100,000: an int64 timestamp array, a list of host + names and three float64 arrays; whole ingest in seconds including wait_completion(), new build over old, below 1 is faster): + 0.370 (95% interval 0.362 to 0.390, median of 11 clean paired runs). The figure is for that shape. + """ + def plain(values): kind = getattr(getattr(values, "dtype", None), "kind", None) if ( @@ -622,20 +639,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 c941feea..58b261ba 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" # ARCADEDB_NATIVE_GB_BIN: the float32 vectors of a GraphBatch op (create_vertices rows, new_edges props) go through arcadedb_n_gb_bin in a float32 side buffer, not as number text GB_BIN = os.environ.get("ARCADEDB_NATIVE_GB_BIN", "1") != "0" @@ -247,6 +249,7 @@ class _Params(ctypes.Structure): "arcadedb_n_gb": [P, L, S, S], "arcadedb_n_gb_bin": [P, L, S, S, S, L], "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], @@ -270,7 +273,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, GB_BIN + global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY, NEXT_BIN, GB_BIN, TS_BIN if not hasattr( lib, "arcadedb_n_exec_bin" ): # a library built before the binary path @@ -291,6 +294,10 @@ class _Params(ctypes.Structure): lib, "arcadedb_n_gb_bin" ): # a library built before the side buffer of GraphBatch ops GB_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 0df9d9ad..fd1603aa 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -77,6 +77,8 @@ collect_ignore.append("test_native_streaming_batches.py") # the Python part of the native point operations (lookup_by_key, edge ends, steps): spies on the codec, the cursor and the row object collect_ignore.append("test_native_point_op_glue.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 00000000..bd26af12 --- /dev/null +++ b/tests/test_native_timeseries_binary.py @@ -0,0 +1,493 @@ +"""``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.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() == [] + + +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, + 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), + ( + 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