Bind cleanup and preflight schema edits

Author: TonisOrmisson

src/openstatspec/sql/inplace_transform.py

@@ -27,6 +27,7 @@
 from .capabilities import effective_profile
 from .dolt_conformance import DoltConformanceSource
 from .normative import catalog as core_catalog
+from .profiles import SqlProfile, preflight
 from .wide import physical_name, require_verified_catalog
 from .workflow import TransformationError
 
@@ -435,6 +436,7 @@ def _apply_plan_on_connection(
     submission: InPlacePlanSubmission,
     actor: str,
     database_profile: str,
+    target_profile: SqlProfile,
     allow_schema_change: bool,
     allow_delete_variable: bool,
     dolt_branch: str | None,
@@ -451,6 +453,22 @@ def _apply_plan_on_connection(
     table_name = _physical_table_name(dataset)
     plan = submission.plan
     bound = bind_transformation_plan(plan, schema)
+    output_used_physical = {"__case_ordinal"}
+    output_variables = [
+        {
+            "ordinal": source_ordinal,
+            "source_name": variable.name,
+            "physical_name": physical_name(
+                variable.name, output_used_physical,
+            ),
+            "storage_kind": variable.storage_kind,
+            "string_width": variable.declared_string_width,
+        }
+        for source_ordinal, variable in enumerate(
+            bound.output_schema.variables, start=1,
+        )
+    ]
+    preflight(target_profile, output_variables)
     output_by_name = {
         variable.name.casefold(): variable
         for variable in bound.output_schema.variables
@@ -963,6 +981,7 @@ def _run_in_place_submission(
                     submission=submission,
                     actor=actor,
                     database_profile=profile.name,
+                    target_profile=profile,
                     allow_schema_change=profile.name in {"sqlite", "postgresql"},
                     allow_delete_variable=allow_delete_variable,
                     dolt_branch=branch,

src/openstatspec/sql/wide.py

@@ -949,7 +949,12 @@ def create_wide_dataset(
     except Exception as error:
         if namespace_owned:
             try:
-                with engine.begin() as cleanup:
+                with _bound_catalog_transaction(
+                    engine=engine, profile_name=profile.name,
+                    active=active_connection,
+                    audit_relations={*audit_relations, data_table.name},
+                    phase="import cleanup",
+                ) as cleanup:
                     create_normative_catalog(cleanup, normative)
                     if cleanup.execute(
                         select(normative.dataset.c.dataset_id).where(

tests/test_atomic_import.py

@@ -1,4 +1,5 @@
 import sqlite3
+from contextlib import contextmanager
 from dataclasses import replace
 
 import pytest
@@ -194,6 +195,55 @@ def test_declared_string_width_preflight_is_atomic_and_diagnostic(tmp_path, monk
     assert '"reason": "declared_string_width_limit"' in details
     assert '"string_width": 4' in details
 
+def test_dolt_import_cleanup_uses_bound_catalog_transaction(
+    tmp_path, monkeypatch,
+) -> None:
+    database_path = tmp_path / "dolt-cleanup.sqlite"
+    database = f"sqlite:///{database_path}"
+    active = {
+        "working_set_binding": {
+            "database": "test",
+            "active_branch": "main",
+        },
+    }
+    monkeypatch.setattr(wide, "effective_profile", lambda _url: (DOLT, active))
+    monkeypatch.setattr(
+        wide, "_capture_dolt_state", lambda *_args, **_kwargs: None,
+    )
+    phases = []
+    real_bound = wide._bound_catalog_transaction
+
+    @contextmanager
+    def capture_bound(**kwargs):
+        phases.append(kwargs["phase"])
+        with real_bound(**kwargs) as connection:
+            yield connection
+
+    monkeypatch.setattr(wide, "_bound_catalog_transaction", capture_bound)
+    real_store = wide.store_normative_dataset
+
+    def fail_after_normative_write(*args, **kwargs):
+        real_store(*args, **kwargs)
+        raise RuntimeError("fault after Dolt normative write")
+
+    monkeypatch.setattr(wide, "store_normative_dataset", fail_after_normative_write)
+    variables = [{
+        "ordinal": 1, "source_name": "score", "physical_name": "score",
+        "storage_kind": "numeric", "string_width": None, "label": "",
+        "format": "F8.0", "measure": "scale", "alignment": "right",
+        "display_width": 8, "value_labels": "{}", "missing_ranges": "[]",
+    }]
+
+    with pytest.raises(RuntimeError, match="fault after Dolt normative write"):
+        create_wide_dataset(
+            database_url=database, dataset_id="dolt-cleanup",
+            source_name="fixture.sav", source_format="SAV",
+            rows=[{"score": 1.0}], variables=variables,
+        )
+
+    assert phases[-2:] == ["import", "import cleanup"]
+
+
 def test_nonatomic_failure_after_normative_write_cleans_both_catalogs_and_data(
     tmp_path, monkeypatch,
 ) -> None:

tests/test_conditional_inplace_transform.py

@@ -2,11 +2,11 @@
 
 import sqlite3
 
-from types import SimpleNamespace
 import pytest
 
 import openstatspec
 import openstatspec.sql.inplace_transform as inplace_transform
+from openstatspec.sql.profiles import DOLT
 from openstatspec.sql.wide import create_wide_dataset
 
 
@@ -254,7 +254,7 @@ def test_dolt_mock_applies_exact_program_to_preexisting_target_without_schema_dd
     monkeypatch.setattr(
         inplace_transform,
         "effective_profile",
-        lambda _url, **_kwargs: (SimpleNamespace(name="dolt"), {}),
+        lambda _url, **_kwargs: (DOLT, {}),
     )
     states = iter([
         ("main", "abc123", 0),
@@ -291,7 +291,7 @@ def test_dolt_mock_rejects_create_target_before_schema_mutation(
     monkeypatch.setattr(
         inplace_transform,
         "effective_profile",
-        lambda _url, **_kwargs: (SimpleNamespace(name="dolt"), {}),
+        lambda _url, **_kwargs: (DOLT, {}),
     )
     monkeypatch.setattr(
         inplace_transform,
@@ -334,7 +334,7 @@ def test_dolt_mock_rechecks_clean_state_after_dataset_lock(
     monkeypatch.setattr(
         inplace_transform,
         "effective_profile",
-        lambda _url, **_kwargs: (SimpleNamespace(name="dolt"), {}),
+        lambda _url, **_kwargs: (DOLT, {}),
     )
     states = iter([
         ("main", "abc123", 0),

tests/test_inplace_transform.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import sqlite3
+from dataclasses import replace
 from types import SimpleNamespace
 
 import pytest
@@ -12,6 +13,9 @@
     InPlacePlanSubmission,
     _apply_plan_on_connection,
 )
+from openstatspec.sql.profiles import (
+    DOLT, SQLITE, TargetCapabilityExceededError,
+)
 from openstatspec.sql.wide import create_wide_dataset, validate_wide_dataset
 
 
@@ -102,6 +106,7 @@ def test_plan_applies_to_same_dataset_and_physical_table_without_copy(
             ),
             actor="test-agent",
             database_profile="sqlite",
+            target_profile=inplace_transform.effective_profile(url)[0],
             allow_schema_change=True,
             allow_delete_variable=True,
             dolt_branch=None,
@@ -318,6 +323,44 @@ def test_public_generic_plan_apply_accepts_object_and_mapping(
     )
 
 
+def test_generic_string_width_is_rejected_before_ddl(
+    catalog, monkeypatch,
+) -> None:
+    url, path, dataset_id, table_name = catalog
+    monkeypatch.setattr(
+        inplace_transform,
+        "effective_profile",
+        lambda _url, **_kwargs: (
+            replace(SQLITE, max_text_value_bytes=3),
+            {"server_version": "3.35.0"},
+        ),
+    )
+    plan = openstatspec.TransformationPlan((
+        openstatspec.CreateVariableOperation("note", "string", 4),
+    ))
+
+    with pytest.raises(TargetCapabilityExceededError, match="permits 3"):
+        openstatspec.apply_transformation_plan_in_place(
+            database_url=url,
+            dataset_id=dataset_id,
+            plan=plan,
+            actor="test-agent",
+        )
+
+    connection = sqlite3.connect(path)
+    assert [
+        row[1] for row in connection.execute(f'PRAGMA table_info("{table_name}")')
+    ] == ["__case_ordinal", "score"]
+    assert connection.execute(
+        "select source_name from variable where dataset_id = ?",
+        (dataset_id,),
+    ).fetchall() == [("score",)]
+    assert connection.execute(
+        "select count(*) from transformation_apply"
+    ).fetchone() == (0,)
+    connection.close()
+
+
 def test_generic_plan_is_bound_to_live_schema_before_mutation(catalog) -> None:
     url, path, dataset_id, table_name = catalog
     plan = openstatspec.TransformationPlan((
@@ -481,7 +524,7 @@ def test_delete_is_rejected_when_drop_column_is_unavailable(
         inplace_transform,
         "effective_profile",
         lambda _url, **_kwargs: (
-            SimpleNamespace(name="sqlite"),
+            SQLITE,
             {"server_version": "3.34.0"},
         ),
     )
@@ -521,6 +564,7 @@ def test_nontransactional_ddl_profile_rejects_create_before_mutation(
                 ),
                 actor="test-agent",
                 database_profile="mysql",
+                target_profile=inplace_transform.effective_profile(url)[0],
                 allow_schema_change=False,
                 allow_delete_variable=True,
                 dolt_branch=None,
@@ -546,7 +590,7 @@ def test_public_apply_binds_expected_dolt_branch_and_head(
     monkeypatch.setattr(
         inplace_transform,
         "effective_profile",
-        lambda _url, **_kwargs: (SimpleNamespace(name="dolt"), {}),
+        lambda _url, **_kwargs: (DOLT, {}),
     )
     states = iter([
         ("feature/recode", "abc123", 0),
@@ -577,7 +621,7 @@ def test_public_apply_rejects_dirty_dolt_working_set_before_mutation(
     monkeypatch.setattr(
         inplace_transform,
         "effective_profile",
-        lambda _url, **_kwargs: (SimpleNamespace(name="dolt"), {}),
+        lambda _url, **_kwargs: (DOLT, {}),
     )
     monkeypatch.setattr(
         inplace_transform,
@@ -613,7 +657,7 @@ def test_public_apply_rejects_dolt_context_mismatch_before_mutation(
     monkeypatch.setattr(
         inplace_transform,
         "effective_profile",
-        lambda _url, **_kwargs: (SimpleNamespace(name="dolt"), {}),
+        lambda _url, **_kwargs: (DOLT, {}),
     )
     monkeypatch.setattr(
         inplace_transform, "_dolt_state", lambda _connection: state