test: expect Recall schema version 4 and compare rewritten views over the previous columns

Derive the expected Recall schema version from RECALL_SCHEMA_VERSION, compare the rewritten ClickHouse views with the previous ones over the columns both have, and check the columns the invocations view adds separately.

Author: kvlonge

tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/_test_types.py

@@ -268,6 +268,7 @@ class ClickHouseRecallViewEquivalenceTestCase:
     replaced_views: tuple[str, ...]
     all_views: tuple[str, ...]
     expected_statuses: list[list[object]]
+    expected_added_columns: tuple[tuple[tuple[object, ...], ...], ...]
 
 
 @dataclass(frozen=True)

tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/helpers.py

@@ -14,6 +14,7 @@
 
 from sqlbuild.adapters.clickhouse.classes.clickhouse_adapter import ClickHouseAdapter
 from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import (
+    REPO_ROOT,
     prepare_inline_project,
     prepare_waffle_shop,
     run_sqb,
@@ -28,6 +29,22 @@
 )
 
 CLICKHOUSE_IMAGE: str = "clickhouse/clickhouse-server:26.1"
+_RECALL_SCHEMA_CONSTANTS: Path = REPO_ROOT / "crates/sqlbuild-recall/src/schema/constants.rs"
+_RECALL_SCHEMA_VERSION_PATTERN: re.Pattern[str] = re.compile(
+    r"pub const RECALL_SCHEMA_VERSION: i64 = (\d+);"
+)
+
+
+def _recall_schema_version() -> int:
+    """The Recall schema version this build writes, as `RECALL_SCHEMA_VERSION` declares it."""
+
+    (version,) = _RECALL_SCHEMA_VERSION_PATTERN.findall(
+        _RECALL_SCHEMA_CONSTANTS.read_text(encoding="utf-8")
+    )
+    return int(version)
+
+
+RECALL_SCHEMA_VERSION: int = _recall_schema_version()
 
 
 def build_unique_schema_name(*, prefix: str) -> str:
@@ -915,16 +932,23 @@ def fetch_with_settings(
 
 
 def differing_rows(
-    *, config: dict[str, object], left: str, right: str, settings: dict[str, object]
+    *,
+    config: dict[str, object],
+    left: str,
+    right: str,
+    settings: dict[str, object],
+    columns: tuple[str, ...] = (),
 ) -> int:
-    """Rows of `left` missing from `right` plus rows of `right` missing from `left`."""
+    """Rows of `left` missing from `right` plus rows of `right` missing from `left`, over `columns`
+    when given, else over every column."""
 
+    selected: str = ", ".join(f"`{column}`" for column in columns) or "*"
     rows: tuple[tuple[object, ...], ...] = fetch_with_settings(
         config=config,
         sql=(
-            f"SELECT (SELECT count() FROM (SELECT * FROM {left} EXCEPT DISTINCT "
-            f"SELECT * FROM {right})) + (SELECT count() FROM (SELECT * FROM {right} "
-            f"EXCEPT DISTINCT SELECT * FROM {left}))"
+            f"SELECT (SELECT count() FROM (SELECT {selected} FROM {left} EXCEPT DISTINCT "
+            f"SELECT {selected} FROM {right})) + (SELECT count() FROM (SELECT {selected} "
+            f"FROM {right} EXCEPT DISTINCT SELECT {selected} FROM {left}))"
         ),
         settings=settings,
     )
@@ -940,6 +964,30 @@ def described(*, config: dict[str, object], relation: str) -> tuple[tuple[object
     )
 
 
+def column_names(*, config: dict[str, object], relation: str) -> tuple[str, ...]:
+    """Each column's name, in order."""
+
+    return tuple(str(name) for name, _ in described(config=config, relation=relation))
+
+
+def described_except(
+    *, config: dict[str, object], relation: str, previous: str
+) -> tuple[tuple[tuple[object, ...], ...], tuple[tuple[object, ...], ...]]:
+    """`relation`'s columns and types that `previous` also has, and those it adds, in order."""
+
+    columns: dict[object, object] = {
+        column[0]: column[1] for column in described(config=config, relation=relation)
+    }
+    kept: tuple[str, ...] = column_names(config=config, relation=previous)
+    added: dict[object, object] = dict(columns)
+    for name in kept:
+        del added[name]
+    return (
+        tuple((name, columns[name]) for name in kept),
+        tuple(added.items()),
+    )
+
+
 def create_previous_views(
     *, config: dict[str, object], database: str, names: tuple[str, ...], prefix: str
 ) -> None:

tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall.py

@@ -32,6 +32,7 @@
 )
 from tests.e2e.src.sqlbuild.cli.commands.main.clickhouse.helpers import (
     FIRST_RELEASE_EDITS,
+    RECALL_SCHEMA_VERSION,
     age_recorded_invocations,
     age_stored_artifacts,
     apply_printed_ddl,
@@ -903,7 +904,7 @@ def test_given_a_large_store_when_filtering_a_view_then_the_filter_reaches_the_t
             description="views of an older version are replaced when its version row expired",
             stale_view="SELECT invocation_id, statement_id FROM {database}.recall_events",
             expected_statement_columns=((1,),),
-            expected_version=((4,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
     ],
     ids=lambda case: case.description,
@@ -1039,12 +1040,12 @@ def test_given_a_newer_store_whose_version_row_expired_when_an_older_writer_buil
         ClickHouseRecallVersionRefreshTestCase(
             description="a version row past half the retention is recorded again",
             row_age_days=20,
-            expected_rows=((2, 3, 1),),
+            expected_rows=((2, RECALL_SCHEMA_VERSION, 1),),
         ),
         ClickHouseRecallVersionRefreshTestCase(
             description="a recent version row is left as it is",
             row_age_days=1,
-            expected_rows=((1, 3, 0),),
+            expected_rows=((1, RECALL_SCHEMA_VERSION, 0),),
         ),
     ],
     ids=lambda case: case.description,

tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_setup.py

@@ -15,6 +15,7 @@
     ClickHouseRecallStaleViewsTestCase,
 )
 from tests.e2e.src.sqlbuild.cli.commands.main.clickhouse.helpers import (
+    RECALL_SCHEMA_VERSION,
     apply_printed_ddl,
     create_previous_views,
     drop_database,
@@ -197,15 +198,15 @@ def test_given_a_missing_database_when_the_writer_may_not_create_it_then_the_sta
             writer_privileges="ALL",
             expected_notices=0,
             expected_stale_views=((0,),),
-            expected_version=((4,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
         ClickHouseRecallStaleViewsTestCase(
             description="previous views recorded as current are replaced by an administrator",
             deleted_version_rows="0",
             writer_privileges="ALL",
             expected_notices=0,
             expected_stale_views=((0,),),
-            expected_version=((4,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
         ClickHouseRecallStaleViewsTestCase(
             description="previous views without a version row make a SELECT, INSERT writer warn",
@@ -374,7 +375,7 @@ def test_given_a_writer_that_may_create_only_the_database_when_building_then_bot
             dropped_column=("recall_invocations", "argv_digest"),
             expected_columns=((1,),),
             expected_stale_views=((0,),),
-            expected_version=((4,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
     ],
     ids=lambda case: case.description,

tests/e2e/src/sqlbuild/cli/commands/main/clickhouse/test_clickhouse_recall_views.py

@@ -17,9 +17,12 @@
     ClickHouseRecallViewEquivalenceTestCase,
 )
 from tests.e2e.src.sqlbuild.cli.commands.main.clickhouse.helpers import (
+    RECALL_SCHEMA_VERSION,
     RECALL_VIEWS,
+    column_names,
     create_previous_views,
     described,
+    described_except,
     differing_rows,
     duplicate_recall_rows,
     execute_clickhouse_sql,
@@ -206,6 +209,19 @@ def test_given_a_reader_when_querying_then_pinned_settings_apply_unless_it_may_n
                 ["failed"],
                 ["running"],
             ],
+            expected_added_columns=(
+                (
+                    ("origin", "Nullable(String)"),
+                    ("command_mode", "Nullable(String)"),
+                    ("integration", "Nullable(String)"),
+                    ("integration_job", "Nullable(String)"),
+                    ("integration_run_id", "Nullable(String)"),
+                    ("integration_retry", "Nullable(Int64)"),
+                ),
+                (),
+                (),
+                (),
+            ),
         ),
     ],
     ids=lambda case: case.description,
@@ -230,22 +246,40 @@ def test_given_representative_invocations_when_views_are_rewritten_then_rows_and
         statuses: list[list[object]] = recall_rows(
             project_dir, "SELECT status FROM invocations ORDER BY status"
         )
+        previous_columns: list[tuple[str, ...]] = [
+            column_names(
+                config=clickhouse_e2e_config, relation=f"{recall_database}.previous_{view}"
+            )
+            for view in test_case.replaced_views
+        ]
         against_previous: list[int] = [
             differing_rows(
                 config=clickhouse_e2e_config,
                 left=f"{recall_database}.{view}",
                 right=f"{recall_database}.previous_{view}",
                 settings={"join_use_nulls": 1},
+                columns=columns,
+            )
+            for view, columns in zip(test_case.replaced_views, previous_columns, strict=True)
+        ]
+        compared: list[tuple[tuple[tuple[object, ...], ...], tuple[tuple[object, ...], ...]]] = [
+            described_except(
+                config=clickhouse_e2e_config,
+                relation=f"{recall_database}.{view}",
+                previous=f"{recall_database}.previous_{view}",
             )
             for view in test_case.replaced_views
         ]
         types: list[bool] = [
-            described(config=clickhouse_e2e_config, relation=f"{recall_database}.{view}")
+            kept
             == described(
                 config=clickhouse_e2e_config, relation=f"{recall_database}.previous_{view}"
             )
-            for view in test_case.replaced_views
+            for view, (kept, _) in zip(test_case.replaced_views, compared, strict=True)
         ]
+        added: tuple[tuple[tuple[object, ...], ...], ...] = tuple(
+            added_columns for _, added_columns in compared
+        )
     finally:
         os.killpg(running.pid, signal.SIGKILL)
         _ = running.wait(timeout=30)
@@ -277,8 +311,9 @@ def test_given_representative_invocations_when_views_are_rewritten_then_rows_and
             left=f"{recall_database}.{view}",
             right=f"{recall_database}.previous_{view}",
             settings={"join_use_nulls": 1},
+            columns=columns,
         )
-        for view in test_case.replaced_views
+        for view, columns in zip(test_case.replaced_views, previous_columns, strict=True)
     ]
     counts_with_duplicates: list[bool] = [
         fetch_clickhouse_rows(
@@ -293,6 +328,7 @@ def test_given_representative_invocations_when_views_are_rewritten_then_rows_and
     assert statuses == test_case.expected_statuses
     assert against_previous == [0 for _ in test_case.replaced_views]
     assert types == [True for _ in test_case.replaced_views]
+    assert added == test_case.expected_added_columns
     assert without_join_nulls == [0 for _ in test_case.all_views]
     assert with_duplicates == [0 for _ in test_case.all_views]
     assert previous_with_duplicates == [0 for _ in test_case.replaced_views]
@@ -354,9 +390,9 @@ def test_given_one_invocation_lookup_when_planned_then_each_table_is_read_once_b
             expected_ddl_fragments=(
                 "CREATE OR REPLACE VIEW `{database}`.`invocations`",
                 "CREATE OR REPLACE VIEW `{database}`.`statements`",
-                "SELECT 'schema', 'schema_version', '4'",
+                f"SELECT 'schema', 'schema_version', '{RECALL_SCHEMA_VERSION}'",
             ),
-            expected_version=((4,),),
+            expected_version=((RECALL_SCHEMA_VERSION,),),
         ),
     ],
     ids=lambda case: case.description,