From cedf3a83b71e4cd9367356a0e7fe51a23c31181f Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Wed, 4 Nov 2026 06:47:58 +0900 Subject: [PATCH 1/3] Native insert_many: pause the cyclic garbage collector for a list or tuple input (draft, ships only if the A/B passes) A decorator on NativeDatabase.insert_many, the native counterpart of the JVM change in #236 (#235): for a list or a tuple the collector is paused for the call and restored in finally (re-enabled only if it was enabled on entry); a generator or any other lazy input is not paused. The decorator leaves the body, the docstring and the signature of insert_many untouched so the change does not overlap the lines that #218 and #232 edit. Tests mirror the JVM ones. Draft: the native build's untimed probes found no collection inside the call, so this ships only if the timed A/B of DECISIONS #240 passes (Closes #237 when it does). Co-Authored-By: Claude Opus 5.5 --- src/arcadedb/_native/database.py | 29 +++ tests/conftest.py | 2 + tests/test_native_insert_many_gc_pause.py | 277 ++++++++++++++++++++++ 3 files changed, 308 insertions(+) create mode 100644 tests/test_native_insert_many_gc_pause.py diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 8fb321f..b77f732 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -1,5 +1,7 @@ """NativeDatabase, NativeDatabaseFactory, NativeResultSet and NativeResult: the DSL-first core over libarcadedb.""" +import functools +import gc import itertools import json import os @@ -65,6 +67,32 @@ def _call(prefix: str, name: str, *args) -> dict: return reply +def _pause_gc_for_sized_rows(insert_many): + """Wrap ``NativeDatabase.insert_many`` so that Python's cyclic garbage collector is paused for the call when ``rows`` is a list or a tuple. + + The previous state is put back in ``finally``: the collector comes back on only when it was on at entry, also after an exception. A + lazy input (a generator or any other iterator) is not paused, because user code that runs between the chunks can create reference + cycles that only the collector frees. ``gc.disable()`` is process wide, so other threads do not collect automatically while a call + with a sized input runs, and with overlapping calls the pause does not last for the whole of each call (it is never left off). + """ + + @functools.wraps(insert_many) + def wrapper(self, *args, **kwargs): + rows = args[1] if len(args) > 1 else kwargs.get("rows") + if not isinstance(rows, (list, tuple)): + return insert_many(self, *args, **kwargs) + self._check_not_closed() + gc_was_enabled = gc.isenabled() + gc.disable() + try: + return insert_many(self, *args, **kwargs) + finally: + if gc_was_enabled: + gc.enable() + + return wrapper + + class NativeResult: """One row. Values are already Python types (datetime, Decimal, lists, dicts, RID text).""" @@ -744,6 +772,7 @@ def lookup_by_key(self, type_name: str, keys: List[str], values: List[Any]): row = NativeResultSet(self, reply).first() return row.get_element() if row is not None else None + @_pause_gc_for_sized_rows def insert_many( self, type_name: str, diff --git a/tests/conftest.py b/tests/conftest.py index 01ab332..c6b1d16 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -63,6 +63,8 @@ collect_ignore.append("test_native_tx_crossings.py") # counts the statement texts the native backend hands to the engine (the JPype backend has no such layer) collect_ignore.append("test_native_statement_texts.py") + # the cyclic collector pause of the native insert_many, observed on the native library (the JVM build's is test_insert_many_gc_pause.py) + collect_ignore.append("test_native_insert_many_gc_pause.py") if not os.path.isfile( os.path.join(os.path.dirname(os.path.dirname(__file__)), "docs", "index.md") ): diff --git a/tests/test_native_insert_many_gc_pause.py b/tests/test_native_insert_many_gc_pause.py new file mode 100644 index 0000000..7201317 --- /dev/null +++ b/tests/test_native_insert_many_gc_pause.py @@ -0,0 +1,277 @@ +"""insert_many pauses CPython's cyclic garbage collector for the call when the input is a list or a tuple (native build). + +A lazy input (a generator or any other iterator) is not paused: user code that runs between the chunks can create reference cycles that +only the collector frees. The previous state is restored in ``finally``: the collector comes back on only when it was on at entry, also +after an exception. Same contract as the JVM build (``test_insert_many_gc_pause.py``, ArcadeData/arcadedb-python#235). + +The state is observed from inside the call: ``NativeDatabase._insert_chunk`` (and ``_insert_many_parallel``) are wrapped to record +``gc.isenabled()`` and then run unchanged, so the rows are really stored and counted. +""" + +import gc +import inspect + +import pytest + +from arcadedb._native.database import NativeDatabase +from arcadedb.exceptions import ArcadeDBError + +ROWS = 25 + + +def _count(db, type_name): + q = f"SELECT count(*) AS n FROM {type_name}" # nosec B608 - test-controlled + return int(db.query("sql", q).to_list()[0]["n"]) + + +@pytest.fixture +def collector_state(): + """Remember the collector's state and put it back, so a failing test cannot leave it off for the rest of the run.""" + was = gc.isenabled() + yield + if was: + gc.enable() + else: + gc.disable() + + +@pytest.fixture +def seen(monkeypatch): + """gc.isenabled() as seen inside each chunk insert and inside the parallel writer path of insert_many.""" + states = [] + real_chunk = NativeDatabase._insert_chunk + real_parallel = NativeDatabase._insert_many_parallel + + def chunk(self, *args, **kwargs): + states.append(gc.isenabled()) + return real_chunk(self, *args, **kwargs) + + def parallel(self, *args, **kwargs): + states.append(gc.isenabled()) + return real_parallel(self, *args, **kwargs) + + monkeypatch.setattr(NativeDatabase, "_insert_chunk", chunk) + monkeypatch.setattr(NativeDatabase, "_insert_many_parallel", parallel) + return states + + +@pytest.fixture +def doc_type(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE GcDoc") + return "GcDoc" + + +def _rows(): + return [{"id": i, "name": f"row_{i}"} for i in range(ROWS)] + + +@pytest.mark.parametrize("make", [list, tuple], ids=["list", "tuple"]) +def test_the_collector_is_paused_during_a_call_with_a_sized_input( + temp_db, doc_type, seen, collector_state, make +): + gc.enable() + + written = temp_db.insert_many(doc_type, make(_rows())) + + assert written == ROWS + assert _count(temp_db, doc_type) == ROWS + assert seen == [False] + + +@pytest.mark.parametrize("make", [list, tuple], ids=["list", "tuple"]) +def test_the_collector_is_enabled_again_after_the_call( + temp_db, doc_type, seen, collector_state, make +): + gc.enable() + + temp_db.insert_many(doc_type, make(_rows())) + + assert gc.isenabled() + + +def test_a_call_that_uses_several_chunks_is_paused_for_all_of_them( + temp_db, doc_type, seen, collector_state +): + gc.enable() + rows = [{"id": i} for i in range(25)] + + # commit_every=10 splits the 25 rows into chunks of 10, 10 and 5 + assert temp_db.insert_many(doc_type, rows, commit_every=10) == 25 + + assert seen == [False, False, False] + assert gc.isenabled() + + +def test_a_parallel_call_with_a_list_is_paused_too( + temp_db, doc_type, seen, collector_state +): + gc.enable() + + written = temp_db.insert_many(doc_type, _rows(), parallel=True) + + assert written == ROWS + assert _count(temp_db, doc_type) == ROWS + assert seen == [False] + assert gc.isenabled() + + +def test_the_collector_is_enabled_again_after_an_exception( + temp_db, doc_type, monkeypatch, collector_state +): + gc.enable() + states = [] + + def failing_chunk(self, *args, **kwargs): + states.append(gc.isenabled()) + raise RuntimeError("chunk failed") + + monkeypatch.setattr(NativeDatabase, "_insert_chunk", failing_chunk) + + with pytest.raises(RuntimeError, match="chunk failed"): + temp_db.insert_many(doc_type, _rows()) + + # the pause was in force when the failure happened, and the exception did not leave it on + assert states == [False] + assert gc.isenabled() + assert not temp_db.is_transaction_active() + + +def test_a_collector_the_caller_disabled_stays_disabled( + temp_db, doc_type, seen, collector_state +): + gc.disable() + + written = temp_db.insert_many(doc_type, _rows()) + + assert written == ROWS + assert seen == [False] + assert not gc.isenabled() + + +def test_a_collector_the_caller_disabled_stays_disabled_after_an_exception( + temp_db, doc_type, monkeypatch, collector_state +): + gc.disable() + + def failing_chunk(self, *args, **kwargs): + raise RuntimeError("chunk failed") + + monkeypatch.setattr(NativeDatabase, "_insert_chunk", failing_chunk) + + with pytest.raises(RuntimeError, match="chunk failed"): + temp_db.insert_many(doc_type, _rows()) + + assert not gc.isenabled() + + +def test_a_generator_is_not_paused(temp_db, doc_type, seen, collector_state): + gc.enable() + in_generator = [] + + def rows(): + for row in _rows(): + in_generator.append(gc.isenabled()) + yield row + + written = temp_db.insert_many(doc_type, rows()) + + assert written == ROWS + assert _count(temp_db, doc_type) == ROWS + assert in_generator == [True] * ROWS + assert seen == [True] + assert gc.isenabled() + + +@pytest.mark.parametrize( + "make", [iter, lambda rows: (r for r in rows)], ids=["iterator", "genexp"] +) +def test_other_lazy_inputs_are_not_paused( + temp_db, doc_type, seen, collector_state, make +): + gc.enable() + + written = temp_db.insert_many(doc_type, make(_rows())) + + assert written == ROWS + assert seen == [True] + assert gc.isenabled() + + +def test_a_parallel_generator_is_not_paused(temp_db, doc_type, seen, collector_state): + gc.enable() + + written = temp_db.insert_many(doc_type, iter(_rows()), parallel=True) + + assert written == ROWS + assert seen == [True] + assert gc.isenabled() + + +def test_an_empty_list_leaves_the_collector_as_it_was( + temp_db, doc_type, collector_state +): + gc.enable() + assert temp_db.insert_many(doc_type, []) == 0 + assert gc.isenabled() + gc.disable() + assert temp_db.insert_many(doc_type, []) == 0 + assert not gc.isenabled() + + +def test_a_closed_database_raises_before_the_collector_is_touched( + temp_db, collector_state +): + gc.enable() + temp_db.close() + + with pytest.raises(ArcadeDBError, match="closed"): + temp_db.insert_many("GcDoc", _rows()) + + assert gc.isenabled() + + +def test_rows_given_by_keyword_are_paused_too(temp_db, doc_type, seen, collector_state): + gc.enable() + + written = temp_db.insert_many(type_name=doc_type, rows=_rows()) + + assert written == ROWS + assert seen == [False] + assert gc.isenabled() + + +def test_a_generator_given_by_keyword_is_not_paused( + temp_db, doc_type, seen, collector_state +): + gc.enable() + + written = temp_db.insert_many(type_name=doc_type, rows=iter(_rows())) + + assert written == ROWS + assert seen == [True] + assert gc.isenabled() + + +def test_the_wrapper_keeps_the_signature_and_the_docstring(temp_db): + signature = inspect.signature(temp_db.insert_many) + + assert list(signature.parameters) == [ + "type_name", + "rows", + "commit_every", + "parallel", + ] + assert signature.parameters["commit_every"].default == 1000 + assert temp_db.insert_many.__name__ == "insert_many" + assert temp_db.insert_many.__doc__.startswith("Bulk insert.") + + +def test_a_call_without_rows_raises_and_leaves_the_collector_as_it_was( + temp_db, doc_type, collector_state +): + gc.enable() + + with pytest.raises(TypeError): + temp_db.insert_many(doc_type) + + assert gc.isenabled() From 3b2eaaa4313d867301d61fbe52a0cb0e5d346bab Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Wed, 4 Nov 2026 07:44:00 +0900 Subject: [PATCH 2/3] Native insert_many: do the pause in the body, no wrapper (insert_many keeps its own __defaults__) The decorator replaced insert_many with a (*args, **kwargs) function, so tests/test_native_columnar_insert.py, which patches insert_many.__defaults__, failed in setup (None is not subscriptable) and patching the wrapper would not have reached the real function. The pause is now six lines at the top of the body: for a list or tuple with the collector on, disable it, re-enter insert_many once (the collector is off, so the body runs) and enable again in finally; a lazy input or a collector the caller already disabled goes straight to the body. Signature, defaults and docstring are unchanged; a test asserts the defaults and that nothing wraps it. Co-Authored-By: Claude Opus 5.5 --- src/arcadedb/_native/database.py | 37 ++++++----------------- tests/test_native_insert_many_gc_pause.py | 16 +++++----- 2 files changed, 18 insertions(+), 35 deletions(-) diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index b77f732..d63d2c2 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -1,6 +1,5 @@ """NativeDatabase, NativeDatabaseFactory, NativeResultSet and NativeResult: the DSL-first core over libarcadedb.""" -import functools import gc import itertools import json @@ -67,32 +66,6 @@ def _call(prefix: str, name: str, *args) -> dict: return reply -def _pause_gc_for_sized_rows(insert_many): - """Wrap ``NativeDatabase.insert_many`` so that Python's cyclic garbage collector is paused for the call when ``rows`` is a list or a tuple. - - The previous state is put back in ``finally``: the collector comes back on only when it was on at entry, also after an exception. A - lazy input (a generator or any other iterator) is not paused, because user code that runs between the chunks can create reference - cycles that only the collector frees. ``gc.disable()`` is process wide, so other threads do not collect automatically while a call - with a sized input runs, and with overlapping calls the pause does not last for the whole of each call (it is never left off). - """ - - @functools.wraps(insert_many) - def wrapper(self, *args, **kwargs): - rows = args[1] if len(args) > 1 else kwargs.get("rows") - if not isinstance(rows, (list, tuple)): - return insert_many(self, *args, **kwargs) - self._check_not_closed() - gc_was_enabled = gc.isenabled() - gc.disable() - try: - return insert_many(self, *args, **kwargs) - finally: - if gc_was_enabled: - gc.enable() - - return wrapper - - class NativeResult: """One row. Values are already Python types (datetime, Decimal, lists, dicts, RID text).""" @@ -772,7 +745,6 @@ def lookup_by_key(self, type_name: str, keys: List[str], values: List[Any]): row = NativeResultSet(self, reply).first() return row.get_element() if row is not None else None - @_pause_gc_for_sized_rows def insert_many( self, type_name: str, @@ -790,6 +762,15 @@ def insert_many( ``commit_every`` keeps fewer rows of a failed load (only the transaction that was open is rolled back). """ self._check_not_closed() + if isinstance(rows, (list, tuple)) and gc.isenabled(): + # A sized input: no user code runs between the chunks, so Python's cyclic collector is paused for the call and put back in + # finally (enabled again only because this call disabled it). The call re-enters once with the collector off and runs the body + # below; a lazy input, or a collector the caller already disabled, goes straight to the body. + gc.disable() + try: + return self.insert_many(type_name, rows, commit_every, parallel) + finally: + gc.enable() if parallel: return self._insert_many_parallel(type_name, rows) sql = f"INSERT INTO `{type_name}` CONTENT :row" diff --git a/tests/test_native_insert_many_gc_pause.py b/tests/test_native_insert_many_gc_pause.py index 7201317..a3ae7c2 100644 --- a/tests/test_native_insert_many_gc_pause.py +++ b/tests/test_native_insert_many_gc_pause.py @@ -252,18 +252,20 @@ def test_a_generator_given_by_keyword_is_not_paused( assert gc.isenabled() -def test_the_wrapper_keeps_the_signature_and_the_docstring(temp_db): - signature = inspect.signature(temp_db.insert_many) - - assert list(signature.parameters) == [ +def test_insert_many_is_not_wrapped_and_keeps_its_defaults(): + """The pause is in insert_many's own body: its defaults are its own (a test fixture patches ``__defaults__``), and nothing wraps it.""" + func = NativeDatabase.insert_many + + assert func.__defaults__ == (1000, False) + assert not hasattr(func, "__wrapped__") + assert list(inspect.signature(func).parameters) == [ + "self", "type_name", "rows", "commit_every", "parallel", ] - assert signature.parameters["commit_every"].default == 1000 - assert temp_db.insert_many.__name__ == "insert_many" - assert temp_db.insert_many.__doc__.startswith("Bulk insert.") + assert func.__doc__.startswith("Bulk insert.") def test_a_call_without_rows_raises_and_leaves_the_collector_as_it_was( From 3bc5f63090c4fef6df7dac9e97132d26506692b6 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Wed, 4 Nov 2026 07:48:25 +0900 Subject: [PATCH 3/3] Native insert_many: the body moves to _insert_many_nogc, insert_many calls it (as the JVM change in #236), no re-entry through the public method Re-entering self.insert_many made an instance-level spy or a subclass override run twice per call. insert_many keeps its signature, __defaults__ and docstring, checks the database is open, pauses the collector for a list or tuple (restoring in finally), and passes commit_every explicitly to the body, so the columnar fixture's patched default still reaches it. A test asserts that a spy on insert_many sees one call for a list. Co-Authored-By: Claude Opus 5.5 --- src/arcadedb/_native/database.py | 10 +++++++--- tests/test_native_insert_many_gc_pause.py | 21 +++++++++++++++++++++ 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index d63d2c2..a8a02a4 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -764,13 +764,17 @@ def insert_many( self._check_not_closed() if isinstance(rows, (list, tuple)) and gc.isenabled(): # A sized input: no user code runs between the chunks, so Python's cyclic collector is paused for the call and put back in - # finally (enabled again only because this call disabled it). The call re-enters once with the collector off and runs the body - # below; a lazy input, or a collector the caller already disabled, goes straight to the body. + # finally (enabled again only because this call disabled it). A lazy input, or a collector the caller already disabled, + # runs the body as it is. gc.disable() try: - return self.insert_many(type_name, rows, commit_every, parallel) + return self._insert_many_nogc(type_name, rows, commit_every, parallel) finally: gc.enable() + return self._insert_many_nogc(type_name, rows, commit_every, parallel) + + def _insert_many_nogc(self, type_name, rows, commit_every, parallel) -> int: + """The body of ``insert_many``, without the pause of the cyclic collector.""" if parallel: return self._insert_many_parallel(type_name, rows) sql = f"INSERT INTO `{type_name}` CONTENT :row" diff --git a/tests/test_native_insert_many_gc_pause.py b/tests/test_native_insert_many_gc_pause.py index a3ae7c2..3021462 100644 --- a/tests/test_native_insert_many_gc_pause.py +++ b/tests/test_native_insert_many_gc_pause.py @@ -277,3 +277,24 @@ def test_a_call_without_rows_raises_and_leaves_the_collector_as_it_was( temp_db.insert_many(doc_type) assert gc.isenabled() + + +def test_an_instance_level_spy_on_insert_many_sees_one_call_for_a_list( + temp_db, doc_type, monkeypatch, collector_state +): + """The pause does not call the public ``insert_many`` again: a spy or a subclass override of it runs once per call.""" + gc.enable() + real = temp_db.insert_many + calls = [] + + def spy(*args, **kwargs): + calls.append(args) + return real(*args, **kwargs) + + monkeypatch.setattr(temp_db, "insert_many", spy) + + written = temp_db.insert_many(doc_type, _rows()) + + assert written == ROWS + assert len(calls) == 1 + assert gc.isenabled()