From 8bf380cc350b3e0d3413794772365547c8afe3d9 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 18:58:18 +0900 Subject: [PATCH 1/4] insert_many and create_vertices: check the rows in the one json.dumps pass, not in a loop before it (refs #318) json_bulk_dumps and json_bulk_dumps_tagged walked every value of every row in Python before json.dumps walked the rows again. The rows are now written once and the checks come from what that pass shows: the text (a surrogate escape, an integer token of 2**63 and above) and a census of the types of the keys and values in C. The exact per-row walk (_check_rows) is the reference and runs only when a value may be bad, and when json.dumps itself refuses the rows, so the exception type and message for the first bad row in row order are the old loop's. Fuzzed against the old code: 60000 cases, no difference. JVM GraphBatch.create_vertices: classify_vertex_rows decides the shapes by type census instead of a json_bulk_scalar_ok call per value; a scalar the text would change (NaN, integer beyond 64 bits, lone surrogate, subnormal) now sends its own chunk through the property-matrix path, as a tagged chunk already did. The untagged path now also refuses a subnormal, as the tagged one did. NativeGraphBatch: the dead _create_vertices_json_bulk is deleted, and the row type check is a C census. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/_native/extras.py | 29 +- src/arcadedb/graph_batch.py | 83 ++--- src/arcadedb/type_conversion.py | 228 +++++++++--- tests/conftest.py | 1 + tests/test_bulk_json_paths.py | 11 +- tests/test_bulk_single_pass_validation.py | 418 ++++++++++++++++++++++ tests/test_graph_batch_single_pass.py | 84 +++++ tests/test_graph_batch_tagged_shapes.py | 2 + 8 files changed, 723 insertions(+), 133 deletions(-) create mode 100644 tests/test_bulk_single_pass_validation.py create mode 100644 tests/test_graph_batch_single_pass.py diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index 5f8a4702..82440e9a 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -28,6 +28,7 @@ _UPCALL_TYPE = ctypes.CFUNCTYPE( None, ctypes.c_longlong, ctypes.c_char_p, ctypes.c_void_p ) +_DICT_OR_NONE = frozenset((dict, type(None))) _callbacks: dict = {} _callback_ids = itertools.count(1) _callbacks_lock = threading.Lock() @@ -229,21 +230,6 @@ def _to_java_varargs(self, properties: dict) -> dict: raise TypeError(f"property names must be str, got {key!r}") return properties - def _create_vertices_json_bulk(self, type_name: str, rows): - """The bulk path (JSON rows in, RIDs out): the RIDs, or None for rows JSON would change (the caller falls back).""" - from ..type_conversion import json_bulk_scalar_ok - - for row in rows: - for key, value in (row or {}).items(): - if type(key) is not str or not json_bulk_scalar_ok(value): - return None - return self._op( - f"Failed to create batch vertices for type '{type_name}'", - "create_vertices", - type=type_name, - rows=[r or {} for r in rows], - )["rids"] - def create_vertices(self, type_name: str, count_or_properties) -> list: """The RIDs of the new vertices. Rows go in chunks of the batch's ``commit_every`` (1000 unless graph_batch() got another value; 0: 10,000), one crossing and one transaction per chunk: chunks that already went in stay stored if a later chunk fails. @@ -274,11 +260,14 @@ def create_vertices(self, type_name: str, count_or_properties) -> list: ] return rids rows = list(count_or_properties) - for row in rows: - if row is not None and not isinstance(row, dict): - raise TypeError( - "create_vertices() property rows must be dicts or None" - ) + if ( + not set(map(type, rows)) <= _DICT_OR_NONE + ): # exact types in C; only an odd row pays the isinstance loop + for row in rows: + if row is not None and not isinstance(row, dict): + raise TypeError( + "create_vertices() property rows must be dicts or None" + ) # chunked, as the JPype wrapper does: one crossing, one JSON text and one reply per chunk, so a 1M-vertex load # never holds the text, the parsed rows and the reply of all of it at once (the Serial GC copies what lives) rids = [] diff --git a/src/arcadedb/graph_batch.py b/src/arcadedb/graph_batch.py index 03495844..efc1aff2 100644 --- a/src/arcadedb/graph_batch.py +++ b/src/arcadedb/graph_batch.py @@ -12,9 +12,11 @@ from .exceptions import ArcadeDBError from .graph import Document, Vertex from .type_conversion import ( + classify_vertex_rows, convert_python_to_java, declared_float_vectors, - is_vector_candidate, + declared_scalars_ok, + json_bulk_dumps, json_bulk_dumps_tagged, json_bulk_scalar_ok, json_bulk_shape_ok, @@ -256,58 +258,40 @@ def _create_vertices_json_bulk(self, type_name, rows): 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.""" + are all scalars is written by json_bulk_dumps. The values are checked + in the one json.dumps pass of each chunk (see type_conversion._dumps_once), + not in a loop over the rows before it: a chunk holding a value the text + cannot carry (a NaN, an integer beyond 64 bits, a lone surrogate, a + subnormal, or a date-time inside a list) goes through the property-matrix + path on its own. Returns None when unavailable/unsuitable (a property + name that is not a str, a type the text does not carry) 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(): - if type(key) is not str: - # a property name that is not a str, which json.dumps would - # write as text ({1: "a"} as "1") where the matrix path - # refuses it - return None - 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): - 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 + tagged = classify_vertex_rows(rows, declared) + if tagged is None: + # a property name that is not a str (json.dumps would write 1 as "1" where the matrix path refuses it), or 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) + return None + dumps = json_bulk_dumps_tagged if tagged else json_bulk_dumps # chunked so huge ingests never materialize one giant JSON string rids = [] for start in range(0, len(rows), self._BULK_CHUNK): 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 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 + text = None + if declared_scalars_ok(rest, declared): + try: + text = dumps([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, or a value the text would change: the @@ -330,25 +314,6 @@ def _create_vertices_json_bulk(self, type_name, rows): rids.extend(joined.split(";")) return rids - @staticmethod - def _json_bulk_ok(rows): - for row in rows: - if row: - for value in row.values(): - if not json_bulk_scalar_ok(value): - 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, diff --git a/src/arcadedb/type_conversion.py b/src/arcadedb/type_conversion.py index 1eca63ed..8f90133c 100644 --- a/src/arcadedb/type_conversion.py +++ b/src/arcadedb/type_conversion.py @@ -5,6 +5,7 @@ developer experience and integration with Python ecosystem (pandas, numpy, etc.). """ +import itertools import json import math import re @@ -804,16 +805,38 @@ def _dumps_checked(rows: Any, **kwargs: Any) -> str: return text -def json_bulk_dumps(rows: Any) -> str: - """json.dumps for the bulk paths: raises ValueError (or TypeError) when - any value in `rows` would change on the way to the engine. +def _int64_overflow_pattern() -> str: + """The digit runs of 2**63 and above, which a JSON integer token beyond 64 bits has. + + A run of 20 digits or more, or of 19 digits that is not below 9223372036854775808. A run that follows a digit or a "." is not an + integer token (the digits after the point of a float such as 0.00012345678901234568 are no token), so only a run start counts. + """ + limit = str(2**63) + parts = [r"[0-9]{20,}"] + for i, digit in enumerate(limit): + low = int(digit) if i == len(limit) - 1 else int(digit) + 1 + if low <= 9: + parts.append(f"{limit[:i]}[{low}-9][0-9]{{{len(limit) - 1 - i}}}") + return r"(? None: + """The per-row walk: raise ValueError (or TypeError) for the first row, in row order, that holds a value the JSON text would change. + + This is the exact reference. The bulk dumps do not run it on every call: `_dumps_once` runs it only when the text or the types of + the rows show that a value may be bad, and when json.dumps itself refused the rows, so the error is always the walk's. + `tagged` is insert_many's format (see json_bulk_dumps_tagged), where the kinds bulk_tag_default writes are allowed in a row. """ int_min, int_max = _INT64_MIN, _INT64_MAX + kinds = _TAGGED_KINDS for row in rows: for key in row: if type(key) is not str: @@ -832,11 +855,87 @@ def json_bulk_dumps(rows: Any) -> str: ) elif kind is float or kind is bool or value is None: continue + elif tagged and kind in kinds: + continue # bulk_tag_default checks it as json.dumps writes it + elif tagged and kind is dict and _TAG in value: + raise ValueError( + "a map keyed by the tag marker cannot go through the JSON bulk path" + ) + elif tagged and ( + isinstance(value, (datetime, date, Decimal, bytes, bytearray)) + or (_np is not None and isinstance(value, _np.generic)) + or _is_pandas_null(kind) + ): + continue # a subclass (pandas.Timestamp), a numpy scalar, pandas NA or NaT else: _json_bulk_check( value ) # nested lists and dicts, and subclasses of the types above - return _dumps_checked(rows) + + +def _kind_is_tagged(kind: type) -> bool: + """Whether bulk_tag_default (not the walk) checks a value of this type: the kinds `_check_rows` lets through in tagged mode.""" + return ( + kind in _TAGGED_KINDS + or issubclass(kind, (datetime, date, Decimal, bytes, bytearray)) + or (_np is not None and issubclass(kind, _np.generic)) + or _is_pandas_null(kind) + ) + + +def _rows_need_walk(rows: Any, tagged: bool) -> bool: + """Whether the types of the rows leave a check that the JSON text cannot show: only exact dict rows with str keys and values of + the exact scalar types (and, tagged, the kinds bulk_tag_default checks) are decided by the text. The census runs in C: one pass + over the keys and one over the values, no Python-level loop per row. + """ + if not set(map(type, rows)) <= _DICT_ONLY: + return True + if not set(map(type, itertools.chain.from_iterable(rows))) <= _STR_ONLY: + return True + extra = ( + set(map(type, itertools.chain.from_iterable(map(dict.values, rows)))) + - _BULK_SCALAR_TYPES + ) + if not extra: + return False + return not (tagged and all(map(_kind_is_tagged, extra))) + + +def _dumps_once(rows: Any, tagged: bool) -> str: + """One json.dumps pass for the bulk paths, with the checks of `_check_rows` decided from what that pass shows. + + A str or int the text would change cannot be seen by a json.dumps hook (the encoder writes both itself), so the text decides: + a surrogate escape (a lone surrogate) or an integer token of 2**63 and above sends the rows to the walk, and so do rows whose + types the text cannot vouch for (`_rows_need_walk`). When json.dumps itself refuses the rows (NaN, an unknown type, a + subnormal, a hook error) the walk runs first, as the validation loop used to run before json.dumps, so the error of the first + bad row in row order is still the walk's; if the walk finds nothing, json.dumps's own error stands. A false alarm costs the walk + and changes nothing else. + """ + try: + text = ( + _dumps_checked(rows, default=bulk_tag_default) + if tagged + else _dumps_checked(rows) + ) + except (TypeError, ValueError): + _check_rows(rows, tagged) + raise + if ( + _rows_need_walk(rows, tagged) + or _SURROGATE_ESCAPE.search(text) + or _INT64_OVERFLOW.search(text) + ): + _check_rows(rows, tagged) + return text + + +def json_bulk_dumps(rows: Any) -> str: + """json.dumps for the bulk paths: raises ValueError (or TypeError) when + any value in `rows` would change on the way to the engine. + + `rows` is a list of dicts (one per record). The rows are written once; see `_dumps_once` for how the values are checked. + """ + return _dumps_once(rows, False) # --------------------------------------------------------------------------- @@ -933,43 +1032,7 @@ def json_bulk_dumps_tagged(rows: Any) -> str: through, written by bulk_tag_default. Raises TypeError or ValueError exactly where json_bulk_dumps does for every other value (a datetime inside a list or map included), so the caller can send that row down the per-row path. """ - int_min, int_max = _INT64_MIN, _INT64_MAX - tagged = _TAGGED_KINDS - for row in rows: - for key in row: - if type(key) is not str: - raise ValueError("a non-str key cannot go through the JSON bulk path") - for value in row.values(): - kind = type(value) - if kind is str: - if not value.isascii() and not _str_survives_json(value): - raise ValueError( - "a string with a lone surrogate cannot go through the JSON bulk path" - ) - elif kind is int: - if not int_min <= value <= int_max: - raise ValueError( - f"an integer beyond 64 bits cannot go through the JSON bulk path: {value}" - ) - elif kind is float or kind is bool or value is None: - continue - elif kind in tagged: - continue # bulk_tag_default checks it as json.dumps writes it - elif kind is dict and _TAG in value: - raise ValueError( - "a map keyed by the tag marker cannot go through the JSON bulk path" - ) - elif ( - isinstance(value, (datetime, date, Decimal, bytes, bytearray)) - or (_np is not None and isinstance(value, _np.generic)) - or _is_pandas_null(kind) - ): - continue # a subclass (pandas.Timestamp), a numpy scalar, pandas NA or NaT - else: - _json_bulk_check( - value - ) # nested lists and dicts, and subclasses of the types above - return _dumps_checked(rows, default=bulk_tag_default) + return _dumps_once(rows, True) def bulk_row_ok(row: Any, skip=()) -> bool: @@ -983,6 +1046,81 @@ def bulk_row_ok(row: Any, skip=()) -> bool: return True +_ROWS_OR_NONE = frozenset((dict, type(None))) + + +def _column(rows: Any, key: str): + """The value of `key` in every non-empty row (None where the row lacks it), as one C-level iterator.""" + return map(dict.get, filter(None, rows), itertools.repeat(key)) + + +def classify_vertex_rows(rows: Any, declared: Any): + """What GraphBatch.create_vertices must know about its rows before it writes any chunk, decided by types. + + Returns None when the rows must all go the property-matrix way (a property name that is not a str, or a value of a type the + tagged JSON does not carry: a set, a numpy or pandas value, a list or tuple of only numbers, any other object); otherwise True when + some value needs the tagged text (json_bulk_dumps_tagged) and False when json_bulk_dumps carries every row. A value of an exact + scalar type is not judged here: whether the text carries it (a NaN, an integer beyond 64 bits, a lone surrogate) is decided in + the dumps of its chunk. The property of `declared` (ARRAY_OF_FLOATS) that holds a list or a numpy array is left to + split_float_vectors. The census over the keys and the values runs in C; a Python-level loop runs only over a column that + holds a type other than the exact scalars. + """ + if not set(map(type, rows)) <= _ROWS_OR_NONE: + return _classify_walk( + rows, declared + ) # a row that is no dict: the walk raises what it always raised + keys = set(itertools.chain.from_iterable(filter(None, rows))) + if not set(map(type, keys)) <= _STR_ONLY: + return None + tagged = False + for key in keys: + odd = set(map(type, _column(rows, key))) - _BULK_SCALAR_TYPES + if not odd: + continue + if key in declared and all( + kind is list or (_np is not None and issubclass(kind, _np.ndarray)) + for kind in odd + ): + continue # the vectors of a declared column + for value in _column(rows, key): + if type(value) in _BULK_SCALAR_TYPES or ( + key in declared and is_vector_candidate(value) + ): + continue + if not json_bulk_scalar_ok(value): + if not json_bulk_shape_ok(value): + return None + tagged = True + return tagged + + +def _classify_walk(rows: Any, declared: Any): + tagged = False + for row in rows: + if row: + for key, value in row.items(): + if type(key) is not str: + return None + if key in declared and is_vector_candidate(value): + continue + if not json_bulk_scalar_ok(value): + if not json_bulk_shape_ok(value): + return None + tagged = True + return tagged + + +def declared_scalars_ok(rows: Any, declared: Any) -> bool: + """False when a declared ARRAY_OF_FLOATS property still holds a list (or any non-scalar) after the vector split.""" + for key in declared: + if set(map(type, _column(rows, key))) <= _BULK_SCALAR_TYPES: + continue # scalars: the dumps of the chunk decides whether the text carries them + for row in rows: + if row and key in row and not json_bulk_scalar_ok(row[key]): + return False + return True + + _FLOAT_VECTORS = None # the bridge's FloatVectors class once resolved; False when the bridge jar lacks it diff --git a/tests/conftest.py b/tests/conftest.py index ec3147c7..3ed7133f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -119,6 +119,7 @@ "test_query_vector_params_jvm.py": "the java.util objects the JVM build's parameter conversion builds", "test_insert_many_gc_pause.py": "the cyclic collector pause is the JVM build's (the native insert_many allocates nothing the collector tracks)", "test_graph_batch_tagged_shapes.py": "the JVM wheel's graph batch bulk path (bridge VertexBatcher and EdgeBatcher, the property-matrix fallback)", + "test_graph_batch_single_pass.py": "the JVM wheel's graph batch bulk path (GraphBatch._create_vertices_json_bulk, the property-matrix fallback)", "test_async_proxy_cache.py": "the JPype proxy classes of the JVM build's async executor", "test_read_crossings.py": "crossing counts of the JVM build's read path (JPype proxies of the Java objects and the bridge classes)", "test_reader_crossings.py": "crossing counts of the JVM build's record and row readers (JPype proxies of the Java objects and the bridge classes)", diff --git a/tests/test_bulk_json_paths.py b/tests/test_bulk_json_paths.py index 8bb7bb07..1c801d00 100644 --- a/tests/test_bulk_json_paths.py +++ b/tests/test_bulk_json_paths.py @@ -104,7 +104,7 @@ def test_a_lone_surrogate_raises_instead_of_becoming_a_question_mark( _write(db, entry, "a\ud800b") -def test_ordinary_rows_stay_on_the_bulk_path(temp_db_path): +def test_ordinary_rows_stay_on_the_bulk_path(): rows = [ {"id": i, "name": f"n{i}", "score": i * 0.5, "flag": i % 2 == 0, "note": None} for i in range(5) @@ -113,14 +113,7 @@ def test_ordinary_rows_stay_on_the_bulk_path(temp_db_path): assert all( type_conversion.json_bulk_scalar_ok(v) for row in rows for v in row.values() ) - with arcadedb.create_database(temp_db_path) as db: - _create_types(db) - with db.graph_batch(parallel_flush=False) as batch: - assert len(batch._create_vertices_json_bulk("V", rows)) == len(rows) - assert batch._create_vertices_json_bulk("V", rows + [{"id": 2**70}]) is None - assert ( - batch._create_vertices_json_bulk("V", rows + [{"x": math.nan}]) is None - ) + # the graph batch's side of this (the JVM wheel's _create_vertices_json_bulk) is in test_graph_batch_single_pass.py @pytest.mark.parametrize( diff --git a/tests/test_bulk_single_pass_validation.py b/tests/test_bulk_single_pass_validation.py new file mode 100644 index 00000000..2610c6d8 --- /dev/null +++ b/tests/test_bulk_single_pass_validation.py @@ -0,0 +1,418 @@ +"""insert_many and create_vertices validate their rows in the one json.dumps pass, not in a Python loop over the rows before it. + +`json_bulk_dumps` and `json_bulk_dumps_tagged` used to walk every value of every row before json.dumps walked it again. Now json.dumps +writes the rows once and the check comes from what that pass shows (the text and the types of the rows), and a walk runs only when it +may find something. The contract is unchanged: the same text for rows that carry, and for a row that does not, the same exception type +and message as the old loop gave for the first bad row in row order. The old loop is copied here as the private reference, so these +tests fail on the old code (they spy on `_check_rows`, which only the new code has) and compare the new code with that reference. +The old messages name no row index, so the first bad row of a large batch is told from a later one by the value in its message. +""" + +import datetime as dt +import decimal +import enum +import json +import math +import random + +import numpy as np +import pytest + +import arcadedb +from arcadedb import type_conversion as tc + +INT64_MAX = 2**63 - 1 +INT64_MIN = -(2**63) + + +def _old_dumps(rows, tagged): + """The json_bulk_dumps and json_bulk_dumps_tagged of 26.11.1: the validation loop, then json.dumps.""" + for row in rows: + for key in row: + if type(key) is not str: + raise ValueError("a non-str key cannot go through the JSON bulk path") + for value in row.values(): + kind = type(value) + if kind is str: + if not value.isascii() and not tc._str_survives_json(value): + raise ValueError( + "a string with a lone surrogate cannot go through the JSON bulk path" + ) + elif kind is int: + if not INT64_MIN <= value <= INT64_MAX: + raise ValueError( + f"an integer beyond 64 bits cannot go through the JSON bulk path: {value}" + ) + elif kind is float or kind is bool or value is None: + continue + elif tagged and kind in tc._TAGGED_KINDS: + continue + elif tagged and kind is dict and tc._TAG in value: + raise ValueError( + "a map keyed by the tag marker cannot go through the JSON bulk path" + ) + elif tagged and ( + isinstance( + value, (dt.datetime, dt.date, decimal.Decimal, bytes, bytearray) + ) + or isinstance(value, np.generic) + or tc._is_pandas_null(kind) + ): + continue + else: + tc._json_bulk_check(value) + if tagged: + return tc._dumps_checked(rows, default=tc.bulk_tag_default) + return tc._dumps_checked(rows) + + +def _outcome(call, rows): + try: + return ("text", call(rows)) + except ( + BaseException + ) as e: # noqa: BLE001 - the type and the message are what is compared + return (type(e).__name__, str(e)) + + +class Colour(enum.IntEnum): + RED = 5 + + +class Name(str): + pass + + +def _scalar_rows(): + return [ + {"id": 1, "name": "alpha", "score": 0.5, "flag": True, "note": None}, + {"id": 2, "name": "", "score": -1e-30, "flag": False, "note": "x"}, + ] + + +CASES = { + "scalars": _scalar_rows(), + "empty": [], + "empty_rows": [{}, {}], + "bounds_fit": [{"a": INT64_MAX, "b": INT64_MIN, "c": 0}], + "above_max": [{"a": INT64_MAX + 1}], + "below_min": [{"a": INT64_MIN - 1}], + "big_negative": [{"a": -(10**30)}], + "huge_digits": [{"a": 10**5000}], + "non_bmp": [{"a": "emoji \U0001f600 pair", "b": "\U00010000"}], + "cjk_and_hangul": [{"a": "中文한글퟿"}], + "nul_and_controls": [{"a": "a\x00b\x1f", "b": "
"}], + "lone_high": [{"a": "a\ud800b"}], + "lone_low": [{"a": "\udc00"}], + "two_lone_halves": [{"a": "𐀀"}], + "literal_backslash_u": [{"a": "\\ud800", "b": "\\u0041"}], + "digits_in_text": [{"a": "9999999999999999999", "b": "1234567890123456789012"}], + "nan": [{"a": math.nan}], + "inf": [{"a": -math.inf}], + "subnormal": [{"a": 5e-324}], + "small_floats": [{"a": 0.00012345678901234568, "b": 1.2345678901234567e19}], + "int_key": [{1: "a"}], + "bool_key": [{True: "a"}], + "none_key": [{None: "a"}], + "list_ok": [{"a": [1, "x", None, [2.5]]}], + "list_big": [{"a": [1, 2**70]}], + "list_nan": [{"a": [1.0, math.nan]}], + "list_lone": [{"a": ["\ud800"]}], + "list_datetime": [{"a": [dt.datetime(2026, 1, 2)]}], + "dict_ok": [{"a": {"b": [1], "c": "d"}}], + "dict_int_key": [{"a": {1: 2}}], + "dict_tag_key": [{"a": {tc._TAG: 1}}], + "tuple_value": [{"a": (1, 2)}], + "datetime": [{"a": dt.datetime(2026, 1, 2, 3, 4, 5, 6)}], + "aware_datetime": [ + {"a": dt.datetime(2026, 1, 2, tzinfo=dt.timezone(dt.timedelta(hours=9)))} + ], + "date": [{"a": dt.date(2026, 1, 2)}], + "decimal": [{"a": decimal.Decimal("1.50")}], + "decimal_nan": [{"a": decimal.Decimal("NaN")}], + "bytes": [{"a": b"\x00\xffab", "b": bytearray(b"x")}], + "numpy_scalars": [ + { + "a": np.int64(5), + "b": np.float64(1.5), + "c": np.bool_(True), + "d": np.float32(2.5), + } + ], + "numpy_big": [{"a": np.uint64(2**64 - 1)}], + "numpy_nan": [{"a": np.float64(math.nan)}], + "numpy_datetime": [{"a": np.datetime64("2026-01-02T03:04:05")}], + "numpy_str": [{"a": np.str_("q")}], + "int_enum": [{"a": Colour.RED}], + "str_subclass": [{"a": Name("s")}], + "str_subclass_lone": [{"a": Name("\ud800")}], + "set_value": [{"a": {1, 2}}], + "object_value": [{"a": object()}], + "none_row": [None], + "list_row": [[1, 2]], + "string_row": ["ab"], + "mixed_good_then_bad": [{"a": 1}, {"a": 2**64}, {"a": math.nan}], +} + + +@pytest.mark.parametrize("tagged", [False, True], ids=["untagged", "tagged"]) +@pytest.mark.parametrize("name", sorted(CASES)) +def test_same_text_or_same_error_as_the_validation_loop(name, tagged): + rows = CASES[name] + new = tc.json_bulk_dumps_tagged if tagged else tc.json_bulk_dumps + assert _outcome(new, rows) == _outcome(lambda r: _old_dumps(r, tagged), rows) + + +@pytest.mark.parametrize("tagged", [False, True], ids=["untagged", "tagged"]) +def test_random_rows_agree_with_the_validation_loop(tagged): + atoms = [ + None, True, 0, -1, INT64_MAX, INT64_MIN, INT64_MAX + 1, 10**30, 1.5, 0.00012345678901234568, + 5e-324, math.nan, math.inf, "", "abc", "a\x00b", "é中", "\U0001f600", "a\ud800b", "\udc00", + "𐀀", "9999999999999999999", [1, 2], [math.nan], {"a": 1}, {1: 2}, (1, 2), + dt.datetime(2026, 1, 2), dt.date(2026, 1, 2), decimal.Decimal("1.5"), b"ab", np.int64(7), + np.float64(math.nan), Colour.RED, Name("s"), {1, 2}, object(), + ] # fmt: skip + rnd = random.Random(318) + new = tc.json_bulk_dumps_tagged if tagged else tc.json_bulk_dumps + for _ in range(3000): + rows = [] + for _ in range(rnd.choice([1, 2, 3, 8])): + keys = rnd.sample(["a", "b", "c", "d"], rnd.randint(0, 4)) + if rnd.random() < 0.03: + keys.append(1) + rows.append({k: rnd.choice(atoms) for k in keys}) + assert _outcome(new, rows) == _outcome(lambda r: _old_dumps(r, tagged), rows) + + +def _batch(n, bad): + """n scalar rows of the shape of a TPC-H part table; `bad` maps a row index to its replacement row.""" + rows = [ + { + "p_partkey": i, + "p_name": f"part {i}", + "p_size": i % 50, + "p_retailprice": 900.0 + (i % 1000) / 100, + "p_comment": None if i % 7 == 0 else "carefully final deposits", + } + for i in range(n) + ] + for index, row in bad.items(): + rows[index] = row + return rows + + +# kind -> (first bad row, a later bad row of another value), so the message tells which row raised +BAD_KINDS = { + "int_beyond_64_bits": ({"p_size": 2**63 + 11}, {"p_size": 2**70 + 22}), + "negative_beyond_64_bits": ({"p_size": INT64_MIN - 33}, {"p_size": 2**64 + 44}), + "lone_surrogate": ({"p_name": "a\ud800"}, {"p_size": 2**65}), + "non_str_key": ({7: "x"}, {"p_size": 2**65}), + "nested_int": ({"p_name": [1, 2**66]}, {"p_size": 2**65 + 55}), + "nan": ({"p_retailprice": math.nan}, {"p_size": 2**65}), + "subnormal": ({"p_retailprice": 5e-324}, {"p_size": 2**65}), +} + + +@pytest.mark.parametrize("tagged", [False, True], ids=["untagged", "tagged"]) +@pytest.mark.parametrize("kind", sorted(BAD_KINDS)) +def test_the_first_bad_row_of_a_large_batch_gives_the_old_error(kind, tagged): + first, later = BAD_KINDS[kind] + rows = _batch(20_000, {12_345: first, 19_000: later}) + new = tc.json_bulk_dumps_tagged if tagged else tc.json_bulk_dumps + expected = _outcome(lambda r: _old_dumps(r, tagged), rows) + assert expected[0] in ("ValueError", "TypeError"), expected + assert _outcome(new, rows) == expected + # the later row alone raises a message of its own. A NaN or a subnormal is refused by json.dumps, after the old loop had walked + # every row, so the old error there is the later row's; for every other kind the first bad row's error comes first + only_later = _batch(20_000, {19_000: later}) + assert _outcome(new, only_later) == _outcome( + lambda r: _old_dumps(r, tagged), only_later + ) + assert (_outcome(new, only_later) == expected) == (kind in ("nan", "subnormal")) + + +@pytest.mark.parametrize("tagged", [False, True], ids=["untagged", "tagged"]) +def test_a_clean_large_batch_is_the_plain_json_text(tagged): + rows = _batch(20_000, {}) + new = tc.json_bulk_dumps_tagged if tagged else tc.json_bulk_dumps + assert new(rows) == json.dumps(rows) + + +def test_scalar_rows_are_not_walked_when_the_text_carries_them(monkeypatch): + # the insert_many shape of the paper's l1tpc Part load and the create_vertices shape of l2, e2 and l3d: dict rows of str, int, float, + # bool and None. Numbers that look alike but fit: 19-digit ints below 2**63, floats whose text has a long run of digits after the + # point, text with digits, Hangul and CJK (BMP characters that json.dumps writes as \uXXXX, some of them as \ud...). + walks = [] + real = tc._check_rows + monkeypatch.setattr( + tc, + "_check_rows", + lambda rows, tagged: walks.append(tagged) or real(rows, tagged), + ) + rows = _batch(2_000, {}) + rows += [ + { + "ts": 1_700_000_000_123_456_789, + "hi": INT64_MAX, + "lo": INT64_MIN + 1, + "f": 0.00012345678901234568, + }, + { + "f": 1.2345678901234567e19, + "s": "id-123456789012345678", + "han": "한글퟿中文", + "nul": "a\x00b", + }, + {"e": "", "n": None, "b": True, "neg": -123456789012345678}, + ] + assert tc.json_bulk_dumps(rows) == json.dumps(rows) + assert tc.json_bulk_dumps_tagged(rows) == json.dumps(rows) + assert walks == [] + + +def test_tagged_kinds_are_not_walked_either(monkeypatch): + walks = [] + real = tc._check_rows + monkeypatch.setattr( + tc, + "_check_rows", + lambda rows, tagged: walks.append(tagged) or real(rows, tagged), + ) + rows = [ + {"d": dt.datetime(2026, 1, 2, 3, 4, 5), "day": dt.date(2026, 1, 2), "m": decimal.Decimal("1.50"), "b": b"ab", "n": np.int64(5), "f": np.float32(2.5)} + for _ in range(100) + ] # fmt: skip + assert tc.json_bulk_dumps_tagged(rows) + assert walks == [] + + +@pytest.mark.parametrize( + "row", + [ + {"a": 2**63}, + {"a": 10**20}, + {"a": INT64_MIN - 1}, + {"a": "a\ud800"}, + {"a": "\U0001f600"}, + {7: "x"}, + {"a": [1]}, + {"a": 5e-324}, + {"a": Colour.RED}, + ], +) +def test_a_value_that_may_be_bad_runs_the_walk(monkeypatch, row): + walks = [] + real = tc._check_rows + monkeypatch.setattr( + tc, + "_check_rows", + lambda rows, tagged: walks.append(tagged) or real(rows, tagged), + ) + _outcome(tc.json_bulk_dumps, [{"id": 1}, row]) + assert walks, "the walk must decide a row the text cannot vouch for" + + +def test_int64_overflow_pattern_matches_exactly_the_tokens_from_2_to_the_63(): + rnd = random.Random(63) + values = [ + 0, + 1, + 10**18, + 2**63 - 2, + 2**63 - 1, + 2**63, + 2**63 + 1, + 10**19 - 1, + 10**19, + 10**20, + 9 * 10**18, + 9_999_999_999_999_999_999, + ] + values += [rnd.randint(10**18, 2**65) for _ in range(5000)] # fmt: skip + for value in values: + for text in (str(value), str(-value), f"[{value},1]", f'{{"a":{value}}}'): + hit = tc._INT64_OVERFLOW.search(text) is not None + assert hit == (value >= 2**63), text + # digits after the point of a float and inside an exponent are no integer token + assert tc._INT64_OVERFLOW.search("[0.00012345678901234568]") is None + assert tc._INT64_OVERFLOW.search("[1.2345678901234567e+19]") is None + + +# --------------------------------------------------------------------------- +# the entry points, on whichever wheel is installed + + +def _types(db): + db.command("sql", "CREATE DOCUMENT TYPE Part") + db.command("sql", "CREATE VERTEX TYPE PartV") + + +def _stored(db, type_name): + rows = db.query("sql", f"SELECT FROM {type_name}").to_list() + return sorted( + ( + r["p_partkey"], + r["p_name"], + r["p_size"], + r["p_retailprice"], + r.get("p_comment"), + ) + for r in rows + ) + + +def _expected(rows): + return sorted( + ( + r["p_partkey"], + r["p_name"], + r["p_size"], + r["p_retailprice"], + r.get("p_comment"), + ) + for r in rows + ) + + +def _rows_with_text(): + rows = _batch(25_000, {}) + rows[10]["p_name"] = "emoji \U0001f600 \U00010000" + rows[11]["p_name"] = "" + rows[12]["p_name"] = "한글 中文 ퟿" + rows[13]["p_size"] = INT64_MAX + rows[14]["p_size"] = INT64_MIN + return rows + + +def test_insert_many_stores_a_scalar_batch_as_before(temp_db_path): + rows = _rows_with_text() + with arcadedb.create_database(temp_db_path) as db: + _types(db) + assert db.insert_many("Part", rows) == len(rows) + assert _stored(db, "Part") == _expected(rows) + + +def test_create_vertices_stores_a_scalar_batch_as_before(temp_db_path): + rows = _rows_with_text() + with arcadedb.create_database(temp_db_path) as db: + _types(db) + with db.graph_batch(parallel_flush=False) as batch: + assert len(batch.create_vertices("PartV", rows)) == len(rows) + assert _stored(db, "PartV") == _expected(rows) + + +@pytest.mark.parametrize("entry", ["insert_many", "create_vertices"]) +@pytest.mark.parametrize("bad", [2**63 + 11, "a\ud800"], ids=["int", "surrogate"]) +def test_a_bad_row_inside_a_large_batch_is_refused(temp_db_path, entry, bad): + column = "p_size" if isinstance(bad, int) else "p_name" + rows = _batch(25_000, {12_345: {column: bad}, 19_000: {"p_size": 2**70 + 22}}) + with arcadedb.create_database(temp_db_path) as db: + _types(db) + with pytest.raises( + Exception + ): # noqa: B017 - the entry points raise their own types + if entry == "insert_many": + db.insert_many("Part", rows) + else: + with db.graph_batch(parallel_flush=False) as batch: + batch.create_vertices("PartV", rows) diff --git a/tests/test_graph_batch_single_pass.py b/tests/test_graph_batch_single_pass.py new file mode 100644 index 00000000..2415c012 --- /dev/null +++ b/tests/test_graph_batch_single_pass.py @@ -0,0 +1,84 @@ +"""GraphBatch.create_vertices (the JVM wheel) decides its rows by their types and writes each chunk once. + +Before, `_create_vertices_json_bulk` called `json_bulk_scalar_ok` on every value of every row before the first chunk was written, and +the chunk was then walked again by json.dumps. Now `type_conversion.classify_vertex_rows` takes a census of the types in C, and the +value checks come from the one json.dumps pass of each chunk (`json_bulk_dumps`, as `json_bulk_dumps_tagged` already did for the tagged +text). What stays: the whole input goes the property-matrix way for a property name that is not a str or a type the text does not carry +(`_create_vertices_json_bulk` returns None); what changes: a scalar the text would change (a subnormal, a NaN, an integer beyond 64 bits, +a lone surrogate) sends its own chunk, not the whole input, through the matrix path, as it already did for a tagged chunk. +""" + +import pytest + +import arcadedb +from arcadedb import graph_batch as gbm +from arcadedb import type_conversion as tc +from arcadedb.graph_batch import GraphBatch + + +def _rows(n): + return [ + {"id": i, "name": f"n{i}", "score": i * 0.5, "flag": i % 2 == 0, "note": None} + for i in range(n) + ] + + +def test_scalar_rows_are_not_judged_value_by_value(temp_db_path, monkeypatch): + def refuse(value): + raise AssertionError("a per-value gate ran over a scalar table") + + monkeypatch.setattr(tc, "json_bulk_scalar_ok", refuse) + monkeypatch.setattr(gbm, "json_bulk_scalar_ok", refuse) + monkeypatch.setattr(tc, "json_bulk_shape_ok", refuse) + monkeypatch.setattr(gbm, "json_bulk_shape_ok", refuse) + rows = _rows(2_000) + 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 len(batch._create_vertices_json_bulk("V", rows)) == len(rows) + count = db.query("sql", "SELECT count(*) AS c FROM V").first().get("c") + assert count == len(rows) + + +def test_the_whole_input_still_goes_the_matrix_way_for_a_name_or_type_the_text_does_not_carry( + temp_db_path, +): + rows = _rows(5) + 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 len(batch._create_vertices_json_bulk("V", rows)) == len(rows) + assert batch._create_vertices_json_bulk("V", rows + [{1: "a"}]) is None + assert batch._create_vertices_json_bulk("V", rows + [{"k": {1, 2}}]) is None + + +@pytest.mark.parametrize("value", [5e-324, float("nan"), 2**70, "a\ud800b"]) +def test_a_scalar_the_text_changes_sends_its_chunk_through_the_matrix_path( + temp_db_path, monkeypatch, value +): + monkeypatch.setattr(GraphBatch, "_BULK_CHUNK", 100) + calls = [] + real = GraphBatch._to_java_property_matrix + + def spy(self, chunk): + calls.append(len(chunk)) + return real(self, chunk) + + monkeypatch.setattr(GraphBatch, "_to_java_property_matrix", spy) + rows = [{"id": i, "x": 1.5} for i in range(250)] + rows[150] = {"id": 150, "x": value} + 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: + try: + rids = batch.create_vertices("V", rows) + except ( + Exception + ): # noqa: BLE001 - the matrix path refuses a value that no entry point stores + rids = None + assert calls == [100], "only the chunk of row 150 takes the matrix path" + if value == 5e-324: + # stored as the double every other entry point stores; the text of a subnormal was read as a DECIMAL + assert len(rids) == 250 + stored = db.query("sql", "SELECT x FROM V WHERE id = 150").first().get("x") + assert stored == 5e-324 and type(stored) is float diff --git a/tests/test_graph_batch_tagged_shapes.py b/tests/test_graph_batch_tagged_shapes.py index 2da99c18..560b2d98 100644 --- a/tests/test_graph_batch_tagged_shapes.py +++ b/tests/test_graph_batch_tagged_shapes.py @@ -438,6 +438,8 @@ def shape(*a, **k): monkeypatch.setattr(gbm, "json_bulk_dumps_tagged", dumps) monkeypatch.setattr(gbm, "json_bulk_shape_ok", shape) + # create_vertices decides the shapes in type_conversion.classify_vertex_rows, which calls the gate there + monkeypatch.setattr(type_conversion, "json_bulk_shape_ok", shape) return calls From 7fddca0a65e864346bc8d12d81a9b8e3fa7feaa4 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 20:02:08 +0900 Subject: [PATCH 2/4] Tests: bandit annotations for the seeded corpus and the test-controlled query (refs #318) Co-Authored-By: Claude Sonnet 5.5 --- tests/test_bulk_single_pass_validation.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/test_bulk_single_pass_validation.py b/tests/test_bulk_single_pass_validation.py index 2610c6d8..5d9afe72 100644 --- a/tests/test_bulk_single_pass_validation.py +++ b/tests/test_bulk_single_pass_validation.py @@ -172,7 +172,7 @@ def test_random_rows_agree_with_the_validation_loop(tagged): dt.datetime(2026, 1, 2), dt.date(2026, 1, 2), decimal.Decimal("1.5"), b"ab", np.int64(7), np.float64(math.nan), Colour.RED, Name("s"), {1, 2}, object(), ] # fmt: skip - rnd = random.Random(318) + rnd = random.Random(318) # nosec B311 - a seeded corpus, not security new = tc.json_bulk_dumps_tagged if tagged else tc.json_bulk_dumps for _ in range(3000): rows = [] @@ -313,7 +313,7 @@ def test_a_value_that_may_be_bad_runs_the_walk(monkeypatch, row): def test_int64_overflow_pattern_matches_exactly_the_tokens_from_2_to_the_63(): - rnd = random.Random(63) + rnd = random.Random(63) # nosec B311 - a seeded corpus, not security values = [ 0, 1, @@ -348,7 +348,8 @@ def _types(db): def _stored(db, type_name): - rows = db.query("sql", f"SELECT FROM {type_name}").to_list() + query = f"SELECT FROM {type_name}" # nosec B608 - test-controlled + rows = db.query("sql", query).to_list() return sorted( ( r["p_partkey"], From 4b317ba2552033a79abd954eb3641c337486b8ae Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 21:23:30 +0900 Subject: [PATCH 3/4] bulk dumps: raise the first bad row's error outside the except block (no json.dumps __context__); test the stored type across chunks (refs #318) Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/type_conversion.py | 14 +++++++-- tests/test_bulk_single_pass_validation.py | 15 +++++++++ tests/test_graph_batch_single_pass.py | 37 +++++++++++++++++++++++ 3 files changed, 63 insertions(+), 3 deletions(-) diff --git a/src/arcadedb/type_conversion.py b/src/arcadedb/type_conversion.py index 8f90133c..2ec8f424 100644 --- a/src/arcadedb/type_conversion.py +++ b/src/arcadedb/type_conversion.py @@ -917,9 +917,17 @@ def _dumps_once(rows: Any, tagged: bool) -> str: if tagged else _dumps_checked(rows) ) - except (TypeError, ValueError): - _check_rows(rows, tagged) - raise + except (TypeError, ValueError) as error: + refused = error + else: + refused = None + if refused is not None: + # outside the except block, so the walk's error does not carry json.dumps's as __context__ (the old loop ran first) + try: + _check_rows(rows, tagged) + raise refused + finally: + del refused if ( _rows_need_walk(rows, tagged) or _SURROGATE_ESCAPE.search(text) diff --git a/tests/test_bulk_single_pass_validation.py b/tests/test_bulk_single_pass_validation.py index 5d9afe72..7fa56ed6 100644 --- a/tests/test_bulk_single_pass_validation.py +++ b/tests/test_bulk_single_pass_validation.py @@ -231,6 +231,21 @@ def test_the_first_bad_row_of_a_large_batch_gives_the_old_error(kind, tagged): assert (_outcome(new, only_later) == expected) == (kind in ("nan", "subnormal")) +@pytest.mark.parametrize("tagged", [False, True], ids=["untagged", "tagged"]) +def test_the_first_bad_rows_error_carries_no_context(tagged): + # json.dumps refuses the NaN of the second row first; the error is still the walk's for the first row, as it was when the + # loop ran before json.dumps, and it does not show the NaN error as the context it was raised in + rows = [{"a": 2**70}, {"b": math.nan}] + new = tc.json_bulk_dumps_tagged if tagged else tc.json_bulk_dumps + with pytest.raises(ValueError) as caught: + new(rows) + assert str(caught.value) == _outcome(lambda r: _old_dumps(r, tagged), rows)[1] + assert caught.value.__context__ is None + # with nothing for the walk to find, json.dumps's own error is the one raised + with pytest.raises(ValueError, match="Out of range float"): + new([{"b": math.nan}]) + + @pytest.mark.parametrize("tagged", [False, True], ids=["untagged", "tagged"]) def test_a_clean_large_batch_is_the_plain_json_text(tagged): rows = _batch(20_000, {}) diff --git a/tests/test_graph_batch_single_pass.py b/tests/test_graph_batch_single_pass.py index 2415c012..3b21bfdd 100644 --- a/tests/test_graph_batch_single_pass.py +++ b/tests/test_graph_batch_single_pass.py @@ -82,3 +82,40 @@ def spy(self, chunk): assert len(rids) == 250 stored = db.query("sql", "SELECT x FROM V WHERE id = 150").first().get("x") assert stored == 5e-324 and type(stored) is float + + +def test_the_stored_type_does_not_depend_on_which_chunk_held_a_bad_value( + temp_db_path, monkeypatch +): + """An undeclared int property is stored as the same type in the chunks that crossed as JSON text and in the chunk that took the + property-matrix path because a NaN sat in another property of one of its rows (the 100,000-row chunking of the real class). + """ + n = 250_000 + rows = [{"id": i, "k": i % 1000} for i in range(n)] + rows[150_000] = {"id": 150_000, "k": 1, "x": float("nan")} # chunk 2 of 3 only + probes = {5: "chunk 1", 150_001: "chunk 2", 249_999: "chunk 3"} + calls = [] + real = GraphBatch._to_java_property_matrix + + def spy(self, chunk): + calls.append(len(chunk)) + return real(self, chunk) + + monkeypatch.setattr(GraphBatch, "_to_java_property_matrix", spy) + 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 len(batch.create_vertices("V", rows)) == n + assert calls == [100_000], "only the chunk with the NaN takes the matrix path" + ids = ", ".join(str(i) for i in probes) + query = ( + f"SELECT id, k, k.javaType() AS t FROM V WHERE id IN [{ids}]" # nosec B608 + ) + got = db.query("sql", query).to_list() + by_id = {r["id"]: (r["k"], str(r["t"])) for r in got} + assert set(by_id) == set(probes) + assert {i: by_id[i][0] for i in probes} == {i: i % 1000 for i in probes} + types = {probes[i]: by_id[i][1] for i in probes} + assert ( + len(set(types.values())) == 1 + ), f"the stored type depends on the chunk: {types}" From 7e811708439fbb2d0db3e160d99f49865554bd99 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 9 Nov 2026 22:25:12 +0900 Subject: [PATCH 4/4] create_vertices: decide the whole input before any chunk is sent when it has several chunks (refs #318) A bad scalar in a later chunk used to send only that chunk through the property-matrix path, which stores a Python int as a Long while the JSON chunks store an Integer: one undeclared property got two Java types in one call. An input of more than one chunk now has its columns of exact scalars screened in C over the whole input first (type census, joined text of a str column, min and max of an int column, one float64 array for a float column, in blocks of 100,000 rows); a column that may hold a bad value is judged value by value as before, so a bad scalar anywhere sends the whole input the matrix way, as before #330. One chunk is decided by its own dumps before anything is sent. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/graph_batch.py | 2 +- src/arcadedb/type_conversion.py | 86 +++++++++++++++++++--- tests/test_bulk_single_pass_validation.py | 30 ++++++++ tests/test_graph_batch_single_pass.py | 87 ++++++++++++++++------- 4 files changed, 168 insertions(+), 37 deletions(-) diff --git a/src/arcadedb/graph_batch.py b/src/arcadedb/graph_batch.py index efc1aff2..4fad9342 100644 --- a/src/arcadedb/graph_batch.py +++ b/src/arcadedb/graph_batch.py @@ -272,7 +272,7 @@ def _create_vertices_json_bulk(self, type_name, rows): except Exception: return None - tagged = classify_vertex_rows(rows, declared) + tagged = classify_vertex_rows(rows, declared, len(rows) > self._BULK_CHUNK) if tagged is None: # a property name that is not a str (json.dumps would write 1 as "1" where the matrix path refuses it), or 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 diff --git a/src/arcadedb/type_conversion.py b/src/arcadedb/type_conversion.py index 2ec8f424..68e3857a 100644 --- a/src/arcadedb/type_conversion.py +++ b/src/arcadedb/type_conversion.py @@ -1062,14 +1062,70 @@ def _column(rows: Any, key: str): return map(dict.get, filter(None, rows), itertools.repeat(key)) -def classify_vertex_rows(rows: Any, declared: Any): +_NONE_KIND = type(None) +_SCREEN_BLOCK = 100_000 # rows per block of the scalar screen, so its transient lists and joined texts stay bounded + + +def _scalar_column_suspect(column: list) -> bool: + """Whether one block of a column of exact scalar values may hold a value the JSON text would change. + + Conservative (a false alarm only sends the column to the exact per-value judgement) and run in C: the type census, the joined + text of a str column (a lone surrogate has no UTF-8 form), min and max of an int column (the 64-bit range), and a float column as + one float64 array (NaN, Infinity, a subnormal). Columns of mixed kinds are suspect. + """ + kinds = set(map(type, column)) + if _NONE_KIND in kinds: + column = [value for value in column if value is not None] + kinds.discard(_NONE_KIND) + if kinds <= {bool}: + return False + if len(kinds) != 1: + return True + (kind,) = kinds + if kind is str: + text = "".join(column) + if text.isascii(): + return False + try: + text.encode("utf-8") + except UnicodeEncodeError: + return True + return False + if kind is int: + return not (_INT64_MIN <= min(column) and max(column) <= _INT64_MAX) + if kind is float and _np is not None: + values = _np.fromiter(column, dtype=_np.float64, count=len(column)) + if not _np.isfinite(values).all(): + return True + values = values[values != 0.0] + return bool((_np.abs(values) < _FLOAT_NORMAL_MIN).any()) + return True + + +def _scalar_screen(rows: Any, key: str) -> bool: + """Whether the column `key` of all `rows`, which holds exact scalars only, may hold a value the JSON text would change.""" + for start in range(0, len(rows), _SCREEN_BLOCK): + block = list(_column(rows[start : start + _SCREEN_BLOCK], key)) + if _scalar_column_suspect(block): + return True + return False + + +def classify_vertex_rows(rows: Any, declared: Any, screen_scalars: bool = False): """What GraphBatch.create_vertices must know about its rows before it writes any chunk, decided by types. + `screen_scalars` is for an input of more than one chunk. One chunk is decided by its own dumps before anything is sent, so a + value the text would change sends all of it the matrix way. Several chunks are sent one after the other, so such a value in + a late chunk would leave the earlier ones stored from the JSON text (an int as an Integer) and the late one from the matrix path + (a Long): the stored type of a property would depend on which chunk held a bad value. With `screen_scalars` the columns of exact + scalars are screened in C over the whole input first (`_scalar_screen`), and a column that may hold a bad value is judged value + by value as the code before did, so a bad scalar anywhere gives None for the whole input, as it always did. + Returns None when the rows must all go the property-matrix way (a property name that is not a str, or a value of a type the tagged JSON does not carry: a set, a numpy or pandas value, a list or tuple of only numbers, any other object); otherwise True when some value needs the tagged text (json_bulk_dumps_tagged) and False when json_bulk_dumps carries every row. A value of an exact - scalar type is not judged here: whether the text carries it (a NaN, an integer beyond 64 bits, a lone surrogate) is decided in - the dumps of its chunk. The property of `declared` (ARRAY_OF_FLOATS) that holds a list or a numpy array is left to + scalar type is not judged here unless `screen_scalars`: whether the text carries it (a NaN, an integer beyond 64 bits, a lone + surrogate) is decided in the dumps of its chunk. The property of `declared` (ARRAY_OF_FLOATS) that holds a list or a numpy array is left to split_float_vectors. The census over the keys and the values runs in C; a Python-level loop runs only over a column that holds a type other than the exact scalars. """ @@ -1083,18 +1139,28 @@ def classify_vertex_rows(rows: Any, declared: Any): tagged = False for key in keys: odd = set(map(type, _column(rows, key))) - _BULK_SCALAR_TYPES - if not odd: - continue - if key in declared and all( - kind is list or (_np is not None and issubclass(kind, _np.ndarray)) - for kind in odd + if screen_scalars: + if not odd and not _scalar_screen(rows, key): + continue + elif not odd or ( + key in declared + and all( + kind is list or (_np is not None and issubclass(kind, _np.ndarray)) + for kind in odd + ) ): - continue # the vectors of a declared column + continue # exact scalars, or the vectors of a declared column for value in _column(rows, key): - if type(value) in _BULK_SCALAR_TYPES or ( + if (type(value) in _BULK_SCALAR_TYPES and not screen_scalars) or ( key in declared and is_vector_candidate(value) ): continue + if ( + screen_scalars + and type(value) is float + and 0.0 < abs(value) < _FLOAT_NORMAL_MIN + ): + return None # a subnormal: the text is read back as a DECIMAL if not json_bulk_scalar_ok(value): if not json_bulk_shape_ok(value): return None diff --git a/tests/test_bulk_single_pass_validation.py b/tests/test_bulk_single_pass_validation.py index 7fa56ed6..c830597b 100644 --- a/tests/test_bulk_single_pass_validation.py +++ b/tests/test_bulk_single_pass_validation.py @@ -432,3 +432,33 @@ def test_a_bad_row_inside_a_large_batch_is_refused(temp_db_path, entry, bad): else: with db.graph_batch(parallel_flush=False) as batch: batch.create_vertices("PartV", rows) + + +def test_the_scalar_screen_flags_what_the_text_would_change_and_passes_the_rest(): + # the whole-input screen of create_vertices: conservative, so a false alarm is allowed, a miss is not + clean = { + "str": ["a", "é中", "\U0001f600", "", None], + "int": [1, -1, INT64_MAX, INT64_MIN, None], + "float": [0.5, -0.0, 0.0, 1e-300, 1.7e308, None], + "bool": [True, False, None], + "nones": [None, None], + } + for name, column in clean.items(): + assert tc._scalar_column_suspect(list(column)) is False, name + bad = { + "surrogate": ["ok", "a\ud800b"], + "above_max": [1, INT64_MAX + 1], + "below_min": [INT64_MIN - 1, 1], + "nan": [1.5, math.nan], + "inf": [math.inf], + "subnormal": [1.5, 5e-324], + "mixed": [1, "a"], + "int_float": [1, 2.5], + } + for name, column in bad.items(): + assert tc._scalar_column_suspect(list(column)) is True, name + # the screen runs over blocks: a bad value in the last block of a large column is found + column = [{"x": i} for i in range(250_000)] + column[249_999] = {"x": INT64_MAX + 1} + assert tc.classify_vertex_rows(column, set(), True) is None + assert tc.classify_vertex_rows(column, set(), False) is False diff --git a/tests/test_graph_batch_single_pass.py b/tests/test_graph_batch_single_pass.py index 3b21bfdd..2a35a768 100644 --- a/tests/test_graph_batch_single_pass.py +++ b/tests/test_graph_batch_single_pass.py @@ -3,9 +3,11 @@ Before, `_create_vertices_json_bulk` called `json_bulk_scalar_ok` on every value of every row before the first chunk was written, and the chunk was then walked again by json.dumps. Now `type_conversion.classify_vertex_rows` takes a census of the types in C, and the value checks come from the one json.dumps pass of each chunk (`json_bulk_dumps`, as `json_bulk_dumps_tagged` already did for the tagged -text). What stays: the whole input goes the property-matrix way for a property name that is not a str or a type the text does not carry -(`_create_vertices_json_bulk` returns None); what changes: a scalar the text would change (a subnormal, a NaN, an integer beyond 64 bits, -a lone surrogate) sends its own chunk, not the whole input, through the matrix path, as it already did for a tagged chunk. +text). The rule that keeps the stored type of a property from depending on which chunk held a bad value: an input of one chunk is decided +by its own dumps before anything is sent; an input of several chunks first has its columns of exact scalars screened in C over the whole +input (`classify_vertex_rows(..., screen_scalars=True)`), and a bad scalar (a NaN, an integer beyond 64 bits, a lone surrogate, a +subnormal) anywhere sends the WHOLE input the property-matrix way, as it did before. A name that is not a str or a type the text does not +carry does the same (`_create_vertices_json_bulk` returns None). """ import pytest @@ -23,10 +25,12 @@ def _rows(n): ] -def test_scalar_rows_are_not_judged_value_by_value(temp_db_path, monkeypatch): +@pytest.mark.parametrize("chunk", [100_000, 500], ids=["one_chunk", "screened"]) +def test_scalar_rows_are_not_judged_value_by_value(temp_db_path, monkeypatch, chunk): def refuse(value): raise AssertionError("a per-value gate ran over a scalar table") + monkeypatch.setattr(GraphBatch, "_BULK_CHUNK", chunk) monkeypatch.setattr(tc, "json_bulk_scalar_ok", refuse) monkeypatch.setattr(gbm, "json_bulk_scalar_ok", refuse) monkeypatch.setattr(tc, "json_bulk_shape_ok", refuse) @@ -53,16 +57,17 @@ def test_the_whole_input_still_goes_the_matrix_way_for_a_name_or_type_the_text_d @pytest.mark.parametrize("value", [5e-324, float("nan"), 2**70, "a\ud800b"]) -def test_a_scalar_the_text_changes_sends_its_chunk_through_the_matrix_path( - temp_db_path, monkeypatch, value +@pytest.mark.parametrize("chunks", [1, 3], ids=["one_chunk", "three_chunks"]) +def test_a_scalar_the_text_changes_sends_the_whole_input_through_the_matrix_path( + temp_db_path, monkeypatch, value, chunks ): - monkeypatch.setattr(GraphBatch, "_BULK_CHUNK", 100) + monkeypatch.setattr(GraphBatch, "_BULK_CHUNK", 100 if chunks == 3 else 1_000) calls = [] real = GraphBatch._to_java_property_matrix - def spy(self, chunk): - calls.append(len(chunk)) - return real(self, chunk) + def spy(self, rows): + calls.append(len(rows)) + return real(self, rows) monkeypatch.setattr(GraphBatch, "_to_java_property_matrix", spy) rows = [{"id": i, "x": 1.5} for i in range(250)] @@ -76,7 +81,7 @@ def spy(self, chunk): Exception ): # noqa: BLE001 - the matrix path refuses a value that no entry point stores rids = None - assert calls == [100], "only the chunk of row 150 takes the matrix path" + assert calls == [250], "the whole input takes the matrix path, once" if value == 5e-324: # stored as the double every other entry point stores; the text of a subnormal was read as a DECIMAL assert len(rids) == 250 @@ -84,16 +89,20 @@ def spy(self, chunk): assert stored == 5e-324 and type(stored) is float +@pytest.mark.parametrize( + "bad_row", [5, 150_000, 249_999], ids=["first_chunk", "middle_chunk", "last_chunk"] +) def test_the_stored_type_does_not_depend_on_which_chunk_held_a_bad_value( - temp_db_path, monkeypatch + temp_db_path, monkeypatch, bad_row ): - """An undeclared int property is stored as the same type in the chunks that crossed as JSON text and in the chunk that took the - property-matrix path because a NaN sat in another property of one of its rows (the 100,000-row chunking of the real class). + """An undeclared int property is stored as the same Java type in every chunk, whichever chunk holds a NaN in another property + (the 100,000-row chunking of the real class): the NaN sends the whole input the matrix way, so no chunk is stored from the JSON text. """ n = 250_000 rows = [{"id": i, "k": i % 1000} for i in range(n)] - rows[150_000] = {"id": 150_000, "k": 1, "x": float("nan")} # chunk 2 of 3 only - probes = {5: "chunk 1", 150_001: "chunk 2", 249_999: "chunk 3"} + rows[bad_row] = {"id": bad_row, "k": 1, "x": float("nan")} + probes = {7: "chunk 1", 150_001: "chunk 2", 249_999: "chunk 3"} + probes.pop(bad_row, None) calls = [] real = GraphBatch._to_java_property_matrix @@ -106,16 +115,42 @@ def spy(self, chunk): db.command("sql", "CREATE VERTEX TYPE V") with db.graph_batch(parallel_flush=False) as batch: assert len(batch.create_vertices("V", rows)) == n - assert calls == [100_000], "only the chunk with the NaN takes the matrix path" - ids = ", ".join(str(i) for i in probes) - query = ( - f"SELECT id, k, k.javaType() AS t FROM V WHERE id IN [{ids}]" # nosec B608 - ) - got = db.query("sql", query).to_list() - by_id = {r["id"]: (r["k"], str(r["t"])) for r in got} - assert set(by_id) == set(probes) - assert {i: by_id[i][0] for i in probes} == {i: i % 1000 for i in probes} - types = {probes[i]: by_id[i][1] for i in probes} + assert calls == [n], "the whole input takes the matrix path, once" + types = _stored_types(db, probes) assert ( len(set(types.values())) == 1 ), f"the stored type depends on the chunk: {types}" + + +def test_a_clean_input_of_several_chunks_is_stored_alike_in_every_chunk( + temp_db_path, monkeypatch +): + n = 250_000 + rows = [{"id": i, "k": i % 1000} for i in range(n)] + probes = {7: "chunk 1", 150_001: "chunk 2", 249_999: "chunk 3"} + calls = [] + real = GraphBatch._to_java_property_matrix + monkeypatch.setattr( + GraphBatch, + "_to_java_property_matrix", + lambda self, chunk: calls.append(len(chunk)) or real(self, chunk), + ) + 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 len(batch.create_vertices("V", rows)) == n + assert calls == [], "a clean input never takes the matrix path" + types = _stored_types(db, probes) + assert ( + len(set(types.values())) == 1 + ), f"the stored type depends on the chunk: {types}" + + +def _stored_types(db, probes): + ids = ", ".join(str(i) for i in probes) + query = f"SELECT id, k, k.javaType() AS t FROM V WHERE id IN [{ids}]" # nosec B608 + got = db.query("sql", query).to_list() + by_id = {r["id"]: (r["k"], str(r["t"])) for r in got} + assert set(by_id) == set(probes) + assert {i: by_id[i][0] for i in probes} == {i: i % 1000 for i in probes} + return {probes[i]: by_id[i][1] for i in probes}