Fix sequential schema operation execution

Author: TonisOrmisson

src/openstatspec/sql/inplace_transform.py

@@ -400,6 +400,7 @@ def _apply_plan_on_connection(
     actor: str,
     database_profile: str,
     allow_schema_change: bool,
+    allow_delete_variable: bool,
     dolt_branch: str | None,
     dolt_head: str | None,
     mutation_journal: dict[str, Any] | None = None,
@@ -450,6 +451,14 @@ def _apply_plan_on_connection(
             "schema_change_not_atomic",
             "This database profile has no coherent new-target strategy.",
         )
+    if any(
+        isinstance(operation, DeleteVariableOperation)
+        for operation in plan.operations
+    ) and not allow_delete_variable:
+        raise TransformationError(
+            "delete_variable_not_supported",
+            "This SQLite runtime does not support ALTER TABLE DROP COLUMN.",
+        )
     unsupported_targets = [
         operation.target for operation in create_operations
         if not isinstance(operation, CreateVariableOperation)
@@ -469,7 +478,7 @@ def _apply_plan_on_connection(
     by_name = {str(row["source_name"]).casefold(): row for row in variables}
     used_physical = {str(row["physical_name"]).casefold() for row in variables}
     next_ordinal = max(int(row["source_ordinal"]) for row in variables) + 1
-    target_rows: list[dict[str, Any]] = []
+    target_rows_by_operation: dict[int, dict[str, Any]] = {}
     quote = connection.dialect.identifier_preparer.quote
     qualified_table = connection.dialect.identifier_preparer.format_table(relation)
     numeric_type = (
@@ -544,24 +553,27 @@ def _apply_plan_on_connection(
             "variable_label": None,
         }
         next_ordinal += 1
-        target_rows.append(target_row)
+        target_rows_by_operation[id(operation)] = target_row
         if mutation_journal is not None:
             mutation_journal["target_rows"].append(dict(target_row))
     if create_operations:
         _failure_boundary("schema")
 
-    for target_row in target_rows:
-        connection.execute(insert(core.variable).values(**target_row))
-        variables.append(target_row)
-        by_name[str(target_row["source_name"]).casefold()] = target_row
-    if target_rows:
-        _failure_boundary("catalog")
+    if create_operations:
         relation = Table(
             table_name, MetaData(), schema=dataset.get("physical_table_schema"),
             autoload_with=connection,
         )
 
     for operation in plan.operations:
+        created_target = target_rows_by_operation.get(id(operation))
+        if created_target is not None:
+            connection.execute(insert(core.variable).values(**created_target))
+            variables.append(created_target)
+            by_name[str(created_target["source_name"]).casefold()] = created_target
+            _failure_boundary("catalog")
+        if isinstance(operation, CreateVariableOperation):
+            continue
         if isinstance(operation, RecodeOperation):
             source_variable = by_name[operation.source.casefold()]
             target_variable = by_name[operation.target.casefold()]
@@ -856,9 +868,16 @@ def _run_in_place_submission(
         raise TransformationError(
             "actor_required", "A non-empty actor identity is mandatory.",
         )
-    profile, _active = effective_profile(
+    profile, active = effective_profile(
         database_url, dolt_conformance_source=dolt_conformance_source,
     )
+    allow_delete_variable = True
+    if profile.name == "sqlite":
+        sqlite_version_parts = tuple(
+            int(part) for part in str(active["server_version"]).split(".")
+        )
+        sqlite_version = (*sqlite_version_parts, 0, 0)[:3]
+        allow_delete_variable = sqlite_version >= (3, 35, 0)
     engine = create_engine(database_url)
     journal: dict[str, Any] = {}
     branch: str | None = None
@@ -905,6 +924,7 @@ def _run_in_place_submission(
                     actor=actor,
                     database_profile=profile.name,
                     allow_schema_change=profile.name in {"sqlite", "postgresql"},
+                    allow_delete_variable=allow_delete_variable,
                     dolt_branch=branch,
                     dolt_head=head,
                     mutation_journal=journal,

tests/test_inplace_transform.py

@@ -103,6 +103,7 @@ def test_plan_applies_to_same_dataset_and_physical_table_without_copy(
             actor="test-agent",
             database_profile="sqlite",
             allow_schema_change=True,
+            allow_delete_variable=True,
             dolt_branch=None,
             dolt_head=None,
         )
@@ -153,6 +154,56 @@ def test_plan_applies_to_same_dataset_and_physical_table_without_copy(
     )
 
 
+def test_string_declaration_creates_column_and_catalog_variable(catalog) -> None:
+    url, path, dataset_id, table_name = catalog
+
+    openstatspec.apply_spss_in_place(
+        database_url=url,
+        dataset_id=dataset_id,
+        source_text="STRING note (A8).",
+        actor="test-agent",
+    )
+
+    connection = sqlite3.connect(path)
+    assert connection.execute(
+        "select source_name, storage_kind, declared_string_width from variable "
+        "where dataset_id = ? order by source_ordinal",
+        (dataset_id,),
+    ).fetchall() == [("score", "numeric", None), ("note", "string", 8)]
+    assert connection.execute(
+        f'SELECT note FROM "{table_name}" ORDER BY __case_ordinal'
+    ).fetchall() == [(None,), (None,), (None,)]
+    connection.close()
+
+
+def test_delete_then_recreate_same_name_resolves_operations_in_order(catalog) -> None:
+    url, path, dataset_id, table_name = catalog
+
+    openstatspec.apply_spss_in_place(
+        database_url=url,
+        dataset_id=dataset_id,
+        source_text=(
+            "COMPUTE other = score. DELETE VARIABLES score. "
+            "RECODE other (2 = 9) (ELSE = COPY) INTO score."
+        ),
+        actor="test-agent",
+    )
+
+    connection = sqlite3.connect(path)
+    variables = connection.execute(
+        "select source_name, physical_name from variable where dataset_id = ? "
+        "order by source_ordinal",
+        (dataset_id,),
+    ).fetchall()
+    assert [row[0] for row in variables] == ["other", "score"]
+    physical = dict(variables)
+    assert connection.execute(
+        f'SELECT "{physical["other"]}", "{physical["score"]}" '
+        f'FROM "{table_name}" ORDER BY __case_ordinal'
+    ).fetchall() == [(1.0, 1.0), (2.0, 9.0), (3.0, 3.0)]
+    connection.close()
+
+
 def test_public_apply_supports_non_dolt_without_building_undo(catalog) -> None:
     url, path, dataset_id, table_name = catalog
     plan = _plan("RECODE score (1 = 0).")
@@ -364,6 +415,38 @@ def test_missing_audit_schema_fails_before_mutation(catalog) -> None:
     ).fetchall() == [(1.0,), (2.0,), (3.0,)]
 
 
+def test_delete_is_rejected_when_drop_column_is_unavailable(
+    catalog, monkeypatch,
+) -> None:
+    url, path, dataset_id, table_name = catalog
+    monkeypatch.setattr(
+        inplace_transform,
+        "effective_profile",
+        lambda _url, **_kwargs: (
+            SimpleNamespace(name="sqlite"),
+            {"server_version": "3.34.0"},
+        ),
+    )
+
+    with pytest.raises(openstatspec.TransformationError) as caught:
+        openstatspec.apply_spss_in_place(
+            database_url=url,
+            dataset_id=dataset_id,
+            source_text="COMPUTE other = score. DELETE VARIABLES score.",
+            actor="test-agent",
+        )
+
+    assert caught.value.code == "delete_variable_not_supported"
+    connection = sqlite3.connect(path)
+    assert connection.execute(
+        f'SELECT score FROM "{table_name}" ORDER BY __case_ordinal'
+    ).fetchall() == [(1.0,), (2.0,), (3.0,)]
+    assert [
+        row[1] for row in connection.execute(f'PRAGMA table_info("{table_name}")')
+    ] == ["__case_ordinal", "score"]
+    connection.close()
+
+
 def test_nontransactional_ddl_profile_rejects_create_before_mutation(
     catalog,
 ) -> None:
@@ -381,6 +464,7 @@ def test_nontransactional_ddl_profile_rejects_create_before_mutation(
                 actor="test-agent",
                 database_profile="mysql",
                 allow_schema_change=False,
+                allow_delete_variable=True,
                 dolt_branch=None,
                 dolt_head=None,
             )
@@ -692,7 +776,10 @@ def test_transactional_ddl_rollback_skips_unlocked_compensation(
     monkeypatch.setattr(
         inplace_transform,
         "effective_profile",
-        lambda _url, **_kwargs: (SimpleNamespace(name=profile_name), {}),
+        lambda _url, **_kwargs: (
+            SimpleNamespace(name=profile_name),
+            {"server_version": "3.35.0"},
+        ),
     )
 
     def fail_after_schema_change(