Handle transient schema create targets

Author: TonisOrmisson

src/openstatspec/sql/inplace_transform.py

@@ -484,10 +484,6 @@ def _apply_plan_on_connection(
         )
     ]
     preflight(target_profile, output_variables)
-    output_by_name = {
-        variable.name.casefold(): variable
-        for variable in bound.output_schema.variables
-    }
     audit = apply_audit_catalog(MetaData())
     if not inspect(connection).has_table("transformation_apply"):
         raise TransformationError(
@@ -528,11 +524,24 @@ def _apply_plan_on_connection(
             "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)
-        and output_by_name[operation.target.casefold()].storage_kind != "numeric"
-    ]
+    unsupported_targets: list[str] = []
+    for operation_index, operation in enumerate(plan.operations):
+        if not isinstance(operation, (RecodeOperation, AssignOperation)):
+            continue
+        if operation.target_mode != "create":
+            continue
+        prefix = TransformationPlan(
+            plan.operations[:operation_index + 1],
+            contract=plan.contract,
+            input_alias=plan.input_alias,
+        )
+        prefix_output = bind_transformation_plan(prefix, schema).output_schema
+        created = next(
+            variable for variable in prefix_output.variables
+            if variable.name.casefold() == operation.target.casefold()
+        )
+        if created.storage_kind != "numeric":
+            unsupported_targets.append(operation.target)
     if unsupported_targets:
         raise TransformationError(
             "in_place_target_type_unsupported",

tests/test_inplace_transform.py

@@ -374,6 +374,55 @@ def test_delete_then_recreate_same_name_resolves_operations_in_order(catalog) ->
     )["valid"] is True
 
 
+def test_temporary_created_target_can_be_deleted_before_final_schema(catalog) -> None:
+    url, path, dataset_id, table_name = catalog
+
+    openstatspec.apply_spss_in_place(
+        database_url=url,
+        dataset_id=dataset_id,
+        source_text="COMPUTE tmp = score. DELETE VARIABLES tmp.",
+        actor="test-agent",
+    )
+
+    connection = sqlite3.connect(path)
+    assert connection.execute(
+        "select source_name, source_ordinal from variable "
+        "where dataset_id = ? order by source_ordinal",
+        (dataset_id,),
+    ).fetchall() == [("score", 1)]
+    assert [
+        row[1] for row in connection.execute(f'PRAGMA table_info("{table_name}")')
+    ] == ["__case_ordinal", "score"]
+    connection.close()
+
+
+def test_temporary_target_type_is_not_taken_from_same_name_recreation(catalog) -> None:
+    url, path, dataset_id, table_name = catalog
+
+    openstatspec.apply_spss_in_place(
+        database_url=url,
+        dataset_id=dataset_id,
+        source_text=(
+            "COMPUTE tmp = score. DELETE VARIABLES tmp. STRING tmp (A4)."
+        ),
+        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), ("tmp", "string", 4)]
+    assert connection.execute(
+        f'SELECT score, tmp FROM "{table_name}" ORDER BY __case_ordinal'
+    ).fetchall() == [(1.0, ""), (2.0, ""), (3.0, "")]
+    connection.close()
+    assert validate_wide_dataset(
+        database_url=url, dataset_id=dataset_id,
+    )["valid"] is True
+
+
 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).")