diff --git a/src/openstatspec/frontends/spss/binding.py b/src/openstatspec/frontends/spss/binding.py index cd37d12..2913115 100644 --- a/src/openstatspec/frontends/spss/binding.py +++ b/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,6 +287,8 @@ 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 @@ -259,6 +296,8 @@ def _bind_recode( 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] diff --git a/src/openstatspec/frontends/spss/syntax.py b/src/openstatspec/frontends/spss/syntax.py index ed78f93..5aadd44 100644 --- a/src/openstatspec/frontends/spss/syntax.py +++ b/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), diff --git a/src/openstatspec/sql/inplace_transform.py b/src/openstatspec/sql/inplace_transform.py index 06ffecc..0f458be 100644 --- a/src/openstatspec/sql/inplace_transform.py +++ b/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, diff --git a/src/openstatspec/sql/profiles.py b/src/openstatspec/sql/profiles.py index 4ddcd57..3c3c8e8 100644 --- a/src/openstatspec/sql/profiles.py +++ b/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 ( diff --git a/src/openstatspec/sql/wide.py b/src/openstatspec/sql/wide.py index bd8488b..da488df 100644 --- a/src/openstatspec/sql/wide.py +++ b/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"), diff --git a/src/openstatspec/transform/validation.py b/src/openstatspec/transform/validation.py index d3021c0..36febc6 100644 --- a/src/openstatspec/transform/validation.py +++ b/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.", diff --git a/tests/test_inplace_transform.py b/tests/test_inplace_transform.py index cd97771..3f12051 100644 --- a/tests/test_inplace_transform.py +++ b/tests/test_inplace_transform.py @@ -191,7 +191,7 @@ def test_string_declaration_creates_column_and_catalog_variable(catalog) -> None )["valid"] is True -def test_delete_recanonicalizes_surviving_collision_columns(tmp_path) -> None: +def test_delete_preserves_surviving_physical_bindings_and_ordinals(tmp_path) -> None: path = tmp_path / "collision-delete.sqlite" url = f"sqlite:///{path}" openstatspec.initialize_catalog(database_url=url) @@ -242,9 +242,9 @@ def test_delete_recanonicalizes_surviving_collision_columns(tmp_path) -> None: "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)] + ).fetchall() == [("a_b", "a_b_2", 2), ("keep", "keep", 3)] assert connection.execute( - "select a_b, keep from data_collision_source order by __case_ordinal" + "select a_b_2, keep from data_collision_source order by __case_ordinal" ).fetchall() == [(2.0, 3.0), (5.0, 6.0)] connection.close() assert validate_wide_dataset( @@ -361,7 +361,7 @@ def test_delete_then_recreate_same_name_resolves_operations_in_order(catalog) -> (dataset_id,), ).fetchall() assert [(row[0], row[2]) for row in variables] == [ - ("other", 1), ("score", 2), + ("other", 2), ("score", 3), ] physical = {row[0]: row[1] for row in variables} assert connection.execute( @@ -414,8 +414,13 @@ def test_temporary_target_type_is_not_taken_from_same_name_recreation(catalog) - "from variable where dataset_id = ? order by source_ordinal", (dataset_id,), ).fetchall() == [("score", "numeric", None), ("tmp", "string", 4)] + physical = dict(connection.execute( + "select source_name, physical_name from variable where dataset_id = ?", + (dataset_id,), + ).fetchall()) assert connection.execute( - f'SELECT score, tmp FROM "{table_name}" ORDER BY __case_ordinal' + f'SELECT "{physical["score"]}", "{physical["tmp"]}" ' + f'FROM "{table_name}" ORDER BY __case_ordinal' ).fetchall() == [(1.0, ""), (2.0, ""), (3.0, "")] connection.close() assert validate_wide_dataset( diff --git a/tests/test_sql_services.py b/tests/test_sql_services.py index 1abbc2f..a442c47 100755 --- a/tests/test_sql_services.py +++ b/tests/test_sql_services.py @@ -6,12 +6,17 @@ import pandas as pd import pyspssio import pytest -from sqlalchemy import create_engine, inspect as inspect_database, text +from sqlalchemy import MetaData, create_engine, inspect as inspect_database, text from sqlalchemy.exc import DBAPIError import openstatspec from openstatspec.core import UnsupportedOperationError from openstatspec.sql.dolt_conformance import DoltConformanceSource +from openstatspec.sql.normative import ( + catalog as normative_catalog, + delete_dataset_representation, +) +from openstatspec.sql.wide import create_wide_dataset from conformance import compare_sav_semantics, write_supported_semantics_fixture @@ -206,4 +211,89 @@ def test_live_dolt_candidate_limit_probe_smoke() -> None: connection.execute(text( f"DROP TABLE IF EXISTS {quote(table_name)}" )) - engine.dispose() \ No newline at end of file + engine.dispose() + +def test_live_postgresql_in_place_create_rejects_exhausted_physical_slots() -> None: + database_url = os.environ.get("OPENSTATSPEC_POSTGRES_URL") + if not database_url: + pytest.skip("OPENSTATSPEC_POSTGRES_URL is not configured") + + token = uuid4().hex[:12] + dataset_name = f"inplace_slots_{token}" + base_variable = { + "storage_kind": "numeric", + "string_width": None, + "label": "", + "format": "F8.0", + "print_format": "[5, 8, 0]", + "write_format": "[5, 8, 0]", + "measure": "scale", + "role": "input", + "alignment": "right", + "display_width": 8, + "attributes": "{}", + "compat_name": None, + "value_labels": "{}", + "missing_ranges": "[]", + } + variables = [ + { + **base_variable, + "ordinal": ordinal, + "source_name": f"v{ordinal}", + "physical_name": f"v{ordinal}", + } + for ordinal in range(1, 1_600) + ] + created = None + engine = create_engine(database_url) + try: + openstatspec.initialize_catalog(database_url=database_url) + created = create_wide_dataset( + database_url=database_url, + dataset_id=dataset_name, + source_name="slot-limit.sav", + source_format="SAV", + source_sha256="0" * 64, + rows=[], + variables=variables, + ) + openstatspec.install_in_place_transformation_schema( + database_url=database_url, + ) + plan = openstatspec.TransformationPlan( + ( + openstatspec.DeleteVariableOperation("v1599"), + openstatspec.CreateVariableOperation("replacement", "numeric"), + ), + contract="openstatspec-transformation-plan-v0.3", + ) + + with pytest.raises(openstatspec.TransformationError) as caught: + openstatspec.apply_transformation_plan_in_place( + database_url=database_url, + dataset_id=created["dataset_id"], + plan=plan, + actor="service-test", + ) + + assert caught.value.code == "source_variable_limit" + with engine.connect() as connection: + assert connection.execute(text( + "SELECT COUNT(*) FROM variable WHERE dataset_id = :dataset_id" + ), {"dataset_id": created["dataset_id"]}).scalar_one() == 1_599 + assert connection.execute(text( + "SELECT COUNT(*) FROM pg_attribute " + "WHERE attrelid = to_regclass(:relation_name) AND attnum > 0" + ), {"relation_name": created["data_table"]}).scalar_one() == 1_600 + finally: + if created is not None: + with engine.begin() as connection: + delete_dataset_representation( + connection, normative_catalog(MetaData()), created["dataset_id"], + ) + quote = connection.dialect.identifier_preparer.quote + connection.execute(text( + f"DROP TABLE IF EXISTS {quote(created['data_table'])}" + )) + engine.dispose() diff --git a/tests/test_transform_frontend.py b/tests/test_transform_frontend.py index 0caeee0..29a1e8f 100644 --- a/tests/test_transform_frontend.py +++ b/tests/test_transform_frontend.py @@ -15,6 +15,7 @@ ) from openstatspec.transform import ( CreateVariableOperation, + DeleteVariableOperation, RecodeMatch, RecodeOperation, RecodeResult, @@ -611,6 +612,64 @@ def test_custom_nonempty_input_alias_is_canonical() -> None: assert plan.input_alias == "survey" +@pytest.mark.parametrize( + ("source", "schema"), + [ + ( + "STRING first TO last (A8).", + _schema( + VariableDefinition("first", "numeric"), + VariableDefinition("last", "numeric"), + ), + ), + ( + "DELETE VARIABLES first TO last.", + _schema( + VariableDefinition("first", "numeric"), + VariableDefinition("last", "numeric"), + ), + ), + ], +) +def test_string_and_delete_variables_reject_unsupported_to_ranges( + source: str, schema: VariableSchema, +) -> None: + assert _error(source, schema).code == "unsupported_spss_feature" + + +def test_delete_final_variable_requires_a_later_explicit_create() -> None: + schema = _schema(VariableDefinition("only", "numeric")) + + assert _error("DELETE VARIABLES only.", schema).code == "cannot_delete_last_variable" + + bound = _compile("DELETE VARIABLES only. STRING replacement (A3).", schema) + assert bound.plan.operations == ( + DeleteVariableOperation("only"), + CreateVariableOperation("replacement", "string", 3), + ) + assert bound.output_schema.variables == ( + VariableDefinition("replacement", "string", declared_string_width=3), + ) + + +def test_recode_string_literals_respect_declared_width() -> None: + schema = _schema(VariableDefinition( + "note", "string", declared_string_width=3, + )) + + assert _error("RECODE note ('a' = 'abcd').", schema).code == "string_width_exceeded" + assert _compile("RECODE note ('a' = '\u00e4b').", schema).plan.operations + + +def test_recode_copy_preserves_source_declared_width() -> None: + schema = _schema(VariableDefinition( + "note", "string", declared_string_width=3, + )) + + bound = _compile("RECODE note ('yes' = COPY).", schema) + assert bound.output_schema == schema + + def test_generic_plan_binding_validates_sequential_schema_state() -> None: plan = TransformationPlan(( RecodeOperation(