From 0927d6a68b5e98e258b5a9ac03c5f75d1a3bfe93 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Tue, 8 Sep 2026 08:58:43 +0300 Subject: [PATCH 1/2] test: reproduce in-place metadata ownership and lineage failures --- tests/test_inplace_transform.py | 201 ++++++++++++++++++++++++++++++++ 1 file changed, 201 insertions(+) diff --git a/tests/test_inplace_transform.py b/tests/test_inplace_transform.py index 3f12051..11e0b32 100644 --- a/tests/test_inplace_transform.py +++ b/tests/test_inplace_transform.py @@ -1,11 +1,13 @@ from __future__ import annotations +import json import sqlite3 from dataclasses import replace from types import SimpleNamespace import pytest from sqlalchemy import create_engine, inspect, text +from sqlalchemy.exc import IntegrityError import openstatspec import openstatspec.sql.inplace_transform as inplace_transform @@ -85,6 +87,205 @@ def catalog(tmp_path): return url, path, dataset_id, table_name +def _publish_score(catalog): + url, _path, dataset_id, _table_name = catalog + openstatspec.apply_spss_in_place( + database_url=url, dataset_id=dataset_id, + source_text="COMPUTE other = score.", actor="test-setup", + ) + return openstatspec.derive_sql_dataset( + database_url=url, parent_dataset_id=dataset_id, + query_sql="SELECT score FROM parent ORDER BY score ASC NULLS LAST", + columns=[{ + "name": "score", "storage_kind": "numeric", "source": "score", + "expression_role": "identity", + }], + transformation_name="published_score", + ) + + +@pytest.mark.parametrize("surface", ["canonical", "spss"]) +def test_shared_value_labels_preserve_sibling_until_last_owner(catalog, surface): + url, path, dataset_id, table_name = catalog + openstatspec.apply_spss_in_place( + database_url=url, dataset_id=dataset_id, + source_text="COMPUTE other = score. VALUE LABELS score 1 'Original'.", + actor="test-setup", + ) + with sqlite3.connect(path) as connection: + connection.execute("PRAGMA foreign_keys = ON") + old_set = connection.execute( + "SELECT value_label_set_id FROM variable_value_label_set" + ).fetchone()[0] + other_id = connection.execute( + "SELECT variable_id FROM variable WHERE source_name = 'other'" + ).fetchone()[0] + connection.execute( + "INSERT INTO variable_value_label_set VALUES (?, ?)", + (other_id, old_set), + ) + original_set = connection.execute("SELECT * FROM value_label_set").fetchall() + original_labels = connection.execute("SELECT * FROM value_label").fetchall() + schema = openstatspec.VariableSchema(tuple( + openstatspec.VariableDefinition(name, "numeric") + for name in ("score", "other") + )) + for name in ("score", "other"): + syntax = f"VALUE LABELS {name} 2 'Replacement'." + if surface == "canonical": + result = openstatspec.apply_transformation_plan_in_place( + database_url=url, dataset_id=dataset_id, actor="test-agent", + plan=openstatspec.compile_spss_syntax(syntax, schema).plan, + ) + else: + result = openstatspec.apply_spss_in_place( + database_url=url, dataset_id=dataset_id, actor="test-agent", + source_text=syntax, + ) + assert (result["dataset_id"], result["physical_table_name"]) == ( + dataset_id, table_name, + ) + _, variables, _rows = read_wide_dataset(database_url=url, dataset_id=dataset_id) + assert {v["source_name"]: json.loads(v["value_labels"]) for v in variables} == { + "score": {"2.0": "Replacement"}, + "other": {"1.0": "Original"} if name == "score" else {"2.0": "Replacement"}, + } + with sqlite3.connect(path) as connection: + assert connection.execute( + f'SELECT __case_ordinal, score, other FROM "{table_name}" ' + "ORDER BY __case_ordinal" + ).fetchall() == [(1, 1.0, 1.0), (2, 2.0, 2.0), (3, 3.0, 3.0)] + assert connection.execute( + "SELECT dataset_id, physical_table_name FROM dataset" + ).fetchall() == [(dataset_id, table_name)] + assert connection.execute( + "SELECT status FROM transformation_apply WHERE apply_id = ?", + (result["apply_id"],), + ).fetchone() == ("succeeded",) + assert connection.execute("PRAGMA foreign_key_check").fetchall() == [] + assert connection.execute( + "SELECT * FROM value_label_set WHERE value_label_set_id = ?", (old_set,), + ).fetchall() == (original_set if name == "score" else []) + assert connection.execute( + "SELECT * FROM value_label WHERE value_label_set_id = ?", (old_set,), + ).fetchall() == (original_labels if name == "score" else []) + assert connection.execute( + "SELECT * FROM variable_value_label_set WHERE value_label_set_id = ?", + (old_set,), + ).fetchall() == ([(other_id, old_set)] if name == "score" else []) + + +@pytest.mark.parametrize("surface", ["canonical", "spss"]) +@pytest.mark.parametrize("recreate", [False, True], ids=["delete", "recreate"]) +def test_published_lineage_blocks_original_deletion_before_mutation( + catalog, monkeypatch, surface, recreate, +): + url, path, dataset_id, _table_name = catalog + _publish_score(catalog) + with sqlite3.connect(path) as connection: + before = tuple(connection.iterdump()) + reached = [] + monkeypatch.setattr(inplace_transform, "_failure_boundary", reached.append) + syntax = ( + "COMPUTE temporary = score. VARIABLE LABELS score 'Must not change'. " + "DELETE VARIABLES ScOrE." + ) + if recreate: + syntax += " COMPUTE score = temporary." + with pytest.raises(openstatspec.TransformationError) as caught: + if surface == "canonical": + openstatspec.apply_transformation_plan_in_place( + database_url=url, dataset_id=dataset_id, actor="test-agent", + plan=_plan(syntax), + ) + else: + openstatspec.apply_spss_in_place( + database_url=url, dataset_id=dataset_id, actor="test-agent", + source_text=syntax, + ) + assert caught.value.code == "variable_has_dependents" + assert "score" in str(caught.value).casefold() + assert reached == [] + with sqlite3.connect(path) as connection: + assert tuple(connection.iterdump()) == before + + +def test_lineage_allows_local_deletes_and_unreferenced_original_recreation(catalog): + url, path, dataset_id, table_name = catalog + published = _publish_score(catalog) + with sqlite3.connect(path) as connection: + original_ids = dict(connection.execute( + "SELECT source_name, variable_id FROM variable" + )) + lineage = connection.execute("SELECT * FROM derived_variable_lineage").fetchall() + tables = connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name" + ).fetchall() + result = openstatspec.apply_spss_in_place( + database_url=url, dataset_id=dataset_id, actor="test-agent", + source_text=( + "COMPUTE tmp = score. DELETE VARIABLES tmp. " + "RECODE score (2 = 9) (ELSE = COPY) INTO band. DELETE VARIABLES band. " + "STRING note (A4). DELETE VARIABLES note. " + "DELETE VARIABLES other. " + "RECODE score (2 = 9) (ELSE = COPY) INTO other." + ), + ) + assert (result["dataset_id"], result["physical_table_name"]) == (dataset_id, table_name) + with sqlite3.connect(path) as connection: + variables = connection.execute( + "SELECT source_name, variable_id, physical_name, source_ordinal " + "FROM variable ORDER BY source_ordinal" + ).fetchall() + assert variables[0] == ("score", original_ids["score"], "score", 1) + assert len(variables) == 2 + assert variables[1][0] == "other" + assert variables[1][1] not in original_ids.values() + assert variables[1][3] == 6 + assert connection.execute( + f'SELECT __case_ordinal, score, "{variables[1][2]}" FROM "{table_name}" ' + "ORDER BY __case_ordinal" + ).fetchall() == [(1, 1.0, 1.0), (2, 2.0, 9.0), (3, 3.0, 3.0)] + assert [row[1] for row in connection.execute( + f'PRAGMA table_info("{table_name}")' + )] == ["__case_ordinal", "score", variables[1][2]] + assert connection.execute( + "SELECT dataset_id, physical_table_name FROM dataset" + ).fetchall() == [(dataset_id, table_name)] + assert connection.execute("SELECT * FROM derived_variable_lineage").fetchall() == lineage + assert connection.execute("PRAGMA foreign_key_check").fetchall() == [] + assert connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name" + ).fetchall() == tables + assert openstatspec.validate_derived( + database_url=url, derived_dataset_id=published["derived_dataset_id"], + )["valid"] is True + + +def test_native_foreign_key_failure_rolls_back_schema_data_metadata_and_audit(catalog): + url, path, dataset_id, _table_name = catalog + openstatspec.apply_spss_in_place( + database_url=url, dataset_id=dataset_id, + source_text="VARIABLE LABELS score 'Before probe'.", actor="test-setup", + ) + with sqlite3.connect(path) as connection: + connection.execute( + "CREATE TRIGGER invalid_variable_reference AFTER UPDATE OF variable_label " + "ON variable BEGIN INSERT INTO missing_rule " + "(missing_rule_id, variable_id, ordinal, rule_kind, code_kind, numeric_value) " + "VALUES ('fk-probe', 'nonexistent-variable', 1, 'discrete', 'numeric', 99); END" + ) + before = tuple(connection.iterdump()) + with pytest.raises(IntegrityError, match="FOREIGN KEY constraint failed") as caught: + openstatspec.apply_transformation_plan_in_place( + database_url=url, dataset_id=dataset_id, actor="test-agent", + plan=_plan("COMPUTE temporary = score. VARIABLE LABELS score 'Trigger probe'."), + ) + assert isinstance(caught.value.orig, sqlite3.IntegrityError) + with sqlite3.connect(path) as connection: + assert tuple(connection.iterdump()) == before + + def test_plan_applies_to_same_dataset_and_physical_table_without_copy( catalog, ) -> None: From 395cbc09f2c3e333d5bbed676fce8a69daa7d0bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Tue, 8 Sep 2026 09:06:24 +0300 Subject: [PATCH 2/2] fix: preserve shared labels and reject referenced variable deletion --- src/openstatspec/sql/inplace_transform.py | 53 +++++++++++++++++++---- 1 file changed, 45 insertions(+), 8 deletions(-) diff --git a/src/openstatspec/sql/inplace_transform.py b/src/openstatspec/sql/inplace_transform.py index 608ed81..022d928 100644 --- a/src/openstatspec/sql/inplace_transform.py +++ b/src/openstatspec/sql/inplace_transform.py @@ -11,7 +11,7 @@ from sqlalchemy import ( Column, DateTime, Float, Integer, MetaData, String, Table, Text, and_, case, create_engine, delete, inspect, insert, literal, null, select, text, update, - or_, + or_, event as sqlalchemy_event, ) from ..transform import ( @@ -29,7 +29,7 @@ from .normative import catalog as core_catalog from .profiles import SqlProfile, preflight from .wide import physical_name, require_verified_catalog -from .workflow import TransformationError +from .workflow import TransformationError, workflow_catalog APPLY_CONTRACT = "openstatspec-in-place-transformation-v0.2" @@ -278,15 +278,21 @@ def _replace_value_labels( ) ).scalar_one_or_none() if old_set is not None: - connection.execute(delete(core.value_label).where( - core.value_label.c.value_label_set_id == old_set - )) connection.execute(delete(core.variable_value_label_set).where( core.variable_value_label_set.c.variable_id == variable_id )) - connection.execute(delete(core.value_label_set).where( - core.value_label_set.c.value_label_set_id == old_set - )) + still_used = connection.execute( + select(core.variable_value_label_set.c.variable_id).where( + core.variable_value_label_set.c.value_label_set_id == old_set + ) + ).first() + if still_used is None: + connection.execute(delete(core.value_label).where( + core.value_label.c.value_label_set_id == old_set + )) + connection.execute(delete(core.value_label_set).where( + core.value_label_set.c.value_label_set_id == old_set + )) label_set_id = str(uuid4()) connection.execute(insert(core.value_label_set).values( value_label_set_id=label_set_id, @@ -476,6 +482,26 @@ def _apply_plan_on_connection( table_name = _physical_table_name(dataset) plan = submission.plan bound = bind_transformation_plan(plan, schema) + deleted_names = { + operation.variable.casefold() for operation in plan.operations + if isinstance(operation, DeleteVariableOperation) + } + original_deletions = { + str(row["variable_id"]): str(row["source_name"]) for row in variables + if str(row["source_name"]).casefold() in deleted_names + } + if original_deletions and inspect(connection).has_table("derived_variable_lineage"): + lineage = workflow_catalog(MetaData()).derived_variable_lineage + blocked_id = connection.execute( + select(lineage.c.core_variable_id).where( + lineage.c.core_variable_id.in_(original_deletions) + ).limit(1) + ).scalar_one_or_none() + if blocked_id is not None: + raise TransformationError( + "variable_has_dependents", + f"Variable {original_deletions[blocked_id]!r} is referenced by published lineage.", + ) output_used_physical = {"__case_ordinal"} output_variables = [ { @@ -972,6 +998,17 @@ def _run_in_place_submission( sqlite_version = (*sqlite_version_parts, 0, 0)[:3] allow_delete_variable = sqlite_version >= (3, 35, 0) engine = create_engine(database_url) + if profile.name == "sqlite": + @sqlalchemy_event.listens_for(engine, "connect") + def enable_sqlite_foreign_keys(dbapi_connection, _connection_record): + cursor = dbapi_connection.cursor() + try: + cursor.execute("PRAGMA foreign_keys = ON") + if cursor.execute("PRAGMA foreign_keys").fetchone() != (1,): + raise RuntimeError("SQLite foreign-key enforcement could not be enabled.") + finally: + cursor.close() + journal: dict[str, Any] = {} branch: str | None = None head: str | None = None