diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 8fb321f..a8a02a4 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -1,5 +1,6 @@ """NativeDatabase, NativeDatabaseFactory, NativeResultSet and NativeResult: the DSL-first core over libarcadedb.""" +import gc import itertools import json import os @@ -761,6 +762,19 @@ 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). A lazy input, or a collector the caller already disabled, + # runs the body as it is. + gc.disable() + try: + 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/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..3021462 --- /dev/null +++ b/tests/test_native_insert_many_gc_pause.py @@ -0,0 +1,300 @@ +"""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_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 func.__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() + + +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()