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..4fad9342 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, 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 + # 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..68e3857a 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. - `rows` is a list of dicts (one per record). The common value types are - checked in this loop without a call per value, because every value of a - large load passes through it; floats need no check here, since - `allow_nan=False` makes json.dumps itself refuse NaN and the infinities. + 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,95 @@ 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) 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) + 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 +1040,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 +1054,147 @@ 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)) + + +_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 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. + """ + 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 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 # exact scalars, or the vectors of a declared column + for value in _column(rows, key): + 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 + 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..c830597b --- /dev/null +++ b/tests/test_bulk_single_pass_validation.py @@ -0,0 +1,464 @@ +"""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) # 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 = [] + 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_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, {}) + 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) # nosec B311 - a seeded corpus, not security + 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): + query = f"SELECT FROM {type_name}" # nosec B608 - test-controlled + rows = db.query("sql", query).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) + + +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 new file mode 100644 index 00000000..2a35a768 --- /dev/null +++ b/tests/test_graph_batch_single_pass.py @@ -0,0 +1,156 @@ +"""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). 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 + +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) + ] + + +@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) + 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"]) +@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 if chunks == 3 else 1_000) + calls = [] + real = GraphBatch._to_java_property_matrix + + 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)] + 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 == [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 + stored = db.query("sql", "SELECT x FROM V WHERE id = 150").first().get("x") + 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, bad_row +): + """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[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 + + 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 == [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} 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