From a215349218d8d195c53a5fb343f1d4e62db16227 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 26 Oct 2026 01:11:57 +0900 Subject: [PATCH] Native build: commit every 1000 rows by default in insert_many, graph_batch and create_vertices A pre-registered laptop A/B on the columnar native wheel (1M documents through insert_many, 200k vertices and 500k edges through GraphBatch) found 1000-row transactions faster (wall 0.822-0.831 and 0.853-0.864 of the old default) and lighter on the young GC. The native defaults become insert_many commit_every 1000 (was 10,000), graph_batch commit_every 1000 (was unset) and a create_vertices chunk equal to the batch's commit_every (was 10,000). An explicit value wins. The JVM defaults are unchanged and pinned by a source-level test. Co-Authored-By: Claude Sonnet 5.5 --- docs/modules/ROOT/nav.adoc | 1 + .../modules/ROOT/pages/how-to/bulk-loads.adoc | 66 +++++++++ src/arcadedb/_native/database.py | 20 ++- src/arcadedb/_native/extras.py | 20 ++- src/arcadedb/core.py | 7 +- src/arcadedb/graph_batch.py | 4 + tests/conftest.py | 2 + tests/test_ingest_defaults.py | 94 ++++++++++++ tests/test_native_columnar_insert.py | 16 ++ tests/test_native_ingest_defaults.py | 138 ++++++++++++++++++ 10 files changed, 358 insertions(+), 10 deletions(-) create mode 100644 docs/modules/ROOT/pages/how-to/bulk-loads.adoc create mode 100644 tests/test_ingest_defaults.py create mode 100644 tests/test_native_ingest_defaults.py diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc index 79c9979..ecafb18 100644 --- a/docs/modules/ROOT/nav.adoc +++ b/docs/modules/ROOT/nav.adoc @@ -6,6 +6,7 @@ ** xref:how-to/databases-and-queries.adoc[Databases and queries] ** xref:how-to/concurrency.adoc[Concurrency, processes and crashes] ** xref:how-to/logs.adoc[Where the logs go] +** xref:how-to/bulk-loads.adoc[Bulk loads and commit size] * Development ** xref:development/ci.adoc[Continuous integration] ** xref:development/release.adoc[Release] diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc new file mode 100644 index 0000000..9d83efb --- /dev/null +++ b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc @@ -0,0 +1,66 @@ += Bulk loads and commit size +:description: The default commit size of insert_many, graph_batch and create_vertices on the arcadedb (native) and arcadedb-jvm packages, what it means for a failed load, and how to change it. + +`insert_many`, `graph_batch` and `GraphBatch.create_vertices` load data in transactions. The commit size is the number of rows in one transaction. It decides how much work the engine holds in memory before it commits, and how many rows stay in the database when a load fails. + +== Defaults + +[cols="2,2,2"] +|=== +| Call | `arcadedb` (native) | `arcadedb-jvm` + +| `db.insert_many(type_name, rows)` +| `commit_every=1000` +| `commit_every=10000` + +| `db.graph_batch()` +| `commit_every=1000` +| `commit_every` not set: one commit for each flush + +| `batch.create_vertices(type_name, rows)` +| 1000 rows for each call into the engine, one transaction for each call (the `commit_every` of the batch) +| 100000 rows for each call into the engine +|=== + +The JVM defaults are the same as before. They stay until someone measures them on the JVM build. + +An explicit value always wins. `commit_every=0` gives one transaction for `insert_many` and one commit for each flush for `graph_batch`, on both packages. In a transaction that you open yourself, `insert_many` commits nothing. + +== Why the native defaults are smaller + +The in-flight transaction is the data that each young garbage collection must copy. A smaller transaction copies less. A laptop A/B on the columnar native wheel measured this with `ab.py --noise`, ABBA order, 12 clean pairs for each figure, and a clean A/A control (95 % interval of the median pair ratio, new default against old default): + +[cols="3,2,2,2"] +|=== +| Load | Wall time | Data copied by the young GC | Peak RSS + +| `insert_many`, 1,000,000 mixed documents, `commit_every=1000` +| 0.822 to 0.831 +| 0.55x +| 0.955x + +| `GraphBatch`, 200,000 vertices and 500,000 edges, `parallel_flush`, one client thread, `commit_every=1000` +| 0.853 to 0.864 +| 0.63x +| 0.718x +|=== + +`commit_every=5000` for `insert_many` also passed (0.903 to 0.914 of the wall time). 1000 had the lower median, so it is the default. The runs used an isolate heap of 80 % of RAM. The default is now 50 %, and smaller transactions lower the heap that a load needs. + +== What changes for you + +Rows that are already committed stay committed when a later chunk fails. Only the transaction that was open is rolled back. With a smaller commit size, a failed load keeps its rows up to the last commit, and that commit is now at most 1000 rows behind the failure instead of 10,000. If you need all-or-nothing, open your own transaction, or pass `commit_every=0`. + +== Change the commit size + +[source,python] +---- +db.insert_many("Doc", rows, commit_every=10_000) # the old native default +db.insert_many("Doc", rows, commit_every=0) # one transaction for all rows + +with db.graph_batch(commit_every=0) as batch: # one commit for each flush, as before + ... + +with db.graph_batch(commit_every=5000) as batch: # also sets the create_vertices chunk + ... +---- diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 3c48c32..d20cb82 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -21,6 +21,9 @@ _LAZY_COMMAND_ROWS = os.environ.get("ARCADEDB_NATIVE_LAZY_CMD", "1") != "0" _BINARY_REPLY = os.environ.get("ARCADEDB_NATIVE_BIN_REPLY", "1") != "0" _INSERT_MANY_CHUNK = 10_000 +# rows per transaction when insert_many gets no commit_every: 1000 on the native build (laptop A/B, smaller commits copy less in the +# young GC and ran faster); the JVM build keeps 10,000 (core.py), which is not measured +DEFAULT_INSERT_MANY_COMMIT_EVERY = 1000 _XML_LIMIT_ERROR = re.compile(r"JAXP0001000[134]") # the JDK's XML entity limits _XML_LIMIT_HINT = ( " The JDK's XML entity limits stopped this import. Pass disable_xml_limits=True to import_documents to lift them for this" @@ -682,14 +685,20 @@ def lookup_by_key(self, type_name: str, keys: List[str], values: List[Any]): raise ArcadeDBError(f"Failed to lookup by key in '{type_name}': {e}") from e def insert_many( - self, type_name: str, rows, commit_every: int = 10_000, parallel: bool = False + self, + type_name: str, + rows, + commit_every: int = DEFAULT_INSERT_MANY_COMMIT_EVERY, + parallel: bool = False, ) -> int: - """Bulk insert. Rows go in chunks of 10,000, one crossing per chunk. A chunk of rows with the same keys whose columns each hold one + """Bulk insert. Rows go in chunks of at most 10,000 (a chunk never crosses a commit), one crossing per chunk. A chunk of rows with the same keys whose columns each hold one value type (ints, floats, bools, strings, equal-length float lists for an ``ARRAY_OF_FLOATS`` property; None is a null) goes as columns by address (``arcadedb_n_insert_colp``, no JSON text); any other chunk goes as one JSON text (``arcadedb_n_insert_rows``: a newDocument/set/save loop inside the image, as ``DocumentBatcher.insertManyJson`` on the JPype path); a chunk with a value the JSON text cannot carry unchanged goes through ``INSERT INTO t CONTENT :row`` for each row, in one crossing. Inside the caller's - transaction nothing is committed; otherwise every ``commit_every`` rows are one transaction (0: all rows in one). + transaction nothing is committed; otherwise every ``commit_every`` rows are one transaction (0: all rows in one). The default is 1000 on + this build (the JVM build's is 10,000). Rows that are already committed stay committed if a later chunk fails, so a smaller + ``commit_every`` keeps fewer rows of a failed load (only the transaction that was open is rolled back). """ self._check_not_closed() if parallel: @@ -872,7 +881,10 @@ def new_document(self, type_name: str): return Document(self, type_name) def graph_batch(self, **options): - """A GraphBatch for bulk graph loads (keywords as :meth:`Database.graph_batch`).""" + """A GraphBatch for bulk graph loads (keywords as :meth:`Database.graph_batch`). ``commit_every`` defaults to 1000 on this build + (the JVM build leaves it unset: one commit per flush); ``commit_every=0`` gives one commit per flush here too, and an explicit value + wins. It is also the number of rows ``create_vertices`` sends, and commits, per call into the image. + """ from .extras import NativeGraphBatch self._check_not_closed() diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index e27a1c7..c374396 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -121,18 +121,27 @@ def _call(prefix: str, name: str, *args) -> dict: class NativeGraphBatch(_BaseGraphBatch): """``GraphBatch`` over the engine's batch builder in the image (same names and behaviour as the JPype wrapper).""" - _BULK_CHUNK = 10_000 # rows per crossing: small enough that a chunk's text, rows and reply die in the young generation + _BULK_CHUNK = 10_000 # rows per crossing of new_edges, and of create_vertices when commit_every is 0 + # commit_every when graph_batch() gets none (the JVM build leaves it unset: one commit per flush). It is also the create_vertices + # chunk: each crossing is one transaction, and 1000 rows per transaction ran faster and copied less in the young GC than 10,000 + DEFAULT_COMMIT_EVERY = 1000 - def __init__(self, database, handle: int): + def __init__(self, database, handle: int, commit_every: int = DEFAULT_COMMIT_EVERY): self._db = database self._gb = handle self._closed = False + self._vertex_chunk = ( + int(commit_every) if int(commit_every) > 0 else self._BULK_CHUNK + ) @classmethod def create(cls, database, **options) -> "NativeGraphBatch": wal_flush = options.get("wal_flush") if wal_flush is not None: options["wal_flush"] = cls._to_java_wal_flush_name(wal_flush) + commit_every = options.get("commit_every") + if commit_every is None: # an explicit value, 0 included, wins + commit_every = options["commit_every"] = cls.DEFAULT_COMMIT_EVERY given = {k: v for k, v in options.items() if v is not None} try: reply = _call( @@ -143,7 +152,7 @@ def create(cls, database, **options) -> "NativeGraphBatch": ) except ArcadeDBError: raise - return cls(database, reply["h"]) + return cls(database, reply["h"], commit_every) @classmethod def _to_java_wal_flush_name(cls, mode: str) -> str: @@ -222,12 +231,15 @@ def _create_vertices_json_bulk(self, type_name: str, 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. + """ self._check_not_closed() db = self._db owns_transaction = not db.is_transaction_active() prefix = f"Failed to create batch vertices for type '{type_name}'" try: - chunk = self._BULK_CHUNK + chunk = self._vertex_chunk if isinstance(count_or_properties, int): rids = [] for start in range(0, count_or_properties, chunk): diff --git a/src/arcadedb/core.py b/src/arcadedb/core.py index 3a36c54..b828ea6 100644 --- a/src/arcadedb/core.py +++ b/src/arcadedb/core.py @@ -708,7 +708,9 @@ def insert_many( 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. + commit_every: Transaction batch size for the synchronous mode. Default 10,000 on this build; the native build (the + ``arcadedb`` package) defaults to 1000, a size measured there and not yet on the JVM. Rows that are already + committed stay committed if a later chunk fails. parallel: If True, route rows through the async executor's parallel bucket writers and wait for completion before returning (out-of-order writes). The maintainers' rule @@ -2018,7 +2020,8 @@ def graph_batch( ArcadeDBError naming the type. Before 26.10.1 it was accepted, and any query the planner walked from the target end returned 0 rows (ArcadeData/arcadedb#8625). - commit_every: Commit cadence within a flush. `0` means one commit per flush. + commit_every: Commit cadence within a flush. `0` means one commit per flush, which is also what an unset value gives on + this build; the native build (the ``arcadedb`` package) defaults to 1000. use_wal: Write-ahead log during the import. Off by default, so a crash mid-import can lose its tail; pass True for a crash-safe import (ArcadeData/arcadedb#8287). wal_flush: WAL flush mode: `"no"`, `"yes_nometadata"`, `"yes_full"`. diff --git a/src/arcadedb/graph_batch.py b/src/arcadedb/graph_batch.py index 1e1f45f..7184752 100644 --- a/src/arcadedb/graph_batch.py +++ b/src/arcadedb/graph_batch.py @@ -190,6 +190,10 @@ def create_vertices( ``ARRAY_OF_FLOATS`` whose value in every row is a list or a numpy array of numbers of one length crosses as one binary float32 matrix instead of JSON text. + + Large inputs cross in chunks of 100,000 rows on this build. On the native build (the ``arcadedb`` package) the chunk is the + batch's ``commit_every``, 1000 unless ``graph_batch()`` got another value, and each chunk is its own transaction. Chunks that + already went in stay stored if a later chunk fails. """ self._check_not_closed() # From engine 26.10.1 the engine refuses to run inside a transaction the diff --git a/tests/conftest.py b/tests/conftest.py index 5cebd98..7094e9e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -41,6 +41,8 @@ collect_ignore.append("test_native_udf.py") # the columnar insert's stored records are compared between the JSON and the columnar entry points of the native library collect_ignore.append("test_native_columnar_insert.py") + # the native build's ingest defaults (insert_many and graph_batch commit size), observed on the native library + collect_ignore.append("test_native_ingest_defaults.py") # the native library's start-up output (#477) can only be checked with the native library loaded collect_ignore.append("test_native_startup_runtime.py") # the XML entity limits of an import, lifted by the native glue (the JVM build lifts them with JVM system properties) diff --git a/tests/test_ingest_defaults.py b/tests/test_ingest_defaults.py new file mode 100644 index 0000000..7003a35 --- /dev/null +++ b/tests/test_ingest_defaults.py @@ -0,0 +1,94 @@ +"""The ingest defaults, read from the source so the JVM is not started and the native library is not loaded. + +The native build (the ``arcadedb`` package) commits every 1000 rows in ``insert_many`` and ``graph_batch`` (a laptop A/B: smaller +transactions copy less in the young GC and ran faster). The JVM build (``arcadedb-jvm``) keeps its old defaults until they are measured: +``insert_many`` 10,000, ``graph_batch`` unset (one commit per flush), ``create_vertices`` 100,000 rows per crossing. These tests pin both +sides, and run on both backends because they only parse the files. +""" + +import ast +from pathlib import Path + +import arcadedb + +SRC = Path(arcadedb.__file__).parent + + +def _tree(rel): + return ast.parse((SRC / rel).read_text(encoding="utf-8")) + + +def _param_default(rel, func, arg): + """The AST node of the default of ``arg`` in the only ``def func`` of the file.""" + funcs = [ + n + for n in ast.walk(_tree(rel)) + if isinstance(n, ast.FunctionDef) and n.name == func + ] + assert len(funcs) == 1, (rel, func, len(funcs)) + a = funcs[0].args + positional = a.posonlyargs + a.args + pairs = list(zip(positional[len(positional) - len(a.defaults) :], a.defaults)) + pairs += [(k, d) for k, d in zip(a.kwonlyargs, a.kw_defaults) if d is not None] + for node, default in pairs: + if node.arg == arg: + return default + raise AssertionError(f"{rel}: {func}() has no default for {arg}") + + +def _module_constant(rel, name): + for node in _tree(rel).body: + if isinstance(node, ast.Assign) and any( + isinstance(t, ast.Name) and t.id == name for t in node.targets + ): + return ast.literal_eval(node.value) + raise AssertionError(f"{rel}: no module constant {name}") + + +def _class_constant(rel, cls, name): + for node in ast.walk(_tree(rel)): + if isinstance(node, ast.ClassDef) and node.name == cls: + for item in node.body: + if isinstance(item, ast.Assign) and any( + isinstance(t, ast.Name) and t.id == name for t in item.targets + ): + return ast.literal_eval(item.value) + raise AssertionError(f"{rel}: no {cls}.{name}") + + +def test_jvm_insert_many_default_is_10000(): + node = _param_default("core.py", "insert_many", "commit_every") + assert ast.literal_eval(node) == 10_000 + + +def test_jvm_graph_batch_commit_every_stays_unset(): + node = _param_default("core.py", "graph_batch", "commit_every") + assert ast.literal_eval(node) is None + # the JVM wrapper passes the builder option only when it is given, so unset stays one commit per flush + node = _param_default("graph_batch.py", "create", "commit_every") + assert ast.literal_eval(node) is None + + +def test_jvm_create_vertices_chunk_stays_100000(): + assert _class_constant("graph_batch.py", "GraphBatch", "_BULK_CHUNK") == 100_000 + + +def test_native_insert_many_default_is_1000(): + assert ( + _module_constant("_native/database.py", "DEFAULT_INSERT_MANY_COMMIT_EVERY") + == 1000 + ) + node = _param_default("_native/database.py", "insert_many", "commit_every") + assert isinstance(node, ast.Name) + assert node.id == "DEFAULT_INSERT_MANY_COMMIT_EVERY" + + +def test_native_graph_batch_default_is_1000_and_the_edge_chunk_is_unchanged(): + assert ( + _class_constant("_native/extras.py", "NativeGraphBatch", "DEFAULT_COMMIT_EVERY") + == 1000 + ) + assert ( + _class_constant("_native/extras.py", "NativeGraphBatch", "_BULK_CHUNK") + == 10_000 + ) diff --git a/tests/test_native_columnar_insert.py b/tests/test_native_columnar_insert.py index 3aac3df..a3591d5 100644 --- a/tests/test_native_columnar_insert.py +++ b/tests/test_native_columnar_insert.py @@ -23,6 +23,22 @@ ) +@pytest.fixture(autouse=True) +def _one_chunk_per_commit(monkeypatch): + """The crossing counts below are for one chunk of up to 10,000 rows, so these tests run insert_many with its old commit_every of + 10,000 (the native default is 1000, which cuts a 2,500-row load into three crossings). The tests are about the columnar entry + points, not the default; tests/test_native_ingest_defaults.py covers the default.""" + from arcadedb._native import database as native_database + + cls = next( + c + for c in vars(native_database).values() + if isinstance(c, type) and "insert_many" in vars(c) + ) + defaults = cls.insert_many.__defaults__ + monkeypatch.setattr(cls.insert_many, "__defaults__", (10_000,) + defaults[1:]) + + class Spy: def __init__(self, monkeypatch): self.calls = [] diff --git a/tests/test_native_ingest_defaults.py b/tests/test_native_ingest_defaults.py new file mode 100644 index 0000000..9234bb7 --- /dev/null +++ b/tests/test_native_ingest_defaults.py @@ -0,0 +1,138 @@ +"""The native build's ingest defaults, observed on the native library (conftest.py collects this file on the native backend only). + +``insert_many`` commits every 1000 rows, ``graph_batch`` commits every 1000 and ``create_vertices`` sends (and commits) 1000 rows per +call into the image. An explicit value wins, and the semantics stay: rows that are already committed stay committed if a later chunk fails. +""" + +import inspect + +import pytest + +import arcadedb +from arcadedb._native import extras +from arcadedb.exceptions import ArcadeDBError + +_COUNT_SQL = { # fixed statements, one for each type the tests create + "Dup": "SELECT count(*) AS c FROM Dup", + "V": "SELECT count(*) AS c FROM V", + "W": "SELECT count(*) AS c FROM W", +} + + +def _count(db, type_name): + return list(db.query("sql", _COUNT_SQL[type_name]))[0].get("c") + + +def _unique_type(db, name): + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.k INTEGER") + db.command("sql", f"CREATE INDEX ON {name} (k) UNIQUE") + + +def test_insert_many_signature_default_is_1000(temp_db): + param = inspect.signature(temp_db.insert_many).parameters["commit_every"] + assert param.default == 1000 + + +@pytest.mark.parametrize( + "kwargs, kept", + [ + ({}, 2000), # the default: 1000-row transactions, the third one fails + ({"commit_every": 1000}, 2000), + ({"commit_every": 1500}, 1500), # an explicit value wins + ( + {"commit_every": 10_000}, + 0, + ), # the old default: nothing was committed before the failure + ({"commit_every": 0}, 0), # one transaction + ], +) +def test_insert_many_commit_size_decides_what_a_failed_load_keeps( + temp_db, kwargs, kept +): + _unique_type(temp_db, "Dup") + rows = [{"k": i} for i in range(2500)] + rows[2400] = {"k": 0} # a duplicate key inside the third 1000 rows + with pytest.raises(ArcadeDBError): + temp_db.insert_many("Dup", rows, **kwargs) + assert temp_db.is_transaction_active() is False + assert _count(temp_db, "Dup") == kept + + +class _Spy: + def __init__(self, monkeypatch): + self.created = ( + [] + ) # the option dicts that went to arcadedb_n_gb_create (and every other call that encodes a dict) + self.vertex_calls = ( + [] + ) # the row or vertex count of each create_vertices crossing + real_enc = extras._enc + real_op = extras.NativeGraphBatch._op + + def enc(value): + if isinstance(value, dict): + self.created.append(dict(value)) + return real_enc(value) + + def op(this, prefix, name, **args): + if name == "create_vertices": + rows = args.get("rows") + self.vertex_calls.append( + len(rows) if rows is not None else args.get("count") + ) + return real_op(this, prefix, name, **args) + + monkeypatch.setattr(extras, "_enc", enc) + monkeypatch.setattr(extras.NativeGraphBatch, "_op", op) + + def commit_every_given(self): + return [d["commit_every"] for d in self.created if "commit_every" in d] + + +def _vertices(db, name): + db.command("sql", f"CREATE VERTEX TYPE {name}") + + +def test_graph_batch_default_commit_every_is_1000(temp_db, monkeypatch): + spy = _Spy(monkeypatch) + _vertices(temp_db, "V") + with temp_db.graph_batch(parallel_flush=False) as gb: + gb.create_vertices("V", [{"id": i} for i in range(2500)]) + gb.create_vertices("V", 2500) + assert spy.commit_every_given() == [1000] + assert spy.vertex_calls == [1000, 1000, 500, 1000, 1000, 500] + assert _count(temp_db, "V") == 5000 + + +@pytest.mark.parametrize( + "commit_every, calls", + [ + ( + 250, + [250] * 10, + ), # an explicit value wins, for the engine and for the create_vertices chunk + ( + 0, + [2500], + ), # 0: one commit per flush, and the create_vertices chunk is the old 10,000 + ], +) +def test_graph_batch_explicit_commit_every_wins( + temp_db, monkeypatch, commit_every, calls +): + spy = _Spy(monkeypatch) + _vertices(temp_db, "W") + with temp_db.graph_batch(commit_every=commit_every, parallel_flush=False) as gb: + gb.create_vertices("W", [{"id": i} for i in range(2500)]) + assert spy.commit_every_given() == [commit_every] + assert spy.vertex_calls == calls + assert _count(temp_db, "W") == 2500 + + +def test_graph_batch_default_constant_and_edges_chunk(): + assert extras.NativeGraphBatch.DEFAULT_COMMIT_EVERY == 1000 + assert ( + extras.NativeGraphBatch._BULK_CHUNK == 10_000 + ) # new_edges is not part of the measured change + assert arcadedb.__backend__ == "native"