insert_many and create_vertices: check the rows in the one json.dumps pass, not in a loop before it (refs #318)

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 = []

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,