diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 16531e4..e31ca4e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,7 +20,7 @@ jobs: uses: actions/checkout@v7 with: &specification-checkout repository: OpenStatSpec/specification - ref: d287c2cde9ade71f04e27dd012caec876901aed5 + ref: 79339ec3d8f8aa81789b7e85f6b8afa6f1374e50 path: openstatspec-specification - name: Checkout required SPSS engine uses: actions/checkout@v7 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index adaadd4..b53eb02 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -47,7 +47,7 @@ jobs: uses: actions/checkout@v7 with: repository: OpenStatSpec/specification - ref: d287c2cde9ade71f04e27dd012caec876901aed5 + ref: 79339ec3d8f8aa81789b7e85f6b8afa6f1374e50 path: openstatspec-specification - name: Checkout required SPSS engine uses: actions/checkout@v7 diff --git a/CHANGELOG.md b/CHANGELOG.md index fe2f2d5..8ee3cfe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,8 +4,21 @@ All notable changes to this reference implementation are documented here. ## Unreleased +### Added + +- Added a bounded SPSS-like `RECODE`, `VARIABLE LABELS`, and `VALUE LABELS` + parser with stable diagnostics and conformance fixtures. +- Added canonical Transformation Plan serialization, RFC 8785 hashes, schema + binding, deterministic typed operations, and an agent-facing CLI/API. +- Added product-neutral in-place execution with same dataset/table identity, + direct data and metadata mutation, and compact operation audit. Dolt adds + expected branch/HEAD and clean-working-set checks. + ### Changed +- Version history, diff, rollback, restoration, and commit remain database + responsibilities; the transformer creates no derived/copy/snapshot/recovery + layer and performs no `DOLT_COMMIT`. - SQL server support now distinguishes conservative family claims from exact CI evidence: PostgreSQL 17.x/18.x at 17.10/18.4, MySQL 8.4.x/9.7.x at 8.4.11/9.7.2, and MariaDB 11.4.x/11.8.x/12.3.x at @@ -19,8 +32,8 @@ All notable changes to this reference implementation are documented here. ### Specification basis - CI, release validation, and machine-readable capabilities use OpenStatSpec - specification release `v0.1.0` at exact commit - `d287c2cde9ade71f04e27dd012caec876901aed5`. + specification release `v0.2.0` at exact commit + `79339ec3d8f8aa81789b7e85f6b8afa6f1374e50`. ## 0.2.0 — 2026-07-30 diff --git a/README.md b/README.md index e2db280..229518b 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,46 @@ See [the SQL transformation workflow](docs/sql-transformation-workflow.md) for Python and CLI examples, migration behavior, hashing, atomicity, and the exact implemented capability boundary. +## SPSS-like transformation frontend + +The transformation frontend accepts `RECODE`, `VARIABLE LABELS`, and +`VALUE LABELS`, lowers them to a canonical OpenStatSpec Transformation Plan, +and mutates the same logical dataset and physical wide table. SQLite, +PostgreSQL, MySQL, MariaDB, and Dolt connections are allowed. It creates no +derived dataset, copied table, snapshot, or OpenStatSpec rollback/version layer. + +Install the compact operation-audit relation separately, then apply syntax: + +```python +from openstatspec import ( + apply_spss_in_place, + install_in_place_transformation_schema, +) + +database_url = "postgresql+psycopg://user:password@host/database" +install_in_place_transformation_schema(database_url=database_url) +result = apply_spss_in_place( + database_url=database_url, + dataset_id="...", + actor="agent@example.org", + source_text=""" + RECODE age (18 THRU 34 = 1) (35 THRU 64 = 2) INTO age_group. + VARIABLE LABELS age_group 'Age group'. + VALUE LABELS age_group 1 '18-34' 2 '35-64'. + """, +) +``` + +Existing-target recodes and metadata mutations use direct DML. SQLite and +PostgreSQL may also add a target column in the same native transaction. MySQL, +MariaDB, and Dolt reject such schema-changing plans before mutation because +their implicit-commit DDL could otherwise leave a partial apply. On Dolt, the +caller additionally supplies expected branch and HEAD identities, and the +working set must be clean. The transformer never calls `DOLT_COMMIT`. + +See [in-place transformations](docs/in-place-transformation.md) for the full +execution boundary and CLI form. + ## Current support status The adapter requires `openstatspec-pyspssio==0.5.1.post2` as its sole SPSS diff --git a/docs/in-place-transformation.md b/docs/in-place-transformation.md new file mode 100644 index 0000000..22d0a01 --- /dev/null +++ b/docs/in-place-transformation.md @@ -0,0 +1,41 @@ +# In-place SPSS-like transformation + +`openstatspec.apply_spss_in_place` is the public execution path for the +SPSS-like frontend. It accepts `RECODE`, `VARIABLE LABELS`, and `VALUE LABELS`, +binds them to the existing core dataset, and applies the canonical plan to that +same SQL wide table and metadata catalog. + +The caller supplies a supported SQL URL, the existing normative `dataset_id`, +and a non-empty actor identity. For Dolt, the caller also supplies the expected +active branch and current `HEAD` hash. + +Install the compact audit relation once with +`openstatspec.install_in_place_transformation_schema(database_url=...)` before +the first apply. Apply never creates schema-management objects itself. + +The adapter uses the engine's ordinary transaction behavior. On Dolt it first +checks branch and HEAD and requires `dolt_status` to be empty. Existing-target +recodes are one direct `UPDATE`. SQLite and PostgreSQL can add a new numeric +`INTO` target to the same table. MySQL, MariaDB, and Dolt require that target +column and variable metadata to exist before apply because their DDL can commit +independently of the following data and metadata changes. Label commands +update/replace the same dataset's normative and compatibility metadata rows. + +One compact `transformation_apply` row records operation identity, canonical +plan/source hashes, actor, status, timestamps, and the observed Dolt branch and +HEAD. It contains no row values and points to no copied table. + +The adapter does not create `derived_dataset` rows, persistent output tables, +full-table copies, staging datasets, snapshots, rollback tables, retirement +records, or a recovery/version catalog. It does not call `DOLT_COMMIT`, change +branches, merge, reset, or tag. After success, the caller reviews `dolt diff` +and independently decides whether to commit or restore the working set. + +The local SQLite tests exercise the public mutation path and assert that +dataset/table counts and identities do not change. Live PostgreSQL/MySQL/MariaDB and +exact-version Dolt service evidence remains required before release execution +claims for those engines. + +```text +openstatspec apply-spss --database-url mysql+pymysql://user:password@host/database --dataset-id ... --actor agent@example.org --expected-branch feature/recode --expected-head ... --syntax-file transform.sps +``` diff --git a/docs/release-readiness.md b/docs/release-readiness.md index 961c33e..13f9224 100644 --- a/docs/release-readiness.md +++ b/docs/release-readiness.md @@ -69,8 +69,8 @@ machine-readable loss report with the export result. environment. 5. Confirm `openstatspec capabilities` reflects the intended support boundary. 6. Confirm the release tag matches the package version and that CI, release - fixtures, and capabilities use OpenStatSpec specification release `v0.1.0` - at exact commit `d287c2cde9ade71f04e27dd012caec876901aed5`. + fixtures, and capabilities use OpenStatSpec specification release `v0.2.0` + at exact commit `79339ec3d8f8aa81789b7e85f6b8afa6f1374e50`. 7. Review this document, the README, and CHANGELOG for accurate scope. The tag-triggered release workflow repeats the non-service test suite, builds diff --git a/src/openstatspec/__init__.py b/src/openstatspec/__init__.py index 4de4594..a93114a 100644 --- a/src/openstatspec/__init__.py +++ b/src/openstatspec/__init__.py @@ -1,20 +1,30 @@ """Public API for the OpenStatSpec Python reference implementation.""" from .api import ( - capabilities, capability_matrix, derive_sql_dataset, execute_sql_transformation, - export_sav, get_dataset, import_sav, inspect, list_datasets, + apply_spss_in_place, capabilities, capability_matrix, derive_sql_dataset, + execute_sql_transformation, + export_sav, get_dataset, import_sav, inspect, + install_in_place_transformation_schema, list_datasets, register_sql_transformation, reconcile_derived_removals, reconcile_sql_transformation_runs, remove_derived_physical_relation, retire_derived, validate, validate_derived, ) from .core import CapabilityDeclaration, LossReport, UnsupportedOperationError from .sql.workflow import TransformationError +from .transform import ( + SpssFrontendCompilation, TransformationFrontendError, + VariableDefinition, VariableSchema, compile_spss_syntax, +) __all__ = [ - "CapabilityDeclaration", "LossReport", "TransformationError", + "CapabilityDeclaration", "LossReport", "SpssFrontendCompilation", + "TransformationError", "TransformationFrontendError", + "VariableDefinition", "VariableSchema", "UnsupportedOperationError", "capabilities", "capability_matrix", - "derive_sql_dataset", "execute_sql_transformation", "export_sav", - "get_dataset", "import_sav", "inspect", "list_datasets", + "apply_spss_in_place", "compile_spss_syntax", "derive_sql_dataset", + "execute_sql_transformation", "export_sav", + "get_dataset", "import_sav", "inspect", + "install_in_place_transformation_schema", "list_datasets", "register_sql_transformation", "reconcile_derived_removals", "reconcile_sql_transformation_runs", "remove_derived_physical_relation", "retire_derived", "validate", diff --git a/src/openstatspec/api.py b/src/openstatspec/api.py index afbae14..6bfa308 100644 --- a/src/openstatspec/api.py +++ b/src/openstatspec/api.py @@ -20,6 +20,11 @@ remove_derived_relation as _remove_derived_relation, retire_derived_dataset as _retire_derived_dataset, ) +from .sql.inplace_transform import ( + apply_spss_in_place as _apply_spss_in_place, + in_place_transformation_capabilities, + install_in_place_transformation_schema as _install_in_place_schema, +) from .sql.capabilities import ( SPECIFICATION_COMMIT, SPECIFICATION_RELEASE, active_connection, catalog_binding, ) @@ -84,6 +89,9 @@ def capability_matrix(database_url: str | None = None) -> Mapping[str, Any]: "sql_profiles": declared_profiles(database_url), "optional_profiles": { "sql_transformation_workflow": transformation_capabilities(database_url), + "spss_in_place_transformation": ( + in_place_transformation_capabilities() + ), }, } return declaration @@ -130,6 +138,27 @@ def derive_sql_dataset(*, database_url: Any, **options: Any) -> Mapping[str, Any return result(_derive_dataset(database_url=str(database_url), **options)) +def apply_spss_in_place( + *, database_url: Any, dataset_id: str, source_text: str, + actor: str, expected_branch: str | None = None, + expected_head: str | None = None, +) -> Mapping[str, Any]: + """Apply supported SPSS-like syntax to the same SQL dataset/table.""" + return result(_apply_spss_in_place( + database_url=str(database_url), + dataset_id=dataset_id, + source_text=source_text, + actor=actor, + expected_branch=expected_branch, + expected_head=expected_head, + )) + + +def install_in_place_transformation_schema(*, database_url: Any) -> None: + """Install the compact apply-audit relation before the first apply.""" + _install_in_place_schema(database_url=str(database_url)) + + def validate_derived(*, database_url: Any, derived_dataset_id: str) -> Mapping[str, Any]: return result(_validate_derived_dataset( database_url=str(database_url), derived_dataset_id=derived_dataset_id, diff --git a/src/openstatspec/cli.py b/src/openstatspec/cli.py index 4d710bc..e1ffc22 100644 --- a/src/openstatspec/cli.py +++ b/src/openstatspec/cli.py @@ -2,9 +2,11 @@ import argparse import json from collections.abc import Sequence +from pathlib import Path from .api import ( - capability_matrix, derive_sql_dataset, execute_sql_transformation, + apply_spss_in_place, capability_matrix, derive_sql_dataset, + execute_sql_transformation, export_sav, get_dataset, import_sav, inspect, list_datasets, register_sql_transformation, validate, validate_derived, ) @@ -73,6 +75,19 @@ def main(argv: Sequence[str] | None = None) -> int: derive.add_argument("--dataset-name") derive.add_argument("--weight-variable") + apply_spss = commands.add_parser( + "apply-spss", + help="apply supported SPSS syntax in-place on a controlled Dolt branch", + ) + apply_spss.add_argument("--database-url", required=True) + apply_spss.add_argument("--dataset-id", required=True) + apply_spss.add_argument("--actor", required=True) + apply_spss.add_argument("--expected-branch") + apply_spss.add_argument("--expected-head") + syntax_source = apply_spss.add_mutually_exclusive_group(required=True) + syntax_source.add_argument("--syntax") + syntax_source.add_argument("--syntax-file") + derived_validator = commands.add_parser("validate-derived", help="validate a derived dataset") derived_validator.add_argument("--database-url", required=True) derived_validator.add_argument("--derived-dataset-id", required=True) @@ -122,6 +137,20 @@ def main(argv: Sequence[str] | None = None) -> int: transformation_name=args.name, dataset_name=args.dataset_name, weight_variable=args.weight_variable, ) + elif args.command == "apply-spss": + source_text = ( + args.syntax + if args.syntax is not None + else Path(args.syntax_file).read_text(encoding="utf-8") + ) + output = apply_spss_in_place( + database_url=args.database_url, + dataset_id=args.dataset_id, + source_text=source_text, + actor=args.actor, + expected_branch=args.expected_branch, + expected_head=args.expected_head, + ) else: output = validate_derived( database_url=args.database_url, derived_dataset_id=args.derived_dataset_id, diff --git a/src/openstatspec/sql/capabilities.py b/src/openstatspec/sql/capabilities.py index 1484bc5..2701660 100644 --- a/src/openstatspec/sql/capabilities.py +++ b/src/openstatspec/sql/capabilities.py @@ -15,8 +15,8 @@ from .profiles import profile_for_url, validate_connection_url from ..core import UnsupportedOperationError -SPECIFICATION_COMMIT = "d287c2cde9ade71f04e27dd012caec876901aed5" -SPECIFICATION_RELEASE: str | None = "v0.1.0" +SPECIFICATION_COMMIT = "79339ec3d8f8aa81789b7e85f6b8afa6f1374e50" +SPECIFICATION_RELEASE: str | None = "v0.2.0" _DOLT_2_2_STABLE_VERSION = re.compile(r"2\.2\.(0|[1-9][0-9]*)") diff --git a/src/openstatspec/sql/inplace_transform.py b/src/openstatspec/sql/inplace_transform.py new file mode 100644 index 0000000..c6a0fe0 --- /dev/null +++ b/src/openstatspec/sql/inplace_transform.py @@ -0,0 +1,556 @@ +"""In-place application without an OpenStatSpec undo or copy layer.""" + +from __future__ import annotations + +from datetime import UTC, datetime +import json +from typing import Any +from uuid import uuid4 + +from sqlalchemy import ( + Column, DateTime, Float, Integer, MetaData, String, Table, Text, case, + create_engine, delete, inspect, insert, literal, null, select, text, update, +) + +from ..transform import ( + RecodeOperation, RecodeResult, ReplaceValueLabelsOperation, + SetVariableLabelOperation, TypedValue, ValueLabel, VariableDefinition, + VariableSchema, compile_spss_syntax, +) +from .capabilities import effective_profile +from .normative import catalog as core_catalog +from .wide import catalog as legacy_catalog +from .wide import normalized_metadata_tables, physical_name +from .workflow import TransformationError + + +APPLY_CONTRACT = "openstatspec-in-place-transformation-v0.1" + + +def in_place_transformation_capabilities() -> dict[str, Any]: + return { + "contract": APPLY_CONTRACT, + "status": "experimental", + "database_products": [ + "sqlite", "postgresql", "mysql", "mariadb", "dolt", + ], + "parent_kinds": ["core"], + "mutation": "same_dataset_same_physical_wide_table", + "commands": ["RECODE", "VARIABLE LABELS", "VALUE LABELS"], + "new_target_column": { + "sqlite": True, + "postgresql": True, + "mysql": False, + "mariadb": False, + "dolt": False, + "reason": "non-transactional DDL must not make apply partially durable", + }, + "creates_derived_dataset": False, + "creates_persistent_data_copy": False, + "openstatspec_rollback_or_version_history": False, + "dolt_requires_clean_working_set": True, + "dolt_requires_expected_branch_and_head": True, + "performs_dolt_commit": False, + "execution_evidence": { + "sqlite": "local_conformance", + "postgresql": "service_conformance_required", + "mysql": "service_conformance_required", + "mariadb": "service_conformance_required", + "dolt": "service_conformance_required", + }, + } + + +def apply_audit_catalog(metadata: MetaData) -> Table: + return Table( + "transformation_apply", + metadata, + Column("apply_id", String(36), primary_key=True), + Column("contract_id", String(128), nullable=False), + Column("dataset_id", String(36), nullable=False), + Column("database_profile", String(32), nullable=False), + Column("physical_table_schema", String(255)), + Column("physical_table_name", String(255), nullable=False), + Column("source_hash", String(64), nullable=False), + Column("plan_hash", String(64), nullable=False), + Column("canonical_plan_json", Text, nullable=False), + Column("actor", String(255), nullable=False), + Column("status", String(16), nullable=False), + Column("dolt_branch", String(255)), + Column("dolt_head_before", String(128)), + Column("dolt_head_after", String(128)), + Column("operation_count", Integer, nullable=False), + Column("started_at", DateTime, nullable=False), + Column("completed_at", DateTime, nullable=False), + ) + + +def _now() -> datetime: + return datetime.now(UTC).replace(tzinfo=None) + + +def _typed_value(value: TypedValue) -> float | str: + return value.number() if value.type == "binary64" else str(value.value) + + +def _result_expression(result: RecodeResult, source: Any) -> Any: + if result.kind == "copy": + return source + if result.kind == "system_missing": + return null() + assert result.value is not None + return literal(_typed_value(result.value)) + + +def _match_expression(match: Any, source: Any) -> Any: + if match.kind == "system_missing": + return source.is_(None) + if match.kind == "range": + return source.between( + _typed_value(match.lower), _typed_value(match.upper) + ) + return source.in_([_typed_value(value) for value in match.values]) + + +def _input_schema( + connection: Any, + dataset_id: str, +) -> tuple[dict[str, Any], list[dict[str, Any]], VariableSchema]: + core = core_catalog(MetaData()) + dataset = connection.execute( + select(core.dataset).where(core.dataset.c.dataset_id == dataset_id) + ).mappings().one_or_none() + if dataset is None: + raise TransformationError( + "dataset_not_found", "The in-place target dataset does not exist." + ) + variables = [dict(row) for row in connection.execute( + select(core.variable) + .where(core.variable.c.dataset_id == dataset_id) + .order_by(core.variable.c.source_ordinal) + ).mappings()] + if not variables: + raise TransformationError( + "dataset_invalid", "The in-place target has no variables." + ) + labels = connection.execute( + select( + core.variable_value_label_set.c.variable_id, + core.value_label.c.ordinal, + core.value_label.c.code_kind, + core.value_label.c.numeric_code, + core.value_label.c.string_code, + core.value_label.c.label, + ).select_from( + core.variable_value_label_set.join( + core.value_label, + core.variable_value_label_set.c.value_label_set_id + == core.value_label.c.value_label_set_id, + ) + ).order_by( + core.variable_value_label_set.c.variable_id, + core.value_label.c.ordinal, + ) + ).mappings() + labels_by_variable: dict[str, list[ValueLabel]] = {} + for row in labels: + typed = ( + TypedValue.binary64(float(row["numeric_code"])) + if row["code_kind"] == "numeric" + else TypedValue.string(str(row["string_code"])) + ) + labels_by_variable.setdefault(str(row["variable_id"]), []).append( + ValueLabel(typed, str(row["label"])) + ) + schema = VariableSchema(tuple( + VariableDefinition( + str(row["source_name"]), + str(row["storage_kind"]), + variable_label=row["variable_label"], + value_labels=tuple(labels_by_variable.get(str(row["variable_id"]), [])), + ) + for row in variables + )) + return dict(dataset), variables, schema + + +def _catalog_identity_counts( + connection: Any, +) -> tuple[int, tuple[tuple[str | None, str], ...]]: + core = core_catalog(MetaData()) + dataset_rows = connection.execute( + select( + core.dataset.c.physical_table_schema, + core.dataset.c.physical_table_name, + ) + ).all() + identities = tuple(sorted( + ( + ( + str(schema) if schema is not None else None, + str(name), + ) + for schema, name in dataset_rows + ), + key=lambda item: (item[0] or "", item[1]), + )) + return len(dataset_rows), identities + + +def _legacy_identifiers(dataset: dict[str, Any]) -> tuple[str, str]: + dataset_name = dataset.get("dataset_name") + table_name = dataset.get("physical_table_name") + if not isinstance(dataset_name, str) or not dataset_name: + raise TransformationError( + "dataset_invalid", "The target lacks its legacy catalog identity." + ) + if not isinstance(table_name, str) or not table_name: + raise TransformationError( + "dataset_invalid", "The target lacks its physical wide-table name." + ) + return dataset_name, table_name + + +def _replace_value_labels( + connection: Any, + *, + core: Any, + legacy_variable: Table, + legacy_labels: Table, + legacy_dataset_id: str, + variable: dict[str, Any], + labels: tuple[ValueLabel, ...], +) -> None: + variable_id = str(variable["variable_id"]) + ordinal = int(variable["source_ordinal"]) + old_set = connection.execute( + select(core.variable_value_label_set.c.value_label_set_id).where( + core.variable_value_label_set.c.variable_id == variable_id + ) + ).scalar_one_or_none() + if old_set is not None: + connection.execute(delete(core.value_label).where( + core.value_label.c.value_label_set_id == old_set + )) + connection.execute(delete(core.variable_value_label_set).where( + core.variable_value_label_set.c.variable_id == variable_id + )) + connection.execute(delete(core.value_label_set).where( + core.value_label_set.c.value_label_set_id == old_set + )) + label_set_id = str(uuid4()) + connection.execute(insert(core.value_label_set).values( + value_label_set_id=label_set_id, + dataset_id=variable["dataset_id"], + name=None, + )) + connection.execute(insert(core.variable_value_label_set).values( + variable_id=variable_id, + value_label_set_id=label_set_id, + )) + for label_ordinal, item in enumerate(labels, start=1): + connection.execute(insert(core.value_label).values( + value_label_id=str(uuid4()), + value_label_set_id=label_set_id, + ordinal=label_ordinal, + code_kind="numeric" if item.value.type == "binary64" else "string", + numeric_code=(item.value.number() if item.value.type == "binary64" else None), + string_code=(str(item.value.value) if item.value.type == "string" else None), + label=item.label, + )) + connection.execute(delete(legacy_labels).where( + legacy_labels.c.dataset_id == legacy_dataset_id, + legacy_labels.c.variable_ordinal == ordinal, + )) + for label_ordinal, item in enumerate(labels, start=1): + connection.execute(insert(legacy_labels).values( + dataset_id=legacy_dataset_id, + variable_ordinal=ordinal, + ordinal=label_ordinal, + value_type="numeric" if item.value.type == "binary64" else "text", + numeric_value=(item.value.number() if item.value.type == "binary64" else None), + text_value=(str(item.value.value) if item.value.type == "string" else None), + label=item.label, + )) + legacy_json = { + str(_typed_value(item.value)): item.label for item in labels + } + connection.execute(update(legacy_variable).where( + legacy_variable.c.dataset_id == legacy_dataset_id, + legacy_variable.c.ordinal == ordinal, + ).values(value_labels=json.dumps(legacy_json, ensure_ascii=False))) + + +def _apply_on_connection( + connection: Any, + *, + dataset_id: str, + source_text: str, + actor: str, + database_profile: str, + allow_schema_change: bool, + dolt_branch: str | None, + dolt_head: str | None, +) -> dict[str, Any]: + before_count, before_tables = _catalog_identity_counts(connection) + dataset, variables, schema = _input_schema(connection, dataset_id) + legacy_dataset_id, table_name = _legacy_identifiers(dataset) + compilation = compile_spss_syntax(source_text, schema, input_alias="parent") + audit = apply_audit_catalog(MetaData()) + if not inspect(connection).has_table("transformation_apply"): + raise TransformationError( + "in_place_audit_schema_missing", + "The compact transformation_apply audit schema must be installed " + "before apply.", + ) + if not allow_schema_change and any( + isinstance(operation, RecodeOperation) + and operation.target_mode == "create" + for operation in compilation.plan.operations + ): + raise TransformationError( + "schema_change_not_atomic", + "This database profile requires RECODE targets to exist before " + "apply because its DDL is not transaction-atomic.", + ) + core = core_catalog(MetaData()) + legacy_metadata = MetaData() + _, legacy_variable, _, _ = legacy_catalog(legacy_metadata) + _, legacy_labels, _, _ = normalized_metadata_tables(legacy_metadata) + relation = Table( + table_name, + MetaData(), + schema=dataset.get("physical_table_schema"), + autoload_with=connection, + ) + by_name = {str(row["source_name"]).casefold(): row for row in variables} + used_physical = {str(row["physical_name"]).casefold() for row in variables} + + for operation in compilation.plan.operations: + if isinstance(operation, RecodeOperation): + source_variable = by_name[operation.source.casefold()] + if operation.target_mode == "create": + if str(source_variable["storage_kind"]) != "numeric": + raise TransformationError( + "in_place_target_type_unsupported", + "This profile creates only numeric RECODE targets.", + ) + target_physical = physical_name(operation.target, used_physical) + quote = connection.dialect.identifier_preparer.quote + numeric_type = ( + "DOUBLE PRECISION" + if connection.dialect.name == "postgresql" + else "DOUBLE" + ) + qualified_table = connection.dialect.identifier_preparer.format_table( + relation + ) + connection.exec_driver_sql( + f"ALTER TABLE {qualified_table} ADD COLUMN " + f"{quote(target_physical)} {numeric_type} NULL" + ) + new_ordinal = max(int(row["source_ordinal"]) for row in variables) + 1 + target_variable = { + "variable_id": str(uuid4()), + "dataset_id": dataset_id, + "source_ordinal": new_ordinal, + "source_name": operation.target, + "physical_name": target_physical, + "storage_kind": "numeric", + "variable_label": None, + } + connection.execute(insert(core.variable).values(**target_variable)) + connection.execute(insert(legacy_variable).values( + dataset_id=legacy_dataset_id, + ordinal=new_ordinal, + source_name=operation.target, + physical_name=target_physical, + storage_kind="numeric", + string_width=None, + label="", + attributes="{}", + value_labels="{}", + missing_ranges="[]", + )) + variables.append(target_variable) + by_name[operation.target.casefold()] = target_variable + relation = Table( + table_name, + MetaData(), + schema=dataset.get("physical_table_schema"), + autoload_with=connection, + ) + target_variable = by_name[operation.target.casefold()] + source_column = relation.c[str(source_variable["physical_name"])] + target_column = relation.c[str(target_variable["physical_name"])] + expression = case( + *[ + ( + _match_expression(rule.match, source_column), + _result_expression(rule.result, source_column), + ) + for rule in operation.rules + ], + else_=_result_expression(operation.unmatched, source_column), + ) + connection.execute(update(relation).values({target_column: expression})) + elif isinstance(operation, SetVariableLabelOperation): + variable = by_name[operation.variable.casefold()] + connection.execute(update(core.variable).where( + core.variable.c.variable_id == variable["variable_id"] + ).values(variable_label=operation.label)) + connection.execute(update(legacy_variable).where( + legacy_variable.c.dataset_id == legacy_dataset_id, + legacy_variable.c.ordinal == variable["source_ordinal"], + ).values(label=operation.label)) + elif isinstance(operation, ReplaceValueLabelsOperation): + _replace_value_labels( + connection, + core=core, + legacy_variable=legacy_variable, + legacy_labels=legacy_labels, + legacy_dataset_id=legacy_dataset_id, + variable=by_name[operation.variable.casefold()], + labels=operation.labels, + ) + else: # pragma: no cover - canonical plan type is closed + raise TransformationError( + "operation_not_supported", "Unsupported in-place plan operation." + ) + + after_count, after_tables = _catalog_identity_counts(connection) + if (after_count, after_tables) != (before_count, before_tables): + raise TransformationError( + "dataset_identity_changed", + "In-place apply changed dataset or physical data-table identity.", + ) + apply_id = str(uuid4()) + started = _now() + connection.execute(insert(audit).values( + apply_id=apply_id, + contract_id=APPLY_CONTRACT, + dataset_id=dataset_id, + database_profile=database_profile, + physical_table_schema=dataset.get("physical_table_schema"), + physical_table_name=table_name, + source_hash=compilation.source_hash, + plan_hash=compilation.plan_hash, + canonical_plan_json=compilation.plan.canonical_json(), + actor=actor, + status="succeeded", + dolt_branch=dolt_branch, + dolt_head_before=dolt_head, + dolt_head_after=dolt_head, + operation_count=len(compilation.plan.operations), + started_at=started, + completed_at=_now(), + )) + forbidden = { + name for name in inspect(connection).get_table_names() + if name.startswith("derived_plan_") + or name.startswith("__openstatspec_plan_staging_") + or name.startswith("openstatspec_rollback_") + or name.startswith("openstatspec_snapshot_") + } + if forbidden: + raise TransformationError( + "forbidden_copy_artifact", + "In-place apply created a forbidden copy/history artifact.", + ) + return { + "apply_id": apply_id, + "status": "succeeded", + "dataset_id": dataset_id, + "database_profile": database_profile, + "physical_table_schema": dataset.get("physical_table_schema"), + "physical_table_name": table_name, + "source_hash": compilation.source_hash, + "plan_hash": compilation.plan_hash, + "dolt_branch": dolt_branch, + "dolt_head_before": dolt_head, + "dolt_head_after": dolt_head, + "dolt_commit_performed": False, + } + + +def _dolt_state(connection: Any) -> tuple[str, str, int]: + branch = str(connection.execute(text("SELECT active_branch()" )).scalar_one()) + head = str(connection.execute(text("SELECT DOLT_HASHOF('HEAD')")).scalar_one()) + dirty = int(connection.execute(text( + "SELECT COUNT(*) FROM dolt_status" + )).scalar_one()) + return branch, head, dirty + + +def install_in_place_transformation_schema(*, database_url: str) -> None: + """Install the compact operation audit separately from any data apply.""" + engine = create_engine(database_url) + try: + with engine.begin() as connection: + apply_audit_catalog(MetaData()).create(connection, checkfirst=True) + finally: + engine.dispose() + + +def apply_spss_in_place( + *, + database_url: str, + dataset_id: str, + source_text: str, + actor: str, + expected_branch: str | None = None, + expected_head: str | None = None, +) -> dict[str, Any]: + """Apply one plan to the same dataset/table; never create undo or copies.""" + if not actor: + raise TransformationError( + "actor_required", "A non-empty actor identity is mandatory.", + ) + profile, _active = effective_profile(database_url) + engine = create_engine(database_url) + try: + with engine.begin() as connection: + branch: str | None = None + head: str | None = None + if profile.name == "dolt": + if not expected_branch or not expected_head: + raise TransformationError( + "dolt_context_required", + "Dolt apply requires expected_branch and expected_head.", + ) + branch, head, dirty = _dolt_state(connection) + if branch != expected_branch: + raise TransformationError( + "dolt_branch_mismatch", + "The active Dolt branch differs from the caller's expectation.", + ) + if head != expected_head: + raise TransformationError( + "dolt_head_mismatch", + "The active Dolt HEAD differs from the caller's expectation.", + ) + if dirty != 0: + raise TransformationError( + "dolt_working_set_dirty", + "The Dolt working set must be clean before in-place apply.", + ) + result = _apply_on_connection( + connection, + dataset_id=dataset_id, + source_text=source_text, + actor=actor, + database_profile=profile.name, + allow_schema_change=profile.name in {"sqlite", "postgresql"}, + dolt_branch=branch, + dolt_head=head, + ) + if profile.name == "dolt": + after_branch, after_head, _dirty_after = _dolt_state(connection) + if after_branch != branch or after_head != head: + raise TransformationError( + "dolt_context_changed", + "Apply must not switch branches or create a Dolt commit.", + ) + return result + finally: + engine.dispose() diff --git a/src/openstatspec/transform/__init__.py b/src/openstatspec/transform/__init__.py new file mode 100644 index 0000000..27ae283 --- /dev/null +++ b/src/openstatspec/transform/__init__.py @@ -0,0 +1,33 @@ +"""Pure transformation models and SPSS-syntax frontend; no database adapter.""" + +from .binding import ( + BoundTransformation, VariableDefinition, VariableSchema, bind_spss_syntax, +) +from .compiler import SpssFrontendCompilation, compile_spss_syntax +from .errors import SourcePosition, SourceSpan, TransformationFrontendError +from .plan import ( + SPSS_FRONTEND_CONTRACT, TRANSFORMATION_PLAN_CONTRACT, + RecodeMatch, RecodeOperation, RecodeResult, RecodeRule, + ReplaceValueLabelsOperation, SetVariableLabelOperation, TransformationPlan, + TypedValue, ValueLabel, canonical_plan_hash, canonical_plan_json, + transformation_plan_from_dict, +) +from .syntax import ( + SpssSyntaxProgram, normalize_spss_source, parse_spss_syntax, + spss_source_hash, tokenize_spss, +) +__all__ = [ + "BoundTransformation", + "RecodeMatch", "RecodeOperation", "RecodeResult", + "RecodeRule", "ReplaceValueLabelsOperation", "SPSS_FRONTEND_CONTRACT", + "SetVariableLabelOperation", "SourcePosition", "SourceSpan", + "SpssFrontendCompilation", + "SpssSyntaxProgram", "TRANSFORMATION_PLAN_CONTRACT", "TransformationFrontendError", + "TransformationPlan", "TypedValue", "ValueLabel", + "VariableDefinition", + "VariableSchema", "bind_spss_syntax", "canonical_plan_hash", + "compile_spss_syntax", + "canonical_plan_json", "normalize_spss_source", "parse_spss_syntax", + "spss_source_hash", "tokenize_spss", + "transformation_plan_from_dict", +] diff --git a/src/openstatspec/transform/binding.py b/src/openstatspec/transform/binding.py new file mode 100644 index 0000000..c2dbdad --- /dev/null +++ b/src/openstatspec/transform/binding.py @@ -0,0 +1,298 @@ +"""Bind catalog-independent SPSS syntax against an explicit in-memory schema.""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from typing import Literal + +from .errors import SourceSpan, frontend_error +from .plan import ( + PlanOperation, RecodeMatch, RecodeOperation, RecodeResult, RecodeRule, + ReplaceValueLabelsOperation, SetVariableLabelOperation, TransformationPlan, + TypedValue, ValueLabel, +) +from .syntax import ( + RecodeCommandSyntax, RecodeMatchSyntax, RecodeResultSyntax, SpssSyntaxProgram, + SyntaxLiteral, ValueLabelsCommandSyntax, VariableLabelsCommandSyntax, +) + + +StorageKind = Literal["numeric", "string"] + + +@dataclass(frozen=True) +class VariableDefinition: + name: str + storage_kind: StorageKind + variable_label: str | None = None + value_labels: tuple[ValueLabel, ...] = () + + def __post_init__(self) -> None: + if not isinstance(self.name, str) or not self.name: + raise ValueError("Variable name must be non-empty text.") + if self.storage_kind not in {"numeric", "string"}: + raise ValueError("storage_kind must be numeric or string.") + expected_type = "binary64" if self.storage_kind == "numeric" else "string" + if any(label.value.type != expected_type for label in self.value_labels): + raise ValueError("Value-label types must match their variable storage kind.") + + +@dataclass(frozen=True) +class VariableSchema: + variables: tuple[VariableDefinition, ...] + + def __post_init__(self) -> None: + names = [variable.name.casefold() for variable in self.variables] + if len(names) != len(set(names)): + raise ValueError("Variable names must be unique case-insensitively.") + + +@dataclass(frozen=True) +class BoundTransformation: + plan: TransformationPlan + output_schema: VariableSchema + operation_spans: tuple[SourceSpan, ...] + + +def _typed(literal: SyntaxLiteral) -> TypedValue: + if literal.kind == "numeric": + assert isinstance(literal.value, float) + return TypedValue.binary64(literal.value) + assert isinstance(literal.value, str) + return TypedValue.string(literal.value) + + +def _expected_type(storage_kind: StorageKind) -> str: + return "binary64" if storage_kind == "numeric" else "string" + + +def _resolve( + variables: list[VariableDefinition], name: str, span: SourceSpan, +) -> tuple[int, VariableDefinition]: + matches = [ + (index, variable) for index, variable in enumerate(variables) + if variable.name.casefold() == name.casefold() + ] + if len(matches) != 1: + raise frontend_error( + "unknown_variable", f"Variable {name!r} is not present in the current schema.", + span=span, variable=name, + ) + return matches[0] + + +def _match( + syntax: RecodeMatchSyntax, source: VariableDefinition, +) -> RecodeMatch: + expected = _expected_type(source.storage_kind) + if syntax.kind == "system_missing": + if source.storage_kind == "string": + raise frontend_error( + "system_missing_for_string", + "SYSMIS cannot match a string variable.", span=syntax.span, + variable=source.name, + ) + return RecodeMatch("system_missing") + if syntax.kind == "range": + assert syntax.lower is not None and syntax.upper is not None + lower, upper = _typed(syntax.lower), _typed(syntax.upper) + if lower.type != "binary64" or upper.type != "binary64" or expected != "binary64": + raise frontend_error( + "type_mismatch", "THRU ranges require a numeric source and endpoints.", + span=syntax.span, variable=source.name, + ) + if lower.number() > upper.number(): + raise frontend_error( + "invalid_numeric_range", + "THRU lower endpoint exceeds its upper endpoint.", + span=syntax.span, variable=source.name, + ) + return RecodeMatch("range", lower=lower, upper=upper) + if syntax.kind != "values": + raise AssertionError("ELSE is lowered separately") + values = tuple(_typed(value) for value in syntax.values) + if any(value.type != expected for value in values): + raise frontend_error( + "type_mismatch", "RECODE match values must match the source storage kind.", + span=syntax.span, variable=source.name, expected_type=expected, + ) + return RecodeMatch("values", values) + + +def _result( + syntax: RecodeResultSyntax, source: VariableDefinition, +) -> RecodeResult: + if syntax.kind == "copy": + return RecodeResult("copy") + if syntax.kind == "system_missing": + if source.storage_kind == "string": + raise frontend_error( + "system_missing_for_string", + "SYSMIS cannot be produced for a string variable.", span=syntax.span, + variable=source.name, + ) + return RecodeResult("system_missing") + assert syntax.value is not None + return RecodeResult("literal", _typed(syntax.value)) + + +def _result_type( + result: RecodeResult, source: VariableDefinition, +) -> Literal["binary64", "string"]: + if result.kind == "copy": + return _expected_type(source.storage_kind) # type: ignore[return-value] + if result.kind == "system_missing": + return "binary64" + assert result.value is not None + return result.value.type + + +def _bind_recode( + command: RecodeCommandSyntax, variables: list[VariableDefinition], +) -> tuple[list[RecodeOperation], list[SourceSpan]]: + sources = [_resolve(variables, token.text, token.span)[1] for token in command.sources] + targets = command.targets + target_mode: Literal["create", "replace"] = "create" if targets is not None else "replace" + target_names = ( + [token.text for token in targets] if targets is not None + else [source.name for source in sources] + ) + operations: list[RecodeOperation] = [] + spans: list[SourceSpan] = [] + for ordinal, (source, target_name) in enumerate(zip(sources, target_names)): + target_span = ( + targets[ordinal].span if targets is not None else command.sources[ordinal].span + ) + if target_mode == "create": + if target_name.startswith("__"): + raise frontend_error( + "reserved_target_name", + f"Target name {target_name!r} is reserved.", + span=target_span, target=target_name, + ) + if any( + variable.name.casefold() == target_name.casefold() + for variable in variables + ): + raise frontend_error( + "target_already_exists", + f"Target name {target_name!r} already exists.", + span=target_span, target=target_name, + ) + rules: list[RecodeRule] = [] + else_result: RecodeResult | None = None + for clause in command.clauses: + result = _result(clause.result, source) + 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" + ) + result_types = { + _result_type(result, source) + for result in [*(rule.result for rule in rules), unmatched] + } + if target_mode == "create" and "string" in result_types: + raise frontend_error( + "string_target_requires_declaration", + "New string targets require a STRING declaration, which is outside the MVP.", + span=target_span, target=target_name, + ) + if len(result_types) != 1: + raise frontend_error( + "mixed_result_types", + "All RECODE results, including unmatched behavior, must have one type.", + span=command.span, source=source.name, result_types=sorted(result_types), + ) + output_type = next(iter(result_types)) + if target_mode == "replace" and output_type != _expected_type(source.storage_kind): + raise frontend_error( + "type_mismatch", "In-place RECODE cannot change the variable storage kind.", + span=command.span, variable=source.name, + ) + operation = RecodeOperation( + source=source.name, + target=source.name if target_mode == "replace" else target_name, + target_mode=target_mode, rules=tuple(rules), unmatched=unmatched, + ) + operations.append(operation) + spans.append(command.span) + if target_mode == "create": + variables.append(VariableDefinition(target_name, "numeric")) + # replace intentionally preserves the existing variable metadata. A later + # VALUE LABELS command replaces value labels explicitly. + return operations, spans + + +def bind_spss_syntax( + program: SpssSyntaxProgram, schema: VariableSchema, *, input_alias: str = "parent", +) -> BoundTransformation: + """Resolve names and sequential semantics into a canonical generic plan.""" + if not program.commands: + raise frontend_error( + "spss_syntax_error", "At least one supported SPSS command is required.", + span=program.span, + ) + variables = list(schema.variables) + operations: list[PlanOperation] = [] + spans: list[SourceSpan] = [] + for command in program.commands: + if isinstance(command, RecodeCommandSyntax): + recodes, recode_spans = _bind_recode(command, variables) + operations.extend(recodes) + spans.extend(recode_spans) + continue + if isinstance(command, VariableLabelsCommandSyntax): + for assignment in command.assignments: + index, variable = _resolve( + variables, assignment.variable.text, assignment.variable.span, + ) + assert isinstance(assignment.label.value, str) + operations.append(SetVariableLabelOperation( + variable.name, assignment.label.value, + )) + spans.append(assignment.span) + variables[index] = replace( + variable, variable_label=assignment.label.value, + ) + continue + if isinstance(command, ValueLabelsCommandSyntax): + for group in command.groups: + for variable_token in group.variables: + index, variable = _resolve( + variables, variable_token.text, variable_token.span, + ) + expected = _expected_type(variable.storage_kind) + labels = tuple(ValueLabel( + _typed(label.value), str(label.label.value), + ) for label in group.labels) + mismatched = next( + (label for label in labels if label.value.type != expected), None + ) + if mismatched is not None: + raise frontend_error( + "type_mismatch", + "VALUE LABELS codes must match the variable storage kind.", + span=group.span, variable=variable.name, + expected_type=expected, + ) + keys = [label.value.canonical_key() for label in labels] + if len(keys) != len(set(keys)): + raise frontend_error( + "duplicate_value_label", + "VALUE LABELS contains duplicate canonical codes.", + span=group.span, variable=variable.name, + ) + operation = ReplaceValueLabelsOperation(variable.name, labels) + operations.append(operation) + spans.append(group.span) + variables[index] = replace(variable, value_labels=labels) + continue + raise AssertionError(f"Unknown syntax command: {type(command)!r}") + return BoundTransformation( + TransformationPlan(tuple(operations), input_alias=input_alias), + VariableSchema(tuple(variables)), + tuple(spans), + ) diff --git a/src/openstatspec/transform/compiler.py b/src/openstatspec/transform/compiler.py new file mode 100644 index 0000000..be98433 --- /dev/null +++ b/src/openstatspec/transform/compiler.py @@ -0,0 +1,50 @@ +"""High-level, deterministic SPSS-like frontend compilation.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .binding import BoundTransformation, VariableSchema, bind_spss_syntax +from .plan import TransformationPlan +from .syntax import ( + normalize_spss_source, + parse_spss_syntax, + spss_source_hash, +) + + +@dataclass(frozen=True) +class SpssFrontendCompilation: + """One source artifact and its fully bound canonical plan.""" + + source_text_lf: str + source_hash: str + bound: BoundTransformation + + @property + def plan(self) -> TransformationPlan: + return self.bound.plan + + @property + def plan_hash(self) -> str: + return self.plan.sha256() + + +def compile_spss_syntax( + source: str, + schema: VariableSchema, + *, + input_alias: str = "parent", +) -> SpssFrontendCompilation: + """Parse and bind source without SQL generation or database mutation.""" + normalized = normalize_spss_source(source) + bound = bind_spss_syntax( + parse_spss_syntax(normalized), + schema, + input_alias=input_alias, + ) + return SpssFrontendCompilation( + source_text_lf=normalized, + source_hash=spss_source_hash(normalized), + bound=bound, + ) diff --git a/src/openstatspec/transform/errors.py b/src/openstatspec/transform/errors.py new file mode 100644 index 0000000..30f9756 --- /dev/null +++ b/src/openstatspec/transform/errors.py @@ -0,0 +1,59 @@ +"""Stable diagnostics and source locations for transformation frontends.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class SourcePosition: + """One zero-based byte-independent offset and one-based display position.""" + + offset: int + line: int + column: int + + def as_dict(self) -> dict[str, int]: + return {"offset": self.offset, "line": self.line, "column": self.column} + + +@dataclass(frozen=True) +class SourceSpan: + """Half-open source range.""" + + start: SourcePosition + end: SourcePosition + + def as_dict(self) -> dict[str, dict[str, int]]: + return {"start": self.start.as_dict(), "end": self.end.as_dict()} + + +class TransformationFrontendError(ValueError): + """A safe, machine-readable frontend or binding failure.""" + + def __init__( + self, code: str, detail: str, *, span: SourceSpan | None = None, + details: dict[str, Any] | None = None, + ) -> None: + super().__init__(f"Transformation frontend failed [{code}]: {detail}") + self.code = code + self.detail = detail + self.span = span + self.details = dict(details or {}) + + def as_dict(self) -> dict[str, Any]: + result: dict[str, Any] = {"code": self.code, "detail": self.detail} + if self.span is not None: + result["span"] = self.span.as_dict() + if self.details: + result["details"] = dict(self.details) + return result + + +def frontend_error( + code: str, detail: str, *, span: SourceSpan | None = None, **details: Any, +) -> TransformationFrontendError: + return TransformationFrontendError( + code, detail, span=span, details=details or None, + ) diff --git a/src/openstatspec/transform/plan.py b/src/openstatspec/transform/plan.py new file mode 100644 index 0000000..c2ab3c4 --- /dev/null +++ b/src/openstatspec/transform/plan.py @@ -0,0 +1,416 @@ +"""Canonical generic OpenStatSpec transformation-plan models.""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import math +import re +import struct +from typing import Any, Literal, Mapping, Sequence + +import rfc8785 + +from .errors import frontend_error + + +TRANSFORMATION_PLAN_CONTRACT = "openstatspec-transformation-plan-v0.1" +SPSS_FRONTEND_CONTRACT = "openstatspec-spss-syntax-frontend-v0.1" +_BINARY64 = re.compile(r"[0-9a-f]{16}") + + +def _invalid(detail: str, **details: Any): + raise frontend_error("invalid_transformation_plan", detail, **details) + + +@dataclass(frozen=True) +class TypedValue: + type: Literal["binary64", "string"] + bits: str | None = None + value: str | None = None + + def __post_init__(self) -> None: + if self.type == "binary64": + if self.value is not None or not isinstance(self.bits, str): + _invalid("A binary64 value requires bits and forbids string value.") + if _BINARY64.fullmatch(self.bits) is None: + _invalid("binary64 bits must be exactly 16 lowercase hexadecimal digits.") + number = struct.unpack(">d", bytes.fromhex(self.bits))[0] + if not math.isfinite(number): + _invalid("binary64 plan values must be finite.") + if self.bits == "8000000000000000": + _invalid("Negative zero is not canonical; use positive-zero bits.") + elif self.type == "string": + if self.bits is not None or not isinstance(self.value, str): + _invalid("A string value requires exact text and forbids bits.") + else: # pragma: no cover - guarded by the public type and from_dict + _invalid("Typed value type must be binary64 or string.") + + @classmethod + def binary64(cls, value: float) -> "TypedValue": + number = float(value) + if not math.isfinite(number): + _invalid("binary64 plan values must be finite.") + if number == 0.0: + number = 0.0 + return cls("binary64", bits=struct.pack(">d", number).hex()) + + @classmethod + def string(cls, value: str) -> "TypedValue": + return cls("string", value=value) + + @classmethod + def from_dict(cls, raw: Mapping[str, Any]) -> "TypedValue": + if not isinstance(raw, Mapping): + _invalid("Typed value must be an object.") + if raw.get("type") == "binary64" and set(raw) == {"type", "bits"}: + return cls("binary64", bits=raw.get("bits")) + if raw.get("type") == "string" and set(raw) == {"type", "value"}: + return cls("string", value=raw.get("value")) + _invalid("Typed value has an unknown type or unexpected fields.") + + def as_dict(self) -> dict[str, str]: + if self.type == "binary64": + assert self.bits is not None + return {"type": "binary64", "bits": self.bits} + assert self.value is not None + return {"type": "string", "value": self.value} + + def canonical_key(self) -> tuple[str, str]: + return ( + self.type, + self.bits if self.type == "binary64" else str(self.value), + ) + + def number(self) -> float: + if self.type != "binary64" or self.bits is None: + raise TypeError("Only binary64 values have a numeric value.") + return struct.unpack(">d", bytes.fromhex(self.bits))[0] + + +@dataclass(frozen=True) +class RecodeMatch: + kind: Literal["values", "range", "system_missing"] + values: tuple[TypedValue, ...] = () + lower: TypedValue | None = None + upper: TypedValue | None = None + + def __post_init__(self) -> None: + if self.kind == "values": + if not self.values or self.lower is not None or self.upper is not None: + _invalid("A values match requires a non-empty values array only.") + if not all(isinstance(value, TypedValue) for value in self.values): + _invalid("Every values-match entry must be a typed value.") + keys = [value.canonical_key() for value in self.values] + if len(keys) != len(set(keys)): + _invalid("A values match cannot contain duplicate typed values.") + elif self.kind == "range": + if self.values or self.lower is None or self.upper is None: + _invalid("A range match requires lower and upper only.") + if self.lower.type != "binary64" or self.upper.type != "binary64": + _invalid("Range endpoints must be binary64 values.") + if self.lower.number() > self.upper.number(): + _invalid("Range lower endpoint cannot exceed its upper endpoint.") + elif self.kind == "system_missing": + if self.values or self.lower is not None or self.upper is not None: + _invalid("A system_missing match has no value fields.") + else: # pragma: no cover + _invalid("Unknown recode match kind.") + + def as_dict(self) -> dict[str, Any]: + if self.kind == "values": + return {"kind": "values", "values": [value.as_dict() for value in self.values]} + if self.kind == "range": + assert self.lower is not None and self.upper is not None + return { + "kind": "range", "lower": self.lower.as_dict(), + "upper": self.upper.as_dict(), + } + return {"kind": "system_missing"} + + +@dataclass(frozen=True) +class RecodeResult: + kind: Literal["literal", "system_missing", "copy"] + value: TypedValue | None = None + + def __post_init__(self) -> None: + if self.kind == "literal": + if not isinstance(self.value, TypedValue): + _invalid("A literal result requires a typed value.") + elif self.kind in {"system_missing", "copy"}: + if self.value is not None: + _invalid(f"A {self.kind} result cannot contain a value.") + else: # pragma: no cover + _invalid("Unknown recode result kind.") + + def as_dict(self) -> dict[str, Any]: + result: dict[str, Any] = {"kind": self.kind} + if self.value is not None: + result["value"] = self.value.as_dict() + return result + + +@dataclass(frozen=True) +class RecodeRule: + match: RecodeMatch + result: RecodeResult + + def __post_init__(self) -> None: + if not isinstance(self.match, RecodeMatch): + _invalid("A recode rule requires a typed match.") + if not isinstance(self.result, RecodeResult): + _invalid("A recode rule requires a typed result.") + + def as_dict(self) -> dict[str, Any]: + return {"match": self.match.as_dict(), "result": self.result.as_dict()} + + +@dataclass(frozen=True) +class RecodeOperation: + source: str + target: str + target_mode: Literal["create", "replace"] + rules: tuple[RecodeRule, ...] + unmatched: RecodeResult + op: Literal["recode"] = "recode" + + def __post_init__(self) -> None: + if not isinstance(self.source, str) or not self.source: + _invalid("Recode source name must be non-empty text.") + if not isinstance(self.target, str) or not self.target: + _invalid("Recode source and target names must be non-empty.") + if self.op != "recode": + _invalid("Recode operation discriminator is invalid.") + if self.target_mode not in {"create", "replace"}: + _invalid("Recode target_mode must be create or replace.") + if not isinstance(self.rules, tuple) or not self.rules: + _invalid("Recode requires at least one non-ELSE rule.") + if not all(isinstance(rule, RecodeRule) for rule in self.rules): + _invalid("Recode rules must be typed rule objects.") + if not isinstance(self.unmatched, RecodeResult): + _invalid("Recode unmatched behavior must be a typed result.") + if self.target_mode == "replace" and self.source != self.target: + _invalid("A replace recode must target its source variable.") + + def as_dict(self) -> dict[str, Any]: + return { + "op": self.op, "source": self.source, "target": self.target, + "target_mode": self.target_mode, + "rules": [rule.as_dict() for rule in self.rules], + "unmatched": self.unmatched.as_dict(), + } + + +@dataclass(frozen=True) +class SetVariableLabelOperation: + variable: str + label: str + op: Literal["set_variable_label"] = "set_variable_label" + + def __post_init__(self) -> None: + if self.op != "set_variable_label": + _invalid("Variable-label operation discriminator is invalid.") + if not isinstance(self.variable, str) or not self.variable or not isinstance(self.label, str): + _invalid("Variable-label operation requires a variable and text label.") + + def as_dict(self) -> dict[str, Any]: + return {"op": self.op, "variable": self.variable, "label": self.label} + + +@dataclass(frozen=True) +class ValueLabel: + value: TypedValue + label: str + + def __post_init__(self) -> None: + if not isinstance(self.value, TypedValue) or not isinstance(self.label, str): + _invalid("A value label must be text.") + + def as_dict(self) -> dict[str, Any]: + return {"value": self.value.as_dict(), "label": self.label} + + +@dataclass(frozen=True) +class ReplaceValueLabelsOperation: + variable: str + labels: tuple[ValueLabel, ...] + op: Literal["replace_value_labels"] = "replace_value_labels" + + def __post_init__(self) -> None: + if self.op != "replace_value_labels": + _invalid("Value-label operation discriminator is invalid.") + if not isinstance(self.variable, str) or not self.variable or not self.labels: + _invalid("replace_value_labels requires a variable and non-empty labels.") + if not isinstance(self.labels, tuple) or not all( + isinstance(label, ValueLabel) for label in self.labels + ): + _invalid("replace_value_labels labels must be typed value-label objects.") + keys = [label.value.canonical_key() for label in self.labels] + if len(keys) != len(set(keys)): + _invalid("replace_value_labels cannot contain duplicate typed values.") + + def as_dict(self) -> dict[str, Any]: + return { + "op": self.op, "variable": self.variable, + "labels": [label.as_dict() for label in self.labels], + } + + +PlanOperation = RecodeOperation | SetVariableLabelOperation | ReplaceValueLabelsOperation + + +@dataclass(frozen=True) +class TransformationPlan: + operations: tuple[PlanOperation, ...] + contract: str = TRANSFORMATION_PLAN_CONTRACT + input_alias: str = "parent" + + def __post_init__(self) -> None: + if self.contract != TRANSFORMATION_PLAN_CONTRACT: + _invalid(f"Plan contract must be {TRANSFORMATION_PLAN_CONTRACT!r}.") + if not isinstance(self.input_alias, str) or not self.input_alias: + _invalid("Plan input_alias must be non-empty text.") + if not isinstance(self.operations, tuple) or not self.operations: + _invalid("A transformation plan requires at least one operation.") + if not all( + isinstance( + operation, + (RecodeOperation, SetVariableLabelOperation, ReplaceValueLabelsOperation), + ) + for operation in self.operations + ): + _invalid("Plan operations must be typed operation objects.") + + def as_dict(self) -> dict[str, Any]: + return { + "contract": self.contract, "input_alias": self.input_alias, + "operations": [operation.as_dict() for operation in self.operations], + } + + def canonical_bytes(self) -> bytes: + return rfc8785.dumps(self.as_dict()) + + def canonical_json(self) -> str: + return self.canonical_bytes().decode("utf-8") + + def sha256(self) -> str: + return hashlib.sha256(self.canonical_bytes()).hexdigest() + + +def _exact(raw: Mapping[str, Any], fields: set[str], label: str) -> None: + if set(raw) != fields: + _invalid(f"{label} fields must be exactly {sorted(fields)!r}.") + + +def _typed(raw: Any) -> TypedValue: + if not isinstance(raw, Mapping): + _invalid("Typed value must be an object.") + return TypedValue.from_dict(raw) + + +def _result(raw: Any) -> RecodeResult: + if not isinstance(raw, Mapping) or not isinstance(raw.get("kind"), str): + _invalid("Recode result must be an object with a kind.") + kind = raw["kind"] + if kind == "literal": + _exact(raw, {"kind", "value"}, "Literal result") + return RecodeResult("literal", _typed(raw["value"])) + if kind in {"system_missing", "copy"}: + _exact(raw, {"kind"}, f"{kind} result") + return RecodeResult(kind) + _invalid("Unknown recode result kind.") + + +def _match(raw: Any) -> RecodeMatch: + if not isinstance(raw, Mapping) or not isinstance(raw.get("kind"), str): + _invalid("Recode match must be an object with a kind.") + kind = raw["kind"] + if kind == "values": + _exact(raw, {"kind", "values"}, "Values match") + values = raw["values"] + if not isinstance(values, Sequence) or isinstance(values, (str, bytes)): + _invalid("Values match values must be an array.") + return RecodeMatch("values", tuple(_typed(value) for value in values)) + if kind == "range": + _exact(raw, {"kind", "lower", "upper"}, "Range match") + return RecodeMatch("range", lower=_typed(raw["lower"]), upper=_typed(raw["upper"])) + if kind == "system_missing": + _exact(raw, {"kind"}, "system_missing match") + return RecodeMatch("system_missing") + _invalid("Unknown recode match kind.") + + +def transformation_plan_from_dict(raw: Mapping[str, Any]) -> TransformationPlan: + """Strictly validate and construct the canonical v0.1 plan document.""" + if not isinstance(raw, Mapping): + _invalid("Transformation plan must be an object.") + _exact(raw, {"contract", "input_alias", "operations"}, "Transformation plan") + operations_raw = raw["operations"] + if not isinstance(operations_raw, Sequence) or isinstance(operations_raw, (str, bytes)): + _invalid("Plan operations must be an array.") + operations: list[PlanOperation] = [] + for raw_operation in operations_raw: + if not isinstance(raw_operation, Mapping): + _invalid("Every plan operation must be an object.") + operation = raw_operation.get("op") + if operation == "recode": + _exact( + raw_operation, + {"op", "source", "target", "target_mode", "rules", "unmatched"}, + "Recode operation", + ) + rules_raw = raw_operation["rules"] + if not isinstance(rules_raw, Sequence) or isinstance(rules_raw, (str, bytes)): + _invalid("Recode rules must be an array.") + rules = [] + for raw_rule in rules_raw: + if not isinstance(raw_rule, Mapping): + _invalid("Every recode rule must be an object.") + _exact(raw_rule, {"match", "result"}, "Recode rule") + rules.append(RecodeRule(_match(raw_rule["match"]), _result(raw_rule["result"]))) + operations.append(RecodeOperation( + source=raw_operation["source"], target=raw_operation["target"], + target_mode=raw_operation["target_mode"], rules=tuple(rules), + unmatched=_result(raw_operation["unmatched"]), + )) + elif operation == "set_variable_label": + _exact(raw_operation, {"op", "variable", "label"}, "Variable-label operation") + operations.append(SetVariableLabelOperation( + raw_operation["variable"], raw_operation["label"], + )) + elif operation == "replace_value_labels": + _exact(raw_operation, {"op", "variable", "labels"}, "Value-label operation") + labels_raw = raw_operation["labels"] + if not isinstance(labels_raw, Sequence) or isinstance(labels_raw, (str, bytes)): + _invalid("Value labels must be an array.") + labels = [] + for raw_label in labels_raw: + if not isinstance(raw_label, Mapping): + _invalid("Every value label must be an object.") + _exact(raw_label, {"value", "label"}, "Value label") + labels.append(ValueLabel(_typed(raw_label["value"]), raw_label["label"])) + operations.append(ReplaceValueLabelsOperation( + raw_operation["variable"], tuple(labels), + )) + else: + _invalid(f"Unknown plan operation {operation!r}.") + return TransformationPlan( + tuple(operations), contract=raw["contract"], input_alias=raw["input_alias"], + ) + + +def canonical_plan_json(plan: TransformationPlan | Mapping[str, Any]) -> str: + normalized = ( + plan if isinstance(plan, TransformationPlan) + else transformation_plan_from_dict(plan) + ) + return normalized.canonical_json() + + +def canonical_plan_hash(plan: TransformationPlan | Mapping[str, Any]) -> str: + normalized = ( + plan if isinstance(plan, TransformationPlan) + else transformation_plan_from_dict(plan) + ) + return normalized.sha256() diff --git a/src/openstatspec/transform/syntax.py b/src/openstatspec/transform/syntax.py new file mode 100644 index 0000000..c4df2e4 --- /dev/null +++ b/src/openstatspec/transform/syntax.py @@ -0,0 +1,478 @@ +"""Tokenizer, catalog-independent AST, and parser for the SPSS MVP subset.""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import math +import re +from typing import Literal + +from .errors import SourcePosition, SourceSpan, frontend_error + + +TokenKind = Literal[ + "identifier", "number", "string", "left_paren", "right_paren", + "equals", "comma", "slash", "period", "eof", +] + + +@dataclass(frozen=True) +class Token: + kind: TokenKind + text: str + value: str | float | None + span: SourceSpan + + +@dataclass(frozen=True) +class SyntaxLiteral: + kind: Literal["numeric", "string"] + value: float | str + span: SourceSpan + + +@dataclass(frozen=True) +class RecodeMatchSyntax: + kind: Literal["values", "range", "system_missing", "else"] + span: SourceSpan + values: tuple[SyntaxLiteral, ...] = () + lower: SyntaxLiteral | None = None + upper: SyntaxLiteral | None = None + + +@dataclass(frozen=True) +class RecodeResultSyntax: + kind: Literal["literal", "system_missing", "copy"] + span: SourceSpan + value: SyntaxLiteral | None = None + + +@dataclass(frozen=True) +class RecodeClauseSyntax: + match: RecodeMatchSyntax + result: RecodeResultSyntax + span: SourceSpan + + +@dataclass(frozen=True) +class RecodeCommandSyntax: + sources: tuple[Token, ...] + clauses: tuple[RecodeClauseSyntax, ...] + targets: tuple[Token, ...] | None + span: SourceSpan + + +@dataclass(frozen=True) +class VariableLabelSyntax: + variable: Token + label: Token + span: SourceSpan + + +@dataclass(frozen=True) +class VariableLabelsCommandSyntax: + assignments: tuple[VariableLabelSyntax, ...] + span: SourceSpan + + +@dataclass(frozen=True) +class ValueLabelSyntax: + value: SyntaxLiteral + label: Token + span: SourceSpan + + +@dataclass(frozen=True) +class ValueLabelsGroupSyntax: + variables: tuple[Token, ...] + labels: tuple[ValueLabelSyntax, ...] + span: SourceSpan + + +@dataclass(frozen=True) +class ValueLabelsCommandSyntax: + groups: tuple[ValueLabelsGroupSyntax, ...] + span: SourceSpan + + +SyntaxCommand = ( + RecodeCommandSyntax | VariableLabelsCommandSyntax | ValueLabelsCommandSyntax +) + + +@dataclass(frozen=True) +class SpssSyntaxProgram: + commands: tuple[SyntaxCommand, ...] + span: SourceSpan + + +_NUMBER = re.compile( + r"[+-]?(?:(?:[0-9]+(?:\.[0-9]*)?)|(?:\.[0-9]+))(?:[Ee][+-]?[0-9]+)?" +) +_IDENTIFIER_START = frozenset("_@$#") +_IDENTIFIER_CONTINUE = frozenset("_@$#") + + +def _position(source: str, offset: int) -> SourcePosition: + prefix = source[:offset] + line = prefix.count("\n") + 1 + last_newline = prefix.rfind("\n") + column = offset + 1 if last_newline < 0 else offset - last_newline + return SourcePosition(offset=offset, line=line, column=column) + + +def _span(source: str, start: int, end: int) -> SourceSpan: + return SourceSpan(_position(source, start), _position(source, end)) + + +def _joined_span(first: SourceSpan, last: SourceSpan) -> SourceSpan: + return SourceSpan(first.start, last.end) + + +def normalize_spss_source(source: str) -> str: + """Normalize source line endings without changing any other source byte.""" + if not isinstance(source, str): + raise TypeError("source must be text") + return source.replace("\r\n", "\n").replace("\r", "\n") + + +def spss_source_hash(source: str) -> str: + """Hash the exact UTF-8 source after normative LF normalization.""" + normalized = normalize_spss_source(source) + return hashlib.sha256(normalized.encode("utf-8")).hexdigest() + + +def tokenize_spss(source: str) -> tuple[Token, ...]: + """Tokenize supported SPSS text without consulting a dataset catalog.""" + if not isinstance(source, str): + raise TypeError("source must be text") + tokens: list[Token] = [] + offset = 0 + punctuation: dict[str, TokenKind] = { + "(": "left_paren", ")": "right_paren", "=": "equals", + ",": "comma", "/": "slash", ".": "period", + } + while offset < len(source): + character = source[offset] + if character.isspace(): + offset += 1 + continue + if character in {"'", '"'}: + start = offset + quote = character + offset += 1 + value: list[str] = [] + while offset < len(source): + character = source[offset] + if character == quote: + if offset + 1 < len(source) and source[offset + 1] == quote: + value.append(quote) + offset += 2 + continue + offset += 1 + tokens.append(Token( + "string", source[start:offset], "".join(value), + _span(source, start, offset), + )) + break + value.append(character) + offset += 1 + else: + raise frontend_error( + "spss_syntax_error", "Unterminated string literal.", + span=_span(source, start, len(source)), + ) + continue + number = _NUMBER.match(source, offset) + if number is not None: + start, offset = offset, number.end() + text = source[start:offset] + try: + value = float(text) + except ValueError as error: # pragma: no cover - guarded by regex + raise frontend_error( + "spss_syntax_error", "Invalid numeric literal.", + span=_span(source, start, offset), + ) from error + if not math.isfinite(value): + raise frontend_error( + "spss_syntax_error", "Numeric literals must be finite binary64 values.", + span=_span(source, start, offset), + ) + tokens.append(Token( + "number", text, value, _span(source, start, offset), + )) + continue + if character.isalpha() or character in _IDENTIFIER_START: + start = offset + offset += 1 + while offset < len(source): + candidate = source[offset] + if candidate.isalnum() or candidate in _IDENTIFIER_CONTINUE: + offset += 1 + continue + if ( + candidate == "." and offset + 1 < len(source) + and ( + source[offset + 1].isalnum() + or source[offset + 1] in _IDENTIFIER_CONTINUE + ) + ): + offset += 1 + continue + break + text = source[start:offset] + tokens.append(Token( + "identifier", text, text, _span(source, start, offset), + )) + continue + if character in punctuation: + tokens.append(Token( + punctuation[character], character, character, + _span(source, offset, offset + 1), + )) + offset += 1 + continue + raise frontend_error( + "spss_syntax_error", f"Unexpected character {character!r}.", + span=_span(source, offset, offset + 1), + ) + end = _position(source, len(source)) + tokens.append(Token("eof", "", None, SourceSpan(end, end))) + return tuple(tokens) + + +class _Parser: + def __init__(self, source: str) -> None: + self.source = source + self.tokens = tokenize_spss(source) + self.index = 0 + + @property + def current(self) -> Token: + return self.tokens[self.index] + + def advance(self) -> Token: + token = self.current + if token.kind != "eof": + self.index += 1 + return token + + def accepts(self, kind: TokenKind) -> Token | None: + if self.current.kind == kind: + return self.advance() + return None + + def accepts_keyword(self, keyword: str) -> Token | None: + token = self.current + if token.kind == "identifier" and token.text.casefold() == keyword.casefold(): + return self.advance() + return None + + def expects(self, kind: TokenKind, detail: str) -> Token: + token = self.accepts(kind) + if token is None: + raise frontend_error("spss_syntax_error", detail, span=self.current.span) + return token + + def expects_keyword(self, keyword: str) -> Token: + token = self.accepts_keyword(keyword) + if token is None: + raise frontend_error( + "spss_syntax_error", f"Expected keyword {keyword}.", + span=self.current.span, + ) + return token + + def variable_list(self, *, stop_kinds: frozenset[str]) -> tuple[Token, ...]: + variables: list[Token] = [] + while self.current.kind not in stop_kinds: + if self.accepts("comma") is not None: + continue + variables.append(self.expects("identifier", "Expected a variable name.")) + if not variables: + raise frontend_error( + "spss_syntax_error", "Expected at least one variable name.", + span=self.current.span, + ) + return tuple(variables) + + def literal(self) -> SyntaxLiteral: + token = self.current + if token.kind == "number": + self.advance() + assert isinstance(token.value, float) + return SyntaxLiteral("numeric", token.value, token.span) + if token.kind == "string": + self.advance() + assert isinstance(token.value, str) + return SyntaxLiteral("string", token.value, token.span) + raise frontend_error( + "spss_syntax_error", "Expected a numeric or string literal.", + span=token.span, + ) + + def recode_result(self) -> RecodeResultSyntax: + if (token := self.accepts_keyword("SYSMIS")) is not None: + return RecodeResultSyntax("system_missing", token.span) + if (token := self.accepts_keyword("COPY")) is not None: + return RecodeResultSyntax("copy", token.span) + literal = self.literal() + return RecodeResultSyntax("literal", literal.span, value=literal) + + def recode_clause(self) -> RecodeClauseSyntax: + left = self.expects("left_paren", "Expected '(' before a RECODE rule.") + if (token := self.accepts_keyword("ELSE")) is not None: + match = RecodeMatchSyntax("else", token.span) + elif (token := self.accepts_keyword("SYSMIS")) is not None: + match = RecodeMatchSyntax("system_missing", token.span) + else: + first = self.literal() + if self.accepts_keyword("THRU") is not None: + upper = self.literal() + match = RecodeMatchSyntax( + "range", _joined_span(first.span, upper.span), + lower=first, upper=upper, + ) + else: + values = [first] + while self.current.kind != "equals": + self.accepts("comma") + if self.current.kind == "equals": + break + values.append(self.literal()) + match = RecodeMatchSyntax( + "values", _joined_span(values[0].span, values[-1].span), + values=tuple(values), + ) + self.expects("equals", "Expected '=' in a RECODE rule.") + result = self.recode_result() + right = self.expects("right_paren", "Expected ')' after a RECODE rule.") + return RecodeClauseSyntax(match, result, _joined_span(left.span, right.span)) + + def recode(self, start: Token) -> RecodeCommandSyntax: + sources = self.variable_list(stop_kinds=frozenset({"left_paren", "period", "eof"})) + clauses: list[RecodeClauseSyntax] = [] + while self.current.kind == "left_paren": + clauses.append(self.recode_clause()) + if not clauses: + raise frontend_error( + "spss_syntax_error", "RECODE requires at least one rule.", + span=self.current.span, + ) + else_indexes = [ + index for index, clause in enumerate(clauses) + if clause.match.kind == "else" + ] + if len(else_indexes) > 1: + duplicate = clauses[else_indexes[1]] + raise frontend_error( + "duplicate_else", "RECODE may contain at most one ELSE rule.", + span=duplicate.match.span, + ) + if else_indexes and else_indexes[0] != len(clauses) - 1: + raise frontend_error( + "else_not_last", "ELSE must be the last RECODE rule.", + span=clauses[else_indexes[0]].match.span, + ) + targets = None + if self.accepts_keyword("INTO") is not None: + targets = self.variable_list(stop_kinds=frozenset({"period", "eof"})) + if len(targets) != len(sources): + raise frontend_error( + "spss_syntax_error", + "RECODE INTO requires one target for every source variable.", + span=_joined_span(targets[0].span, targets[-1].span), + ) + end = self.expects("period", "Expected '.' after RECODE.") + return RecodeCommandSyntax( + sources, tuple(clauses), targets, _joined_span(start.span, end.span), + ) + + def variable_labels(self, start: Token) -> VariableLabelsCommandSyntax: + self.expects_keyword("LABELS") + assignments: list[VariableLabelSyntax] = [] + while self.current.kind not in {"period", "eof"}: + self.accepts("slash") + variable = self.expects("identifier", "Expected a variable name.") + label = self.expects("string", "Expected a quoted variable label.") + assignments.append(VariableLabelSyntax( + variable, label, _joined_span(variable.span, label.span), + )) + if not assignments: + raise frontend_error( + "spss_syntax_error", "VARIABLE LABELS requires an assignment.", + span=self.current.span, + ) + end = self.expects("period", "Expected '.' after VARIABLE LABELS.") + return VariableLabelsCommandSyntax( + tuple(assignments), _joined_span(start.span, end.span), + ) + + def value_labels(self, start: Token) -> ValueLabelsCommandSyntax: + self.expects_keyword("LABELS") + groups: list[ValueLabelsGroupSyntax] = [] + while self.current.kind not in {"period", "eof"}: + self.accepts("slash") + group_start = self.current + variables = self.variable_list( + stop_kinds=frozenset({"number", "string", "period", "slash", "eof"}) + ) + labels: list[ValueLabelSyntax] = [] + while self.current.kind not in {"period", "slash", "eof"}: + value = self.literal() + label = self.expects("string", "Expected a quoted value label.") + labels.append(ValueLabelSyntax( + value, label, _joined_span(value.span, label.span), + )) + if not labels: + raise frontend_error( + "spss_syntax_error", "VALUE LABELS requires at least one value-label pair.", + span=self.current.span, + ) + groups.append(ValueLabelsGroupSyntax( + variables, tuple(labels), + _joined_span(group_start.span, labels[-1].span), + )) + end = self.expects("period", "Expected '.' after VALUE LABELS.") + return ValueLabelsCommandSyntax( + tuple(groups), _joined_span(start.span, end.span), + ) + + def parse(self) -> SpssSyntaxProgram: + commands: list[SyntaxCommand] = [] + while self.current.kind != "eof": + start = self.expects("identifier", "Expected an SPSS command.") + command = start.text.casefold() + if command == "recode": + commands.append(self.recode(start)) + elif command == "variable": + commands.append(self.variable_labels(start)) + elif command == "value": + commands.append(self.value_labels(start)) + else: + raise frontend_error( + "unsupported_spss_command", + f"Unsupported SPSS command {start.text!r}.", span=start.span, + command=start.text, + ) + if commands: + program_span = _joined_span(commands[0].span, commands[-1].span) + else: + program_span = self.current.span + return SpssSyntaxProgram(tuple(commands), program_span) + + +def parse_spss_syntax(source: str) -> SpssSyntaxProgram: + """Parse the supported command subset into a catalog-independent AST.""" + normalized = normalize_spss_source(source) + comment = re.search(r"(?m)^[ \t]*\*", normalized) + if comment is not None: + raise frontend_error( + "unsupported_spss_command", + "SPSS comment statements are outside the v0.1 subset.", + span=_span(normalized, comment.start(), comment.start() + 1), + command="*", + ) + return _Parser(normalized).parse() diff --git a/tests/test_cli.py b/tests/test_cli.py index 6b84ba9..5e3b026 100755 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -39,8 +39,8 @@ def test_cli_import_inspect_validate_and_export_emit_json(tmp_path, capsys) -> N def test_capability_matrix_is_public_and_cli_matches_engine_boundary(capsys) -> None: matrix = openstatspec.capability_matrix() assert matrix["specification_status"] == "released" - assert matrix["specification_release"] == "v0.1.0" - assert matrix["specification_commit"] == "d287c2cde9ade71f04e27dd012caec876901aed5" + assert matrix["specification_release"] == "v0.2.0" + assert matrix["specification_commit"] == "79339ec3d8f8aa81789b7e85f6b8afa6f1374e50" assert matrix["directions"] == ["import", "export", "semantic_round_trip"] assert matrix["active_connection"] is None assert matrix["engine"]["package"] == "openstatspec-pyspssio" diff --git a/tests/test_inplace_transform.py b/tests/test_inplace_transform.py new file mode 100644 index 0000000..ea2db14 --- /dev/null +++ b/tests/test_inplace_transform.py @@ -0,0 +1,309 @@ +from __future__ import annotations + +import sqlite3 +from types import SimpleNamespace + +import pytest +from sqlalchemy import create_engine, inspect, text + +import openstatspec +import openstatspec.sql.inplace_transform as inplace_transform +from openstatspec.sql.inplace_transform import _apply_on_connection +from openstatspec.sql.wide import create_wide_dataset + + +def _variables() -> list[dict[str, object]]: + return [{ + "ordinal": 1, + "source_name": "score", + "physical_name": "score", + "storage_kind": "numeric", + "string_width": None, + "label": "Score", + "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": "[]", + }] + + +@pytest.fixture +def catalog(tmp_path): + path = tmp_path / "in-place.sqlite" + url = f"sqlite:///{path}" + create_wide_dataset( + database_url=url, + dataset_id="in_place_source", + source_name="source.sav", + source_format="SAV", + source_sha256="d" * 64, + rows=[{"score": 1.0}, {"score": 2.0}, {"score": 3.0}], + variables=_variables(), + ) + openstatspec.install_in_place_transformation_schema(database_url=url) + connection = sqlite3.connect(path) + dataset_id, table_name = connection.execute( + "SELECT dataset_id, physical_table_name FROM dataset" + ).fetchone() + return url, path, dataset_id, table_name + + +def test_plan_applies_to_same_dataset_and_physical_table_without_copy( + catalog, +) -> None: + url, path, dataset_id, table_name = catalog + engine = create_engine(url) + with engine.begin() as connection: + before_datasets = connection.execute(text( + "SELECT COUNT(*) FROM dataset" + )).scalar_one() + before_data_tables = { + name for name in inspect(connection).get_table_names() + if name.startswith("data_") + } + result = _apply_on_connection( + connection, + dataset_id=dataset_id, + source_text=( + "RECODE score (1,2 = 0) (3 = 1) INTO score_band. " + "VARIABLE LABELS score_band 'Score band'. " + "VALUE LABELS score_band 0 'Lower' 1 'Upper'." + ), + actor="test-agent", + database_profile="sqlite", + allow_schema_change=True, + dolt_branch="feature/recode", + dolt_head="abc123", + ) + after_datasets = connection.execute(text( + "SELECT COUNT(*) FROM dataset" + )).scalar_one() + after_data_tables = { + name for name in inspect(connection).get_table_names() + if name.startswith("data_") + } + tables = set(inspect(connection).get_table_names()) + + assert result["dataset_id"] == dataset_id + assert result["physical_table_name"] == table_name + assert before_datasets == after_datasets == 1 + assert before_data_tables == after_data_tables == {table_name} + assert not { + name for name in tables + if name.startswith("derived_") + or "staging" in name + or "rollback" in name + or "snapshot" in name + or name.startswith("transformation_plan_") + } + + connection = sqlite3.connect(path) + assert connection.execute( + f'SELECT __case_ordinal, score, score_band FROM "{table_name}" ' + "ORDER BY __case_ordinal" + ).fetchall() == [(1, 1.0, 0.0), (2, 2.0, 0.0), (3, 3.0, 1.0)] + assert connection.execute( + "SELECT dataset_id, physical_table_name FROM dataset" + ).fetchone() == (dataset_id, table_name) + assert connection.execute( + "SELECT variable_label FROM variable WHERE source_name = 'score_band'" + ).fetchone() == ("Score band",) + assert connection.execute( + "SELECT numeric_code, label FROM value_label ORDER BY ordinal" + ).fetchall() == [(0.0, "Lower"), (1.0, "Upper")] + assert connection.execute( + "SELECT label FROM variable_catalog WHERE source_name = 'score_band'" + ).fetchone() == ("Score band",) + assert connection.execute( + "SELECT database_profile, dolt_branch, dolt_head_before, " + "dolt_head_after, actor, status " + "FROM transformation_apply" + ).fetchone() == ( + "sqlite", "feature/recode", "abc123", "abc123", "test-agent", + "succeeded", + ) + + +def test_public_apply_supports_non_dolt_without_building_undo(catalog) -> None: + url, path, dataset_id, table_name = catalog + result = openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text="RECODE score (1 = 0).", + actor="test-agent", + ) + assert result["dolt_branch"] is None + assert result["dolt_commit_performed"] is False + assert sqlite3.connect(path).execute( + f'SELECT score FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(0.0,), (2.0,), (3.0,)] + + +def test_missing_audit_schema_fails_before_mutation(catalog) -> None: + url, path, dataset_id, table_name = catalog + connection = sqlite3.connect(path) + connection.execute("DROP TABLE transformation_apply") + connection.commit() + connection.close() + with pytest.raises(openstatspec.TransformationError) as caught: + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text="RECODE score (1 = 0).", + actor="test-agent", + ) + assert caught.value.code == "in_place_audit_schema_missing" + assert sqlite3.connect(path).execute( + f'SELECT score FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(1.0,), (2.0,), (3.0,)] + + +def test_nontransactional_ddl_profile_rejects_create_before_mutation( + catalog, +) -> None: + url, path, dataset_id, table_name = catalog + engine = create_engine(url) + with pytest.raises(openstatspec.TransformationError) as caught: + with engine.begin() as connection: + _apply_on_connection( + connection, + dataset_id=dataset_id, + source_text=( + "VARIABLE LABELS score 'Changed'. " + "RECODE score (1 = 0) INTO score_band." + ), + actor="test-agent", + database_profile="mysql", + allow_schema_change=False, + dolt_branch=None, + dolt_head=None, + ) + engine.dispose() + assert caught.value.code == "schema_change_not_atomic" + connection = sqlite3.connect(path) + assert connection.execute( + "SELECT variable_label FROM variable WHERE source_name = 'score'" + ).fetchone() == ("Score",) + assert "score_band" not in { + row[1] for row in connection.execute( + f'PRAGMA table_info("{table_name}")' + ) + } + + +def test_public_apply_binds_expected_dolt_branch_and_head( + catalog, monkeypatch, +) -> None: + url, path, dataset_id, table_name = catalog + monkeypatch.setattr( + inplace_transform, + "effective_profile", + lambda _url: (SimpleNamespace(name="dolt"), {}), + ) + states = iter([ + ("feature/recode", "abc123", 0), + ("feature/recode", "abc123", 4), + ]) + monkeypatch.setattr( + inplace_transform, "_dolt_state", lambda _connection: next(states) + ) + result = openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text="RECODE score (1 = 9).", + actor="test-agent", + expected_branch="feature/recode", + expected_head="abc123", + ) + assert result["dolt_commit_performed"] is False + assert sqlite3.connect(path).execute( + f'SELECT score FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(9.0,), (2.0,), (3.0,)] + + +def test_public_apply_rejects_dirty_dolt_working_set_before_mutation( + catalog, monkeypatch, +) -> None: + url, path, dataset_id, table_name = catalog + monkeypatch.setattr( + inplace_transform, + "effective_profile", + lambda _url: (SimpleNamespace(name="dolt"), {}), + ) + monkeypatch.setattr( + inplace_transform, + "_dolt_state", + lambda _connection: ("feature/recode", "abc123", 1), + ) + with pytest.raises(openstatspec.TransformationError) as caught: + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text="RECODE score (1 = 9).", + actor="test-agent", + expected_branch="feature/recode", + expected_head="abc123", + ) + assert caught.value.code == "dolt_working_set_dirty" + assert sqlite3.connect(path).execute( + f'SELECT score FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(1.0,), (2.0,), (3.0,)] + + +@pytest.mark.parametrize( + ("state", "expected_error"), + [ + (("main", "abc123", 0), "dolt_branch_mismatch"), + (("feature/recode", "other-head", 0), "dolt_head_mismatch"), + ], +) +def test_public_apply_rejects_dolt_context_mismatch_before_mutation( + catalog, monkeypatch, state, expected_error, +) -> None: + url, path, dataset_id, table_name = catalog + monkeypatch.setattr( + inplace_transform, + "effective_profile", + lambda _url: (SimpleNamespace(name="dolt"), {}), + ) + monkeypatch.setattr( + inplace_transform, "_dolt_state", lambda _connection: state + ) + with pytest.raises(openstatspec.TransformationError) as caught: + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text="RECODE score (1 = 9).", + actor="test-agent", + expected_branch="feature/recode", + expected_head="abc123", + ) + assert caught.value.code == expected_error + assert sqlite3.connect(path).execute( + f'SELECT score FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(1.0,), (2.0,), (3.0,)] + + +def test_capability_declares_dolt_owned_versioning() -> None: + declaration = openstatspec.capability_matrix()["optional_profiles"][ + "spss_in_place_transformation" + ] + assert declaration["database_products"] == [ + "sqlite", "postgresql", "mysql", "mariadb", "dolt", + ] + assert declaration["status"] == "experimental" + assert declaration["execution_evidence"]["sqlite"] == "local_conformance" + assert declaration["execution_evidence"]["dolt"] == ( + "service_conformance_required" + ) + assert declaration["creates_derived_dataset"] is False + assert declaration["creates_persistent_data_copy"] is False + assert declaration["openstatspec_rollback_or_version_history"] is False + assert declaration["performs_dolt_commit"] is False diff --git a/tests/test_sql_profiles.py b/tests/test_sql_profiles.py index 54d6c5f..5988265 100755 --- a/tests/test_sql_profiles.py +++ b/tests/test_sql_profiles.py @@ -28,10 +28,10 @@ def test_profile_detection_tracks_supported_dialect_urls() -> None: def test_profile_declarations_publish_released_specification_provenance() -> None: for declaration in capabilities.profile_declarations().values(): assert declaration["specification_status"] == "released" - assert declaration["specification_release"] == "v0.1.0" + assert declaration["specification_release"] == "v0.2.0" assert ( declaration["specification_commit"] - == "d287c2cde9ade71f04e27dd012caec876901aed5" + == "79339ec3d8f8aa81789b7e85f6b8afa6f1374e50" ) def test_profile_preflight_fails_without_transforming_a_wide_dataset() -> None: diff --git a/tests/test_transform_frontend.py b/tests/test_transform_frontend.py new file mode 100644 index 0000000..400f685 --- /dev/null +++ b/tests/test_transform_frontend.py @@ -0,0 +1,330 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest +from openstatspec.transform import ( + TransformationFrontendError, + TypedValue, + ValueLabel, + VariableDefinition, + VariableSchema, + bind_spss_syntax, + canonical_plan_hash, + canonical_plan_json, + compile_spss_syntax, + normalize_spss_source, + parse_spss_syntax, + spss_source_hash, + transformation_plan_from_dict, +) + + +def _schema(*variables: VariableDefinition) -> VariableSchema: + return VariableSchema(tuple(variables)) + + +def _compile(source: str, schema: VariableSchema): + return bind_spss_syntax(parse_spss_syntax(source), schema) + + +def _error(source: str, schema: VariableSchema) -> TransformationFrontendError: + with pytest.raises(TransformationFrontendError) as caught: + _compile(source, schema) + return caught.value + + +def _frontend_conformance_manifest() -> Path: + configured = os.environ.get("OPENSTATSPEC_SPECIFICATION_DIR") + candidates = [ + ( + Path(configured) / "conformance/spss-syntax-frontend-0.1.json" + if configured + else None + ), + Path(__file__).resolve().parents[1] + / "openstatspec-specification/conformance/spss-syntax-frontend-0.1.json", + Path(__file__).resolve().parents[2] + / "specification/conformance/spss-syntax-frontend-0.1.json", + ] + for candidate in candidates: + if candidate and candidate.is_file(): + return candidate + raise RuntimeError( + "The SPSS frontend conformance fixture is required; " + "set OPENSTATSPEC_SPECIFICATION_DIR." + ) + + +def test_official_spss_frontend_conformance_manifest() -> None: + manifest_path = _frontend_conformance_manifest() + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + plan_manifest = json.loads( + (manifest_path.parent / "transformation-plan-0.1.json").read_text( + encoding="utf-8" + ) + ) + plan_cases = {case["id"]: case for case in plan_manifest["cases"]} + + for case in manifest["cases"]: + request = case["request"] + schema = VariableSchema(tuple( + VariableDefinition( + variable["name"], + variable["storage_kind"], + variable_label=variable.get("variable_label"), + value_labels=tuple( + ValueLabel( + TypedValue.from_dict(label["value"]), + label["label"], + ) + for label in variable.get("value_labels", []) + ), + ) + for variable in request["input_schema"]["variables"] + )) + assert spss_source_hash(request["source_text"]) == case["expected_source_hash"] + if case["expected_error"] is not None: + with pytest.raises(TransformationFrontendError) as caught: + compile_spss_syntax( + request["source_text"], + schema, + input_alias=request["input_alias"], + ) + assert caught.value.code == case["expected_error"], case["id"] + continue + + compilation = compile_spss_syntax( + request["source_text"], + schema, + input_alias=request["input_alias"], + ) + if "expected_plan_case" in case: + expected_plan = plan_cases[case["expected_plan_case"]]["plan"] + expected_hash = plan_cases[case["expected_plan_case"]][ + "expected_plan_hash" + ] + else: + expected_plan = case["expected_plan"] + expected_hash = case["expected_plan_hash"] + assert compilation.plan.as_dict() == expected_plan, case["id"] + assert compilation.plan_hash == expected_hash, case["id"] + if "expected_output_metadata" in case: + actual_metadata = { + variable.name: { + "variable_label": variable.variable_label, + "value_labels": [ + label.as_dict() for label in variable.value_labels + ], + } + for variable in compilation.bound.output_schema.variables + } + assert actual_metadata == case["expected_output_metadata"], case["id"] + + +def test_recode_and_labels_lower_to_exact_canonical_plan() -> None: + source = ( + "RECODE q1 (1,2 = 0) (3 THRU 5 = 1) (ELSE = SYSMIS) " + "INTO q1_binary.\n" + "VARIABLE LABELS q1_binary 'Positive response'.\n" + "VALUE LABELS q1_binary 0 'No' 1 'Yes'." + ) + bound = _compile(source, _schema(VariableDefinition("q1", "numeric"))) + + manifest_path = ( + Path(__file__).parents[2] + / "worktree-spec-transform-plan" + / "conformance" + / "transformation-plan-0.1.json" + ) + if manifest_path.exists(): + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + expected = next( + case["plan"] + for case in manifest["cases"] + if case["id"] == "numeric-recode-and-declared-labels" + ) + assert bound.plan.as_dict() == expected + + assert canonical_plan_json(bound.plan) == canonical_plan_json( + bound.plan.as_dict() + ) + assert canonical_plan_hash(bound.plan) == bound.plan.sha256() + assert bound.output_schema.variables[-1].variable_label == "Positive response" + assert [label.label for label in bound.output_schema.variables[-1].value_labels] == [ + "No", + "Yes", + ] + + +def test_recode_varlists_lower_positionally_from_precommand_state() -> None: + bound = _compile( + "RECODE first second (1 = 9) INTO first_new second_new.", + _schema( + VariableDefinition("first", "numeric"), + VariableDefinition("second", "numeric"), + ), + ) + assert [ + (operation.source, operation.target) + for operation in bound.plan.operations + ] == [("first", "first_new"), ("second", "second_new")] + assert [variable.name for variable in bound.output_schema.variables] == [ + "first", + "second", + "first_new", + "second_new", + ] + + +def test_recode_varlist_target_count_and_collisions_are_rejected() -> None: + schema = _schema( + VariableDefinition("first", "numeric"), + VariableDefinition("second", "numeric"), + ) + assert _error( + "RECODE first second (1 = 9) INTO only_one.", schema + ).code == "spss_syntax_error" + assert _error( + "RECODE first second (1 = 9) INTO new NEW.", schema + ).code == "target_already_exists" + assert _error( + "RECODE first (1 = 9) INTO second.", schema + ).code == "target_already_exists" + + +def test_value_label_varlists_and_slash_groups_preserve_source_order() -> None: + bound = _compile( + "VALUE LABELS a b 0 'No' 1 'Yes' / color 'R' 'Red'.", + _schema( + VariableDefinition("a", "numeric"), + VariableDefinition("b", "numeric"), + VariableDefinition("color", "string"), + ), + ) + assert [operation.variable for operation in bound.plan.operations] == [ + "a", + "b", + "color", + ] + assert [ + label.label + for label in bound.output_schema.variables[0].value_labels + ] == ["No", "Yes"] + assert bound.output_schema.variables[2].value_labels[0].value == TypedValue.string( + "R" + ) + + +def test_in_place_recode_preserves_all_existing_metadata() -> None: + labels = ( + ValueLabel(TypedValue.binary64(1), "One"), + ValueLabel(TypedValue.binary64(2), "Two"), + ) + original = VariableDefinition( + "score", "numeric", variable_label="Score", value_labels=labels + ) + bound = _compile("RECODE score (1 = 2).", _schema(original)) + assert bound.output_schema.variables == (original,) + + changed = _compile( + "RECODE score (1 = 2). " + "VARIABLE LABELS score 'Changed'. " + "VALUE LABELS score 2 'Two only'.", + _schema(original), + ) + output = changed.output_schema.variables[0] + assert output.variable_label == "Changed" + assert [label.label for label in output.value_labels] == ["Two only"] + + +def test_strings_quotes_case_and_exact_catalog_spelling() -> None: + bound = _compile( + "variable labels NAME 'O''Brien – nimi'. " + "value labels Name 'x' 'Täpselt'.", + _schema(VariableDefinition("Name", "string")), + ) + assert [operation.variable for operation in bound.plan.operations] == [ + "Name", + "Name", + ] + assert bound.output_schema.variables[0].variable_label == "O'Brien – nimi" + + +def test_source_normalization_hash_and_positions_are_stable() -> None: + lf = "VARIABLE LABELS q1 'One'.\nVALUE LABELS q1 1 'Yes'." + crlf = lf.replace("\n", "\r\n") + assert normalize_spss_source(crlf) == lf + assert spss_source_hash(crlf) == spss_source_hash(lf) + program = parse_spss_syntax(crlf) + assert program.commands[1].span.start.line == 2 + compilation = compile_spss_syntax( + crlf, _schema(VariableDefinition("q1", "numeric")) + ) + assert compilation.source_text_lf == lf + assert compilation.source_hash == spss_source_hash(lf) + assert compilation.plan_hash == compilation.plan.sha256() + + +@pytest.mark.parametrize( + ("source", "code"), + [ + ("FREQUENCIES q1.", "unsupported_spss_command"), + ("COMMENT ignored.", "unsupported_spss_command"), + ("* ignored.", "unsupported_spss_command"), + ("RECODE q1 (1 = 0) /* ignored */.", "spss_syntax_error"), + ("RECODE q1 (ELSE = 0) (1 = 1).", "else_not_last"), + ("RECODE q1 (1 = 0) (ELSE = 1) (ELSE = 2).", "duplicate_else"), + ("RECODE q1 (5 THRU 3 = 1).", "invalid_numeric_range"), + ("VARIABLE LABELS missing 'No'.", "unknown_variable"), + ], +) +def test_stable_failures(source: str, code: str) -> None: + assert _error( + source, _schema(VariableDefinition("q1", "numeric")) + ).code == code + + +def test_string_create_requires_declaration_before_mixed_type_diagnostic() -> None: + error = _error( + "RECODE color ('R' = 'red') INTO normalized.", + _schema(VariableDefinition("color", "string")), + ) + assert error.code == "string_target_requires_declaration" + + +def test_negative_zero_canonicalizes_and_collides_with_positive_zero() -> None: + error = _error( + "VALUE LABELS q1 -0 'Minus' 0 'Plus'.", + _schema(VariableDefinition("q1", "numeric")), + ) + assert error.code == "duplicate_value_label" + assert TypedValue.binary64(-0.0).bits == "0000000000000000" + + +def test_strict_plan_loader_rejects_runtime_type_confusion() -> None: + raw = { + "contract": "openstatspec-transformation-plan-v0.1", + "input_alias": "parent", + "operations": [ + { + "op": "set_variable_label", + "variable": 7, + "label": "bad", + } + ], + } + with pytest.raises(TransformationFrontendError) as caught: + transformation_plan_from_dict(raw) + assert caught.value.code == "invalid_transformation_plan" + + +def test_custom_nonempty_input_alias_is_canonical() -> None: + plan = bind_spss_syntax( + parse_spss_syntax("VARIABLE LABELS q1 'One'."), + _schema(VariableDefinition("q1", "numeric")), + input_alias="survey", + ).plan + assert plan.input_alias == "survey"