diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc index 148c077..4ad3841 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, 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, which property shapes a graph batch loads in bulk on the JVM build, what it means for a failed load, what to do after an interrupted graph batch, how to change it, 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. @@ -67,6 +67,36 @@ with db.graph_batch(commit_every=5000) as batch: # also sets the create_vertice ... ---- +== Property shapes in a graph batch (`arcadedb-jvm`) + +On the `arcadedb-jvm` package, `GraphBatch.create_vertices` and `GraphBatch.new_edges` load these property values through the bulk path (the same tagged JSON text as `insert_many`): a list, a tuple, a dict, a `datetime`, a `date`, a `Decimal` and `bytes` (also `bytearray`). A value is stored as the one-by-one path stores it. A list or tuple of only `int` and `float` keeps the older path, which crosses it as one `double[]`. + +A value that the text cannot carry sends work to the older path on its own: a `NaN`, an integer beyond 64 bits, a `datetime` inside a list or a dict, or a dict key that is not a `str`. `create_vertices` falls back for the chunk of at most 100,000 rows that holds the value, and `new_edges` for the whole call. A value of another type (a `set`, a numpy or pandas value, a subclass of the types above) also takes the older path. Nothing is lost on a fallback, and the data is the same. Only the speed changes. + +The table is one laptop study of the `arcadedb-jvm` package: per-row time of this build against the build before it, 20,000 vertices and 20,000 edges for each shape in chunks of 5,000, one property on every row. Each cell is the median of the pair ratios (below 1 means faster) with its 95 % interval, from 11 to 12 clean ABBA pairs for each figure of `ab.py --noise`, with a clean A/A control. The `arcadedb` (native) package does not change. + +[cols="3,2,2"] +|=== +| Property shape | `create_vertices` | `new_edges` + +| List of strings +| 0.403 (0.378 to 0.420) +| 0.409 (0.389 to 0.420) + +| Dict +| 0.265 (0.243 to 0.286) +| 0.278 (0.269 to 0.298) + +| Datetime +| 0.515 (0.503 to 0.550) +| 0.482 (0.469 to 0.518) + +| List of floats (keeps the old path) +| 1.024 (0.963 to 1.109) +| 1.010 (0.979 to 1.027) + +|=== + == 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/src/arcadedb/graph_batch.py b/src/arcadedb/graph_batch.py index 7184752..938d179 100644 --- a/src/arcadedb/graph_batch.py +++ b/src/arcadedb/graph_batch.py @@ -15,7 +15,9 @@ convert_python_to_java, declared_float_vectors, is_vector_candidate, + json_bulk_dumps_tagged, json_bulk_scalar_ok, + json_bulk_shape_ok, split_float_vectors, ) @@ -190,6 +192,11 @@ def create_vertices( ``ARRAY_OF_FLOATS`` whose value in every row is a list or a numpy array of numbers of one length crosses as one binary float32 matrix instead of JSON text. + A list (not one of only numbers), tuple, dict, date-time, date, + Decimal or bytes value also stays on the bulk path: it crosses + inside the JSON as a tagged value and is stored as the + per-element path stores it. A list of only numbers keeps the + property-matrix path, which crosses it as one double[]. Large inputs cross in chunks of 100,000 rows on this build. On the native build (the ``arcadedb`` package) the chunk is the batch's ``commit_every``, 1000 unless ``graph_batch()`` got another value, and each chunk is its own transaction. Chunks that @@ -244,14 +251,23 @@ def _create_vertices_json_bulk(self, type_name, rows): string (two bulk copies instead of per-value marshaling, measured ~2.4x). The float vectors of ARRAY_OF_FLOATS properties cross as one binary matrix beside the JSON instead of as text (split_float_vectors). - Returns None when unavailable/unsuitable so the caller falls back to - the property-matrix path.""" + A list (not one of only numbers), tuple, dict, date-time, date, Decimal + or bytes value crosses inside the JSON as the tagged value + json_bulk_dumps_tagged writes and VertexBatcher decodes (the stored + value is what the property-matrix path stores). An input whose values + are all scalars takes the untagged json.dumps path as before. A chunk + holding a value the text cannot carry (a NaN, an integer beyond 64 + bits or a date-time inside a list, a subnormal) goes through the + property-matrix path on its own. Returns None when + unavailable/unsuitable so the caller falls back to the property-matrix + path for the whole input.""" try: vertex_batcher = jpype.JClass("com.arcadedb.python.VertexBatcher") declared = set(declared_float_vectors(self._java_db, type_name)) except Exception: return None + tagged = False # a value that needs the tagged text; False keeps the untagged code of before for row in rows: if row: for key, value in row.items(): @@ -263,10 +279,14 @@ def _create_vertices_json_bulk(self, type_name, rows): if key in declared and is_vector_candidate(value): continue # the vector split below takes it, or its chunk goes the matrix way if not json_bulk_scalar_ok(value): - # e.g. datetime/bytes (the matrix path preserves types), or a - # value JSON would change: an integer beyond 64 bits, NaN or - # Infinity, a lone surrogate - return None + if not json_bulk_shape_ok(value): + # a type the tagged JSON does not carry (a set, a numpy + # or pandas value, any subclass, a list of only numbers: + # the matrix path converts those), or a scalar JSON + # would change: an integer beyond 64 bits, NaN or + # Infinity, a lone surrogate + return None + tagged = True import json @@ -276,15 +296,26 @@ def _create_vertices_json_bulk(self, type_name, rows): chunk = rows[start : start + self._BULK_CHUNK] split = split_float_vectors(self._java_db, type_name, chunk, declared) rest = split[0] if split else chunk - if declared and not self._json_bulk_ok(rest): + if not tagged: + text = None + if not declared or self._json_bulk_ok(rest): + text = json.dumps([row or {} for row in rest]) + else: + text = None + if self._json_bulk_declared_ok(rest, declared): + try: + text = json_bulk_dumps_tagged([row or {} for row in rest]) + except (TypeError, ValueError): + text = None # a value the text cannot carry, see the docstring + if text is None: # a list the matrix cannot carry (ragged, a null, not numbers) - # is left in this chunk: the property-matrix path takes the chunk + # is left in this chunk, or a value the text would change: the + # property-matrix path takes the chunk java_rids = self._java_graph_batch.createVertices( type_name, self._to_java_property_matrix(chunk) ) rids.extend(str(rid) for rid in java_rids) continue - text = json.dumps([row or {} for row in rest]) if split: joined = vertex_batcher.createVerticesJson( self._java_graph_batch, type_name, text, split[1], split[2] @@ -307,6 +338,16 @@ def _json_bulk_ok(rows): return False return True + @staticmethod + def _json_bulk_declared_ok(rows, declared): + """False when a declared ARRAY_OF_FLOATS property still holds a list (or any non-scalar) after the vector split.""" + for row in rows: + if row: + for key in declared: + if key in row and not json_bulk_scalar_ok(row[key]): + return False + return True + def new_edge( self, source_vertex_or_rid, @@ -354,8 +395,13 @@ def new_edges( edge_type: Edge type name. destination_rids: Sequence of destination RIDs (same length). properties: Optional sequence of per-edge property dicts (same - length). JSON-representable values take the bulk path; - anything else falls back to per-edge buffering. + length). Scalars and a list (not one of only numbers), tuple, + dict, date-time, date, Decimal or bytes value take the bulk + path (the last ones as tagged values inside the JSON). A list + of only numbers, a value the text cannot carry (a NaN, an + integer beyond 64 bits, a date-time inside a list or dict, a + non-str dict key) or any other type sends the whole call to + per-edge buffering. """ self._check_not_closed() from .results import _bridge_class @@ -419,18 +465,20 @@ def new_edges( return self rows = [] + tagged = False # a value that needs the tagged text; False keeps the untagged code of before for src, dst, p in zip(source_rids, destination_rids, properties): row = {"_src": str(src), "_dst": str(dst)} for key, value in (p or {}).items(): - if ( - type(key) is not str - or key in ("_src", "_dst") - or not json_bulk_scalar_ok(value) - ): - # non-JSON value (datetime/bytes), or one JSON would change - # (an integer beyond 64 bits, NaN or Infinity, a lone - # surrogate), or a property name that is not a str (JSON - # would write 1 as "1"): per-edge fallback preserves + ok = type(key) is str and key not in ("_src", "_dst") + if ok and not json_bulk_scalar_ok(value): + ok = json_bulk_shape_ok(value) + tagged = tagged or ok + if not ok: + # a type the tagged JSON does not carry (a set, a numpy or + # pandas value, a list of only numbers), or a scalar JSON + # would change (an integer beyond 64 bits, NaN or Infinity, + # a lone surrogate), or a property name that is not a str + # (JSON would write 1 as "1"): per-edge fallback preserves # types exactly and refuses the name props_iter = properties for s2, d2, p2 in zip( @@ -441,14 +489,32 @@ def new_edges( row[key] = value rows.append(row) - import json + if not tagged: + import json - for start in range(0, len(rows), self._BULK_CHUNK): - edge_batcher.newEdgesJson( - self._java_graph_batch, - edge_type, - json.dumps(rows[start : start + self._BULK_CHUNK]), - ) + for start in range(0, len(rows), self._BULK_CHUNK): + edge_batcher.newEdgesJson( + self._java_graph_batch, + edge_type, + json.dumps(rows[start : start + self._BULK_CHUNK]), + ) + return self + + # Every chunk is written before the first one crosses, so a value the + # text cannot carry (a NaN, an integer beyond 64 bits or a date-time + # inside a list or map, a subnormal) can still start the per-edge + # path over with nothing buffered. + try: + texts = [ + json_bulk_dumps_tagged(rows[start : start + self._BULK_CHUNK]) + for start in range(0, len(rows), self._BULK_CHUNK) + ] + except (TypeError, ValueError): + for s2, d2, p2 in zip(source_rids, destination_rids, properties): + self.new_edge(s2, edge_type, d2, **(p2 or {})) + return self + for text in texts: + edge_batcher.newEdgesJson(self._java_graph_batch, edge_type, text) return self except Exception as e: raise ArcadeDBError( diff --git a/src/arcadedb/type_conversion.py b/src/arcadedb/type_conversion.py index 7df69f8..1eca63e 100644 --- a/src/arcadedb/type_conversion.py +++ b/src/arcadedb/type_conversion.py @@ -697,6 +697,41 @@ def json_bulk_scalar_ok(value: Any) -> bool: return False +_BULK_TAGGED_SHAPES = frozenset( + (list, tuple, dict, datetime, date, Decimal, bytes, bytearray) +) + + +_NUMBER_TYPES = frozenset((int, float)) + + +def json_bulk_shape_ok(value: Any) -> bool: + """Whether a value that json_bulk_scalar_ok refused may still go through json_bulk_dumps_tagged in the graph batches (the cheap, per-value gate). + + True for the exact types the tagged JSON carries: a list, tuple or dict, a date-time, date, Decimal, bytes or bytearray. The deep + check (a NaN, an integer beyond 64 bits or a date-time inside a list, a subnormal) is json_bulk_dumps_tagged's own, which raises + ValueError or TypeError so the caller sends the rows down the per-element path. False for a non-empty list or tuple of nothing but + int and float: the per-element path crosses it as one double[] and measured about 3.8 times faster than the text for 128 doubles + (step-0 probe, 2026-11-04), so it keeps that path. A subclass (pandas.Timestamp), a numpy scalar or array, a set and pandas NA stay + refused: the per-element path converts those. + """ + kind = type(value) + if kind not in _BULK_TAGGED_SHAPES: + return False + if ( + (kind is list or kind is tuple) + and value + and set(map(type, value)) <= _NUMBER_TYPES + ): + return False + return True + + +def json_bulk_tagged_ok(value: Any) -> bool: + """A scalar json_bulk_scalar_ok accepts, or a shape json_bulk_shape_ok accepts.""" + return json_bulk_scalar_ok(value) or json_bulk_shape_ok(value) + + def _json_bulk_check(value: Any) -> None: """Raise ValueError for a value JSON would change on the way to the engine. diff --git a/src/java/com/arcadedb/python/DocumentBatcher.java b/src/java/com/arcadedb/python/DocumentBatcher.java index a3598e6..bbcae4f 100644 --- a/src/java/com/arcadedb/python/DocumentBatcher.java +++ b/src/java/com/arcadedb/python/DocumentBatcher.java @@ -294,7 +294,20 @@ public static Object[] boxLongs(final long[] a) { * A tagged value from the Python side, {"\u0001": [kind, payload]}: T LocalDateTime (UTC wall clock), D LocalDate, n BigDecimal, * b byte[] (base64): the objects the per-row path (type_conversion.convert_python_to_java) builds. Anything else is returned unchanged. */ - private static Object untag(final Object v) { + /** + * One value of a parsed JSON row, as the engine stores it: a nested object or array becomes a plain Map or List (a parsed + * JSONArray stored as it is reads back as a JSONArray inside the transaction that wrote it) and a tagged value is decoded. A + * scalar comes back as it is, after two type tests: the graph batches call this per value, so it adds nothing to a scalar row. + */ + static Object value(final Object v) { + if (v instanceof JSONObject o) + return untag(o.toMap()); + if (v instanceof JSONArray a) + return a.toList(); + return v; + } + + static Object untag(final Object v) { if (v instanceof Map m && m.size() == 1 && m.get("\u0001") instanceof java.util.List t && t.size() == 2 && t.get(0) instanceof String kind && t.get(1) instanceof String s) { switch (kind) { diff --git a/src/java/com/arcadedb/python/EdgeBatcher.java b/src/java/com/arcadedb/python/EdgeBatcher.java index 1d023c2..e215296 100644 --- a/src/java/com/arcadedb/python/EdgeBatcher.java +++ b/src/java/com/arcadedb/python/EdgeBatcher.java @@ -6,8 +6,8 @@ * whole edge list as String[] RIDs (marshaled in one bulk array copy) and * loops Java-side, so Python pays one crossing per batch. * - * Property-less edges only — the common bulk-ingest shape; edges with - * properties keep the per-edge path. + * newEdgesJson takes edges with properties too, in the value shapes of + * VertexBatcher.createVerticesJson (JSON values and tagged date-times). */ package com.arcadedb.python; @@ -44,8 +44,8 @@ public static void newEdgesJoined(final GraphBatch batch, final String joinedSou /** * Bulk edges WITH properties: one JSON array of rows, each - * {"_src": "#1:0", "_dst": "#1:1", ...properties}. JSON-representable - * property values only; the Python side falls back per-edge otherwise. + * {"_src": "#1:0", "_dst": "#1:1", ...properties}. Property values are JSON values and the tagged values + * DocumentBatcher.untag decodes; the Python side falls back per-edge otherwise. */ public static void newEdgesJson(final GraphBatch batch, final String edgeType, final String jsonRows) { final com.arcadedb.serializer.json.JSONArray rows = new com.arcadedb.serializer.json.JSONArray(jsonRows); @@ -59,7 +59,7 @@ public static void newEdgesJson(final GraphBatch batch, final String edgeType, f if (key.equals("_src") || key.equals("_dst")) continue; props[j++] = key; - props[j++] = row.isNull(key) ? null : row.get(key); + props[j++] = DocumentBatcher.value(row.isNull(key) ? null : row.get(key)); } if (props.length == 0) batch.newEdge(src, edgeType, dst); diff --git a/src/java/com/arcadedb/python/VertexBatcher.java b/src/java/com/arcadedb/python/VertexBatcher.java index 5c52998..5943f37 100644 --- a/src/java/com/arcadedb/python/VertexBatcher.java +++ b/src/java/com/arcadedb/python/VertexBatcher.java @@ -7,9 +7,10 @@ * ~0.2us/vertex Python-side) and returns the created RIDs as ONE * semicolon-joined string, so the whole batch costs two bulk string copies. * - * JSON-representable property values only (str/int/float/bool/null and - * nested lists/maps thereof) — the Python side falls back to the matrix - * path for anything else (e.g. datetime, bytes). + * Property values are str/int/float/bool/null, nested lists and maps thereof, + * and the tagged values DocumentBatcher.untag decodes (date-time, date, + * BigDecimal, byte[]) sitting directly in a row; the Python side falls back to + * the matrix path for anything else. */ package com.arcadedb.python; @@ -44,7 +45,7 @@ public static String createVerticesJson(final GraphBatch batch, final String typ int j = 0; for (final String key : row.keySet()) { flat[j++] = key; - flat[j++] = row.isNull(key) ? null : row.get(key); + flat[j++] = DocumentBatcher.value(row.isNull(key) ? null : row.get(key)); } for (int c = 0; c < extra; c++) { flat[j++] = vectorNames[c]; diff --git a/tests/conftest.py b/tests/conftest.py index 01ab332..fa5078b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -97,6 +97,7 @@ "test_build_no_uv_refresh.py": "build scripts of the JVM wheel", "test_build_wheel_check.py": "build scripts of the JVM wheel", "test_query_vector_params_jvm.py": "the java.util objects the JVM build's parameter conversion builds", + "test_graph_batch_tagged_shapes.py": "the JVM wheel's graph batch bulk path (bridge VertexBatcher and EdgeBatcher, the property-matrix fallback)", } if BACKEND == "native" and not os.environ.get("NATIVE_PARITY_ALL"): collect_ignore.extend(NATIVE_NOT_APPLICABLE) diff --git a/tests/test_graph_batch.py b/tests/test_graph_batch.py index 43b52ed..dc95fa9 100644 --- a/tests/test_graph_batch.py +++ b/tests/test_graph_batch.py @@ -461,8 +461,9 @@ def test_graph_batch_create_vertices_failure_rolls_back(temp_db_path, bulk): any transaction was accepted and silently lost at close (#121).""" with arcadedb.create_database(temp_db_path) as db: _unique_person_db(db) - # a datetime is not JSON-safe, so it routes to the property-matrix path - extra = {} if bulk else {"seen": datetime(2026, 10, 3)} + # a NaN is not JSON-safe, so it routes to the property-matrix path (a datetime + # crosses inside the JSON as a tagged value since the graph batches carry it) + extra = {} if bulk else {"seen": float("nan")} rows = [{"id": 2, **extra}, {"id": 2, **extra}] with db.graph_batch(parallel_flush=False) as batch: @@ -627,7 +628,7 @@ def test_graph_batch_outside_the_callers_transactions(temp_db_path): @pytest.mark.parametrize( "rows", - [2, [{"x": 1}, {"x": 2}], [{"seen": datetime(2026, 10, 5)}, None]], + [2, [{"x": 1}, {"x": 2}], [{"seen": float("nan")}, None]], ids=["count", "json-bulk", "property-matrix"], ) def test_graph_batch_create_vertices_leaves_the_callers_transaction(temp_db_path, rows): diff --git a/tests/test_graph_batch_tagged_shapes.py b/tests/test_graph_batch_tagged_shapes.py new file mode 100644 index 0000000..2da99c1 --- /dev/null +++ b/tests/test_graph_batch_tagged_shapes.py @@ -0,0 +1,477 @@ +"""GraphBatch.create_vertices and new_edges keep a list, tuple, dict, date-time, date, Decimal or bytes property on the bulk path. + +Those values made ``json_bulk_scalar_ok`` refuse the whole call, which sent every row through ``_to_java_property_matrix`` (one JPype +conversion per element, measured 91.4 against 9.3 us per row for a list property). They now cross inside the JSON text as the tagged +values ``json_bulk_dumps_tagged`` writes (insert_many's format) and the bridge classes ``VertexBatcher`` and ``EdgeBatcher`` decode with +``DocumentBatcher.untag``. Each test stores the same rows through the property-matrix path (``_create_vertices_json_bulk`` switched +off, or ``new_edge`` per edge) and through the bulk path, compares the stored values AND their Python types, and spies on the +fallback entry points to prove which path ran. A value the text cannot carry (NaN, an integer beyond 64 bits, a date-time inside a list or +map, a subnormal, a non-str dict key) or a type the tags do not carry (a set, a numpy scalar) still takes the old path. +""" + +import datetime as dt +import decimal +import math + +import numpy as np +import pytest + +import arcadedb +from arcadedb import graph_batch as gbm +from arcadedb import type_conversion +from arcadedb.graph_batch import GraphBatch + +UTC9 = dt.timezone(dt.timedelta(hours=9)) + +# name -> value; each is stored by the property-matrix path (convert_python_to_java), so the bulk path has a stored value to match +VALUES = { + "list_strings": ["a", "b", "é中\U0001f600", ""], + "list_mixed": [1, 2.5, "x", None, True], + "list_nested": [[1, 2], [3.5], []], + "list_empty": [], + "dict_flat": {"a": 1, "b": "two", "c": 3.5, "d": None, "e": False}, + "dict_nested": {"a": [1, 2], "b": {"c": "x", "d": [0.5]}}, + "dict_empty": {}, + "naive_dt": dt.datetime(2026, 5, 6, 7, 8, 9, 123456), + "aware_dt": dt.datetime(2026, 5, 6, 7, 8, 9, 654321, tzinfo=UTC9), + "dt_whole_second": dt.datetime(2026, 5, 6, 7, 8, 9), + "date": dt.date(2026, 5, 6), + "decimal": decimal.Decimal("12345.6789"), + "bytes": b"\x00\xff abc\x7f", + "plain_int": 7, + "plain_str": "text", +} + +# lists and tuples of nothing but numbers keep main's property-matrix path (one double[] per list, about 3.8 times faster than the text) +NUMERIC_LISTS = { + "list_floats": [0.1, 1 / 3, 2.5e-3, 1e300, -0.0, 7.0], + "list_floats_long": [i * 0.37 for i in range(300)], + "list_ints": [1, 2, 3, 2**40, -(2**63)], + "list_int_float": [1, 2.5], + "tuple_ints": (4, 5, 6), +} + +# a declared property per value where the type matters, so the engine's conversion to the declared type is compared too +DECLARED = { + "list_floats": "LIST", + "list_strings": "LIST", + "dict_flat": "MAP", + "naive_dt": "DATETIME_MICROS", + "aware_dt": "DATETIME_NANOS", + "dt_whole_second": "DATETIME", + "date": "DATE", + "decimal": "DECIMAL", + "bytes": "BINARY", +} + + +@pytest.fixture +def matrix_calls(monkeypatch): + """Count the property-matrix path: the only caller of GraphBatch._to_java_property_matrix.""" + calls = [] + real = GraphBatch._to_java_property_matrix + + def counting(self, rows): + calls.append(None) + return real(self, rows) + + monkeypatch.setattr(GraphBatch, "_to_java_property_matrix", counting) + return calls + + +@pytest.fixture +def edge_calls(monkeypatch): + """Count the per-edge path: GraphBatch.new_edge, which new_edges falls back to.""" + calls = [] + real = GraphBatch.new_edge + + def counting(self, *args, **kwargs): + calls.append(None) + return real(self, *args, **kwargs) + + monkeypatch.setattr(GraphBatch, "new_edge", counting) + return calls + + +def _schema(db, declared): + for name in ("Ref", "Bulk"): + db.command("sql", f"CREATE VERTEX TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + if declared: + for prop, ty in DECLARED.items(): + db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}") + for name in ("ERef", "EBulk"): + db.command("sql", f"CREATE EDGE TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + if declared: + for prop, ty in DECLARED.items(): + db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}") + + +def _stored(db, type_name, names): + """Every record by id as {property: (python type name, repr)}, so a changed type is as visible as a changed value.""" + out = [] + sql = f"SELECT FROM {type_name} ORDER BY id" # nosec B608 - test-owned type name + for rec in db.query("sql", sql).to_list(): + out.append( + { + k: (type(v).__name__, repr(v)) + for k, v in sorted(rec.items()) + if k in names + } + ) + return out + + +def _names(rows): + return {k for row in rows for k in row} + + +def _rows(names): + return [{"id": i, name: VALUES[name]} for i, name in enumerate(names)] + + +def _vertices(db, type_name, rows): + with db.graph_batch(parallel_flush=False) as batch: + return batch.create_vertices(type_name, rows) + + +def _reference_vertices(monkeypatch, db, type_name, rows): + """The reference: create_vertices without its JSON bulk path, which is the path it took for these values before.""" + with monkeypatch.context() as mp: + mp.setattr(GraphBatch, "_create_vertices_json_bulk", lambda self, t, r: None) + return _vertices(db, type_name, rows) + + +def _edges(db, edge_type, rows, per_edge=False): + with db.transaction(): + src = [db.new_vertex("Bulk").save().get_rid()] + dst = [db.new_vertex("Bulk").save().get_rid()] + with db.graph_batch(parallel_flush=False) as batch: + if per_edge: + for row in rows: + batch.new_edge(src[0], edge_type, dst[0], **row) + else: + batch.new_edges(src * len(rows), edge_type, dst * len(rows), rows) + + +@pytest.mark.parametrize("declared", [True, False], ids=["declared", "schemaless"]) +def test_vertices_every_shape_is_stored_as_the_matrix_path_stores_it( + temp_db_path, monkeypatch, matrix_calls, declared +): + rows = _rows(list(VALUES)) + with arcadedb.create_database(temp_db_path) as db: + _schema(db, declared) + _reference_vertices(monkeypatch, db, "Ref", rows) + assert len(matrix_calls) == 1 + matrix_calls.clear() + rids = _vertices(db, "Bulk", rows) + assert matrix_calls == [], "the chunk left the bulk path" + assert len(rids) == len(rows) and all(r.startswith("#") for r in rids) + names = _names(rows) + assert _stored(db, "Bulk", names) == _stored(db, "Ref", names) + assert len(_stored(db, "Bulk", names)) == len(rows) + + +@pytest.mark.parametrize("declared", [True, False], ids=["declared", "schemaless"]) +def test_vertices_one_row_with_every_shape_at_once( + temp_db_path, monkeypatch, matrix_calls, declared +): + row = {"id": 0, **VALUES} + with arcadedb.create_database(temp_db_path) as db: + _schema(db, declared) + _reference_vertices(monkeypatch, db, "Ref", [row, None, {}]) + matrix_calls.clear() + _vertices(db, "Bulk", [row, None, {}]) + assert matrix_calls == [] + names = set(row) + assert _stored(db, "Bulk", names)[0] == _stored(db, "Ref", names)[0] + + +@pytest.mark.parametrize("declared", [True, False], ids=["declared", "schemaless"]) +def test_edges_every_shape_is_stored_as_the_per_edge_path_stores_it( + temp_db_path, edge_calls, declared +): + rows = _rows(list(VALUES)) + with arcadedb.create_database(temp_db_path) as db: + _schema(db, declared) + _edges(db, "ERef", rows, per_edge=True) + assert len(edge_calls) == len(rows) + edge_calls.clear() + _edges(db, "EBulk", rows) + assert edge_calls == [], "the edges left the bulk path" + names = _names(rows) + assert _stored(db, "EBulk", names) == _stored(db, "ERef", names) + assert len(_stored(db, "EBulk", names)) == len(rows) + + +# a value the text cannot carry, or whose type the tags do not carry: the old path takes it, and stores what it always stored +OFF_THE_BULK_PATH = { + "nan_in_list": ["x", math.nan], + "nan_in_number_list": [1.0, math.nan], + "inf_in_dict": {"a": math.inf}, + "subnormal_in_list": ["x", 5e-324], + "datetime_in_list": [dt.datetime(2026, 5, 6, 7, 8, 9)], + "datetime_in_dict": {"when": dt.datetime(2026, 5, 6, 7, 8, 9)}, + "int_key_dict": {1: "a"}, + "tag_marker_dict": {"\u0001": ["T", "2026-05-06T00:00:00"]}, + "nan_scalar": math.nan, +} + + +@pytest.mark.parametrize("name", list(OFF_THE_BULK_PATH)) +def test_vertices_a_value_the_text_cannot_carry_keeps_the_old_path( + temp_db_path, monkeypatch, matrix_calls, name +): + value = OFF_THE_BULK_PATH[name] + rows = [{"id": 0, "k": ["a", "b"]}, {"id": 1, "k": value}, {"id": 2, "k": "x"}] + with arcadedb.create_database(temp_db_path) as db: + _schema(db, False) + _reference_vertices(monkeypatch, db, "Ref", rows) + matrix_calls.clear() + rids = _vertices(db, "Bulk", rows) + assert len(matrix_calls) >= 1, "the value went through the JSON text" + assert len(rids) == 3 + assert _stored(db, "Bulk", {"id", "k"}) == _stored(db, "Ref", {"id", "k"}) + + +@pytest.mark.parametrize("name", list(OFF_THE_BULK_PATH)) +def test_edges_a_value_the_text_cannot_carry_keeps_the_old_path( + temp_db_path, edge_calls, name +): + value = OFF_THE_BULK_PATH[name] + rows = [{"id": 0, "k": ["a", "b"]}, {"id": 1, "k": value}, {"id": 2, "k": "x"}] + with arcadedb.create_database(temp_db_path) as db: + _schema(db, False) + _edges(db, "ERef", rows, per_edge=True) + edge_calls.clear() + _edges(db, "EBulk", rows) + assert len(edge_calls) == 3, "the edges went through the JSON text" + assert _stored(db, "EBulk", {"id", "k"}) == _stored(db, "ERef", {"id", "k"}) + count = db.query("sql", "SELECT count(*) AS c FROM EBulk").first().get("c") + assert count == 3 + + +def test_edges_a_late_value_the_text_cannot_carry_buffers_each_edge_once( + temp_db_path, edge_calls, monkeypatch +): + """Every chunk is written before the first crosses: a bad value in the last chunk starts the per-edge path over with nothing buffered.""" + monkeypatch.setattr(GraphBatch, "_BULK_CHUNK", 3) + rows = [{"id": i, "k": [f"s{i}", "z"]} for i in range(8)] + rows[7]["k"] = ["x", math.nan] + with arcadedb.create_database(temp_db_path) as db: + _schema(db, False) + _edges(db, "EBulk", rows) + assert len(edge_calls) == 8 + count = db.query("sql", "SELECT count(*) AS c FROM EBulk").first().get("c") + assert count == 8 + + +def test_vertices_only_the_chunk_with_the_bad_value_leaves_the_bulk_path( + temp_db_path, monkeypatch, matrix_calls +): + monkeypatch.setattr(GraphBatch, "_BULK_CHUNK", 3) + rows = [{"id": i, "k": [f"s{i}", "z"], "d": {"x": i}} for i in range(8)] + rows[4]["k"] = ["x", math.nan] + with arcadedb.create_database(temp_db_path) as db: + _schema(db, False) + rids = _vertices(db, "Bulk", rows) + assert len(rids) == 8 and len(set(rids)) == 8 + assert len(matrix_calls) == 1, "only rows 3 to 5 needed the old path" + got = db.query("sql", "SELECT FROM Bulk ORDER BY id").to_list() + assert [r["id"] for r in got] == list(range(8)) + assert got[0]["k"] == ["s0", "z"] and got[0]["d"] == {"x": 0} + assert got[4]["k"][0] == "x" and math.isnan(got[4]["k"][1]) + assert got[5]["k"] == ["s5", "z"] and got[7]["d"] == {"x": 7} + + +def test_vertices_a_ragged_list_in_a_declared_vector_property_keeps_the_old_path( + temp_db_path, matrix_calls +): + """An ARRAY_OF_FLOATS property whose lists the float matrix cannot carry still goes to the matrix path with its chunk.""" + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE VERTEX TYPE Vx") + db.command("sql", "CREATE PROPERTY Vx.id INTEGER") + db.command("sql", "CREATE PROPERTY Vx.embedding ARRAY_OF_FLOATS") + db.command("sql", "CREATE PROPERTY Vx.tags LIST") + rows = [ + {"id": 0, "embedding": [1.0, 2.0], "tags": ["a"]}, + {"id": 1, "embedding": [1.0], "tags": ["b", "c"]}, + ] + _vertices(db, "Vx", rows) + assert len(matrix_calls) == 1 + got = db.query("sql", "SELECT FROM Vx ORDER BY id").to_list() + assert [list(r["embedding"]) for r in got] == [[1.0, 2.0], [1.0]] + assert [r["tags"] for r in got] == [["a"], ["b", "c"]] + + +def test_vertices_a_declared_vector_property_and_a_list_property_share_a_chunk( + temp_db_path, matrix_calls +): + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE VERTEX TYPE Vx") + db.command("sql", "CREATE PROPERTY Vx.id INTEGER") + db.command("sql", "CREATE PROPERTY Vx.embedding ARRAY_OF_FLOATS") + rows = [ + { + "id": i, + "embedding": [float(i), 0.5, 0.25], + "tags": [f"t{i}", "z"], + "meta": {"n": i}, + "seen": dt.datetime(2026, 1, 1, 0, 0, i), + } + for i in range(4) + ] + _vertices(db, "Vx", rows) + assert matrix_calls == [] + got = db.query("sql", "SELECT FROM Vx ORDER BY id").to_list() + assert [list(r["embedding"]) for r in got] == [ + [float(i), 0.5, 0.25] for i in range(4) + ] + assert [r["tags"] for r in got] == [[f"t{i}", "z"] for i in range(4)] + assert [r["meta"] for r in got] == [{"n": i} for i in range(4)] + assert [r["seen"] for r in got] == [ + dt.datetime(2026, 1, 1, 0, 0, i) for i in range(4) + ] + + +@pytest.mark.parametrize( + "value, ok", + [ + (None, True), + (1, True), + (1.5, True), + ("text", True), + (["a", 1], True), + (("a", 1), True), + ([[1, 2]], True), + ([None, 1.5], True), + ([True, 1], True), + ([], True), + ([1, 2], False), + ((1, 2.5), False), + ({"a": 1}, True), + (dt.datetime(2026, 5, 6), True), + (dt.date(2026, 5, 6), True), + (decimal.Decimal("1.5"), True), + (b"bytes", True), + (bytearray(b"ab"), True), + ({1, 2}, False), + (dt.time(1, 2), False), + (math.nan, False), + (2**63, False), + ("a\ud800b", False), + ], +) +def test_json_bulk_tagged_ok(value, ok): + assert type_conversion.json_bulk_tagged_ok(value) is ok + + +def test_json_bulk_tagged_ok_refuses_subclasses_and_numpy_values(): + pd = pytest.importorskip("pandas") + assert not type_conversion.json_bulk_tagged_ok(pd.Timestamp("2026-05-06")) + assert not type_conversion.json_bulk_tagged_ok(np.float32(1.5)) + assert not type_conversion.json_bulk_tagged_ok(np.arange(3)) + + +def test_the_bulk_method_returns_none_for_a_type_the_tags_do_not_carry(temp_db_path): + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE VERTEX TYPE V") + with db.graph_batch(parallel_flush=False) as batch: + assert batch._create_vertices_json_bulk("V", [{"k": {1, 2}}]) is None + rids = batch._create_vertices_json_bulk( + "V", [{"k": ["a"]}, {"k": {"a": 1}}] + ) + assert rids is not None and len(rids) == 2 + + +@pytest.mark.parametrize("name", list(NUMERIC_LISTS)) +def test_vertices_a_list_of_only_numbers_keeps_the_matrix_path( + temp_db_path, monkeypatch, matrix_calls, name +): + """One double[] per list beats the text (probe 2026-11-04: 30.8 against 117.4 us per row for 128 doubles): the input takes main's path.""" + rows = [{"id": i, "k": NUMERIC_LISTS[name]} for i in range(3)] + mixed = rows + [ + {"id": 3, "k": {"a": 1}} + ] # a shape that would take the text, beside it + with arcadedb.create_database(temp_db_path) as db: + _schema(db, False) + _reference_vertices(monkeypatch, db, "Ref", mixed) + matrix_calls.clear() + _vertices(db, "Bulk", rows) + assert len(matrix_calls) == 1, "a numeric list left the matrix path" + names = {"id", "k"} + assert _stored(db, "Bulk", names) == _stored(db, "Ref", names)[:3] + _vertices(db, "Bulk", mixed) + assert ( + len(matrix_calls) == 2 + ), "one numeric list sent the whole call down the text" + + +@pytest.mark.parametrize("name", list(NUMERIC_LISTS)) +def test_edges_a_list_of_only_numbers_keeps_the_per_edge_path( + temp_db_path, edge_calls, name +): + rows = [{"id": i, "k": NUMERIC_LISTS[name]} for i in range(3)] + with arcadedb.create_database(temp_db_path) as db: + _schema(db, False) + _edges(db, "ERef", rows, per_edge=True) + edge_calls.clear() + _edges(db, "EBulk", rows) + assert len(edge_calls) == 3 + assert _stored(db, "EBulk", {"id", "k"}) == _stored(db, "ERef", {"id", "k"}) + + +@pytest.fixture +def untagged_only(monkeypatch): + """Spy on what an input of scalars must not touch: the tagged writer and the shape gate (main's code has neither).""" + calls = {"tagged": 0, "shape": 0} + real_dumps, real_shape = gbm.json_bulk_dumps_tagged, gbm.json_bulk_shape_ok + + def dumps(*a, **k): + calls["tagged"] += 1 + return real_dumps(*a, **k) + + def shape(*a, **k): + calls["shape"] += 1 + return real_shape(*a, **k) + + monkeypatch.setattr(gbm, "json_bulk_dumps_tagged", dumps) + monkeypatch.setattr(gbm, "json_bulk_shape_ok", shape) + return calls + + +def test_vertices_an_input_of_scalars_takes_mains_code( + temp_db_path, matrix_calls, untagged_only +): + rows = [ + {"id": i, "p": i, "s": f"t{i}", "f": i * 0.5, "b": i % 2 == 0, "n": None} + for i in range(50) + ] + with arcadedb.create_database(temp_db_path) as db: + _schema(db, False) + assert len(_vertices(db, "Bulk", rows)) == 50 + assert matrix_calls == [] + assert untagged_only == {"tagged": 0, "shape": 0} + + +def test_edges_an_input_of_scalars_takes_mains_code( + temp_db_path, edge_calls, untagged_only +): + rows = [{"id": i, "p": i, "s": f"t{i}"} for i in range(50)] + with arcadedb.create_database(temp_db_path) as db: + _schema(db, False) + _edges(db, "EBulk", rows) + assert edge_calls == [] + assert untagged_only == {"tagged": 0, "shape": 0} + + +def test_vertices_a_shape_takes_the_tagged_writer( + temp_db_path, matrix_calls, untagged_only +): + rows = [{"id": i, "p": ["a", i]} for i in range(5)] + with arcadedb.create_database(temp_db_path) as db: + _schema(db, False) + _vertices(db, "Bulk", rows) + assert matrix_calls == [] + assert untagged_only["tagged"] == 1 and untagged_only["shape"] == 5