Preserve canonical schema change contracts

Author: TonisOrmisson

src/openstatspec/__init__.py

@@ -27,7 +27,7 @@
 __all__ = [
     "AssignOperation", "BooleanExpression", "ComparisonExpression",
     "ConditionalAssignOperation", "ExecuteOperation", "Operand",
-    "ConditionalAssignOperation", "CreateVariableOperation", "DeleteVariableOperation",
+    "PredicateExpression", "CreateVariableOperation", "DeleteVariableOperation",
     "CapabilityDeclaration", "DoltConformanceSource", "LossReport",
     "SpssFrontendCompilation", "TransformationError",
     "TransformationFrontendError",

src/openstatspec/frontends/spss/binding.py

@@ -13,6 +13,7 @@
     PredicateExpression, RecodeMatch, RecodeOperation, RecodeResult, RecodeRule,
     ReplaceValueLabelsOperation, SetFormatOperation,
     SetMeasurementLevelOperation, SetVariableLabelOperation,
+    TRANSFORMATION_PLAN_SCHEMA_CHANGE_CONTRACT,
     TRANSFORMATION_PLAN_V1_CONTRACT,
     TransformationPlan, TypedValue, ValueLabel,
 )
@@ -491,10 +492,15 @@ def bind_spss_syntax(
     v01_types = (
         RecodeOperation, SetVariableLabelOperation, ReplaceValueLabelsOperation,
     )
+    schema_change_types = (CreateVariableOperation, DeleteVariableOperation)
     contract = (
-        TRANSFORMATION_PLAN_V1_CONTRACT
-        if all(isinstance(operation, v01_types) for operation in operations)
-        else "openstatspec-transformation-plan-v0.2"
+        TRANSFORMATION_PLAN_SCHEMA_CHANGE_CONTRACT
+        if any(isinstance(operation, schema_change_types) for operation in operations)
+        else (
+            TRANSFORMATION_PLAN_V1_CONTRACT
+            if all(isinstance(operation, v01_types) for operation in operations)
+            else "openstatspec-transformation-plan-v0.2"
+        )
     )
     plan = TransformationPlan(
         tuple(operations), contract=contract, input_alias=input_alias,

src/openstatspec/sql/inplace_transform.py

@@ -709,6 +709,28 @@ def _apply_plan_on_connection(
             _compact_variable_ordinals(
                 connection, core=core, variables=variables,
             )
+            canonical_physical = {"__case_ordinal"}
+            for remaining_variable in variables:
+                expected_physical = physical_name(
+                    str(remaining_variable["source_name"]), canonical_physical,
+                )
+                current_physical = str(remaining_variable["physical_name"])
+                if current_physical == expected_physical:
+                    continue
+                connection.exec_driver_sql(
+                    f"ALTER TABLE {qualified_table} RENAME COLUMN "
+                    f"{quote(current_physical)} TO {quote(expected_physical)}"
+                )
+                connection.execute(
+                    update(core.variable)
+                    .where(
+                        core.variable.c.variable_id
+                        == remaining_variable["variable_id"]
+                    )
+                    .values(physical_name=expected_physical)
+                )
+                remaining_variable["physical_name"] = expected_physical
+            used_physical = canonical_physical
             relation = Table(
                 table_name, MetaData(), schema=dataset.get("physical_table_schema"),
                 autoload_with=connection,

src/openstatspec/transform/plan.py

@@ -588,6 +588,19 @@ class TransformationPlan:
     def __post_init__(self) -> None:
         if not isinstance(self.contract, str) or self.contract not in _TRANSFORMATION_PLAN_CONTRACTS:
             _invalid("Plan contract is not a supported transformation-plan contract.")
+        if (
+            self.contract != TRANSFORMATION_PLAN_SCHEMA_CHANGE_CONTRACT
+            and any(
+                isinstance(operation, (
+                    CreateVariableOperation, DeleteVariableOperation,
+                ))
+                for operation in self.operations
+            )
+        ):
+            _invalid(
+                "Create/delete schema operations require "
+                "openstatspec-transformation-plan-v0.3."
+            )
         if self.contract == TRANSFORMATION_PLAN_V1_CONTRACT and any(
             isinstance(operation, (
                 AssignOperation, ConditionalAssignOperation, SetFormatOperation,
@@ -710,7 +723,7 @@ def _match(raw: Any) -> RecodeMatch:
     _invalid("Unknown recode match kind.")
 
 def transformation_plan_from_dict(raw: Mapping[str, Any]) -> TransformationPlan:
-    """Strictly validate canonical v0.1 or additive v0.2 plan documents."""
+    """Strictly validate canonical v0.1, v0.2, or schema-change v0.3 plans."""
     if not isinstance(raw, Mapping):
         _invalid("Transformation plan must be an object.")
     _exact(raw, {"contract", "input_alias", "operations"}, "Transformation plan")

tests/test_inplace_transform.py

@@ -189,6 +189,67 @@ def test_string_declaration_creates_column_and_catalog_variable(catalog) -> None
     )["valid"] is True
 
 
+def test_delete_recanonicalizes_surviving_collision_columns(tmp_path) -> None:
+    path = tmp_path / "collision-delete.sqlite"
+    url = f"sqlite:///{path}"
+    openstatspec.initialize_catalog(database_url=url)
+    base = _variables()[0]
+    variables = [
+        {
+            **base,
+            "ordinal": ordinal,
+            "source_name": source_name,
+            "physical_name": physical,
+        }
+        for ordinal, (source_name, physical) in enumerate((
+            ("a-b", "a_b"), ("a_b", "a_b_2"), ("keep", "keep"),
+        ), start=1)
+    ]
+    create_wide_dataset(
+        database_url=url,
+        dataset_id="collision_source",
+        source_name="collision.sav",
+        source_format="SAV",
+        source_sha256="c" * 64,
+        rows=[
+            {"a_b": 1.0, "a_b_2": 2.0, "keep": 3.0},
+            {"a_b": 4.0, "a_b_2": 5.0, "keep": 6.0},
+        ],
+        variables=variables,
+    )
+    openstatspec.install_in_place_transformation_schema(database_url=url)
+    connection = sqlite3.connect(path)
+    dataset_id = connection.execute(
+        "select dataset_id from dataset where dataset_name = 'collision_source'"
+    ).fetchone()[0]
+    connection.close()
+    plan = openstatspec.TransformationPlan(
+        (openstatspec.DeleteVariableOperation("a-b"),),
+        contract="openstatspec-transformation-plan-v0.3",
+    )
+
+    openstatspec.apply_transformation_plan_in_place(
+        database_url=url,
+        dataset_id=dataset_id,
+        plan=plan,
+        actor="test-agent",
+    )
+
+    connection = sqlite3.connect(path)
+    assert connection.execute(
+        "select source_name, physical_name, source_ordinal from variable "
+        "where dataset_id = ? order by source_ordinal",
+        (dataset_id,),
+    ).fetchall() == [("a_b", "a_b", 1), ("keep", "keep", 2)]
+    assert connection.execute(
+        "select a_b, keep from data_collision_source order by __case_ordinal"
+    ).fetchall() == [(2.0, 3.0), (5.0, 6.0)]
+    connection.close()
+    assert validate_wide_dataset(
+        database_url=url, dataset_id=dataset_id,
+    )["valid"] is True
+
+
 def test_delete_prunes_an_empty_multiple_response_set(catalog) -> None:
     url, path, dataset_id, _table_name = catalog
     connection = sqlite3.connect(path)
@@ -335,9 +396,10 @@ def test_generic_string_width_is_rejected_before_ddl(
             {"server_version": "3.35.0"},
         ),
     )
-    plan = openstatspec.TransformationPlan((
-        openstatspec.CreateVariableOperation("note", "string", 4),
-    ))
+    plan = openstatspec.TransformationPlan(
+        (openstatspec.CreateVariableOperation("note", "string", 4),),
+        contract="openstatspec-transformation-plan-v0.3",
+    )
 
     with pytest.raises(TargetCapabilityExceededError, match="permits 3"):
         openstatspec.apply_transformation_plan_in_place(

tests/test_transform_frontend.py

@@ -13,6 +13,7 @@
     spss_source_hash,
 )
 from openstatspec.transform import (
+    CreateVariableOperation,
     RecodeMatch,
     RecodeOperation,
     RecodeResult,
@@ -559,6 +560,28 @@ def test_v02_plan_and_schema_reject_decimal_format_that_cannot_fit() -> None:
         )
 
 
+def test_schema_operations_require_v03_contract() -> None:
+    operation = CreateVariableOperation("note", "string", 8)
+    with pytest.raises(TransformationFrontendError) as caught:
+        TransformationPlan((operation,))
+    assert caught.value.code == "invalid_transformation_plan"
+
+    plan = TransformationPlan(
+        (operation,), contract="openstatspec-transformation-plan-v0.3",
+    )
+    assert plan.contract == "openstatspec-transformation-plan-v0.3"
+
+
+def test_spss_schema_commands_emit_v03_contract() -> None:
+    schema = _schema(VariableDefinition("q1", "numeric"))
+    assert bind_spss_syntax(
+        parse_spss_syntax("STRING note (A8)."), schema,
+    ).plan.contract == "openstatspec-transformation-plan-v0.3"
+    assert bind_spss_syntax(
+        parse_spss_syntax("COMPUTE other = q1. DELETE VARIABLES q1."), schema,
+    ).plan.contract == "openstatspec-transformation-plan-v0.3"
+
+
 def test_custom_nonempty_input_alias_is_canonical() -> None:
     plan = bind_spss_syntax(
         parse_spss_syntax("VARIABLE LABELS q1 'One'."),