test(index): share harness loads and read backticked raw indexes

tests/integration/test_schema/test_index_use_cases_ch.py

@@ -26,7 +26,8 @@
 """
 
 import json
-from collections.abc import Iterator
+from collections.abc import Callable, Generator, Iterator
+from contextlib import contextmanager, suppress
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from pathlib import Path
@@ -38,6 +39,8 @@
 from dfe_schemas.ddl_catalogue import DDLCatalogue, IndexSpec
 from dfe_schemas.loader import DEFAULT_CARDINALITY
 from dfe_schemas.render import split_use_case
+from scalo.deployment.test_support.pytest_plugin import suite_for
+from scalo.deployment.test_support.shared_service import locked
 
 from dfe_engine.schema.schema_builder_v2 import SchemaBuilderV2
 from dfe_engine.schema.schema_loader import SchemaLoader
@@ -58,7 +61,9 @@
 )
 
 # Loading both tables took 263 s on a remote docker host, against the suite's 300 s default.
-pytestmark = [pytest.mark.integration, pytest.mark.timeout(900)]
+_TIMEOUT_S = 900
+
+pytestmark = [pytest.mark.integration, pytest.mark.timeout(_TIMEOUT_S)]
 
 HEADER = ("timeseries", "1.0.3")
 
@@ -167,37 +172,92 @@ def _insert(client: Any, table: Table) -> None:
     client.command(f"OPTIMIZE TABLE {table.qualified} FINAL")
 
 
+@contextmanager
+def _shared(
+    state: Path, load: Callable[[str], None], drop: Callable[[str], None]
+) -> Generator[str]:
+    """The run's one loaded database on a server, loaded by the first process to ask.
+
+    The suite runs under xdist, which hands this module's tests to several workers.
+    Each loading its own copy put four loads on one server and timed them all out.
+    Every worker holding the database is counted in *state*, and the last one out
+    drops it.
+    """
+    lock = state.with_suffix(".lock")
+    with locked(lock):
+        if state.exists():
+            held = json.loads(state.read_text(encoding="utf-8"))
+            database: str = held["database"]
+            users: int = held["users"]
+        else:
+            database = f"dfe_index_{datetime.now(tz=UTC):%H%M%S%f}"
+            try:
+                load(database)
+            except BaseException:
+                with suppress(Exception):
+                    drop(database)
+                raise
+            users = 0
+        state.write_text(json.dumps({"database": database, "users": users + 1}), encoding="utf-8")
+    try:
+        yield database
+    finally:
+        with locked(lock):
+            users = json.loads(state.read_text(encoding="utf-8"))["users"] - 1
+            if users:
+                state.write_text(
+                    json.dumps({"database": database, "users": users}), encoding="utf-8"
+                )
+            else:
+                state.unlink()
+                drop(database)
+
+
 @pytest.fixture(scope="module")
 def loaded(
-    module_ch_params: dict, tmp_path_factory: pytest.TempPathFactory
+    request: pytest.FixtureRequest,
+    module_ch_params: dict,
+    tmp_path_factory: pytest.TempPathFactory,
 ) -> Iterator[tuple[Any, dict[str, Table]]]:
     """The client, and both tables built through the deploy path with ``ROWS`` rows each."""
     import clickhouse_connect
 
-    client = clickhouse_connect.get_client(**module_ch_params)
-    database = f"dfe_index_{datetime.now(tz=UTC):%H%M%S%f}"
-    client.command(create_database(database, PlatformProbe(client).platform))
+    # The load outlasts the driver's 300 s read default. No session, so a query left
+    # running after a client error cannot lock out the cleanup.
+    client = clickhouse_connect.get_client(
+        **module_ch_params, send_receive_timeout=_TIMEOUT_S, autogenerate_session_id=False
+    )
     meta = tmp_path_factory.mktemp("index-use-cases") / "index_use_cases.yaml"
     meta.write_text(yaml.safe_dump(meta_schema(), sort_keys=False), encoding="utf-8")
     sources = {
         "header": _source("index-probe", meta, header=True),
         "plain": _source("index-probe-plain", meta, header=False),
     }
     windows = {"header": HEADER_WINDOW, "plain": PLAIN_WINDOW}
-    tables = {
-        key: Table(database, source.table_name, key == "header", windows[key])
-        for key, source in sources.items()
-    }
-    try:
-        for key, table in tables.items():
+
+    def tables(database: str) -> dict[str, Table]:
+        return {
+            key: Table(database, source.table_name, key == "header", windows[key])
+            for key, source in sources.items()
+        }
+
+    def load(database: str) -> None:
+        client.command(create_database(database, PlatformProbe(client).platform))
+        for key, table in tables(database).items():
             _deploy(client, database, sources[key])
             _insert(client, table)
-        yield client, tables
+
+    run_dir = suite_for(request.config).run_dir
+    run_dir.mkdir(parents=True, exist_ok=True)
+    server = "".join(c if c.isalnum() else "-" for c in str(module_ch_params["host"]))
+    state = run_dir / f"index-use-cases-{server}-{module_ch_params['port']}.json"
+    try:
+        with _shared(
+            state, load, lambda database: drop_database_and_policies(client, database)
+        ) as database:
+            yield client, tables(database)
     finally:
-        try:
-            drop_database_and_policies(client, database)
-        finally:
-            client.close()
+        client.close()
 
 
 # -- reading the plan -------------------------------------------------------

tests/integration/test_schema/test_source_landing_ddl_ch.py

@@ -33,6 +33,9 @@
 
 pytestmark = pytest.mark.integration
 
+# The renderer may backtick the index name, and a text index type nests parentheses.
+_RAW_INDEX = re.compile(r"INDEX\s+`?idx__raw`?\s.*?\bTYPE\s+(text\(.*\))\s+GRANULARITY\b")
+
 
 @pytest.fixture
 def source_database(clickhouse_test_database):
@@ -115,7 +118,7 @@ def test_the_declared_raw_index_survives_onto_the_server(ch_client, source_datab
     source = _source("raw-index-probe")
     statements = _deploy(ch_client, source_database, source)
 
-    declared = re.search(r"INDEX\s+idx__raw\s+.*?TYPE\s+(text\(.*?\))", statements[0])
+    declared = _RAW_INDEX.search(statements[0])
     assert declared, f"the deploy DDL declares no _raw text index:\n{statements[0]}"
     tokenizer = re.search(r"tokenizer\s*=\s*'?([A-Za-z0-9_]+)", declared.group(1))
     assert tokenizer, f"no tokenizer in {declared.group(1)!r}"