From 75a8b6922adac5456b3d140cb3abca50eee02428 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 7 Nov 2026 09:54:26 +0900 Subject: [PATCH 1/2] Test the COLUMNAR index on both builds and document its settings (refs #292) Why no change to reachability-metadata.json: I read every class of the engine's index/columnar and query/facet packages (26.12.1 snapshot, ArcadeData/arcadedb main 4a91847) for reflection, ServiceLoader, resources and Class.forName. There is none: the projection is plain JDK (FileChannel, VarHandle byte-array views, java.util.concurrent pools and a daemon ticker thread), the index handler is registered by instance in LocalSchema, facet.searchValues by instance in DefaultSQLFunctionFactory, and the only parser nodes the COLUMNAR and FACETS statements add are FacetsClause, FacetItem and FacetRangeItem, which PR #299 already listed. Upstream's own reachability-metadata.json lists no columnar class either. The entries upstream has and ours lacks (SQLFunctionEnsureLabelTypes, antlr StatementExpression and SubqueryExpression, three Ratis gRPC classes) are not columnar and are left for a change of their own. Why a child process for the unclean close: DatabaseInternal.kill(), which the engine's ColumnarCrashRecoveryTest uses, is not reachable from the native package, and a JVM-only hook would not test both. A writer that ends with os._exit after its commits leaves database.lck behind and the commits in the write-ahead log, which is the state the engine's recovery reads (a crash of the process, not a power loss). Two variants: commits unflushed, and commits flushed to published segments before the rest is written. Why no new setter for arcadedb.columnar.*: ALTER DATABASE already takes the DATABASE-scope keys (immediately, stored with the database) and refuses the JVM and SERVER ones; those are system properties through ARCADEDB_JVM_ARGS or ARCADEDB_NATIVE_ISOLATE_ARGS, as for arcadedb.fileSync. The test pins that, and that GROUP BY returns the same rows with the acceleration on and off (PROFILE names the path). tests/test_columnar_index.py is version-gated like the smoke test (an older engine must refuse the index and the keys with its own error), runs on both packages with the public API, and polls states with deadlines. Nothing was run locally (the laptop guard blocks workloads); CI is the first run. Co-Authored-By: Claude Sonnet 5.5 --- docs/modules/ROOT/nav.adoc | 1 + .../ROOT/pages/how-to/columnar-index.adoc | 66 ++ tests/test_columnar_index.py | 622 ++++++++++++++++++ 3 files changed, 689 insertions(+) create mode 100644 docs/modules/ROOT/pages/how-to/columnar-index.adoc create mode 100644 tests/test_columnar_index.py diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc index 4e223ed..48d1813 100644 --- a/docs/modules/ROOT/nav.adoc +++ b/docs/modules/ROOT/nav.adoc @@ -15,6 +15,7 @@ ** xref:how-to/dataframes-and-columns.adoc[DataFrames, Arrow and columns] ** xref:how-to/functions-and-events.adoc[Python functions and record events] ** xref:how-to/vectors.adoc[Vectors] +** xref:how-to/columnar-index.adoc[The COLUMNAR index and its settings] * Development ** xref:development/ci.adoc[Continuous integration] ** xref:development/release.adoc[Release] diff --git a/docs/modules/ROOT/pages/how-to/columnar-index.adoc b/docs/modules/ROOT/pages/how-to/columnar-index.adoc new file mode 100644 index 0000000..68c77b6 --- /dev/null +++ b/docs/modules/ROOT/pages/how-to/columnar-index.adoc @@ -0,0 +1,66 @@ += The COLUMNAR index and its settings +:description: Create and drop a COLUMNAR index from Python on both builds, what a close or a crash does to it, and how to set the arcadedb.columnar.* settings, including the GROUP BY acceleration. + +// Sources: ArcadeDB 26.12.1: GlobalConfiguration.java (COLUMNAR_*), AlterDatabaseStatement.java, ColumnarGroupByStep.java, ColumnarCrashRecoveryTest.java. The statements about Python are pinned by tests/test_columnar_index.py, which runs on arcadedb-jvm and on the native arcadedb. + +A `COLUMNAR` index keeps a column projection of a type next to its records. The engine applies, flushes and merges it in threads of its own, and the `FACETS` clause and `GROUP BY` read it. It needs engine 26.12.1 or later: an older engine refuses `CREATE INDEX ... COLUMNAR` with `Index type 'COLUMNAR' is not supported`. The facets themselves are read with `ResultSet.facets`. + +The index is a statement, so it is the same on both builds: + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Item") +db.command("sql", "CREATE PROPERTY Item.category STRING") +db.command("sql", "CREATE PROPERTY Item.score INTEGER") +db.command("sql", "CREATE PROPERTY Item.tags LIST OF STRING") +db.command("sql", "CREATE INDEX ON Item (category) COLUMNAR") +db.command("sql", "CREATE INDEX ON Item (score) COLUMNAR") +db.command("sql", "CREATE INDEX ON Item (tags BY ITEM) COLUMNAR") # a LIST needs BY ITEM +---- + +A COLUMNAR index is listed in `schema:indexes` as `Item[category]_columnar`, with one more row per bucket of the type (`Item_0_`). Drop it by the name of the type-level row. The engine refuses to drop a per-bucket row alone. After the drop, `FACETS` answers from a scan of the records and `GROUP BY` runs its normal plan. + +[source,python] +---- +db.command("sql", "DROP INDEX `Item[category]_columnar`") +---- + +== Close, reopen and crash + +The projection survives a close. After a clean close, and after a process that ended without closing the database (the lock file `database.lck` is left behind and the next open replays the write-ahead log), the facets of the reopened database are exact. They come from the projection, patched with the records it has not applied yet, or from a scan while the engine rebuilds the projection by itself, and `source` in each facet says which. A test ends a child process with `os._exit` after committing, once with the commits still in the write-ahead log and once with part of them already in published segments, and checks the facets and `GROUP BY` of the reopened database against what Python inserted. That is a crash of the process. It is not a power loss, which no test here can make (see xref:how-to/durability.adoc[Durability settings]). + +== The settings + +The engine reads the settings of the projection from `arcadedb.columnar.*`. Which route sets one depends on its scope in `GlobalConfiguration`, and there is no setter of its own in Python: the routes that exist already are enough. + +* Scope database (`flushInterval`, `flushMaxRows`, `flushMinRows`, `flushMaxAge`, `mergeDeadRatio`, `mergeMaxSegments`, `maxCardinality`, `segmentBlockRows`, `applyInterval`, `applyMaxLag`, `dictionaryCompactRatio`, `maxFallbackScans` and `groupByAcceleration`): `ALTER DATABASE`. It takes effect at once, is stored with the database and is still there after a reopen. `SELECT settings FROM schema:database` lists the value and the default of each. +* Scope JVM (`applyThreads`, `maintenanceThreads`, `rebuildConcurrency`, `queryMemoryBudget`) and scope server (`patchBudget`, `patchBudgetRatio`, `queryBudgetWait`, `parallelCountTasks`, `parallelMinMatches`, `topMax`): `ALTER DATABASE` refuses them with `cannot be set per database`. Set them as system properties before the engine starts, as for `arcadedb.fileSync` on xref:how-to/durability.adoc[Durability settings]: `-Darcadedb.columnar.applyThreads=4` in `jvm_kwargs={"jvm_args": [...]}` or in `ARCADEDB_JVM_ARGS` for `arcadedb-jvm`, and in `ARCADEDB_NATIVE_ISOLATE_ARGS` for `arcadedb`. The pool settings are read once, when the pool starts. No test here sets a system property. + +[source,python] +---- +db.command("sql", "ALTER DATABASE `arcadedb.columnar.flushInterval` 50") +db.command("sql", "ALTER DATABASE `arcadedb.columnar.mergeDeadRatio` 0.5") +---- + +=== GROUP BY acceleration + +`GROUP BY` over a type with a COLUMNAR index on the key, and on each aggregated property, is answered from the projection. `arcadedb.columnar.groupByAcceleration` is on by default. The engine checks it each time a statement runs, so `ALTER DATABASE` changes the next statement: + +[source,python] +---- +sql = "SELECT category, count(*) AS n FROM Item GROUP BY category ORDER BY category" +db.command("sql", "ALTER DATABASE `arcadedb.columnar.groupByAcceleration` false") +off = db.query("sql", sql).to_list() +db.command("sql", "ALTER DATABASE `arcadedb.columnar.groupByAcceleration` true") +on = db.query("sql", sql).to_list() +print(on == off) +print("path: columnar" in db.query("sql", f"PROFILE {sql}").first().get("executionPlanAsString")) +---- + +[source,text] +---- +True +True +---- + +Both paths return the same rows. `PROFILE` runs the statement and its plan names the path: `path: columnar` for the projection, or `path: normal (reason: ...)` with the reason, such as `disabled by arcadedb.columnar.groupByAcceleration`. A statement that the projection cannot answer, for example one inside a transaction that already changed the type, runs the normal plan and returns the same rows. diff --git a/tests/test_columnar_index.py b/tests/test_columnar_index.py new file mode 100644 index 0000000..8f1153c --- /dev/null +++ b/tests/test_columnar_index.py @@ -0,0 +1,622 @@ +"""The COLUMNAR index on both builds: create, ingest, facets and GROUP BY, close and reopen, drop, and the settings (issue #292). + +``CREATE INDEX ... COLUMNAR`` (ArcadeDB 26.12.1) keeps a column projection of a type next to its records. The projection is +applied, flushed and merged by threads of the engine, and ``FACETS`` queries and ``GROUP BY`` read it. This file runs on +``arcadedb-jvm`` and on the native ``arcadedb`` with only the public Python API. On the native build it is the proof that those +threads and the index code run inside the GraalVM shared library: the engine's classes that the index and the facets touch by +reflection are listed in native/src/main/resources/META-INF/native-image/org.humemai/arcadedb-native-lib/reachability-metadata.json +(the audit is in the pull request of issue #292). + +The engine of a lane is whatever the lane links (see tests/test_columnar_engine_smoke.py): an engine older than 26.12.1 must +refuse the index and the settings with its own error, a newer one must run them. The version is read through the test's database +(``SELECT version()``), so the same check runs on both builds and nothing is skipped. + +What the tests state: + +* the facets of a type with a COLUMNAR index equal what Python counts over the rows it inserted, for a single-valued property, a + numeric one and a multi-valued one (``BY ITEM``), and ``GROUP BY`` gives the same counts, sums and extremes; +* a clean close and an unclean one (a child process that ends without closing the database, so ``database.lck`` is left behind and + the next open replays the write-ahead log) both leave the facets exact, and the projection serves them again by itself; +* ``arcadedb.columnar.groupByAcceleration`` is a database setting that ``ALTER DATABASE`` changes at once and stores with the + database, ``PROFILE`` shows which path ran, and both paths return the same rows; +* the settings with the scope of the JVM or of a server are not database settings: ``ALTER DATABASE`` refuses them. + +Waiting for the projection polls a state with a deadline that is a hang detector, never a latency bound. +""" + +import glob +import json +import os +import re +import subprocess # nosec B404 - test-controlled child process, no untrusted input +import sys +import textwrap +import time +from collections import Counter + +import pytest + +from arcadedb import create_database, open_database + +FACETS_SINCE = (26, 12, 1) + +CATEGORIES = ["red", "green", "blue", "red", "green", "red"] +TAG_SETS = [["x", "y"], ["y"], ["z", "x", "x"], ["x"], ["y", "z"]] + +SCHEMA_SQL = [ + "CREATE DOCUMENT TYPE Item", + "CREATE PROPERTY Item.id INTEGER", + "CREATE PROPERTY Item.category STRING", + "CREATE PROPERTY Item.score INTEGER", + "CREATE PROPERTY Item.tags LIST OF STRING", +] +INDEX_SQL = [ + "CREATE INDEX ON Item (category) COLUMNAR", + "CREATE INDEX ON Item (score) COLUMNAR", + "CREATE INDEX ON Item (tags BY ITEM) COLUMNAR", +] + +FACETS_Q = ( + "SELECT id FROM Item ORDER BY id " + "FACETS category TERMS TOP 10, tags TERMS TOP 10, s: score STATS" +) +GROUP_BY_Q = ( + "SELECT category, count(*) AS n, sum(score) AS total, min(score) AS lo, " + "max(score) AS hi FROM Item GROUP BY category ORDER BY category" +) + +GROUP_BY_KEY = "arcadedb.columnar.groupByAcceleration" +# the settings of a database, by scope (GlobalConfiguration 26.12.1): these are DATABASE scope +DATABASE_SETTINGS = [ + "arcadedb.columnar.flushInterval", + "arcadedb.columnar.flushMaxRows", + "arcadedb.columnar.flushMinRows", + "arcadedb.columnar.flushMaxAge", + "arcadedb.columnar.mergeDeadRatio", + "arcadedb.columnar.mergeMaxSegments", + "arcadedb.columnar.maxCardinality", + "arcadedb.columnar.segmentBlockRows", + "arcadedb.columnar.applyInterval", + "arcadedb.columnar.applyMaxLag", + "arcadedb.columnar.dictionaryCompactRatio", + "arcadedb.columnar.maxFallbackScans", + GROUP_BY_KEY, +] +JVM_SETTINGS = [ + "arcadedb.columnar.applyThreads", + "arcadedb.columnar.maintenanceThreads", + "arcadedb.columnar.rebuildConcurrency", + "arcadedb.columnar.queryMemoryBudget", +] +SERVER_SETTINGS = [ + "arcadedb.columnar.patchBudget", + "arcadedb.columnar.patchBudgetRatio", + "arcadedb.columnar.queryBudgetWait", + "arcadedb.columnar.parallelCountTasks", + "arcadedb.columnar.parallelMinMatches", + "arcadedb.columnar.topMax", +] + + +@pytest.fixture +def test_db(tmp_path): + """Create a temporary test database.""" + db = create_database(str(tmp_path / "test_columnar_index_db")) + yield db + db.drop() + + +def _engine_version(db): + """The engine's x.y.z through SQL `SELECT version()`; a SNAPSHOT or a build suffix is dropped.""" + raw = str(db.query("sql", "SELECT version() AS v").first().get("v")) + parts = [ + int(p) if p.isdigit() else 0 + for p in raw.split()[0].split("-")[0].split(".")[:3] + ] + return tuple(parts + [0] * (3 - len(parts))) + + +def _is_older_engine(path): + """True when the engine predates the COLUMNAR index. Read through a database of its own, so a test can decide before it + starts a child process or reopens its database.""" + probe = create_database(path) + try: + return _engine_version(probe) < FACETS_SINCE + finally: + probe.drop() + + +def _make_rows(start, count): + return [ + { + "id": i, + "category": CATEGORIES[i % len(CATEGORIES)], + "score": (i * 7) % 50 + 1, + "tags": TAG_SETS[i % len(TAG_SETS)], + } + for i in range(start, start + count) + ] + + +def _insert_statement(row): + """The INSERT of one row and its positional arguments; the list is a literal, so the same text runs on both builds.""" + tags = "[" + ", ".join(f"'{t}'" for t in row["tags"]) + "]" + sql = f"INSERT INTO Item SET id = ?, category = ?, score = ?, tags = {tags}" # nosec B608 - test-owned text + return sql, [row["id"], row["category"], row["score"]] + + +def _insert(db, rows): + with db.transaction(): + for row in rows: + sql, args = _insert_statement(row) + db.command("sql", sql, *args) + + +def _create_schema(db): + for sql in SCHEMA_SQL: + db.command("sql", sql) + + +def _create_indexes(db): + for sql in INDEX_SQL: + db.command("sql", sql) + + +def _refused_by_older_engine(db): + """True when the engine predates the COLUMNAR index: the statement is then refused with the engine's error, and the caller + has nothing more to check on this engine.""" + if _engine_version(db) >= FACETS_SINCE: + return False + with pytest.raises( + Exception, match=re.escape("Index type 'COLUMNAR' is not supported") + ): + db.command("sql", INDEX_SQL[0]) + return True + + +def _columnar_names(db): + """The names of the type-level COLUMNAR indexes of Item (the per-bucket rows are named `Item_0_`).""" + rows = db.query("sql", "SELECT FROM schema:indexes").to_list() + return sorted( + r.get("name") for r in rows if str(r.get("name")).endswith("_columnar") + ) + + +def _wait_for(predicate, what, timeout=120.0): + """The value of `predicate()` once it is truthy. The timeout is a hang detector.""" + deadline = time.monotonic() + timeout + while True: + value = predicate() + if value: + return value + if time.monotonic() >= deadline: + pytest.fail(f"timed out after {timeout}s waiting for {what}") + time.sleep(0.1) + + +def _counts(facet): + """A terms facet as {value: count}, the bucket of the records without a value left out.""" + return {b["value"]: b["count"] for b in facet["buckets"] if not b.get("missing")} + + +def _expected_group_by(rows): + groups = {} + for row in rows: + group = groups.setdefault(row["category"], []) + group.append(row["score"]) + return [ + {"category": c, "n": len(s), "total": sum(s), "lo": min(s), "hi": max(s)} + for c, s in sorted(groups.items()) + ] + + +def _check_facets(db, rows, sources=("index",)): + """The facets of FACETS_Q equal what Python counts over `rows`; returns the sources the facets were answered from.""" + rs = db.query("sql", FACETS_Q) + got = rs.to_list() + facets = rs.facets + assert [r["id"] for r in got] == [r["id"] for r in rows] + assert facets["meta"]["exact"] is True, facets["meta"] + + category, tags, score = facets["category"], facets["tags"], facets["s"] + assert category["source"] in sources, category + assert tags["source"] in sources, tags + assert score["source"] in sources, score + assert _counts(category) == dict(Counter(r["category"] for r in rows)) + # a record counts once for a tag it holds twice + assert _counts(tags) == dict(Counter(t for r in rows for t in set(r["tags"]))) + scores = [r["score"] for r in rows] + assert score["count"] == len(rows) + assert (score["min"], score["max"], score["sum"]) == ( + min(scores), + max(scores), + sum(scores), + ) + return {category["source"], tags["source"], score["source"]} + + +def _facet_sources(db): + """Where the facets of FACETS_Q are answered from right now, for a test that waits for the projection to serve them.""" + rs = db.query("sql", FACETS_Q) + rs.to_list() + facets = rs.facets + return {facets[alias]["source"] for alias in ("category", "tags", "s")} + + +def _check_group_by(db, rows): + assert db.query("sql", GROUP_BY_Q).to_list() == _expected_group_by(rows) + + +def _plan(db, sql): + """The plan of a PROFILE run of `sql`: it names the path the GROUP BY took (`path: columnar` or `path: normal (reason: ...)`).""" + return str(db.query("sql", f"PROFILE {sql}").first().get("executionPlanAsString")) + + +def _settings(db): + row = db.query("sql", "SELECT settings FROM schema:database").first() + return {s["key"]: s for s in row.get("settings")} + + +def _text(value): + return str(value).lower() + + +def _alter(db, key, value): + db.command("sql", f"ALTER DATABASE `{key}` {value}") + + +def _segment_rows(db_path): + """The rows the published segments of every bucket hold, read from the manifests the engine writes under `columnar/`.""" + total = 0 + for manifest in glob.glob(os.path.join(db_path, "columnar", "b*", "manifest.json")): + try: + with open(manifest, encoding="utf-8") as handle: + total += sum(s["rows"] for s in json.load(handle)["segments"]) + except (OSError, ValueError): + pass # replaced while it was read: the next poll sees the new one + return total + + +class TestColumnarIndexLifecycle: + def test_create_ingest_query_drop(self, test_db): + _create_schema(test_db) + if _refused_by_older_engine(test_db): + return + + _create_indexes(test_db) + names = _columnar_names(test_db) + assert "Item[category]_columnar" in names, names + assert "Item[score]_columnar" in names, names + assert len(names) == 3 and any("tags" in n for n in names), names + assert test_db.schema.exists_index("Item[category]_columnar") + + rows = _make_rows(0, 60) + for start in range(0, 60, 20): + _insert(test_db, rows[start : start + 20]) + + assert _check_facets(test_db, rows) == {"index"} + _check_group_by(test_db, rows) + _wait_for( + lambda: "path: columnar" in _plan(test_db, GROUP_BY_Q), + "GROUP BY to be answered from the projection", + ) + _check_group_by(test_db, rows) + + # updates and deletes reach the projection too + with test_db.transaction(): + test_db.command("sql", "UPDATE Item SET category = 'blue' WHERE id < 10") + test_db.command("sql", "DELETE FROM Item WHERE id >= 50") + rows = [dict(r, category="blue") if r["id"] < 10 else r for r in rows[:50]] + assert _check_facets(test_db, rows) == {"index"} + _check_group_by(test_db, rows) + + # dropped: the same query is answered by a scan, and GROUP BY by the normal plan + for name in names: + test_db.command("sql", f"DROP INDEX `{name}`") + assert _columnar_names(test_db) == [] + assert not test_db.schema.exists_index("Item[category]_columnar") + assert _check_facets(test_db, rows, sources=("scan",)) == {"scan"} + _check_group_by(test_db, rows) + assert "path: columnar" not in _plan(test_db, GROUP_BY_Q) + + def test_index_created_on_existing_records_is_rebuilt_by_the_engine(self, test_db): + _create_schema(test_db) + if _refused_by_older_engine(test_db): + return + + rows = _make_rows(0, 80) + _insert(test_db, rows) + _create_indexes(test_db) + + # the maintenance pool builds the projection of the records that were there: until then a scan answers, exactly + assert _check_facets(test_db, rows, sources=("index", "scan")) + _wait_for( + lambda: _facet_sources(test_db) == {"index"}, + "the projection of the existing records", + ) + assert _check_facets(test_db, rows) == {"index"} + _check_group_by(test_db, rows) + + def test_flush_and_merge_run_in_the_background(self, test_db): + """The flush and merge threads: with a short flush interval and few segments allowed, a stream of small commits ends up + in published segments (the manifests under columnar/), merged, and the facets stay exact. + """ + _create_schema(test_db) + if _refused_by_older_engine(test_db): + return + + _alter(test_db, "arcadedb.columnar.flushInterval", 20) + _alter(test_db, "arcadedb.columnar.applyInterval", 10) + _alter(test_db, "arcadedb.columnar.mergeMaxSegments", 2) + assert ( + str(_settings(test_db)["arcadedb.columnar.flushInterval"]["value"]) == "20" + ) + _create_indexes(test_db) + + rows = _make_rows(0, 120) + for start in range(0, 120, 10): + _insert(test_db, rows[start : start + 10]) + time.sleep(0.05) # lets a flush land between commits + + path = test_db.get_database_path() + _wait_for( + lambda: _segment_rows(path) >= len(rows), + "the flush to publish every row in a segment", + ) + assert _check_facets(test_db, rows) == {"index"} + _check_group_by(test_db, rows) + + def test_clean_close_and_reopen(self, tmp_path): + path = str(tmp_path / "clean_reopen_db") + if _is_older_engine(str(tmp_path / "version_probe_db")): + with create_database(path) as db: + _create_schema(db) + assert _refused_by_older_engine(db) + return + + rows = _make_rows(0, 70) + with create_database(path) as db: + _create_schema(db) + _create_indexes(db) + _insert(db, rows) + assert _check_facets(db, rows) == {"index"} + + with open_database(path) as db: + assert len(_columnar_names(db)) == 3 + # exact at once, from the projection or from a scan, and served from the projection again by itself + assert _check_facets(db, rows, sources=("index", "scan")) + _wait_for( + lambda: _facet_sources(db) == {"index"}, + "the projection after the reopen", + ) + assert _check_facets(db, rows) == {"index"} + _check_group_by(db, rows) + + # the reopened projection takes new commits + more = _make_rows(70, 30) + _insert(db, more) + rows = rows + more + assert _check_facets(db, rows) == {"index"} + + with open_database(path) as db: + _wait_for( + lambda: _facet_sources(db) == {"index"}, + "the projection after the second reopen", + ) + assert _check_facets(db, rows) == {"index"} + # dropped after a reopen: the index and its projection are gone + for name in _columnar_names(db): + db.command("sql", f"DROP INDEX `{name}`") + assert _columnar_names(db) == [] + assert _check_facets(db, rows, sources=("scan",)) == {"scan"} + + with open_database(path) as db: + assert _columnar_names(db) == [] + assert _check_facets(db, rows, sources=("scan",)) == {"scan"} + _check_group_by(db, rows) + + +_CHILD = textwrap.dedent(""" + import glob, json, os, sys, time + + import arcadedb + + path, spec_file = sys.argv[1], sys.argv[2] + with open(spec_file, encoding="utf-8") as handle: + spec = json.load(handle) + + db = arcadedb.create_database(path) + for sql in spec["schema"]: + db.command("sql", sql) + for index, batch in enumerate(spec["batches"]): + if index == spec["wait_before_batch"]: + # the earlier batches must be in published segments before the rest is written + deadline = time.monotonic() + 120 + while True: + total = 0 + for manifest in glob.glob(os.path.join(path, "columnar", "b*", "manifest.json")): + try: + with open(manifest, encoding="utf-8") as handle: + total += sum(s["rows"] for s in json.load(handle)["segments"]) + except (OSError, ValueError): + pass + if total >= spec["wait_rows"]: + break + if time.monotonic() > deadline: + sys.exit("the projection never flushed") + time.sleep(0.1) + with db.transaction(): + for sql, args in batch: + db.command("sql", sql, *args) + print("READY", flush=True) + # no close: the process ends at once, as a crash of the process does. What was committed is in the write-ahead log. + os._exit(0) + """) + + +class TestColumnarIndexUncleanClose: + """The process that wrote the database ends without closing it. Python cannot make the engine's own kill (the Java + ``DatabaseInternal.kill()`` of the engine's crash tests) on the native build, and a JVM-only hook would not test both + packages, so the writer is a child process that calls ``os._exit``: the same state the engine's recovery reads, a database + with its lock file left behind and committed pages not written to the data files (a crash of the process, not a power loss). + """ + + @pytest.mark.parametrize( + "flushed_first", [False, True], ids=["unflushed", "flushed"] + ) + def test_projection_after_an_unclean_close(self, tmp_path, flushed_first): + path = str(tmp_path / "unclean_db") + if _is_older_engine(str(tmp_path / "version_probe_db")): + with create_database(path) as db: + _create_schema(db) + assert _refused_by_older_engine(db) + return + + first, second = _make_rows(0, 60), _make_rows(60, 20) + spec = { + "schema": SCHEMA_SQL + INDEX_SQL, + "batches": [ + [list(_insert_statement(r)) for r in first], + [list(_insert_statement(r)) for r in second], + ], + # flushed: the first batch is waited for in published segments, so the recovery has the second one to replay on top of them + "wait_before_batch": 1 if flushed_first else -1, + "wait_rows": len(first), + } + spec_file = tmp_path / "spec.json" + spec_file.write_text(json.dumps(spec), encoding="utf-8") + + proc = subprocess.run( # nosec B603 - fixed argv, no shell, test-owned + [sys.executable, "-c", _CHILD, path, str(spec_file)], + capture_output=True, + text=True, + timeout=300, + ) + assert proc.returncode == 0, f"{proc.stdout}\n{proc.stderr}" + assert "READY" in proc.stdout, f"{proc.stdout}\n{proc.stderr}" + # a clean close deletes the lock file (tests/test_concurrency.py): one left behind marks an unclean close + assert os.path.exists(os.path.join(path, "database.lck")) + + rows = first + second + with open_database(path) as db: + assert len(_columnar_names(db)) == 3 + # every committed row is there, and the facets are exact at once: from the projection, patched, or from a scan + assert db.query("sql", "SELECT count(*) AS n FROM Item").first().get( + "n" + ) == len(rows) + assert _check_facets(db, rows, sources=("index", "scan")) + _check_group_by(db, rows) + # the projection catches up or is rebuilt by the engine, with nobody asking + _wait_for( + lambda: _facet_sources(db) == {"index"}, + "the projection after the unclean close", + ) + assert _check_facets(db, rows) == {"index"} + more = _make_rows(80, 20) + _insert(db, more) + rows = rows + more + assert _check_facets(db, rows) == {"index"} + + with open_database(path) as db: + _wait_for( + lambda: _facet_sources(db) == {"index"}, + "the projection after the clean close that followed", + ) + assert _check_facets(db, rows) == {"index"} + _check_group_by(db, rows) + + +class TestColumnarSettings: + def test_database_settings_are_listed_with_their_defaults(self, test_db): + if _engine_version(test_db) < FACETS_SINCE: + with pytest.raises(Exception, match="not found"): + _alter(test_db, GROUP_BY_KEY, "false") + return + + settings = _settings(test_db) + for key in DATABASE_SETTINGS: + assert key in settings, key + for key in JVM_SETTINGS + SERVER_SETTINGS: + assert key not in settings, key + assert _text(settings[GROUP_BY_KEY]["default"]) == "true" + assert _text(settings[GROUP_BY_KEY]["value"]) == "true" + + def test_settings_of_the_jvm_and_of_a_server_are_not_database_settings( + self, test_db + ): + if _engine_version(test_db) < FACETS_SINCE: + with pytest.raises(Exception, match="not found"): + _alter(test_db, JVM_SETTINGS[0], 4) + return + + with pytest.raises(Exception) as jvm: + _alter(test_db, "arcadedb.columnar.applyThreads", 4) + assert "JVM scope" in str(jvm.value) + assert "cannot be set per database" in str(jvm.value) + + with pytest.raises(Exception) as server: + _alter(test_db, "arcadedb.columnar.patchBudget", 500) + assert "SERVER scope" in str(server.value) + assert "cannot be set per database" in str(server.value) + + def test_group_by_gives_the_same_rows_with_and_without_the_acceleration( + self, test_db + ): + _create_schema(test_db) + if _refused_by_older_engine(test_db): + return + + _create_indexes(test_db) + rows = _make_rows(0, 90) + _insert(test_db, rows) + expected = _expected_group_by(rows) + + accelerated = test_db.query("sql", GROUP_BY_Q).to_list() + assert accelerated == expected + _wait_for( + lambda: "path: columnar" in _plan(test_db, GROUP_BY_Q), + "GROUP BY to be answered from the projection", + ) + + _alter(test_db, GROUP_BY_KEY, "false") + assert _text(_settings(test_db)[GROUP_BY_KEY]["value"]) == "false" + plan = _plan(test_db, GROUP_BY_Q) + assert "path: columnar" not in plan + assert f"disabled by {GROUP_BY_KEY}" in plan + assert test_db.query("sql", GROUP_BY_Q).to_list() == accelerated + + _alter(test_db, GROUP_BY_KEY, "true") + assert _text(_settings(test_db)[GROUP_BY_KEY]["value"]) == "true" + assert "path: columnar" in _plan(test_db, GROUP_BY_Q) + assert test_db.query("sql", GROUP_BY_Q).to_list() == accelerated + + def test_the_settings_are_stored_with_the_database(self, tmp_path): + path = str(tmp_path / "settings_db") + if _is_older_engine(str(tmp_path / "version_probe_db")): + with create_database(path) as db: + with pytest.raises(Exception, match="not found"): + _alter(db, GROUP_BY_KEY, "false") + return + + rows = _make_rows(0, 40) + with create_database(path) as db: + _create_schema(db) + _create_indexes(db) + _insert(db, rows) + _alter(db, GROUP_BY_KEY, "false") + _alter(db, "arcadedb.columnar.flushInterval", 50) + _alter(db, "arcadedb.columnar.mergeDeadRatio", 0.5) + + with open_database(path) as db: + settings = _settings(db) + assert _text(settings[GROUP_BY_KEY]["value"]) == "false" + assert str(settings["arcadedb.columnar.flushInterval"]["value"]) == "50" + assert float(settings["arcadedb.columnar.mergeDeadRatio"]["value"]) == 0.5 + assert f"disabled by {GROUP_BY_KEY}" in _plan(db, GROUP_BY_Q) + _check_group_by(db, rows) + _alter(db, GROUP_BY_KEY, "true") + _wait_for( + lambda: "path: columnar" in _plan(db, GROUP_BY_Q), + "GROUP BY to be answered from the projection after the reopen", + ) + _check_group_by(db, rows) From 82d8cec62bd7e376e10e37c558a99f9097e111e1 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 7 Nov 2026 10:29:40 +0900 Subject: [PATCH 2/2] Return a Python str from Database.get_database_path(), get_name() and list_vector_indexes() on arcadedb-jvm (refs #292) The first CI run of tests/test_columnar_index.py failed on arcadedb-jvm only: Database.get_database_path() returned the raw java.lang.String of getDatabasePath(), which os.path.join refuses (TypeError: expected str, bytes or os.PathLike object, not String). The native build returns a str, so the lanes disagreed. Same family as #286 (get_type_name, fixed in #287): the binding wraps the Java getter in str(). Sweep of src/arcadedb/*.py for a method that documents a str and returns a Java getter result unconverted (an AST pass over every def annotated with str or Path, looking at what each return expression is): Database.get_name(), Database.get_database_path() and Schema.list_vector_indexes() (a list of java.lang.String). Fixed with str(). Left alone because they are documented as returning Java objects: Document.get_identity(), Schema.get_type(), get_types(), get_index_by_name(), Document.get_raw(), Result.get_raw(). Result.get_rid(), Document.get_rid(), get_type_name() and the vector index names already convert. tests/test_str_returns.py asserts type(x) is str on both packages for the four str returns that can be read through the public API (name, path, vector index names, Result.get_rid), plus that the path joins. Nothing was run locally; CI is the first run. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/core.py | 4 +-- src/arcadedb/schema.py | 4 +-- tests/test_str_returns.py | 55 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 59 insertions(+), 4 deletions(-) create mode 100644 tests/test_str_returns.py diff --git a/src/arcadedb/core.py b/src/arcadedb/core.py index c177781..e685888 100644 --- a/src/arcadedb/core.py +++ b/src/arcadedb/core.py @@ -1400,7 +1400,7 @@ def get_name(self) -> str: """Get the database name.""" self._check_not_closed() try: - return self._java_db.getName() + return str(self._java_db.getName()) except Exception as e: raise ArcadeDBError(f"Failed to get database name: {e}") from e @@ -1422,7 +1422,7 @@ def get_database_path(self) -> str: """Get the database path.""" self._check_not_closed() try: - return self._java_db.getDatabasePath() + return str(self._java_db.getDatabasePath()) except Exception as e: raise ArcadeDBError(f"Failed to get database path: {e}") from e diff --git a/src/arcadedb/schema.py b/src/arcadedb/schema.py index 88b7de5..b53e14b 100644 --- a/src/arcadedb/schema.py +++ b/src/arcadedb/schema.py @@ -742,7 +742,7 @@ def list_vector_indexes(self) -> List[str]: # Check for LSM Vector (direct) if "LSMVectorIndex" in index_class_name: - indexes.append(java_index.getName()) + indexes.append(str(java_index.getName())) # Check for TypeIndex (wrapper) elif "TypeIndex" in index_class_name: @@ -751,7 +751,7 @@ def list_vector_indexes(self) -> List[str]: if sub_indexes and not sub_indexes.isEmpty(): first_sub = sub_indexes.get(0) if "LSMVectorIndex" in first_sub.getClass().getName(): - indexes.append(java_index.getName()) + indexes.append(str(java_index.getName())) except Exception: log_swallowed_exception(_LOGGER, "while listing vector indexes") diff --git a/tests/test_str_returns.py b/tests/test_str_returns.py new file mode 100644 index 0000000..53c2d9d --- /dev/null +++ b/tests/test_str_returns.py @@ -0,0 +1,55 @@ +"""Methods that document a ``str`` return one: a plain Python ``str``, not a ``java.lang.String``, on both packages. + +``arcadedb-jvm`` reads these from Java getters, and JPype returns a ``java.lang.String`` for a getter result unless the binding +wraps it with ``str()``. A ``java.lang.String`` equals the same text but is not accepted by ``os.path.join`` or ``pathlib.Path`` +(``TypeError: expected str, bytes or os.PathLike object, not String``, found by tests/test_columnar_index.py), so the type is +asserted with ``type(x) is str``. The native build builds its answers in Python and must give the same type. Same family as +ArcadeData/arcadedb-python#286. +""" + +import os + +import pytest + + +def _db_name(db): + return db.get_name() + + +def _db_path(db): + return db.get_database_path() + + +def _vector_index_name(db): + db.command( + "opencypher", + "CREATE VECTOR INDEX str_return_embedding FOR (n:StrReturn) ON n.embedding " + "OPTIONS {indexConfig: {`vector.dimensions`: 3}}", + ) + names = db.schema.list_vector_indexes() + assert "str_return_embedding" in names + assert all(type(n) is str for n in names), [type(n) for n in names] + return names[0] + + +def _result_rid(db): + db.command("sql", "CREATE DOCUMENT TYPE StrReturnDoc") + with db.transaction(): + db.command("sql", "INSERT INTO StrReturnDoc SET n = 1") + return db.query("sql", "SELECT FROM StrReturnDoc").first().get_rid() + + +@pytest.mark.parametrize( + "read", + [_db_name, _db_path, _vector_index_name, _result_rid], + ids=lambda f: f.__name__.lstrip("_"), +) +def test_documented_str_returns_are_python_str(temp_db, read): + value = read(temp_db) + assert type(value) is str, type(value) + + +def test_database_path_can_be_joined(temp_db): + path = temp_db.get_database_path() + assert os.path.isdir(path) + assert os.path.join(path, "database.lck").startswith(path)