diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc index c620ea2..5cc3639 100644 --- a/docs/modules/ROOT/nav.adoc +++ b/docs/modules/ROOT/nav.adoc @@ -7,6 +7,8 @@ ** xref:how-to/transactions.adoc[Transactions] ** xref:how-to/values-and-records.adoc[Values and records] ** xref:how-to/concurrency.adoc[Concurrency, processes and crashes] +** xref:how-to/durability.adoc[Durability settings] +** xref:how-to/async-executor.adoc[The async executor] ** xref:how-to/logs.adoc[Where the logs go] ** xref:how-to/bulk-loads.adoc[Bulk loads and commit size] ** xref:how-to/vectors.adoc[Vectors] diff --git a/docs/modules/ROOT/pages/how-to/async-executor.adoc b/docs/modules/ROOT/pages/how-to/async-executor.adoc new file mode 100644 index 0000000..23cf062 --- /dev/null +++ b/docs/modules/ROOT/pages/how-to/async-executor.adoc @@ -0,0 +1,195 @@ += The async executor +:description: What db.async_executor() returns, how to submit SQL and transaction blocks without waiting, how to wait for them, how errors and callbacks reach you, what close() does to every caller, and what is thread-safe, on both builds. + +// Sources: src/arcadedb/async_executor.py (the JVM build), src/arcadedb/_native/extras.py (NativeAsyncExecutor, the native build), core.py (async_executor), and ArcadeDB 26.11.1 DatabaseAsyncExecutorImpl.java. Pinned by tests/test_async_executor.py, tests/test_async_transaction.py and tests/test_async_executor_behavior.py. + +You need this page if you want to hand statements to the engine and carry on, and learn about the result in a callback. The short version: + +* `executor = db.async_executor()` returns a wrapper of the one executor of the database. Every call returns another wrapper of the same executor, with the same settings. +* `executor.command(...)`, `executor.query(...)` and `executor.transaction(...)` queue work and return at once. `executor.wait_completion()` blocks until everything queued before it is stored. +* A callback runs on a thread of the engine, not on yours. +* `executor.close()` shuts the executor down for every caller. Do it once, at the end. +* Do not use it to load many rows. Use `db.insert_many(...)` for documents and `db.graph_batch()` for graphs: see xref:how-to/bulk-loads.adoc[Bulk loads and commit size]. + +The statements are SQL or openCypher, as on xref:how-to/databases-and-queries.adoc[Databases and queries]. The executor, its worker threads and its queue belong to the engine. This page covers what the Python wrappers add. + +The blocks on this page run one after the other against one empty database, and a test in the repository runs them that way (`tests/test_docs_async_page.py`) on the JVM build and on the native build. Each block that prints something is followed by its output. Every block assumes that `db` is an open database. The last block closes the executor, so nothing can follow it. + +== Submit and wait + +Each worker of the executor opens a transaction of its own and commits it after `commit_every` operations, and at `wait_completion()`. You open no transaction for it. At parallel level 1 there is one worker, and every statement of the next block is stored once `wait_completion()` returns. + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Item") + + +def count(): + return db.query("sql", "SELECT count(*) AS c FROM Item").first().get("c") + + +executor = db.async_executor().set_parallel_level(1).set_commit_every(1000) +for n in range(50): + executor.command("sql", "INSERT INTO Item SET n = ?", args=[n]) +executor.wait_completion() +print(count()) +---- + +[source,text] +---- +50 +---- + +The setters return the executor, so they chain. `wait_completion(timeout_ms)` raises `TimeoutError` when the queue is not done in time, and `wait_completion(0)` raises it at once if work is left. A negative timeout raises `ValueError`. Returning from the wait does not make `is_pending()` `False` at the same instant: poll it until it is, if you need an idle executor. + +[source,python] +---- +try: + executor.wait_completion(-1) +except ValueError as error: + print(error) +---- + +[source,text] +---- +timeout_ms must be None or >= 0, got -1 +---- + +`command` takes the parameters as `args=[...]` for `?`, or as named keywords, and not both: the call raises `ValueError` before it queues anything. + +== One executor for the database + +The engine has one executor for each database. `db.async_executor()` does not make a new one: it returns a new wrapper of the same executor, and the settings (parallel level, commit size and flush type, among others) are the executor's, not the wrapper's. + +[source,python] +---- +other = db.async_executor() +print(other is executor, other.get_commit_every(), other.get_parallel_level()) +---- + +[source,text] +---- +False 1000 1 +---- + +So a library that sets the commit size changes it for your code too. Set what you need right after you get the executor, and do not share an executor with code you do not control. The durability settings of the executor are on xref:how-to/durability.adoc[Durability settings]. + +== Callbacks and errors + +A result callback gets one row at a time. It runs on a thread of the engine, so it must be thread-safe, and anything it shares with your thread needs a lock or a queue. + +[source,python] +---- +import threading + +me = threading.get_ident() +rows = [] +on_my_thread = [] + + +def on_row(row): + rows.append(row.get("n")) + on_my_thread.append(threading.get_ident() == me) + + +executor.query("sql", "SELECT n FROM Item WHERE n < 3 ORDER BY n", on_row) +executor.wait_completion() +print(rows, set(on_my_thread)) +---- + +[source,text] +---- +[0, 1, 2] {False} +---- + +A statement that fails does not raise in your thread. It goes to the `error_callback` of that call. A statement without one fails silently: `wait_completion()` returns normally and the statement stored nothing. Statements queued after a failed one still run. + +[source,python] +---- +errors = [] +executor.command("sql", "INSERT INTO Missing SET n = 1", error_callback=errors.append) +executor.command("sql", "INSERT INTO Item SET n = 100") +executor.wait_completion() +print(len(errors), count()) + +executor.command("sql", "INSERT INTO Missing SET n = 1") +executor.wait_completion() +print("no error raised", count()) +---- + +[source,text] +---- +1 51 +no error raised 51 +---- + +Give every call whose result you need an `error_callback`. `executor.on_error(callback)` sets one handler for the whole executor, and `executor.on_ok(callback)` one that runs after each batch commit; see their docstrings for what each receives. A per-record callback of `create_record` runs when the record is applied to the batch, before the commit, and `on_ok` runs after it (`tests/test_async_executor.py` pins this order). + +== Blocks in a transaction + +`executor.transaction(block)` runs `block` on a worker inside a transaction: the engine begins it, commits it, and rolls it back when the block raises. Inside the block, use `db` as you do in your own thread. `retries`, `ok_callback` and `error_callback` are optional, and `slot=n` sends the block to the worker with that number. Blocks sent to the same slot run in the order you submitted them, and the tests pin no other order, so do not depend on the order of `command` calls. + +[source,python] +---- +executor.transaction(lambda: db.command("sql", "INSERT INTO Item SET n = 200")) +executor.wait_completion() +print(count()) +---- + +[source,text] +---- +52 +---- + +== Threads + +Any thread can submit to one executor. Four threads that submit ten statements each, at parallel level 1, give forty rows: + +[source,python] +---- +def submit(worker): + for _ in range(10): + executor.command("sql", "INSERT INTO Item SET worker = ?", args=[worker]) + + +threads = [threading.Thread(target=submit, args=(w,)) for w in range(4)] +for thread in threads: + thread.start() +for thread in threads: + thread.join() +executor.wait_completion() +print(count()) +---- + +[source,text] +---- +92 +---- + +== Close + +`executor.close()` shuts down the worker threads of the executor, and the executor is the one that every wrapper shares. After it, another wrapper still answers `is_closed()` with `False`, but submitting work to it raises an error that says the executor has been shut down. `with db.async_executor() as executor:` closes at the end of the block. `db.close()` closes every executor that the database handed out, and `close()` of a wrapper is safe to call twice. + +The class of the error that a submit raises after a close is not part of the Python API, so the block catches `Exception` and reads the text: + +[source,python] +---- +with db.async_executor() as closing: + closing.command("sql", "INSERT INTO Item SET n = 300") + closing.wait_completion() +print(closing.is_closed(), executor.is_closed()) +try: + executor.transaction(lambda: None) +except Exception as error: + print("shut down" in str(error)) +print(count()) +---- + +[source,text] +---- +True False +True +93 +---- + +Call `wait_completion()` before `close()`, as the block does. diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc index 8acbadf..6433e1d 100644 --- a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc +++ b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc @@ -24,7 +24,7 @@ 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. `graph_batch` refuses to run in one: see xref:how-to/transactions.adoc#graph-batch[GraphBatch inside a transaction]. +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. `graph_batch` refuses to run in one: see xref:how-to/transactions.adoc#graph-batch[GraphBatch inside a transaction]. The durability that a load gets is on xref:how-to/durability.adoc[Durability settings], and the async executor on xref:how-to/async-executor.adoc[The async executor]. == Why the native defaults are smaller diff --git a/docs/modules/ROOT/pages/how-to/concurrency.adoc b/docs/modules/ROOT/pages/how-to/concurrency.adoc index c49fb61..a658637 100644 --- a/docs/modules/ROOT/pages/how-to/concurrency.adoc +++ b/docs/modules/ROOT/pages/how-to/concurrency.adoc @@ -306,6 +306,8 @@ If a process must stop and it does not react to a signal, `SIGKILL` cost no ackn By default the engine does not flush its write-ahead log at commit: `arcadedb.txWalFlush` is `0`. A commit that returned then survives a crash of the process, as the kill probe above shows, but not a power loss or an operating system crash. Setting `arcadedb.txWalFlush` to `1` flushes the log data to stable storage at every commit, and `2` flushes the file metadata as well. What each level guarantees is in the main ArcadeDB docs: see {arcadedb-docs-url}/concepts/transactions#wal-flush-durability[WAL flush and durability]. The setting is described in {arcadedb-docs-url}/reference/settings[Settings]. +The call `db.set_wal_flush(...)`, the group commit and file sync settings, and what a bulk load does with them are on xref:how-to/durability.adoc[Durability settings]. + The setting belongs to the database. The same SQL turns it on for `arcadedb` (native) and `arcadedb-jvm`, and the value is stored with the database, so it is still in force after a reopen: [source,python,subs="attributes+"] diff --git a/docs/modules/ROOT/pages/how-to/durability.adoc b/docs/modules/ROOT/pages/how-to/durability.adoc new file mode 100644 index 0000000..ee87c8e --- /dev/null +++ b/docs/modules/ROOT/pages/how-to/durability.adoc @@ -0,0 +1,210 @@ += Durability settings +:description: What set_wal_flush, arcadedb.txWalFlush, arcadedb.txWalGroupCommit and arcadedb.fileSync do, what the engine documents about a crash at each level, how to set them from Python on both builds, and how the async executor and GraphBatch keep durability settings of their own. + +// Sources: src/arcadedb/core.py (set_wal_flush), graph_batch.py, async_executor.py, and ArcadeDB 26.11.1: GlobalConfiguration.java (TX_WAL_FLUSH, TX_WAL_GROUP_COMMIT, FILE_SYNC), WALFile.java (the flush), TransactionContext.java, LocalDatabase.java, GraphBatch.java, ArcadeDBServer.java. The statements about Python are pinned by tests/test_durability_behavior.py. What a crash loses is the engine's statement: no test here can lose power. + +You need this page if a commit that returned must survive a power loss or an operating system crash, or if you load data in bulk and want to know which durability it gets. The short version: + +* By default a commit that returned survives a crash of the process, but the engine does not flush its write-ahead log (WAL) at commit, so it can lose recent commits in a power loss. +* `db.set_wal_flush("yes_nometadata")` or `db.set_wal_flush("yes_full")` makes every commit flush the WAL. It applies to the open `Database`, to every thread, and is not saved. +* `ALTER DATABASE` with `arcadedb.txWalFlush` stores the same choice with the database, as the level `0`, `1` or `2`. +* `GraphBatch` and the async executor do not follow either. Each has its own setting, which this page lists. +* This page gives no timing. A flush waits for the disk, and what that costs on your storage is yours to measure. No default is recommended here. + +What the levels mean for the engine is in the main manual: {arcadedb-docs-url}/concepts/transactions#wal-flush-durability[WAL flush and durability], and the settings are listed in {arcadedb-docs-url}/reference/settings[Settings]. The crash probes on xref:how-to/concurrency.adoc#durability[Concurrency, processes and crashes] show what a killed process loses, which is nothing that had returned from a commit. + +The blocks on this page run one after the other against one empty database, and a test in the repository runs them that way (`tests/test_docs_durability_page.py`) on the JVM build and on the native build. Each block that prints something is followed by its output. Every block assumes that `db` is an open database. + +== The settings + +[cols="2,1,1,4"] +|=== +| Setting | Scope | Default | What it does + +| `arcadedb.txWalFlush` +| database +| `0` +| `0` does not flush the WAL at commit, `1` flushes the data of the WAL file, `2` flushes the data and the file metadata. + +| `arcadedb.txWalGroupCommit` +| database +| `true` +| With `txWalFlush` at `1` or `2`, one flush covers the commits that wait for it at the same time. + +| `arcadedb.fileSync` +| JVM process +| `true` +| For testing only. `false` skips every flush of files that the engine issues. +|=== + +Both builds start from the engine's defaults, and the bindings change none of them. A database reports the settings that belong to it, with their defaults, in `schema:database`: + +[source,python] +---- +from arcadedb import ArcadeDBError + + +def setting(key): + row = db.query("sql", "SELECT settings FROM schema:database").first() + return next(s for s in row.get("settings") if s["key"] == key) + + +print(setting("arcadedb.txWalFlush")["default"]) +print(str(setting("arcadedb.txWalGroupCommit")["default"]).lower()) +---- + +[source,text] +---- +0 +true +---- + +The value that a database reports can differ from the default. A server that you start in the same process with `config={"mode": "production"}` sets `arcadedb.txWalFlush` to `1` for the databases opened after it, unless the setting was changed before (the engine does this in `ArcadeDBServer.java`, and the docstring of `set_wal_flush` says so). + +== What a crash can lose + +This is what the engine documents. No test on this page checks it, because a test cannot cut the power of the machine. + +[cols="1,4"] +|=== +| Level | What the engine documents + +| `0` (`"no"`) +| No flush at commit. The operating system still writes the log, so a commit survives a crash of the process, but not a power failure or an operating system crash. + +| `1` (`"yes_nometadata"`) +| The commit returns after the data of the WAL file is flushed (`FileChannel.force(false)`, the `fdatasync` call). + +| `2` (`"yes_full"`) +| The commit returns after the data and the metadata of the WAL file are flushed (`force(true)`, the `fsync` call). +|=== + +Two more statements come from the text of the settings in `GlobalConfiguration.java`: + +* With group commit on (the default) and `txWalFlush` at `1` or `2`, a commit still returns only once its WAL is on disk. The transaction becomes visible to other transactions a few milliseconds before that. A crash in that window loses a transaction that other clients may already have read, while the client that committed it has not been answered yet. `arcadedb.txWalGroupCommit` set to `false` makes a transaction visible only after its WAL is on disk. +* `arcadedb.fileSync` set to `false` skips every flush, the flush of the WAL at levels `1` and `2` included. Data then survives a crash of the process but not a power failure. Only the exact text `false` turns it off. The engine marks it for testing only. + +== set_wal_flush + +`db.set_wal_flush(mode)` takes `"no"`, `"yes_nometadata"` or `"yes_full"`, which are the levels `0`, `1` and `2`. Any other value raises `ValueError` and names the three. The names are lower case, and a closed database raises `ArcadeDBError`. + +The call changes the open `Database` object and nothing is written to the database. The next block shows that the stored level stays as it was. Call it again after every open: a reopened database follows its configuration again. + +[source,python] +---- +before = setting("arcadedb.txWalFlush")["value"] +db.set_wal_flush("yes_nometadata") +print("stored level changed:", setting("arcadedb.txWalFlush")["value"] != before) +db.set_wal_flush("no") + +try: + db.set_wal_flush("fsync") +except ValueError as error: + print(error) +---- + +[source,text] +---- +stored level changed: False +Invalid WAL flush mode: fsync. Must be one of: ['no', 'yes_nometadata', 'yes_full'] +---- + +It applies to the transactions of every thread of that `Database`, also of a thread that never called it, and not to another database in the process. The JVM build has tests that read the flush type of a thread's next transaction from the engine: after the call, a second thread and the calling thread both use the chosen type, a second open database keeps its own, and a reopened database goes back to the configured one. The native build calls the same engine method. Python cannot read the effective flush type there, so the native tests check that the call is accepted, refused or stored as described above, and no more. + +== The settings of the engine + +`ALTER DATABASE` stores a database setting in the configuration of the database, so it is still there after a reopen. It is the same statement on both builds: + +[source,python] +---- +db.command("sql", "ALTER DATABASE `arcadedb.txWalFlush` 1") +print(setting("arcadedb.txWalFlush")["value"]) +db.command("sql", "ALTER DATABASE `arcadedb.txWalFlush` 0") +---- + +[source,text] +---- +1 +---- + +A test reopens a database after `ALTER DATABASE` and reads the level again on both builds. On the JVM build, a thread that starts after the statement commits with the stored level. The engine reads the configured level when a thread gets its first transaction context, so this page does not say what the statement does to a thread that has a transaction context already. To change the flush of the running process at once, use `set_wal_flush`. + +Group commit is a database setting too: + +[source,python] +---- +db.command("sql", "ALTER DATABASE `arcadedb.txWalGroupCommit` false") +print(str(setting("arcadedb.txWalGroupCommit")["value"]).lower()) +db.command("sql", "ALTER DATABASE `arcadedb.txWalGroupCommit` true") +---- + +[source,text] +---- +false +---- + +`arcadedb.fileSync` has the scope of the JVM process, so it is not in the list of a database and `ALTER DATABASE` refuses it: + +[source,python] +---- +keys = [s["key"] for s in db.query("sql", "SELECT settings FROM schema:database").first().get("settings")] +print("arcadedb.fileSync" in keys) +try: + db.command("sql", "ALTER DATABASE `arcadedb.fileSync` false") +except ArcadeDBError as error: + print("refused:", "cannot be set per database" in str(error)) +---- + +[source,text] +---- +False +refused: True +---- + +A setting with the scope of the process is a system property. Set it before the engine starts, because the engine reads it when it starts: for `arcadedb-jvm` as `-Darcadedb.fileSync=false` in `jvm_kwargs={"jvm_args": [...]}` or in `ARCADEDB_JVM_ARGS`, and for `arcadedb` in `ARCADEDB_NATIVE_ISOLATE_ARGS`. The same route sets the process default of `arcadedb.txWalFlush` for every database. The pages xref:how-to/logs.adoc[Where the logs go] and xref:how-to/concurrency.adoc#durability[Concurrency, processes and crashes] show both. This page has no test for the system property. + +== Bulk loads keep settings of their own + +The async executor and `GraphBatch` do not read `arcadedb.txWalFlush` and `set_wal_flush` does not change them. A fresh executor writes to the WAL and does not flush it. Its settings are shared by every `db.async_executor()` call of the database (see xref:how-to/async-executor.adoc[The async executor]): + +[source,python] +---- +db.command("sql", "CREATE VERTEX TYPE City") +executor = db.async_executor() +print(executor.get_transaction_sync(), executor.is_transaction_use_wal()) +db.set_wal_flush("yes_full") +print(executor.get_transaction_sync()) +db.set_wal_flush("no") +---- + +[source,text] +---- +no True +no +---- + +A `GraphBatch` builds with `use_wal=False` and `wal_flush="no"`, whatever the database says, and applies them to the shared executor for as long as it is open. It restores the executor's earlier settings when it closes, also when the `with` block raises. The engine logs the change at level INFO. A batch that you give `use_wal` or `wal_flush` uses those instead, and `wal_flush` takes the same three names as `set_wal_flush`: + +[source,python] +---- +executor.set_transaction_sync("yes_nometadata") +with db.graph_batch() as batch: + print(executor.get_transaction_sync(), executor.is_transaction_use_wal()) + batch.create_vertices("City", 2) +print(executor.get_transaction_sync(), executor.is_transaction_use_wal()) + +with db.graph_batch(use_wal=True, wal_flush="yes_full") as batch: + print(executor.get_transaction_sync(), executor.is_transaction_use_wal()) + batch.create_vertices("City", 2) +print(executor.get_transaction_sync(), executor.is_transaction_use_wal()) +---- + +[source,text] +---- +no False +yes_nometadata True +yes_full True +yes_nometadata True +---- + +An unknown `wal_flush` raises `ValueError` when you call `db.graph_batch()`. With the batch open, a transaction that you run yourself between its calls still uses your flush type, and it does after `close()`. The JVM build has a test for this, as it needs the engine's flush type of the thread. A batch cannot run inside a transaction that you opened: see xref:how-to/transactions.adoc#graph-batch[Transactions]. What an interrupted batch leaves behind is on xref:how-to/bulk-loads.adoc[Bulk loads and commit size]. diff --git a/docs/modules/ROOT/pages/how-to/transactions.adoc b/docs/modules/ROOT/pages/how-to/transactions.adoc index e072ad9..ea07591 100644 --- a/docs/modules/ROOT/pages/how-to/transactions.adoc +++ b/docs/modules/ROOT/pages/how-to/transactions.adoc @@ -238,7 +238,7 @@ added [[graph-batch]] == GraphBatch inside a transaction -`GraphBatch` (`db.graph_batch()`) manages its own transactions. From engine 26.10.1 on, its `create_vertices`, `flush` and `close`, and `new_edge` and `new_edges` when the buffer is full, refuse to run inside a transaction that you opened. They raise `ArcadeDBError` and leave your transaction as it was. Engines before 26.10.1 committed your transaction instead. Commit or roll back your own work, then call the batch. The commit sizes of a batch are on xref:how-to/bulk-loads.adoc[Bulk loads and commit size]. +`GraphBatch` (`db.graph_batch()`) manages its own transactions. From engine 26.10.1 on, its `create_vertices`, `flush` and `close`, and `new_edge` and `new_edges` when the buffer is full, refuse to run inside a transaction that you opened. They raise `ArcadeDBError` and leave your transaction as it was. Engines before 26.10.1 committed your transaction instead. Commit or roll back your own work, then call the batch. The commit sizes of a batch are on xref:how-to/bulk-loads.adoc[Bulk loads and commit size]. What a batch does with the durability settings of the database is on xref:how-to/durability.adoc[Durability settings]. [source,python] ---- diff --git a/tests/conftest.py b/tests/conftest.py index 0ae9b36..9a309f2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -177,6 +177,10 @@ "test_legacy_temporal_implementations.py::test_the_engine_hands_back_the_configured_class_holding_the_utc_instant": JAVA_DB, "test_graph_batch.py::test_graph_batch_leaves_the_callers_wal_setting_alone": JAVA_DB, "test_transaction_config.py::test_set_wal_flush_is_database_wide": JAVA_DB, + "test_durability_behavior.py::test_set_wal_flush_reaches_the_commits_of_this_database_only": JAVA_DB, + "test_durability_behavior.py::test_set_wal_flush_ends_with_the_open_database": JAVA_DB, + "test_durability_behavior.py::test_a_thread_that_starts_after_alter_database_commits_with_the_stored_level": JAVA_DB, + "test_durability_behavior.py::test_a_graph_batch_leaves_the_flush_of_your_transactions_alone": JAVA_DB, "test_run_in_transaction_retry.py::test_contended_commits_spend_their_budget_before_failing": JAVA_DB, "test_float_vector_bulk.py::TestAllowedRids::test_rid_objects_and_text_mixed_still_work": "builds a Java RID (db.to_java_rid) and reads the Java RID set of the index", "test_float_vector_bulk.py::TestDecision::test_a_declared_vector_column_is_taken_out": "decides on the Java database (get_java_database) and builds the bridge's FloatVectors.Matrix objects", diff --git a/tests/test_async_executor_behavior.py b/tests/test_async_executor_behavior.py new file mode 100644 index 0000000..0ca13cd --- /dev/null +++ b/tests/test_async_executor_behavior.py @@ -0,0 +1,175 @@ +"""The async executor, as the how-to page ``docs/modules/ROOT/pages/how-to/async-executor.adoc`` says it (ArcadeData/arcadedb-python#267, row 11). + +``test_async_executor.py`` and ``test_async_transaction.py`` already pin most of the calls: exactness at parallel level 1, the +callbacks, ``wait_completion`` and its timeouts, ``close``, the ``transaction`` block and slot order. This file pins what the page +adds: one engine executor behind every ``db.async_executor()`` call, what ``close()`` does to the other callers, that callbacks run +off the calling thread, that a statement failure without an error callback is silent, and submitting from several threads. The +tests use SQL and no Java object, so they run on the JVM build and the native build alike. +""" + +import threading + +import pytest + +from arcadedb import ArcadeDBError + +TIMEOUT_MS = 30_000 + + +@pytest.fixture +def db(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Item") + return temp_db + + +def _count(db): + return int(db.query("sql", "SELECT count(*) AS c FROM Item").first().get("c")) + + +def test_every_call_wraps_the_one_executor_of_the_database(db): + first = db.async_executor() + second = db.async_executor() + assert first is not second + + first.set_parallel_level(3).set_commit_every(77).set_transaction_sync( + "yes_nometadata" + ) + assert second.get_parallel_level() == 3 + assert second.get_commit_every() == 77 + assert second.get_transaction_sync() == "yes_nometadata" + + second.set_commit_every(5) + assert first.get_commit_every() == 5 + + +def test_close_shuts_the_executor_down_for_every_wrapper_of_it(db): + first = db.async_executor() + second = db.async_executor() + first.close() + + assert first.is_closed() is True + assert second.is_closed() is False + with pytest.raises(Exception) as info: + second.transaction(lambda: None) + assert "shut down" in str(info.value) + + +def test_the_with_statement_closes_the_executor_at_the_end(db): + with db.async_executor() as executor: + executor.command("sql", "INSERT INTO Item SET n = 1") + executor.wait_completion(TIMEOUT_MS) + assert executor.is_closed() is False + assert executor.is_closed() is True + assert _count(db) == 1 + + +def test_commands_are_stored_once_wait_completion_returns(db): + executor = db.async_executor().set_parallel_level(1).set_commit_every(1000) + for n in range(50): + executor.command("sql", "INSERT INTO Item SET n = ?", args=[n]) + executor.wait_completion(TIMEOUT_MS) + assert _count(db) == 50 + + +def test_callbacks_run_on_a_thread_of_the_engine_not_the_one_that_submitted(db): + caller = threading.get_ident() + threads = {} + + def on_row(_row): + threads["result"] = threading.get_ident() + + def on_error(_error): + threads["error"] = threading.get_ident() + + executor = db.async_executor().set_parallel_level(1) + executor.command("sql", "INSERT INTO Item SET n = 1", callback=on_row) + executor.command("sql", "INSERT INTO Missing SET n = 1", error_callback=on_error) + executor.wait_completion(TIMEOUT_MS) + + assert threads["result"] != caller + assert threads["error"] != caller + + +def test_a_failed_statement_without_an_error_callback_raises_nowhere(db): + executor = db.async_executor().set_parallel_level(1) + executor.command("sql", "INSERT INTO Missing SET n = 1") + executor.command("sql", "INSERT INTO Item SET n = 2") + executor.wait_completion(TIMEOUT_MS) + assert _count(db) == 1 + + +def test_a_failed_statement_does_not_stop_the_ones_after_it(db): + errors = [] + executor = db.async_executor().set_parallel_level(1) + executor.command( + "sql", "INSERT INTO Missing SET n = 1", error_callback=errors.append + ) + executor.command("sql", "INSERT INTO Item SET n = 2") + executor.wait_completion(TIMEOUT_MS) + assert len(errors) == 1 + assert _count(db) == 1 + + +def test_positional_and_named_parameters_together_raise_before_anything_is_queued(db): + executor = db.async_executor() + with pytest.raises(ValueError, match="positional args or named params"): + executor.command("sql", "INSERT INTO Item SET n = ?", args=[1], n=2) + with pytest.raises(ValueError, match="positional args or named params"): + executor.query("sql", "SELECT FROM Item", lambda row: None, args=[1], n=2) + executor.wait_completion(TIMEOUT_MS) + assert _count(db) == 0 + + +def test_a_query_callback_gets_every_row_of_the_result(db): + for n in range(5): + with db.transaction(): + db.command("sql", "INSERT INTO Item SET n = ?", n) + seen = [] + executor = db.async_executor().set_parallel_level(1) + executor.query( + "sql", "SELECT n FROM Item ORDER BY n", lambda row: seen.append(row.get("n")) + ) + executor.wait_completion(TIMEOUT_MS) + assert seen == [0, 1, 2, 3, 4] + + +def test_several_threads_can_submit_to_one_executor(db): + executor = db.async_executor().set_parallel_level(1).set_commit_every(1000) + errors = [] + + def submit(worker): + try: + for n in range(25): + executor.command( + "sql", "INSERT INTO Item SET worker = ?, n = ?", args=[worker, n] + ) + except Exception as error: # noqa: BLE001 - reported by the assert below + errors.append(error) + + threads = [threading.Thread(target=submit, args=(w,)) for w in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + executor.wait_completion(TIMEOUT_MS) + + assert errors == [] + assert _count(db) == 100 + + +def test_closing_the_database_closes_every_executor_it_handed_out(temp_db_path): + import arcadedb + + db = arcadedb.create_database(temp_db_path) + executors = [db.async_executor(), db.async_executor()] + db.close() + assert [e.is_closed() for e in executors] == [True, True] + + +def test_an_executor_of_a_closed_database_cannot_be_asked_for(temp_db_path): + import arcadedb + + db = arcadedb.create_database(temp_db_path) + db.close() + with pytest.raises(ArcadeDBError, match="closed"): + db.async_executor() diff --git a/tests/test_docs_async_page.py b/tests/test_docs_async_page.py new file mode 100644 index 0000000..053ddc2 --- /dev/null +++ b/tests/test_docs_async_page.py @@ -0,0 +1,60 @@ +"""Every Python block of the Antora page ``docs/modules/ROOT/pages/how-to/async-executor.adoc`` runs, and prints what the page says. + +Same pattern as ``test_docs_vectors_page.py``: the blocks are chained on purpose, they run in the order of the page, in one +namespace, against one fresh temporary database that the namespace holds as ``db``. A block that is followed by a +``[source,text]`` block must print exactly that text. What each statement of the page claims is pinned on its own in +``test_async_executor_behavior.py`` and ``test_async_executor.py``. The test uses no Java object, so it runs on the JVM build and the native build alike. +""" + +import contextlib +import io +import re +from pathlib import Path + +PAGE = ( + Path(__file__).resolve().parents[1] + / "docs" + / "modules" + / "ROOT" + / "pages" + / "how-to" + / "async-executor.adoc" +) + +# a Python listing, then (optionally, directly after it) the text listing with its output +BLOCK_RE = re.compile( + r"\[source,python[^\]\n]*\]\n----\n(?P.*?)\n----\n" + r"(?:\n\[source,text\]\n----\n(?P.*?)\n----\n)?", + re.DOTALL, +) + + +def _blocks(): + text = PAGE.read_text(encoding="utf-8") + found = [(m.group("code"), m.group("out")) for m in BLOCK_RE.finditer(text)] + # no Python listing may escape the extraction (a different spelling of the block header would run nowhere) + assert text.count("[source,python") == len(found) + return found + + +def test_the_page_has_its_blocks(): + blocks = _blocks() + assert len(blocks) >= 7 + assert sum(1 for _, out in blocks if out is not None) >= 7 + + +def test_every_block_runs_in_order_and_prints_what_the_page_says(temp_db): + namespace = {"db": temp_db, "__name__": "async_page"} + for number, (code, expected) in enumerate(_blocks(), start=1): + captured = io.StringIO() + compiled = compile(code, f"async-executor.adoc python block {number}", "exec") + with contextlib.redirect_stdout(captured): + exec(compiled, namespace) # nosec B102 - the page of this repository + if expected is not None: + assert ( + captured.getvalue().strip() == expected.strip() + ), f"block {number} printed other text than the page shows:\n{code}" + else: + assert ( + captured.getvalue() == "" + ), f"block {number} prints, the page shows no output" diff --git a/tests/test_docs_durability_page.py b/tests/test_docs_durability_page.py new file mode 100644 index 0000000..f9b3fbe --- /dev/null +++ b/tests/test_docs_durability_page.py @@ -0,0 +1,60 @@ +"""Every Python block of the Antora page ``docs/modules/ROOT/pages/how-to/durability.adoc`` runs, and prints what the page says. + +Same pattern as ``test_docs_vectors_page.py``: the blocks are chained on purpose, they run in the order of the page, in one +namespace, against one fresh temporary database that the namespace holds as ``db``. A block that is followed by a +``[source,text]`` block must print exactly that text. What each statement of the page claims is pinned on its own in +``test_durability_behavior.py``. The test uses no Java object, so it runs on the JVM build and the native build alike. +""" + +import contextlib +import io +import re +from pathlib import Path + +PAGE = ( + Path(__file__).resolve().parents[1] + / "docs" + / "modules" + / "ROOT" + / "pages" + / "how-to" + / "durability.adoc" +) + +# a Python listing, then (optionally, directly after it) the text listing with its output +BLOCK_RE = re.compile( + r"\[source,python[^\]\n]*\]\n----\n(?P.*?)\n----\n" + r"(?:\n\[source,text\]\n----\n(?P.*?)\n----\n)?", + re.DOTALL, +) + + +def _blocks(): + text = PAGE.read_text(encoding="utf-8") + found = [(m.group("code"), m.group("out")) for m in BLOCK_RE.finditer(text)] + # no Python listing may escape the extraction (a different spelling of the block header would run nowhere) + assert text.count("[source,python") == len(found) + return found + + +def test_the_page_has_its_blocks(): + blocks = _blocks() + assert len(blocks) >= 7 + assert sum(1 for _, out in blocks if out is not None) >= 7 + + +def test_every_block_runs_in_order_and_prints_what_the_page_says(temp_db): + namespace = {"db": temp_db, "__name__": "durability_page"} + for number, (code, expected) in enumerate(_blocks(), start=1): + captured = io.StringIO() + compiled = compile(code, f"durability.adoc python block {number}", "exec") + with contextlib.redirect_stdout(captured): + exec(compiled, namespace) # nosec B102 - the page of this repository + if expected is not None: + assert ( + captured.getvalue().strip() == expected.strip() + ), f"block {number} printed other text than the page shows:\n{code}" + else: + assert ( + captured.getvalue() == "" + ), f"block {number} prints, the page shows no output" diff --git a/tests/test_durability_behavior.py b/tests/test_durability_behavior.py new file mode 100644 index 0000000..b2018e8 --- /dev/null +++ b/tests/test_durability_behavior.py @@ -0,0 +1,271 @@ +"""The durability settings, as the how-to page ``docs/modules/ROOT/pages/how-to/durability.adoc`` says them (ArcadeData/arcadedb-python#267, row 10). + +Each statement of that page is pinned here first: what ``set_wal_flush`` accepts and where it applies, what the database reports for +``arcadedb.txWalFlush``, ``arcadedb.txWalGroupCommit`` and ``arcadedb.fileSync``, how ``ALTER DATABASE`` treats them, and how the +async executor and ``GraphBatch`` keep durability settings of their own and put them back. The tests read the settings through +``SELECT settings FROM schema:database`` and the async executor's getters, which both builds have, so they run on the JVM build and +the native build alike. The few tests that read the engine's effective flush type of a thread need the Java ``Database`` and are +listed in ``conftest.NATIVE_NA_JVM_TESTS``. + +What a crash loses at each level cannot be tested here, because a test process cannot lose power. The page states it only as what +the engine documents, with the source of each statement (``GlobalConfiguration.java`` and ``WALFile.java`` of engine 26.11.1). +""" + +import threading + +import pytest + +from arcadedb import ArcadeDBError, create_database, open_database + +FLUSH = "arcadedb.txWalFlush" +GROUP = "arcadedb.txWalGroupCommit" +FILE_SYNC = "arcadedb.fileSync" + + +def _settings(db): + row = db.query("sql", "SELECT settings FROM schema:database").first() + return {s["key"]: s for s in row.get("settings")} + + +def _alter(db, key, value): + db.command("sql", f"ALTER DATABASE `{key}` {value}") + + +def _text(value): + return str(value).lower() + + +# --- set_wal_flush + + +@pytest.mark.parametrize("mode", ["no", "yes_nometadata", "yes_full"]) +def test_set_wal_flush_takes_these_three_modes(temp_db, mode): + temp_db.set_wal_flush(mode) + temp_db.set_wal_flush("no") + + +@pytest.mark.parametrize("mode", ["fsync", "YES_FULL", "", "1", 1]) +def test_set_wal_flush_refuses_any_other_mode_with_the_list(temp_db, mode): + with pytest.raises(ValueError) as info: + temp_db.set_wal_flush(mode) + assert ( + f"Invalid WAL flush mode: {mode}. " + "Must be one of: ['no', 'yes_nometadata', 'yes_full']" + ) == str(info.value) + + +def test_set_wal_flush_on_a_closed_database_raises_arcadedb_error(temp_db_path): + db = create_database(temp_db_path) + db.close() + with pytest.raises(ArcadeDBError, match="closed"): + db.set_wal_flush("no") + + +def test_set_wal_flush_is_not_a_stored_setting(temp_db): + """It changes the open Database object (LocalDatabase.walFlushSetting, a field), not the saved configuration.""" + before = _settings(temp_db)[FLUSH]["value"] + for mode in ("yes_nometadata", "yes_full", "no"): + temp_db.set_wal_flush(mode) + assert _settings(temp_db)[FLUSH]["value"] == before + + +def test_set_wal_flush_does_not_change_the_executors_own_flush(temp_db): + """The async writers stamp their own flush on their transactions (engine default "no").""" + executor = temp_db.async_executor() + assert executor.get_transaction_sync() == "no" + temp_db.set_wal_flush("yes_full") + try: + assert executor.get_transaction_sync() == "no" + finally: + temp_db.set_wal_flush("no") + + +def test_set_wal_flush_reaches_the_commits_of_this_database_only(temp_db, temp_db_path): + """JVM only: the flush type a thread's next transaction will use, read from the Java database. + + The process default is not always "NO" (a production-mode server anywhere in the process sets it to 1), so the second database + is compared with itself before and after, not with a constant. + """ + other_path = temp_db_path + "_other" + with create_database(other_path) as other: + + def flush_of(db): + with db.transaction(): + return str(db._java_db.getTransaction().getWALFlush()) + + other_before = flush_of(other) + chosen = "YES_FULL" if flush_of(temp_db) != "YES_FULL" else "NO" + temp_db.set_wal_flush({"YES_FULL": "yes_full", "NO": "no"}[chosen]) + try: + assert flush_of(temp_db) == chosen + assert flush_of(other) == other_before + finally: + temp_db.set_wal_flush("no") + + +def test_set_wal_flush_ends_with_the_open_database(temp_db_path): + """JVM only: a reopened database follows the configuration again, not the earlier set_wal_flush.""" + + def flush_of(db): + with db.transaction(): + return str(db._java_db.getTransaction().getWALFlush()) + + with create_database(temp_db_path) as db: + configured = flush_of(db) + chosen = "YES_FULL" if configured != "YES_FULL" else "NO" + db.set_wal_flush({"YES_FULL": "yes_full", "NO": "no"}[chosen]) + assert flush_of(db) == chosen + + with open_database(temp_db_path) as db: + assert flush_of(db) == configured + + +# --- the engine settings, read and written through SQL + + +def test_the_database_reports_the_defaults_of_the_settings_it_has(temp_db): + settings = _settings(temp_db) + assert str(settings[FLUSH]["default"]) == "0" + assert _text(settings[GROUP]["default"]) == "true" + + +def test_file_sync_is_not_a_database_setting_and_alter_database_refuses_it(temp_db): + """arcadedb.fileSync has JVM scope: it is not in the database's list and ALTER DATABASE does not take it.""" + assert FILE_SYNC not in _settings(temp_db) + assert FLUSH in _settings(temp_db) and GROUP in _settings(temp_db) + with pytest.raises(ArcadeDBError) as info: + _alter(temp_db, FILE_SYNC, "false") + assert "cannot be set per database" in str(info.value) + assert "JVM scope" in str(info.value) + + +def test_alter_database_stores_the_flush_level_and_it_survives_a_reopen(temp_db_path): + with create_database(temp_db_path) as db: + _alter(db, FLUSH, 2) + assert str(_settings(db)[FLUSH]["value"]) == "2" + _alter(db, FLUSH, 1) + assert str(_settings(db)[FLUSH]["value"]) == "1" + db.command("sql", "CREATE DOCUMENT TYPE Note") + with db.transaction(): + db.command("sql", "INSERT INTO Note SET text = 'kept'") + + with open_database(temp_db_path) as db: + assert str(_settings(db)[FLUSH]["value"]) == "1" + assert db.query("sql", "SELECT text FROM Note").first().get("text") == "kept" + + +def test_alter_database_stores_group_commit(temp_db): + _alter(temp_db, GROUP, "false") + assert _text(_settings(temp_db)[GROUP]["value"]) == "false" + _alter(temp_db, GROUP, "true") + assert _text(_settings(temp_db)[GROUP]["value"]) == "true" + + +def test_a_thread_that_starts_after_alter_database_commits_with_the_stored_level( + temp_db_path, +): + """JVM only: ALTER DATABASE writes the stored level, and a thread that has no transaction context yet takes it from there. + + This does not say what happens to the thread that ran the ALTER DATABASE itself: its context exists already and the engine + reads the configured level when a context is created (TransactionContext.configuredWALFlush). The page does not claim it. + """ + with create_database(temp_db_path) as db: + _alter(db, FLUSH, 2) + + seen = {} + + def read(): + with db.transaction(): + seen["flush"] = str(db._java_db.getTransaction().getWALFlush()) + + thread = threading.Thread(target=read) + thread.start() + thread.join() + assert seen["flush"] == "YES_FULL" + + +# --- the async executor and GraphBatch keep settings of their own + + +def test_a_fresh_executor_writes_to_the_wal_and_does_not_flush(temp_db): + executor = temp_db.async_executor() + assert executor.is_transaction_use_wal() is True + assert executor.get_transaction_sync() == "no" + + +def test_alter_database_does_not_change_the_executors_flush(temp_db): + executor = temp_db.async_executor() + _alter(temp_db, FLUSH, 2) + assert executor.get_transaction_sync() == "no" + + +def test_a_graph_batch_relaxes_the_executors_durability_and_puts_it_back(temp_db): + """A batch builds with use_wal=False and wal_flush="no"; parallel_flush (the default) applies them to the shared async executor.""" + temp_db.command("sql", "CREATE VERTEX TYPE City") + executor = temp_db.async_executor() + executor.set_transaction_use_wal(True) + executor.set_transaction_sync("yes_full") + + with temp_db.graph_batch() as batch: + assert executor.is_transaction_use_wal() is False + assert executor.get_transaction_sync() == "no" + batch.create_vertices("City", 2) + + assert executor.is_transaction_use_wal() is True + assert executor.get_transaction_sync() == "yes_full" + + +def test_a_graph_batch_takes_the_durability_you_ask_for_and_puts_it_back(temp_db): + temp_db.command("sql", "CREATE VERTEX TYPE City") + executor = temp_db.async_executor() + executor.set_transaction_use_wal(True) + executor.set_transaction_sync("no") + + with temp_db.graph_batch(use_wal=True, wal_flush="yes_full") as batch: + assert executor.is_transaction_use_wal() is True + assert executor.get_transaction_sync() == "yes_full" + batch.create_vertices("City", 2) + + assert executor.is_transaction_use_wal() is True + assert executor.get_transaction_sync() == "no" + + +def test_a_graph_batch_puts_the_settings_back_when_the_block_raises(temp_db): + executor = temp_db.async_executor() + executor.set_transaction_use_wal(True) + executor.set_transaction_sync("yes_nometadata") + + with pytest.raises(RuntimeError, match="stop"): + with temp_db.graph_batch(): + assert executor.get_transaction_sync() == "no" + raise RuntimeError("stop") + + assert executor.is_transaction_use_wal() is True + assert executor.get_transaction_sync() == "yes_nometadata" + + +def test_a_graph_batch_refuses_a_wal_flush_mode_it_does_not_know(temp_db): + with pytest.raises(ValueError, match="Invalid wal_flush mode"): + temp_db.graph_batch(wal_flush="fsync") + + +def test_a_graph_batch_leaves_the_flush_of_your_transactions_alone(temp_db): + """JVM only: with the batch open, a transaction of yours commits with the flush you set, and after close() too.""" + temp_db.command("sql", "CREATE VERTEX TYPE City") + + def flush_of_a_transaction_of_mine(): + temp_db.begin() + try: + return str(temp_db._java_db.getTransaction().getWALFlush()) + finally: + temp_db.rollback() + + temp_db.set_wal_flush("yes_full") + try: + assert flush_of_a_transaction_of_mine() == "YES_FULL" + with temp_db.graph_batch(parallel_flush=False) as batch: + batch.create_vertices("City", 2) + assert flush_of_a_transaction_of_mine() == "YES_FULL" + assert flush_of_a_transaction_of_mine() == "YES_FULL" + finally: + temp_db.set_wal_flush("no")