From 2ac9d51935ed8d7561c01cae784d6561abcc447d Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 21:06:41 +1100 Subject: [PATCH 1/2] test(index): share harness loads and read backticked raw indexes The full tier failed release run 42988646142 on two integration modules. The landing DDL test looked for bare `INDEX idx__raw`. The dfe-schemas 0.2.29 renderer backticks the name, so the detector missed a correct index. The detector takes the name with or without backticks. It reads the whole nested `text(...)` type. The index harness loaded its two 1.44M row tables in a module fixture. Under xdist worksteal, each worker running its tests loaded a copy. Four text-index loads ran at once on one ClickHouse. The first hit the driver's 300 s read timeout. Its cleanup failed with SESSION_IS_LOCKED. Three more ran past the 900 s pytest-timeout. Their workers were killed. The other 38 tests errored. The first worker loads one database under a lock in the run's shared directory. The rest wait and reuse it. The last worker drops it. State is keyed on the server, so a tier with a server per worker loads once per server. The loader client has no session. Its read timeout matches the module's. Data size and assertions are unchanged. The xdist_group marker fails with pytest 9.2 and pytest-xdist 3.8.0. On pytest 9.2 a node id lives in `_id`. pytest-xdist 3.8.0 writes its group suffix to `_nodeid`, so loadgroup ignores the group (pytest-dev/pytest-xdist#1407). --- .../test_schema/test_index_use_cases_ch.py | 89 +++++++++++++++---- .../test_schema/test_source_landing_ddl_ch.py | 5 +- 2 files changed, 76 insertions(+), 18 deletions(-) diff --git a/tests/integration/test_schema/test_index_use_cases_ch.py b/tests/integration/test_schema/test_index_use_cases_ch.py index 938e7faa..239f0d1c 100644 --- a/tests/integration/test_schema/test_index_use_cases_ch.py +++ b/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, 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,16 +172,56 @@ 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]) -> Iterator[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): + held = json.loads(state.read_text(encoding="utf-8")) if state.exists() else None + if held is None: + database = f"dfe_index_{datetime.now(tz=UTC):%H%M%S%f}" + try: + load(database) + except BaseException: + with suppress(Exception): + drop(database) + raise + held = {"database": database, "users": 0} + held["users"] += 1 + state.write_text(json.dumps(held), encoding="utf-8") + try: + yield held["database"] + finally: + with locked(lock): + held = json.loads(state.read_text(encoding="utf-8")) + held["users"] -= 1 + if held["users"]: + state.write_text(json.dumps(held), encoding="utf-8") + else: + state.unlink() + drop(held["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 = { @@ -184,20 +229,30 @@ def loaded( "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 ------------------------------------------------------- diff --git a/tests/integration/test_schema/test_source_landing_ddl_ch.py b/tests/integration/test_schema/test_source_landing_ddl_ch.py index 1a9a78f4..d075aef8 100644 --- a/tests/integration/test_schema/test_source_landing_ddl_ch.py +++ b/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}" From 2ec885b8fc69030b8d0bf56c0817b9a4179e268a Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 22:11:01 +1100 Subject: [PATCH 2/2] test(index): type the shared harness load state One dict held the database name and holder count, giving `+=` and the yield mixed types. ty rejected the shared-load helper. The name and count are separate typed values. The helper is annotated as a Generator. --- .../test_schema/test_index_use_cases_ch.py | 31 +++++++++++-------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/tests/integration/test_schema/test_index_use_cases_ch.py b/tests/integration/test_schema/test_index_use_cases_ch.py index 239f0d1c..6773eb59 100644 --- a/tests/integration/test_schema/test_index_use_cases_ch.py +++ b/tests/integration/test_schema/test_index_use_cases_ch.py @@ -26,7 +26,7 @@ """ import json -from collections.abc import Callable, Iterator +from collections.abc import Callable, Generator, Iterator from contextlib import contextmanager, suppress from dataclasses import dataclass, field from datetime import UTC, datetime @@ -173,7 +173,9 @@ def _insert(client: Any, table: Table) -> None: @contextmanager -def _shared(state: Path, load: Callable[[str], None], drop: Callable[[str], None]) -> Iterator[str]: +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. @@ -183,8 +185,11 @@ def _shared(state: Path, load: Callable[[str], None], drop: Callable[[str], None """ lock = state.with_suffix(".lock") with locked(lock): - held = json.loads(state.read_text(encoding="utf-8")) if state.exists() else None - if held is None: + 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) @@ -192,20 +197,20 @@ def _shared(state: Path, load: Callable[[str], None], drop: Callable[[str], None with suppress(Exception): drop(database) raise - held = {"database": database, "users": 0} - held["users"] += 1 - state.write_text(json.dumps(held), encoding="utf-8") + users = 0 + state.write_text(json.dumps({"database": database, "users": users + 1}), encoding="utf-8") try: - yield held["database"] + yield database finally: with locked(lock): - held = json.loads(state.read_text(encoding="utf-8")) - held["users"] -= 1 - if held["users"]: - state.write_text(json.dumps(held), encoding="utf-8") + 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(held["database"]) + drop(database) @pytest.fixture(scope="module")