Add in-place SPSS schema transformations

Add in-place SPSS schema transformations

Author: TonisOrmisson

src/openstatspec/frontends/spss/binding.py

@@ -216,6 +216,41 @@ def _result_type(
     return result.value.type
 
 
+def _validate_recode_string_width(
+    result: RecodeResult,
+    source: VariableDefinition,
+    target: VariableDefinition,
+    span: SourceSpan,
+) -> None:
+    if target.storage_kind != "string" or target.declared_string_width is None:
+        return
+    if result.kind == "literal":
+        assert result.value is not None
+        if result.value.type != "string":
+            return
+        assert isinstance(result.value.value, str)
+        if len(result.value.value.encode("utf-8")) > target.declared_string_width:
+            raise frontend_error(
+                "string_width_exceeded",
+                "A RECODE string literal exceeds the target's declared string width.",
+                span=span,
+                variable=target.name,
+                declared_string_width=target.declared_string_width,
+            )
+        return
+    if result.kind == "copy" and source.storage_kind == "string":
+        source_width = source.declared_string_width
+        if source_width is None or source_width > target.declared_string_width:
+            raise frontend_error(
+                "string_width_exceeded",
+                "A RECODE COPY result can exceed the target's declared string width.",
+                span=span,
+                source=source.name,
+                variable=target.name,
+                declared_string_width=target.declared_string_width,
+            )
+
+
 def _bind_recode(
     command: RecodeCommandSyntax, variables: list[VariableDefinition],
 ) -> tuple[list[RecodeOperation], list[SourceSpan]]:
@@ -252,13 +287,17 @@ def _bind_recode(
         else_result: RecodeResult | None = None
         for clause in command.clauses:
             result = _result(clause.result, source)
+            if target_mode == "replace":
+                _validate_recode_string_width(result, source, source, clause.result.span)
             if clause.match.kind == "else":
                 else_result = result
                 continue
             rules.append(RecodeRule(_match(clause.match, source), result))
         unmatched = else_result or RecodeResult(
             "system_missing" if target_mode == "create" else "copy"
         )
+        if target_mode == "replace" and else_result is None:
+            _validate_recode_string_width(unmatched, source, source, command.span)
         result_types = {
             _result_type(result, source)
             for result in [*(rule.result for rule in rules), unmatched]
@@ -339,13 +378,6 @@ def bind_spss_syntax(
                 index, variable = _resolve(
                     variables, variable_token.text, variable_token.span,
                 )
-                if len(variables) == 1:
-                    raise frontend_error(
-                        "cannot_delete_last_variable",
-                        "DELETE VARIABLES cannot remove the final dataset variable.",
-                        span=variable_token.span,
-                        variable=variable.name,
-                    )
                 operations.append(DeleteVariableOperation(variable.name))
                 spans.append(command.span)
                 del variables[index]

src/openstatspec/frontends/spss/syntax.py

@@ -563,10 +563,24 @@ def execute(self, start: Token) -> ExecuteCommandSyntax:
         end = self.expects("period", "Expected '.' after EXECUTE.")
         return ExecuteCommandSyntax(_joined_span(start.span, end.span))
 
+    @staticmethod
+    def reject_to_range(variables: tuple[Token, ...], command: str) -> None:
+        range_token = next(
+            (variable for variable in variables if variable.text.casefold() == "to"),
+            None,
+        )
+        if range_token is not None:
+            raise frontend_error(
+                "unsupported_spss_feature",
+                f"{command} variable ranges using TO are not supported.",
+                span=range_token.span,
+            )
+
     def string(self, start: Token) -> StringCommandSyntax:
         variables = self.variable_list(
             stop_kinds=frozenset({"left_paren", "period", "eof"}),
         )
+        self.reject_to_range(variables, "STRING")
         self.expects("left_paren", "Expected '(' before a STRING width.")
         width_token = self.expects(
             "identifier", "STRING requires a width such as A20.",
@@ -595,6 +609,7 @@ def string(self, start: Token) -> StringCommandSyntax:
     def delete_variables(self, start: Token) -> DeleteVariablesCommandSyntax:
         self.expects_keyword("VARIABLES")
         variables = self.variable_list(stop_kinds=frozenset({"period", "eof"}))
+        self.reject_to_range(variables, "DELETE VARIABLES")
         end = self.expects("period", "Expected '.' after DELETE VARIABLES.")
         return DeleteVariablesCommandSyntax(
             variables, _joined_span(start.span, end.span),

src/openstatspec/sql/inplace_transform.py

@@ -389,22 +389,30 @@ def _delete_variable_metadata(
     ))
 
 
-def _compact_variable_ordinals(
+def _assert_postgresql_column_slot_available(
     connection: Any,
     *,
-    core: Any,
-    variables: list[dict[str, Any]],
+    relation: Table,
+    target_profile: SqlProfile,
 ) -> None:
-    """Keep normative variable source order contiguous after deletion."""
-    for source_ordinal, variable in enumerate(variables, start=1):
-        if int(variable["source_ordinal"]) == source_ordinal:
-            continue
-        connection.execute(
-            update(core.variable)
-            .where(core.variable.c.variable_id == variable["variable_id"])
-            .values(source_ordinal=source_ordinal)
+    """Reject PostgreSQL creates when dropped columns exhaust attribute slots."""
+    if connection.dialect.name != "postgresql":
+        return
+    qualified_relation = connection.dialect.identifier_preparer.format_table(relation)
+    attribute_slots = connection.execute(text(
+        "SELECT COUNT(*) FROM pg_attribute "
+        "WHERE attrelid = to_regclass(:relation_name) AND attnum > 0"
+    ), {"relation_name": qualified_relation}).scalar_one()
+    if not isinstance(attribute_slots, int) or attribute_slots < 1:
+        raise TransformationError(
+            "physical_table_missing",
+            "The target dataset's physical wide table does not exist.",
+        )
+    if attribute_slots >= target_profile.max_source_variables + 1:
+        raise TransformationError(
+            "source_variable_limit",
+            "PostgreSQL physical column slots are exhausted for this dataset table.",
         )
-        variable["source_ordinal"] = source_ordinal
 
 
 def _failure_boundary(_name: str) -> None:
@@ -555,6 +563,9 @@ 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_source_ordinal = max(
+        (int(row["source_ordinal"]) for row in variables), default=0,
+    ) + 1
     quote = connection.dialect.identifier_preparer.quote
     qualified_table = connection.dialect.identifier_preparer.format_table(relation)
     numeric_type = (
@@ -603,6 +614,9 @@ def _apply_plan_on_connection(
             )
         )
         if creates_target:
+            _assert_postgresql_column_slot_available(
+                connection, relation=relation, target_profile=target_profile,
+            )
             target_name = (
                 operation.variable
                 if isinstance(operation, CreateVariableOperation)
@@ -631,7 +645,7 @@ def _apply_plan_on_connection(
             created_target = {
                 "variable_id": str(uuid4()),
                 "dataset_id": dataset_id,
-                "source_ordinal": len(variables) + 1,
+                "source_ordinal": next_source_ordinal,
                 "source_name": target_name,
                 "physical_name": target_physical,
                 "storage_kind": storage_kind,
@@ -648,6 +662,7 @@ def _apply_plan_on_connection(
             )
             connection.execute(insert(core.variable).values(**created_target))
             variables.append(created_target)
+            next_source_ordinal += 1
             by_name[str(created_target["source_name"]).casefold()] = created_target
             _failure_boundary("catalog")
         if isinstance(operation, CreateVariableOperation):
@@ -725,36 +740,10 @@ def _apply_plan_on_connection(
             )
             _delete_variable_metadata(connection, core=core, variable=variable)
             by_name.pop(operation.variable.casefold(), None)
-            used_physical.discard(str(variable["physical_name"]).casefold())
             variables = [
                 row for row in variables
                 if row["variable_id"] != variable["variable_id"]
             ]
-            _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/sql/profiles.py

@@ -124,6 +124,7 @@ def preflight(
     variables_or_count: int | Iterable[Mapping[str, Any]],
     *,
     rows: Iterable[Mapping[str, Any]] | None = None,
+    require_canonical_mapping: bool = True,
 ) -> None:
     """Validate strict target capabilities before any source dataset is created."""
     variables = None if isinstance(variables_or_count, int) else list(variables_or_count)
@@ -143,6 +144,7 @@ def preflight(
 
     used = {"__case_ordinal"}
     source_names: set[str] = set()
+    physical_names = {"__case_ordinal"}
     for expected_ordinal, variable in enumerate(variables, start=1):
         source_name = variable.get("source_name")
         if not isinstance(source_name, str) or not source_name or source_name in source_names:
@@ -151,18 +153,36 @@ def preflight(
                 source_name=source_name,
             )
         source_names.add(source_name)
-        expected_name = _physical_name(source_name, used)
         actual_name = variable.get("physical_name")
-        if variable.get("ordinal") != expected_ordinal or actual_name != expected_name:
+        if not isinstance(actual_name, str) or not actual_name:
             raise _exceeded(
                 "physical_identifier_mapping_invalid",
-                f"{source_name!r} must map deterministically to {expected_name!r} in source order.",
-                source_name=source_name, expected_physical_name=expected_name,
-                actual_physical_name=actual_name,
+                f"{source_name!r} has no physical variable identifier.",
+                source_name=source_name, actual_physical_name=actual_name,
+            )
+        if actual_name.casefold() in physical_names:
+            raise _exceeded(
+                "physical_identifier_collision",
+                "physical variable identifiers must be unique.",
+                source_name=source_name, physical_name=actual_name,
+            )
+        physical_names.add(actual_name.casefold())
+        if require_canonical_mapping:
+            expected_name = _physical_name(source_name, used)
+            if variable.get("ordinal") != expected_ordinal or actual_name != expected_name:
+                raise _exceeded(
+                    "physical_identifier_mapping_invalid",
+                    f"{source_name!r} must map deterministically to {expected_name!r} in source order.",
+                    source_name=source_name, expected_physical_name=expected_name,
+                    actual_physical_name=actual_name,
+                )
+            preflight_identifier(
+                profile, expected_name, role="physical variable identifier",
+            )
+        else:
+            preflight_identifier(
+                profile, actual_name, role="physical variable identifier",
             )
-        preflight_identifier(
-            profile, expected_name, role="physical variable identifier",
-        )
         if variable.get("storage_kind") == "string":
             declared_width = variable.get("string_width")
             if declared_width is not None and (

src/openstatspec/sql/wide.py

@@ -1239,7 +1239,9 @@ def read_wide_dataset(
         )
         for row in response_sets
     }, ensure_ascii=False)
-    preflight(profile, variables, rows=rows)
+    preflight(
+        profile, variables, rows=rows, require_canonical_mapping=False,
+    )
     return dataset, variables, rows
 
 
@@ -1705,7 +1707,9 @@ def validate_wide_dataset(
         database_url=database_url, dataset_id=dataset_id, profile=profile,
         dolt_conformance_source=dolt_conformance_source,
     )
-    preflight(profile, variables, rows=rows)
+    preflight(
+        profile, variables, rows=rows, require_canonical_mapping=False,
+    )
     validate_spss_catalog(
         variables,
         case_weight_variable=dataset.get("case_weight_variable"),

src/openstatspec/transform/validation.py

@@ -100,6 +100,38 @@ def _result_type(
     return result.value.type
 
 
+def _validate_recode_string_width(
+    result: RecodeResult,
+    source: VariableDefinition,
+    target: VariableDefinition,
+) -> None:
+    if target.storage_kind != "string" or target.declared_string_width is None:
+        return
+    if result.kind == "literal":
+        assert result.value is not None
+        if result.value.type != "string":
+            return
+        assert isinstance(result.value.value, str)
+        if len(result.value.value.encode("utf-8")) > target.declared_string_width:
+            raise frontend_error(
+                "string_width_exceeded",
+                "A RECODE string literal exceeds the target's declared string width.",
+                variable=target.name,
+                declared_string_width=target.declared_string_width,
+            )
+        return
+    if result.kind == "copy" and source.storage_kind == "string":
+        source_width = source.declared_string_width
+        if source_width is None or source_width > target.declared_string_width:
+            raise frontend_error(
+                "string_width_exceeded",
+                "A RECODE COPY result can exceed the target's declared string width.",
+                source=source.name,
+                variable=target.name,
+                declared_string_width=target.declared_string_width,
+            )
+
+
 def _bind_recode(
     operation: RecodeOperation, variables: list[VariableDefinition]
 ) -> None:
@@ -122,6 +154,10 @@ def _bind_recode(
             )
     for rule in operation.rules:
         _validate_match(rule.match, source)
+        if operation.target_mode == "replace":
+            _validate_recode_string_width(rule.result, source, source)
+    if operation.target_mode == "replace":
+        _validate_recode_string_width(operation.unmatched, source, source)
     result_types = {
         _result_type(result, source)
         for result in [
@@ -282,12 +318,16 @@ def bind_transformation_plan(
     if not isinstance(schema, VariableSchema):
         raise TypeError("schema must be a VariableSchema.")
     variables = list(schema.variables)
-    for operation in plan.operations:
+    for operation_index, operation in enumerate(plan.operations):
+        later_create = any(
+            isinstance(later_operation, CreateVariableOperation)
+            for later_operation in plan.operations[operation_index + 1:]
+        )
         if isinstance(operation, CreateVariableOperation):
             _bind_create(operation, variables)
             continue
         if isinstance(operation, DeleteVariableOperation):
-            _bind_delete(operation, variables)
+            _bind_delete(operation, variables, allow_empty=later_create)
             continue
         if isinstance(operation, RecodeOperation):
             _bind_recode(operation, variables)
@@ -367,9 +407,11 @@ def _bind_create(
 def _bind_delete(
     operation: DeleteVariableOperation,
     variables: list[VariableDefinition],
+    *,
+    allow_empty: bool,
 ) -> None:
     index, variable = _resolve(variables, operation.variable)
-    if len(variables) == 1:
+    if len(variables) == 1 and not allow_empty:
         raise frontend_error(
             "cannot_delete_last_variable",
             "A dataset must retain at least one variable.",