Native build: commit every 1000 rows by default in insert_many, graph_batch and create_vertices

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]

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
+    ...
+----

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()

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):

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"`.

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

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)

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
+    )

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