diff --git a/src/arcadedb/core.py b/src/arcadedb/core.py index 998aa37b..76d2d77e 100644 --- a/src/arcadedb/core.py +++ b/src/arcadedb/core.py @@ -696,7 +696,10 @@ def insert_many( stays in the JSON. Args: - type_name: Target document type (must exist). + type_name: Target type (must exist): a document type gets + documents, a vertex type gets vertices (use ``graph_batch()`` + when edges follow), and an edge type raises ``ArcadeDBError`` + because an edge needs both endpoints. rows: Iterable of dicts, one per document. Read lazily, one chunk at a time. commit_every: Transaction batch size for the synchronous mode. @@ -716,7 +719,7 @@ def insert_many( 10,240); ``commit_every`` does not apply to this mode. Returns: - Number of documents inserted. + Number of records inserted. Raises: Exception: An exception raised by ``rows`` itself propagates @@ -797,6 +800,14 @@ def insert_many( pass raise + def _is_vertex_type(self, type_name): + """True when ``type_name`` is an existing vertex type (a bulk load then creates vertices).""" + java_schema = self._java_db.getSchema() + return java_schema.existsType(type_name) and isinstance( + java_schema.getType(type_name), + _java_class("com.arcadedb.schema.VertexType"), + ) + def _insert_many_chunk(self, type_name, chunk): """Insert one chunk of insert_many rows inside the open transaction.""" # Float vectors for ARRAY_OF_FLOATS properties cross as one binary matrix @@ -809,9 +820,15 @@ def _insert_many_chunk(self, type_name, chunk): # scalars, which json.dumps rejects; an integer beyond 64 bits, # NaN or Infinity, a non-str dict key, a lone surrogate, which the # engine's JSON parser would store as a different value): this - # chunk goes row by row. + # chunk goes row by row (vertices for a vertex type, as the Java + # helper does). + new_record = ( + self.new_vertex + if self._is_vertex_type(type_name) + else self.new_document + ) for row in chunk: - doc = self.new_document(type_name) + doc = new_record(type_name) for k, v in row.items(): doc.set(k, v) doc.save() @@ -935,7 +952,9 @@ def insert_columns( property). Args: - type_name: Target document type (must exist). + type_name: Target type (must exist): a document type gets + documents, a vertex type gets vertices, an edge type raises + ``ArcadeDBError``. columns: ``{property name: column}``, or a pandas ``DataFrame``. Every column has the same length. A column is a numpy array (integer kinds cross as ``long[]``, float kinds as ``double[]``, @@ -955,7 +974,7 @@ def insert_columns( columns instead of one JSON text per batch. Returns: - Number of documents inserted. + Number of records inserted. Raises: ValueError: If ``columns`` is empty, a column has a different diff --git a/src/java/com/arcadedb/python/DocumentBatcher.java b/src/java/com/arcadedb/python/DocumentBatcher.java index 848f18f4..fa36a708 100644 --- a/src/java/com/arcadedb/python/DocumentBatcher.java +++ b/src/java/com/arcadedb/python/DocumentBatcher.java @@ -18,12 +18,19 @@ * JSON-representable property values only (str/int/float/bool/null and * nested lists/maps thereof) — the Python side falls back to the per-row * new_document path for anything else (e.g. datetime, bytes). + * + * Every entry point loads a document type or a vertex type: a vertex type gets + * real vertices (db.newVertex), as INSERT INTO does on the SQL path and on the + * native wheel; an edge type is refused, because an edge needs both endpoints. */ package com.arcadedb.python; import com.arcadedb.database.Database; import com.arcadedb.database.MutableDocument; import com.arcadedb.database.async.ErrorCallback; +import com.arcadedb.schema.DocumentType; +import com.arcadedb.schema.EdgeType; +import com.arcadedb.schema.VertexType; import com.arcadedb.serializer.json.JSONArray; import com.arcadedb.serializer.json.JSONObject; @@ -58,12 +65,13 @@ public static long insertManyJson(final Database db, final String typeName, fina // A caller's own transaction is the caller's to commit or roll back: batch // commits and the failure rollback apply only to the transactions opened // here, matching the Python per-row fallback (#7882). + final boolean vertex = n > 0 && isVertexType(db, typeName); final boolean wasActive = db.isTransactionActive(); if (!wasActive) db.begin(); try { for (int i = 0; i < n; i++) { - final MutableDocument doc = db.newDocument(typeName); + final MutableDocument doc = newRecord(db, typeName, vertex); fill(doc, rows.getJSONObject(i), i, vectorNames, vectors); doc.save(); if (!wasActive && commitEvery > 0 && (i + 1) % commitEvery == 0) { @@ -101,12 +109,13 @@ public static long insertManyJson(final Database db, final String typeName, fina public static long insertColumns(final Database db, final String typeName, final String[] names, final Object[] columns, final int n, final int commitEvery) { checkColumns(names, columns, n); + final boolean vertex = n > 0 && isVertexType(db, typeName); final boolean wasActive = db.isTransactionActive(); if (!wasActive) db.begin(); try { for (int i = 0; i < n; i++) { - final MutableDocument doc = db.newDocument(typeName); + final MutableDocument doc = newRecord(db, typeName, vertex); fillRow(doc, names, columns, i); doc.save(); if (!wasActive && commitEvery > 0 && (i + 1) % commitEvery == 0) { @@ -139,14 +148,33 @@ public static AsyncFailures insertColumnsParallel(final Database db, final Strin checkColumns(names, columns, n); final AsyncFailures failures = new AsyncFailures(); final ErrorCallback onError = failures::record; + final boolean vertex = n > 0 && isVertexType(db, typeName); for (int i = 0; i < n; i++) { - final MutableDocument doc = db.newDocument(typeName); + final MutableDocument doc = newRecord(db, typeName, vertex); fillRow(doc, names, columns, i); db.async().createRecord(doc, null, onError); } return failures; } + /** + * Whether the bulk load into the type creates vertices (true) or documents (false). Resolved once per call, so a row costs no + * schema lookup. A type that does not exist raises the schema's own SchemaException, as newDocument did. + * + * @throws IllegalArgumentException for an edge type: an edge cannot be created without its two endpoints + */ + private static boolean isVertexType(final Database db, final String typeName) { + final DocumentType type = db.getSchema().getType(typeName); + if (type instanceof EdgeType) + throw new IllegalArgumentException("Cannot bulk-insert into '" + typeName + + "' because it is an edge type; load the vertices, then create the edges (graph_batch().new_edges() or CREATE EDGE)"); + return type instanceof VertexType; + } + + private static MutableDocument newRecord(final Database db, final String typeName, final boolean vertex) { + return vertex ? db.newVertex(typeName) : db.newDocument(typeName); + } + private static void checkColumns(final String[] names, final Object[] columns, final int n) { final int k = names.length; if (columns.length != k) @@ -233,8 +261,9 @@ private static AsyncFailures insertManyJsonParallel(final Database db, final Str final AsyncFailures failures = new AsyncFailures(); final ErrorCallback onError = failures::record; final int n = rows.length(); + final boolean vertex = n > 0 && isVertexType(db, typeName); for (int i = 0; i < n; i++) { - final MutableDocument doc = db.newDocument(typeName); + final MutableDocument doc = newRecord(db, typeName, vertex); fill(doc, rows.getJSONObject(i), i, vectorNames, vectors); db.async().createRecord(doc, null, onError); } diff --git a/tests/test_bulk_insert.py b/tests/test_bulk_insert.py index 36c13bb5..124c4271 100644 --- a/tests/test_bulk_insert.py +++ b/tests/test_bulk_insert.py @@ -4,6 +4,8 @@ import pytest +import arcadedb + def _count(db, type_name): q = f"SELECT count(*) AS n FROM {type_name}" # nosec B608 - test-controlled @@ -146,6 +148,104 @@ def test_inside_open_transaction(self, temp_db): assert _count(temp_db, "Tx") == 2 +class TestInsertManyVertexType: + """insert_many and insert_columns into a VERTEX type store real vertices. + + The JVM wheel raised "Cannot create a document of type ... because is not a + document type" (the Java helper called newDocument), while the native wheel + accepted the same call (found 2026-10-25 by a parity probe between the two + wheels). A record counts as a vertex when an edge can join it: CREATE EDGE + refuses a record that is not one. + """ + + ROWS = 25 + + def _vertex_db(self, db, name="Vtx", buckets=None): + suffix = f" BUCKETS {buckets}" if buckets else "" + db.command("sql", f"CREATE VERTEX TYPE {name}{suffix}") + db.command("sql", "CREATE EDGE TYPE Knows") + + def _assert_vertices(self, db, name, expected): + assert _count(db, name) == expected + with db.transaction(): + db.command( + "sql", + f"CREATE EDGE Knows FROM (SELECT FROM {name} WHERE id = 0) " # nosec B608 - test-owned type name, no user input + f"TO (SELECT FROM {name} WHERE id = 1)", + ) + assert _count(db, "Knows") == 1 + out = db.query( + "sql", + f"SELECT out('Knows').size() AS n FROM {name} WHERE id = 0", # nosec B608 - test-owned type name, no user input + ).to_list()[0]["n"] + assert out == 1 + + def test_insert_many(self, temp_db): + self._vertex_db(temp_db) + rows = [{"id": i, "name": f"v{i}", "tags": ["a"]} for i in range(self.ROWS)] + assert temp_db.insert_many("Vtx", rows, commit_every=10) == self.ROWS + self._assert_vertices(temp_db, "Vtx", self.ROWS) + got = temp_db.query("sql", "SELECT FROM Vtx WHERE id = 7").to_list()[0] + assert got["name"] == "v7" + assert list(got["tags"]) == ["a"] + + def test_insert_many_parallel(self, temp_db): + self._vertex_db(temp_db, buckets=2) + rows = [{"id": i} for i in range(2000)] + assert temp_db.insert_many("Vtx", rows, parallel=True) == 2000 + self._assert_vertices(temp_db, "Vtx", 2000) + + def test_insert_many_non_json_fallback(self, temp_db): + self._vertex_db(temp_db) + rows = [ + {"id": i, "when": datetime.datetime(2026, 7, 25, 12, 0, i)} + for i in range(3) + ] + assert temp_db.insert_many("Vtx", rows) == 3 + self._assert_vertices(temp_db, "Vtx", 3) + + def test_insert_many_inside_a_caller_transaction(self, temp_db): + self._vertex_db(temp_db) + with pytest.raises(RuntimeError): + with temp_db.transaction(): + temp_db.insert_many("Vtx", [{"id": 0}, {"id": 1}], commit_every=1) + raise RuntimeError("roll back") + assert _count(temp_db, "Vtx") == 0 + with temp_db.transaction(): + temp_db.insert_many("Vtx", [{"id": 0}, {"id": 1}], commit_every=1) + self._assert_vertices(temp_db, "Vtx", 2) + + def test_insert_columns(self, temp_db): + import numpy as np + + self._vertex_db(temp_db) + assert temp_db.insert_columns("Vtx", {"id": np.arange(5, dtype="int64")}) == 5 + self._assert_vertices(temp_db, "Vtx", 5) + + def test_insert_columns_parallel(self, temp_db): + import numpy as np + + self._vertex_db(temp_db, buckets=2) + n = 1000 + assert ( + temp_db.insert_columns( + "Vtx", {"id": np.arange(n, dtype="int64")}, parallel=True + ) + == n + ) + self._assert_vertices(temp_db, "Vtx", n) + + def test_an_edge_type_is_refused(self, temp_db): + self._vertex_db(temp_db) + with pytest.raises(arcadedb.ArcadeDBError, match="Knows"): + temp_db.insert_many("Knows", [{"id": 1}]) + assert _count(temp_db, "Knows") == 0 + + def test_a_missing_type_still_raises(self, temp_db): + with pytest.raises(arcadedb.ArcadeDBError, match="Nope"): + temp_db.insert_many("Nope", [{"id": 1}]) + + class TestInsertManyParallelReportsFailures: """A record the parallel writers fail to store must fail the call.