From 051e12f763954364c3259028ea0aefbafd874be8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Fri, 31 Jul 2026 18:53:56 +0300 Subject: [PATCH] Package transformation frontends --- CHANGELOG.md | 31 +++ README.md | 52 ++--- docs/in-place-transformation.md | 46 +--- docs/release-readiness.md | 32 ++- docs/transformations.md | 204 +++++++++++++++++ pyproject.toml | 2 +- src/openstatspec/__init__.py | 18 +- src/openstatspec/api.py | 21 +- src/openstatspec/cli.py | 38 ++- src/openstatspec/frontends/__init__.py | 1 + src/openstatspec/frontends/sas/.gitkeep | 0 src/openstatspec/frontends/spss/__init__.py | 26 +++ .../{transform => frontends/spss}/binding.py | 55 +---- .../{transform => frontends/spss}/compiler.py | 5 +- src/openstatspec/frontends/spss/execution.py | 48 ++++ .../{transform => frontends/spss}/syntax.py | 2 +- src/openstatspec/frontends/stata/.gitkeep | 0 src/openstatspec/sql/inplace_transform.py | 216 ++++++++++++++---- src/openstatspec/transform/__init__.py | 41 +++- src/openstatspec/transform/plan.py | 1 - src/openstatspec/transform/schema.py | 46 ++++ src/openstatspec/transform/validation.py | 178 +++++++++++++++ tests/test_cli.py | 63 +++++ tests/test_inplace_transform.py | 215 ++++++++++++++++- tests/test_transform_frontend.py | 47 +++- 25 files changed, 1184 insertions(+), 204 deletions(-) create mode 100644 docs/transformations.md create mode 100644 src/openstatspec/frontends/__init__.py create mode 100644 src/openstatspec/frontends/sas/.gitkeep create mode 100644 src/openstatspec/frontends/spss/__init__.py rename src/openstatspec/{transform => frontends/spss}/binding.py (87%) rename src/openstatspec/{transform => frontends/spss}/compiler.py (88%) create mode 100644 src/openstatspec/frontends/spss/execution.py rename src/openstatspec/{transform => frontends/spss}/syntax.py (99%) create mode 100644 src/openstatspec/frontends/stata/.gitkeep create mode 100644 src/openstatspec/transform/schema.py create mode 100644 src/openstatspec/transform/validation.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e7e27f9..edb3058 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,37 @@ All notable changes to this reference implementation are documented here. +## 0.4.0 — 2026-07-31 + +### Added + +- Added a public canonical transformation-plan in-place API accepting either + typed plan objects or strictly validated mappings. +- Added apply-plan and install-in-place-schema CLI workflows alongside the + compatible apply-spss command. +- Added explicit audit provenance distinguishing canonical plan documents from + SPSS syntax sources. +- Added a transformation manual covering the canonical core, frontend boundary, + execution invariants, CLI/API workflows, and extension rules. + +### Changed + +- Moved the SPSS parser, binder, compiler, and convenience execution adapter to + the dedicated openstatspec.frontends.spss package. +- Kept openstatspec.transform language-neutral with canonical plan, schema, and + live-schema validation modules while preserving previous SPSS import paths. +- Reserved empty Stata and SAS frontend directories without exposing parsers, + capabilities, CLI choices, or support claims. +- Strengthened pre-mutation target-type checks and target-scoped physical table + identity guards. The executor continues to modify the same dataset and table + without creating OpenStatSpec rollback, copy, snapshot, or history layers. + +### Specification basis + +- Conformance and release validation continue to use OpenStatSpec specification + release v0.2.0 at exact commit + 79339ec3d8f8aa81789b7e85f6b8afa6f1374e50. + ## 0.3.0 — 2026-07-31 ### Added diff --git a/README.md b/README.md index 229518b..dead014 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,10 @@ import. capabilities, and loss reports. - `openstatspec.sql`: database connection and wide-table/catalog operations. - `openstatspec.spss`: SAV/ZSAV adapter boundary. +- `openstatspec.transform`: canonical plans, frontend-neutral schema concepts, + and plan validation. +- `openstatspec.frontends.spss`: the SPSS-like syntax frontend and convenience + execution adapter. ## Intended workflow @@ -63,43 +67,17 @@ 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. +The SPSS-like frontend lowers supported `RECODE`, `VARIABLE LABELS`, and +`VALUE LABELS` syntax into a language-neutral canonical plan. The in-place +path applies it to the same logical dataset, physical wide table, and metadata +catalog without a derived dataset, copied table, snapshot, or separate +rollback/history layer. Dolt remains the sole versioning layer for Dolt-backed +edits, and the transformer never calls `DOLT_COMMIT`. + +See the [dataset transformations manual](docs/transformations.md) for schema +installation, Python and CLI surfaces, database invariants, audit provenance, +package layout, and extension guidance. Stata and SAS are unimplemented +placeholders. ## Current support status diff --git a/docs/in-place-transformation.md b/docs/in-place-transformation.md index 22d0a01..2b6039e 100644 --- a/docs/in-place-transformation.md +++ b/docs/in-place-transformation.md @@ -1,41 +1,9 @@ -# In-place SPSS-like transformation +# In-place transformations -`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. +This page has moved to the comprehensive +[dataset transformations manual](transformations.md). -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 -``` +The manual covers the canonical plan, SPSS-like frontend, executor, audit +schema, Python and CLI surfaces, database and Dolt invariants, provenance, +package layout, and future frontend boundary. Stata and SAS remain +unimplemented placeholders. diff --git a/docs/release-readiness.md b/docs/release-readiness.md index 50c1aad..51bca43 100644 --- a/docs/release-readiness.md +++ b/docs/release-readiness.md @@ -1,4 +1,4 @@ -# 0.3.0 release readiness +# 0.4.0 release readiness This page records the expected release contract, not a publication event. Creating a version tag remains a separate maintainer action. @@ -56,6 +56,36 @@ otherwise the strict export policy reports and blocks the loss. A caller that supplies consent for an available loss code receives the machine-readable loss report with the export result. +## Transformation release gates + +The canonical transformation core and SPSS syntax frontend are separate public +surfaces. A release must run the specification-owned canonical-plan and SPSS +frontend conformance fixtures, then exercise both the generic plan apply API +and the SPSS compatibility apply path. + +The gate must prove that: + +- a TransformationPlan object and its strict JSON mapping produce the same + plan hash and in-place result; +- top-level SPSS compiler imports and legacy openstatspec.transform re-exports + still load from an installed wheel; +- install-in-place-schema, apply-plan, and apply-spss execute their documented + CLI workflows; +- invalid or unsupported plans fail before the first data or metadata mutation; +- successful applies retain the same dataset ID, physical schema/table + identity, dataset count, and persistent physical data-table count; +- audit rows distinguish canonical plans from SPSS syntax, preserve correct + source/plan hashes and frontend contract, and contain no copied data; +- no OpenStatSpec rollback, snapshot, staging, copy, derived-dataset, or + parallel history artifacts are created; and +- Dolt checks expected branch, HEAD, and a clean working set without committing + or changing HEAD; other supported SQL connections remain allowed. + +The built wheel must contain the generic openstatspec.transform modules and the +implemented openstatspec.frontends.spss package. Stata and SAS remain empty +source-tree placeholders and must expose no compiler, apply API, CLI choice, +capability claim, or implied support. + ## Maintainer checks before tagging 1. Publish the pinned `openstatspec-pyspssio==0.5.1.post2` engine distribution diff --git a/docs/transformations.md b/docs/transformations.md new file mode 100644 index 0000000..8f2bb65 --- /dev/null +++ b/docs/transformations.md @@ -0,0 +1,204 @@ +# Dataset transformations + +OpenStatSpec separates transformation syntax, canonical meaning, and database +mutation. This lets multiple language frontends produce the same plan without +coupling the executor to any one language. + +The implemented frontend accepts a small SPSS-like subset: `RECODE`, +`VARIABLE LABELS`, and `VALUE LABELS`. Stata and SAS are not implemented. + +## Architecture + +The path has three layers: + +1. A **frontend** parses source text and binds names and types against an + explicit input schema. It performs no database mutation or SQL generation. +2. The **canonical plan** contains ordered, typed, language-neutral operations. + Its canonical JSON and hash are independent of SQL dialect and source + formatting. +3. The **in-place executor** validates the complete plan against the live + dataset, then directly mutates that dataset's existing wide table and + metadata catalog. It does not parse frontend syntax. + +This is a trust boundary. JSON plans must pass +`transformation_plan_from_dict()` and live-schema validation before mutation; +they are never treated as arbitrary SQL. + +## Install the audit schema + +Install the compact audit relation once before the first apply: + +```python +import openstatspec + +openstatspec.install_in_place_transformation_schema( + database_url="sqlite:///survey.sqlite", +) +``` + +Installation is separate because schema DDL may commit independently on some +engines. Apply fails before changing data or metadata when the audit relation is +absent, and never creates or migrates schema-management objects itself. + +The CLI equivalent is: + +```text +openstatspec install-in-place-schema --database-url sqlite:///survey.sqlite +``` + +## Generic canonical-plan apply + +`openstatspec.transform.TransformationPlan` is the generic model. Numeric +constants retain exact binary64 bits, strings retain exact Unicode text, and +operation order is significant. Use `transformation_plan_from_dict()` for +untrusted mappings. `canonical_plan_json()` and `canonical_plan_hash()` +produce stable audit identities. + +The public API is: + +```python +result = openstatspec.apply_transformation_plan_in_place( + database_url="sqlite:///survey.sqlite", + dataset_id="responses", + plan=plan, + actor="agent@example.org", + expected_branch=None, + expected_head=None, +) +``` + +`plan` accepts a `TransformationPlan` or a mapping handled by the strict +loader. The executor recomputes the hash and validates the whole plan before +mutation. + +The corresponding CLI is: + +```text +openstatspec apply-plan --database-url sqlite:///survey.sqlite \ + --dataset-id responses --actor agent@example.org --plan-file plan.json +``` + +## SPSS-like frontend + +The pure compiler works without a database: + +```python +from openstatspec import VariableDefinition, VariableSchema, compile_spss_syntax + +schema = VariableSchema((VariableDefinition("age", "numeric"),)) +compilation = compile_spss_syntax( + "RECODE age (18 THRU 34 = 1) (35 THRU 64 = 2).", + schema, +) +print(compilation.plan.canonical_json()) +print(compilation.plan_hash) +``` + +It normalizes line endings, records the source hash, parses and sequentially +binds the supported subset, and returns a canonical plan plus output schema. + +For database-connected use, the compatibility wrapper loads the live schema, +compiles the source, and invokes the in-place path: + +```python +result = openstatspec.apply_spss_in_place( + database_url="sqlite:///survey.sqlite", + dataset_id="responses", + actor="agent@example.org", + source_text=""" + RECODE age (18 THRU 34 = 1) (35 THRU 64 = 2). + VARIABLE LABELS age 'Age group'. + VALUE LABELS age 1 '18-34' 2 '35-64'. + """, +) +``` + +`openstatspec.compile_spss_syntax` and +`openstatspec.apply_spss_in_place` remain compatibility APIs. + +The current CLI wrapper is: + +```text +openstatspec apply-spss --database-url sqlite:///survey.sqlite \ + --dataset-id responses --actor agent@example.org --syntax-file transform.sps +``` + +On Dolt, also pass `--expected-branch` and `--expected-head`. + +## Database and Dolt invariants + +Every successful apply preserves the logical `dataset_id` and physical +schema/table identity. It creates no derived dataset, output table, full-table +copy, staging table, snapshot, rollback artifact, or recovery/history layer. +Existing-target recodes use direct `UPDATE`; label operations mutate existing +catalog rows. + +SQLite and PostgreSQL may add a numeric target where native transactions make +the complete operation atomic. MySQL, MariaDB, and Dolt reject target-creating +plans before the first mutation because implicit-commit DDL could leave a +partial apply. Their target column and metadata must already exist. + +Dolt is the sole history, diff, branch, and rollback layer for Dolt-backed +datasets. Before mutation, the executor verifies the expected branch and +`HEAD` and requires clean `dolt_status`. Success changes the same working set +without changing `HEAD`. OpenStatSpec does not call `DOLT_COMMIT`, switch +branches, merge, reset, tag, or create a hidden recovery commit. The caller +reviews `dolt diff` and separately decides whether to commit or restore. + +## Audit and provenance + +Each success writes one compact `transformation_apply` row in the data and +metadata transaction. It records dataset and relation identity, actor, plan +hash, operation count, timestamps, database profile, and relevant Dolt +branch/HEAD. SPSS source also has a normalized source hash. + +The row stores no case values and references no copied state. It is provenance, +not an undo log or substitute for Dolt history. `source_kind` distinguishes a +direct `canonical_plan` from `spss_syntax`. For a direct plan, the canonical +JSON document is itself the source artifact, so its source hash equals the plan +hash and `frontend_contract` is null. SPSS applies record the normalized +syntax hash and SPSS frontend contract. + +## Package layout + +The intended boundary is: + +```text +openstatspec +├── transform +│ ├── plan.py +│ ├── schema.py +│ ├── validation.py +│ └── errors.py +└── frontends + ├── spss + │ ├── syntax.py + │ ├── binding.py + │ ├── compiler.py + │ └── execution.py + ├── stata # empty placeholder; not implemented + └── sas # empty placeholder; not implemented +``` + +Current imports from `openstatspec` and `openstatspec.transform` remain +compatibility contracts if implementation files move. Stata and SAS provide no +parser, compiler, apply API, CLI choice, or support claim. + +## Extension guidance + +There is no plugin discovery or Python entry-point protocol. A future built-in +frontend must parse with source spans, bind against `VariableSchema` without +database access, emit only canonical operations, produce deterministic hashes, +share the generic validator/executor, and add specification-owned conformance +fixtures. + +An external plugin protocol should be added only for a real external frontend. +It would need explicit identity, contract compatibility, deterministic +compilation, capabilities, stable error semantics, and a trust policy. + +Transformation conformance covers canonical plan fixtures, strict invalid-plan +cases, frontend source/plan hashes and diagnostics, wrapper-to-plan equivalence, +preflight-before-mutation, dataset/table identity, metadata and audit rows, CLI +compatibility, and service evidence before database execution support is +claimed. Stata and SAS need their own fixtures and implementations before their +placeholder status can change. diff --git a/pyproject.toml b/pyproject.toml index 7a674c2..d16b5f5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "openstatspec" -version = "0.3.0" +version = "0.4.0" description = "Reference adapter for the OpenStatSpec relational contract" readme = "README.md" requires-python = ">=3.11" diff --git a/src/openstatspec/__init__.py b/src/openstatspec/__init__.py index a93114a..d9754f0 100644 --- a/src/openstatspec/__init__.py +++ b/src/openstatspec/__init__.py @@ -1,7 +1,8 @@ """Public API for the OpenStatSpec Python reference implementation.""" from .api import ( - apply_spss_in_place, capabilities, capability_matrix, derive_sql_dataset, + apply_spss_in_place, apply_transformation_plan_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, @@ -11,17 +12,24 @@ ) from .core import CapabilityDeclaration, LossReport, UnsupportedOperationError from .sql.workflow import TransformationError +from .frontends.spss import SpssFrontendCompilation, compile_spss_syntax from .transform import ( - SpssFrontendCompilation, TransformationFrontendError, - VariableDefinition, VariableSchema, compile_spss_syntax, + RecodeMatch, RecodeOperation, RecodeResult, RecodeRule, + ReplaceValueLabelsOperation, SetVariableLabelOperation, + TransformationFrontendError, TransformationPlan, TypedValue, ValueLabel, + VariableDefinition, VariableSchema, transformation_plan_from_dict, ) __all__ = [ "CapabilityDeclaration", "LossReport", "SpssFrontendCompilation", "TransformationError", "TransformationFrontendError", - "VariableDefinition", "VariableSchema", + "RecodeMatch", "RecodeOperation", "RecodeResult", "RecodeRule", + "ReplaceValueLabelsOperation", "SetVariableLabelOperation", + "TransformationPlan", "TypedValue", "ValueLabel", + "VariableDefinition", "VariableSchema", "transformation_plan_from_dict", "UnsupportedOperationError", "capabilities", "capability_matrix", - "apply_spss_in_place", "compile_spss_syntax", "derive_sql_dataset", + "apply_spss_in_place", "apply_transformation_plan_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", diff --git a/src/openstatspec/api.py b/src/openstatspec/api.py index 6bfa308..9de2e21 100644 --- a/src/openstatspec/api.py +++ b/src/openstatspec/api.py @@ -20,11 +20,13 @@ remove_derived_relation as _remove_derived_relation, retire_derived_dataset as _retire_derived_dataset, ) +from .frontends.spss.execution import apply_spss_in_place as _apply_spss_in_place from .sql.inplace_transform import ( - apply_spss_in_place as _apply_spss_in_place, + apply_transformation_plan_in_place as _apply_transformation_plan_in_place, in_place_transformation_capabilities, install_in_place_transformation_schema as _install_in_place_schema, ) +from .transform import TransformationPlan from .sql.capabilities import ( SPECIFICATION_COMMIT, SPECIFICATION_RELEASE, active_connection, catalog_binding, ) @@ -154,6 +156,23 @@ def apply_spss_in_place( )) +def apply_transformation_plan_in_place( + *, database_url: Any, dataset_id: str, + plan: TransformationPlan | Mapping[str, Any], + actor: str, expected_branch: str | None = None, + expected_head: str | None = None, +) -> Mapping[str, Any]: + """Apply a canonical plan to the same logical dataset and physical table.""" + return result(_apply_transformation_plan_in_place( + database_url=str(database_url), + dataset_id=dataset_id, + plan=plan, + 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)) diff --git a/src/openstatspec/cli.py b/src/openstatspec/cli.py index e1ffc22..cc9d5c7 100644 --- a/src/openstatspec/cli.py +++ b/src/openstatspec/cli.py @@ -5,10 +5,12 @@ from pathlib import Path from .api import ( - apply_spss_in_place, capability_matrix, derive_sql_dataset, + apply_spss_in_place, apply_transformation_plan_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, + install_in_place_transformation_schema, register_sql_transformation, + validate, validate_derived, ) @@ -77,7 +79,7 @@ def main(argv: Sequence[str] | None = None) -> int: apply_spss = commands.add_parser( "apply-spss", - help="apply supported SPSS syntax in-place on a controlled Dolt branch", + help="compile supported SPSS syntax and apply it in-place", ) apply_spss.add_argument("--database-url", required=True) apply_spss.add_argument("--dataset-id", required=True) @@ -88,6 +90,23 @@ def main(argv: Sequence[str] | None = None) -> int: syntax_source.add_argument("--syntax") syntax_source.add_argument("--syntax-file") + apply_plan = commands.add_parser( + "apply-plan", + help="apply a canonical transformation plan in-place", + ) + apply_plan.add_argument("--database-url", required=True) + apply_plan.add_argument("--dataset-id", required=True) + apply_plan.add_argument("--actor", required=True) + apply_plan.add_argument("--expected-branch") + apply_plan.add_argument("--expected-head") + apply_plan.add_argument("--plan-file", required=True) + + install_in_place = commands.add_parser( + "install-in-place-schema", + help="install or upgrade the compact in-place apply audit schema", + ) + install_in_place.add_argument("--database-url", required=True) + 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) @@ -151,6 +170,19 @@ def main(argv: Sequence[str] | None = None) -> int: expected_branch=args.expected_branch, expected_head=args.expected_head, ) + elif args.command == "apply-plan": + plan = json.loads(Path(args.plan_file).read_text(encoding="utf-8")) + output = apply_transformation_plan_in_place( + database_url=args.database_url, + dataset_id=args.dataset_id, + plan=plan, + actor=args.actor, + expected_branch=args.expected_branch, + expected_head=args.expected_head, + ) + elif args.command == "install-in-place-schema": + install_in_place_transformation_schema(database_url=args.database_url) + output = {"status": "installed"} else: output = validate_derived( database_url=args.database_url, derived_dataset_id=args.derived_dataset_id, diff --git a/src/openstatspec/frontends/__init__.py b/src/openstatspec/frontends/__init__.py new file mode 100644 index 0000000..ff848fb --- /dev/null +++ b/src/openstatspec/frontends/__init__.py @@ -0,0 +1 @@ +"""Syntax frontends that lower external languages to canonical plans.""" diff --git a/src/openstatspec/frontends/sas/.gitkeep b/src/openstatspec/frontends/sas/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/src/openstatspec/frontends/spss/__init__.py b/src/openstatspec/frontends/spss/__init__.py new file mode 100644 index 0000000..0d537df --- /dev/null +++ b/src/openstatspec/frontends/spss/__init__.py @@ -0,0 +1,26 @@ +"""SPSS syntax frontend for canonical OpenStatSpec transformation plans.""" + +from .binding import bind_spss_syntax +from .compiler import SpssFrontendCompilation, compile_spss_syntax +from .syntax import ( + SpssSyntaxProgram, + normalize_spss_source, + parse_spss_syntax, + spss_source_hash, + tokenize_spss, +) + + +SPSS_FRONTEND_CONTRACT = "openstatspec-spss-syntax-frontend-v0.1" + +__all__ = [ + "SPSS_FRONTEND_CONTRACT", + "SpssFrontendCompilation", + "SpssSyntaxProgram", + "bind_spss_syntax", + "compile_spss_syntax", + "normalize_spss_source", + "parse_spss_syntax", + "spss_source_hash", + "tokenize_spss", +] diff --git a/src/openstatspec/transform/binding.py b/src/openstatspec/frontends/spss/binding.py similarity index 87% rename from src/openstatspec/transform/binding.py rename to src/openstatspec/frontends/spss/binding.py index c2dbdad..77f4b1b 100644 --- a/src/openstatspec/transform/binding.py +++ b/src/openstatspec/frontends/spss/binding.py @@ -2,58 +2,25 @@ from __future__ import annotations -from dataclasses import dataclass, replace +from dataclasses import replace from typing import Literal -from .errors import SourceSpan, frontend_error -from .plan import ( +from ...transform.errors import SourceSpan, frontend_error +from ...transform.plan import ( PlanOperation, RecodeMatch, RecodeOperation, RecodeResult, RecodeRule, ReplaceValueLabelsOperation, SetVariableLabelOperation, TransformationPlan, TypedValue, ValueLabel, ) +from ...transform.schema import ( + BoundTransformation, StorageKind, VariableDefinition, VariableSchema, +) +from ...transform.validation import bind_transformation_plan 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) @@ -291,8 +258,8 @@ def bind_spss_syntax( 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), + plan = TransformationPlan(tuple(operations), input_alias=input_alias) + return bind_transformation_plan( + plan, + schema, ) diff --git a/src/openstatspec/transform/compiler.py b/src/openstatspec/frontends/spss/compiler.py similarity index 88% rename from src/openstatspec/transform/compiler.py rename to src/openstatspec/frontends/spss/compiler.py index be98433..4e003c0 100644 --- a/src/openstatspec/transform/compiler.py +++ b/src/openstatspec/frontends/spss/compiler.py @@ -4,8 +4,9 @@ from dataclasses import dataclass -from .binding import BoundTransformation, VariableSchema, bind_spss_syntax -from .plan import TransformationPlan +from ...transform.plan import TransformationPlan +from ...transform.schema import BoundTransformation, VariableSchema +from .binding import bind_spss_syntax from .syntax import ( normalize_spss_source, parse_spss_syntax, diff --git a/src/openstatspec/frontends/spss/execution.py b/src/openstatspec/frontends/spss/execution.py new file mode 100644 index 0000000..30c672a --- /dev/null +++ b/src/openstatspec/frontends/spss/execution.py @@ -0,0 +1,48 @@ +"""SPSS convenience adapter over the generic canonical-plan executor.""" + +from __future__ import annotations + +from typing import Any + +from ...sql.inplace_transform import ( + InPlacePlanSubmission, + _run_in_place_submission, + load_transformation_schema, +) +from . import SPSS_FRONTEND_CONTRACT +from .compiler import compile_spss_syntax + + +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]: + """Compile SPSS syntax and apply its canonical plan in one transaction.""" + + def prepare(connection: Any, live_dataset_id: str) -> InPlacePlanSubmission: + schema = load_transformation_schema(connection, live_dataset_id) + compilation = compile_spss_syntax( + source_text, + schema, + input_alias="parent", + ) + return InPlacePlanSubmission( + plan=compilation.plan, + source_kind="spss_syntax", + source_hash=compilation.source_hash, + frontend_contract=SPSS_FRONTEND_CONTRACT, + ) + + return _run_in_place_submission( + database_url=database_url, + dataset_id=dataset_id, + actor=actor, + prepare=prepare, + expected_branch=expected_branch, + expected_head=expected_head, + ) diff --git a/src/openstatspec/transform/syntax.py b/src/openstatspec/frontends/spss/syntax.py similarity index 99% rename from src/openstatspec/transform/syntax.py rename to src/openstatspec/frontends/spss/syntax.py index c4df2e4..e4507f1 100644 --- a/src/openstatspec/transform/syntax.py +++ b/src/openstatspec/frontends/spss/syntax.py @@ -8,7 +8,7 @@ import re from typing import Literal -from .errors import SourcePosition, SourceSpan, frontend_error +from ...transform.errors import SourcePosition, SourceSpan, frontend_error TokenKind = Literal[ diff --git a/src/openstatspec/frontends/stata/.gitkeep b/src/openstatspec/frontends/stata/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/src/openstatspec/sql/inplace_transform.py b/src/openstatspec/sql/inplace_transform.py index c6a0fe0..e763a02 100644 --- a/src/openstatspec/sql/inplace_transform.py +++ b/src/openstatspec/sql/inplace_transform.py @@ -2,6 +2,8 @@ from __future__ import annotations +from collections.abc import Callable, Mapping +from dataclasses import dataclass from datetime import UTC, datetime import json from typing import Any @@ -14,8 +16,9 @@ from ..transform import ( RecodeOperation, RecodeResult, ReplaceValueLabelsOperation, - SetVariableLabelOperation, TypedValue, ValueLabel, VariableDefinition, - VariableSchema, compile_spss_syntax, + SetVariableLabelOperation, TransformationPlan, TypedValue, ValueLabel, + VariableDefinition, VariableSchema, bind_transformation_plan, + transformation_plan_from_dict, ) from .capabilities import effective_profile from .normative import catalog as core_catalog @@ -27,6 +30,20 @@ APPLY_CONTRACT = "openstatspec-in-place-transformation-v0.1" +@dataclass(frozen=True) +class InPlacePlanSubmission: + """A canonical plan plus compact provenance for one atomic apply.""" + + plan: TransformationPlan + source_kind: str + source_hash: str + frontend_contract: str | None = None + + def __post_init__(self) -> None: + if not self.source_kind or not self.source_hash: + raise ValueError("source_kind and source_hash must be non-empty") + + def in_place_transformation_capabilities() -> dict[str, Any]: return { "contract": APPLY_CONTRACT, @@ -71,7 +88,9 @@ def apply_audit_catalog(metadata: MetaData) -> Table: Column("database_profile", String(32), nullable=False), Column("physical_table_schema", String(255)), Column("physical_table_name", String(255), nullable=False), + Column("source_kind", String(64)), Column("source_hash", String(64), nullable=False), + Column("frontend_contract", String(128)), Column("plan_hash", String(64), nullable=False), Column("canonical_plan_json", Text, nullable=False), Column("actor", String(255), nullable=False), @@ -115,11 +134,16 @@ def _match_expression(match: Any, source: Any) -> Any: def _input_schema( connection: Any, dataset_id: str, + *, + lock_dataset: bool = False, ) -> 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() + dataset_query = select(core.dataset).where( + core.dataset.c.dataset_id == dataset_id + ) + if lock_dataset: + dataset_query = dataset_query.with_for_update() + dataset = connection.execute(dataset_query).mappings().one_or_none() if dataset is None: raise TransformationError( "dataset_not_found", "The in-place target dataset does not exist." @@ -174,27 +198,35 @@ def _input_schema( return dict(dataset), variables, schema -def _catalog_identity_counts( +def _target_identity_state( connection: Any, -) -> tuple[int, tuple[tuple[str | None, str], ...]]: + dataset_id: str, + *, + lock_dataset: bool = False, +) -> tuple[str, str | None, str, int]: + """Return one locked catalog identity plus its actual relation count.""" core = core_catalog(MetaData()) - dataset_rows = connection.execute( + query = ( select( + core.dataset.c.dataset_id, 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 + .where(core.dataset.c.dataset_id == dataset_id) + ) + if lock_dataset: + query = query.with_for_update() + row = connection.execute(query).one_or_none() + if row is None: + raise TransformationError( + "dataset_not_found", "The in-place target dataset does not exist." + ) + schema = str(row.physical_table_schema) if row.physical_table_schema else None + table_name = str(row.physical_table_name) + relation_count = int( + inspect(connection).has_table(table_name, schema=schema) + ) + return str(row.dataset_id), schema, table_name, relation_count def _legacy_identifiers(dataset: dict[str, Any]) -> tuple[str, str]: @@ -281,21 +313,35 @@ def _replace_value_labels( ).values(value_labels=json.dumps(legacy_json, ensure_ascii=False))) -def _apply_on_connection( +def _apply_plan_on_connection( connection: Any, *, dataset_id: str, - source_text: str, + submission: InPlacePlanSubmission, 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) + before_identity = _target_identity_state( + connection, + dataset_id, + lock_dataset=True, + ) + if before_identity[3] != 1: + raise TransformationError( + "physical_table_missing", + "The target dataset's physical wide table does not exist.", + ) 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") + plan = submission.plan + bound = bind_transformation_plan(plan, schema) + output_by_name = { + variable.name.casefold(): variable + for variable in bound.output_schema.variables + } audit = apply_audit_catalog(MetaData()) if not inspect(connection).has_table("transformation_apply"): raise TransformationError( @@ -303,16 +349,42 @@ def _apply_on_connection( "The compact transformation_apply audit schema must be installed " "before apply.", ) + audit_columns = { + str(column["name"]) + for column in inspect(connection).get_columns("transformation_apply") + } + required_audit_columns = {"source_kind", "frontend_contract"} + if not required_audit_columns.issubset(audit_columns): + raise TransformationError( + "in_place_audit_schema_outdated", + "Re-run install_in_place_transformation_schema before apply.", + ) if not allow_schema_change and any( isinstance(operation, RecodeOperation) and operation.target_mode == "create" - for operation in compilation.plan.operations + for operation in 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.", ) + unsupported_targets = [ + operation.target + for operation in plan.operations + if ( + isinstance(operation, RecodeOperation) + and operation.target_mode == "create" + and output_by_name[operation.target.casefold()].storage_kind + != "numeric" + ) + ] + if unsupported_targets: + raise TransformationError( + "in_place_target_type_unsupported", + "This executor cannot create string targets without an explicit " + "storage-width operation.", + ) core = core_catalog(MetaData()) legacy_metadata = MetaData() _, legacy_variable, _, _ = legacy_catalog(legacy_metadata) @@ -326,15 +398,10 @@ def _apply_on_connection( by_name = {str(row["source_name"]).casefold(): row for row in variables} used_physical = {str(row["physical_name"]).casefold() for row in variables} - for operation in compilation.plan.operations: + for operation in 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 = ( @@ -418,11 +485,12 @@ def _apply_on_connection( "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): + after_identity = _target_identity_state(connection, dataset_id) + if after_identity != before_identity: raise TransformationError( "dataset_identity_changed", - "In-place apply changed dataset or physical data-table identity.", + "In-place apply changed the target dataset or its physical " + "data-table identity.", ) apply_id = str(uuid4()) started = _now() @@ -433,15 +501,17 @@ def _apply_on_connection( 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(), + source_kind=submission.source_kind, + source_hash=submission.source_hash, + frontend_contract=submission.frontend_contract, + plan_hash=plan.sha256(), + canonical_plan_json=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), + operation_count=len(plan.operations), started_at=started, completed_at=_now(), )) @@ -464,8 +534,10 @@ def _apply_on_connection( "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, + "source_kind": submission.source_kind, + "source_hash": submission.source_hash, + "frontend_contract": submission.frontend_contract, + "plan_hash": plan.sha256(), "dolt_branch": dolt_branch, "dolt_head_before": dolt_head, "dolt_head_after": dolt_head, @@ -488,20 +560,40 @@ def install_in_place_transformation_schema(*, database_url: str) -> None: try: with engine.begin() as connection: apply_audit_catalog(MetaData()).create(connection, checkfirst=True) + columns = { + str(column["name"]) + for column in inspect(connection).get_columns("transformation_apply") + } + additions = { + "source_kind": "VARCHAR(64) NULL", + "frontend_contract": "VARCHAR(128) NULL", + } + quote = connection.dialect.identifier_preparer.quote + for name, sql_type in additions.items(): + if name not in columns: + connection.exec_driver_sql( + f"ALTER TABLE {quote('transformation_apply')} " + f"ADD COLUMN {quote(name)} {sql_type}" + ) finally: engine.dispose() -def apply_spss_in_place( +def load_transformation_schema(connection: Any, dataset_id: str) -> VariableSchema: + """Read the live canonical variable schema within the caller's transaction.""" + return _input_schema(connection, dataset_id)[2] + + +def _run_in_place_submission( *, database_url: str, dataset_id: str, - source_text: str, actor: str, + prepare: Callable[[Any, str], InPlacePlanSubmission], 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.""" + """Prepare and apply one canonical plan in the same controlled transaction.""" if not actor: raise TransformationError( "actor_required", "A non-empty actor identity is mandatory.", @@ -534,10 +626,13 @@ def apply_spss_in_place( "dolt_working_set_dirty", "The Dolt working set must be clean before in-place apply.", ) - result = _apply_on_connection( + submission = prepare(connection, dataset_id) + if not isinstance(submission, InPlacePlanSubmission): + raise TypeError("prepare must return InPlacePlanSubmission") + result = _apply_plan_on_connection( connection, dataset_id=dataset_id, - source_text=source_text, + submission=submission, actor=actor, database_profile=profile.name, allow_schema_change=profile.name in {"sqlite", "postgresql"}, @@ -554,3 +649,34 @@ def apply_spss_in_place( return result finally: engine.dispose() + + +def apply_transformation_plan_in_place( + *, + database_url: str, + dataset_id: str, + plan: TransformationPlan | Mapping[str, Any], + actor: str, + expected_branch: str | None = None, + expected_head: str | None = None, +) -> dict[str, Any]: + """Apply a canonical plan without knowing which frontend produced it.""" + normalized = ( + plan + if isinstance(plan, TransformationPlan) + else transformation_plan_from_dict(plan) + ) + plan_hash = normalized.sha256() + submission = InPlacePlanSubmission( + plan=normalized, + source_kind="canonical_plan", + source_hash=plan_hash, + ) + return _run_in_place_submission( + database_url=database_url, + dataset_id=dataset_id, + actor=actor, + prepare=lambda _connection, _dataset_id: submission, + expected_branch=expected_branch, + expected_head=expected_head, + ) diff --git a/src/openstatspec/transform/__init__.py b/src/openstatspec/transform/__init__.py index 27ae283..6ef1124 100644 --- a/src/openstatspec/transform/__init__.py +++ b/src/openstatspec/transform/__init__.py @@ -1,33 +1,52 @@ -"""Pure transformation models and SPSS-syntax frontend; no database adapter.""" +"""Pure canonical transformation models; no syntax or 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, + 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, +from .schema import ( + BoundTransformation, StorageKind, VariableDefinition, VariableSchema, ) +from .validation import bind_transformation_plan __all__ = [ "BoundTransformation", "RecodeMatch", "RecodeOperation", "RecodeResult", "RecodeRule", "ReplaceValueLabelsOperation", "SPSS_FRONTEND_CONTRACT", - "SetVariableLabelOperation", "SourcePosition", "SourceSpan", + "SetVariableLabelOperation", "SourcePosition", "SourceSpan", "StorageKind", "SpssFrontendCompilation", "SpssSyntaxProgram", "TRANSFORMATION_PLAN_CONTRACT", "TransformationFrontendError", "TransformationPlan", "TypedValue", "ValueLabel", "VariableDefinition", - "VariableSchema", "bind_spss_syntax", "canonical_plan_hash", + "VariableSchema", "bind_spss_syntax", "bind_transformation_plan", + "canonical_plan_hash", "compile_spss_syntax", "canonical_plan_json", "normalize_spss_source", "parse_spss_syntax", "spss_source_hash", "tokenize_spss", "transformation_plan_from_dict", ] + + +_SPSS_COMPAT_EXPORTS = { + "SPSS_FRONTEND_CONTRACT", + "SpssFrontendCompilation", + "SpssSyntaxProgram", + "bind_spss_syntax", + "compile_spss_syntax", + "normalize_spss_source", + "parse_spss_syntax", + "spss_source_hash", + "tokenize_spss", +} + + +def __getattr__(name: str): + """Temporarily preserve the v0.3 SPSS re-export surface.""" + if name in _SPSS_COMPAT_EXPORTS: + from ..frontends import spss + + return getattr(spss, name) + raise AttributeError(name) diff --git a/src/openstatspec/transform/plan.py b/src/openstatspec/transform/plan.py index c2ab3c4..6305ff8 100644 --- a/src/openstatspec/transform/plan.py +++ b/src/openstatspec/transform/plan.py @@ -15,7 +15,6 @@ 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}") diff --git a/src/openstatspec/transform/schema.py b/src/openstatspec/transform/schema.py new file mode 100644 index 0000000..15cc638 --- /dev/null +++ b/src/openstatspec/transform/schema.py @@ -0,0 +1,46 @@ +"""Generic in-memory variable schema and bound transformation models.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +from .plan import TransformationPlan, ValueLabel + + +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 diff --git a/src/openstatspec/transform/validation.py b/src/openstatspec/transform/validation.py new file mode 100644 index 0000000..79338b2 --- /dev/null +++ b/src/openstatspec/transform/validation.py @@ -0,0 +1,178 @@ +"""Validate canonical transformation plans against an explicit live schema.""" + +from __future__ import annotations + +from dataclasses import replace +from typing import Literal + +from .errors import frontend_error +from .plan import ( + RecodeMatch, + RecodeOperation, + RecodeResult, + ReplaceValueLabelsOperation, + SetVariableLabelOperation, + TransformationPlan, +) +from .schema import ( + BoundTransformation, + StorageKind, + VariableDefinition, + VariableSchema, +) + + +ValueType = Literal["binary64", "string"] + + +def _expected_type(storage_kind: StorageKind) -> ValueType: + return "binary64" if storage_kind == "numeric" else "string" + + +def _resolve( + variables: list[VariableDefinition], name: str +) -> 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.", + variable=name, + ) + return matches[0] + + +def _validate_match(match: RecodeMatch, source: VariableDefinition) -> None: + expected = _expected_type(source.storage_kind) + if match.kind == "system_missing": + if source.storage_kind == "string": + raise frontend_error( + "system_missing_for_string", + "SYSMIS cannot match a string variable.", + variable=source.name, + ) + return + if match.kind == "range": + if expected != "binary64": + raise frontend_error( + "type_mismatch", + "Ranges require a numeric source.", + variable=source.name, + ) + return + if any(value.type != expected for value in match.values): + raise frontend_error( + "type_mismatch", + "RECODE match values must match the source storage kind.", + variable=source.name, + expected_type=expected, + ) + + +def _result_type( + result: RecodeResult, source: VariableDefinition +) -> ValueType: + if result.kind == "copy": + return _expected_type(source.storage_kind) + if result.kind == "system_missing": + if source.storage_kind == "string": + raise frontend_error( + "system_missing_for_string", + "SYSMIS cannot be produced for a string variable.", + variable=source.name, + ) + return "binary64" + assert result.value is not None + return result.value.type + + +def _bind_recode( + operation: RecodeOperation, variables: list[VariableDefinition] +) -> None: + _, source = _resolve(variables, operation.source) + if operation.target_mode == "create": + if operation.target.startswith("__"): + raise frontend_error( + "reserved_target_name", + f"Target name {operation.target!r} is reserved.", + target=operation.target, + ) + if any( + variable.name.casefold() == operation.target.casefold() + for variable in variables + ): + raise frontend_error( + "target_already_exists", + f"Target name {operation.target!r} already exists.", + target=operation.target, + ) + for rule in operation.rules: + _validate_match(rule.match, source) + result_types = { + _result_type(result, source) + for result in [ + *(rule.result for rule in operation.rules), + operation.unmatched, + ] + } + if len(result_types) != 1: + raise frontend_error( + "mixed_result_types", + "All RECODE results, including unmatched behavior, must have one type.", + source=source.name, + result_types=sorted(result_types), + ) + output_type = next(iter(result_types)) + if ( + operation.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.", + variable=source.name, + ) + if operation.target_mode == "create": + variables.append( + VariableDefinition( + operation.target, + "numeric" if output_type == "binary64" else "string", + ) + ) + + +def bind_transformation_plan( + plan: TransformationPlan, schema: VariableSchema +) -> BoundTransformation: + """Validate sequential plan semantics and return the resulting schema.""" + if not isinstance(plan, TransformationPlan): + raise TypeError("plan must be a TransformationPlan.") + if not isinstance(schema, VariableSchema): + raise TypeError("schema must be a VariableSchema.") + variables = list(schema.variables) + for operation in plan.operations: + if isinstance(operation, RecodeOperation): + _bind_recode(operation, variables) + continue + if isinstance(operation, SetVariableLabelOperation): + index, variable = _resolve(variables, operation.variable) + variables[index] = replace(variable, variable_label=operation.label) + continue + if isinstance(operation, ReplaceValueLabelsOperation): + index, variable = _resolve(variables, operation.variable) + expected = _expected_type(variable.storage_kind) + if any(label.value.type != expected for label in operation.labels): + raise frontend_error( + "type_mismatch", + "Value-label codes must match the variable storage kind.", + variable=variable.name, + expected_type=expected, + ) + variables[index] = replace(variable, value_labels=operation.labels) + continue + raise AssertionError(f"Unknown plan operation: {type(operation)!r}") + return BoundTransformation(plan, VariableSchema(tuple(variables))) diff --git a/tests/test_cli.py b/tests/test_cli.py index 5e3b026..26d3ff8 100755 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,4 +1,5 @@ import json +import sqlite3 import openstatspec import openstatspec.cli @@ -93,3 +94,65 @@ def test_capability_matrix_reports_active_sqlite_limits(tmp_path) -> None: "source": "OpenStatSpec profile boundary; SQLite has no fixed native identifier limit", "repertoire": "generated ASCII [a-z0-9_] identifiers", } + + +def test_cli_installs_schema_and_applies_plan_or_spss( + tmp_path, capsys, +) -> None: + source = tmp_path / "transform-source.sav" + database_path = tmp_path / "transform.sqlite" + database_url = f"sqlite:///{database_path}" + pyspssio.write_sav(str(source), pd.DataFrame({"answer": [1.0]})) + openstatspec.import_sav( + source, + database_url=database_url, + dataset_id="transform-cli", + ) + connection = sqlite3.connect(database_path) + live_dataset_id = connection.execute( + "SELECT dataset_id FROM dataset" + ).fetchone()[0] + + assert openstatspec.cli.main([ + "install-in-place-schema", + "--database-url", database_url, + ]) == 0 + assert json.loads(capsys.readouterr().out) == {"status": "installed"} + + plan = openstatspec.compile_spss_syntax( + "RECODE answer (1 = 2).", + openstatspec.VariableSchema(( + openstatspec.VariableDefinition("answer", "numeric"), + )), + ).plan + plan_file = tmp_path / "plan.json" + plan_file.write_text( + json.dumps(plan.as_dict()), + encoding="utf-8", + ) + assert openstatspec.cli.main([ + "apply-plan", + "--database-url", database_url, + "--dataset-id", live_dataset_id, + "--actor", "cli-test", + "--plan-file", str(plan_file), + ]) == 0 + generic = json.loads(capsys.readouterr().out) + assert generic["source_kind"] == "canonical_plan" + + assert openstatspec.cli.main([ + "apply-spss", + "--database-url", database_url, + "--dataset-id", live_dataset_id, + "--actor", "cli-test", + "--syntax", "RECODE answer (2 = 3).", + ]) == 0 + spss = json.loads(capsys.readouterr().out) + assert spss["source_kind"] == "spss_syntax" + + table_name = connection.execute( + "SELECT physical_table_name FROM dataset" + ).fetchone()[0] + assert connection.execute( + f'SELECT answer FROM "{table_name}"' + ).fetchone() == (3.0,) diff --git a/tests/test_inplace_transform.py b/tests/test_inplace_transform.py index ea2db14..3c4cda4 100644 --- a/tests/test_inplace_transform.py +++ b/tests/test_inplace_transform.py @@ -8,7 +8,10 @@ import openstatspec import openstatspec.sql.inplace_transform as inplace_transform -from openstatspec.sql.inplace_transform import _apply_on_connection +from openstatspec.sql.inplace_transform import ( + InPlacePlanSubmission, + _apply_plan_on_connection, +) from openstatspec.sql.wide import create_wide_dataset @@ -34,6 +37,26 @@ def _variables() -> list[dict[str, object]]: }] +def _plan(source_text: str): + schema = openstatspec.VariableSchema(( + openstatspec.VariableDefinition( + "score", + "numeric", + variable_label="Score", + ), + )) + return openstatspec.compile_spss_syntax(source_text, schema).plan + + +def _submission(source_text: str) -> InPlacePlanSubmission: + plan = _plan(source_text) + return InPlacePlanSubmission( + plan=plan, + source_kind="canonical_plan", + source_hash=plan.sha256(), + ) + + @pytest.fixture def catalog(tmp_path): path = tmp_path / "in-place.sqlite" @@ -68,10 +91,10 @@ def test_plan_applies_to_same_dataset_and_physical_table_without_copy( name for name in inspect(connection).get_table_names() if name.startswith("data_") } - result = _apply_on_connection( + result = _apply_plan_on_connection( connection, dataset_id=dataset_id, - source_text=( + submission=_submission( "RECODE score (1,2 = 0) (3 = 1) INTO score_band. " "VARIABLE LABELS score_band 'Score band'. " "VALUE LABELS score_band 0 'Lower' 1 'Upper'." @@ -79,8 +102,8 @@ def test_plan_applies_to_same_dataset_and_physical_table_without_copy( actor="test-agent", database_profile="sqlite", allow_schema_change=True, - dolt_branch="feature/recode", - dolt_head="abc123", + dolt_branch=None, + dolt_head=None, ) after_datasets = connection.execute(text( "SELECT COUNT(*) FROM dataset" @@ -126,13 +149,14 @@ def test_plan_applies_to_same_dataset_and_physical_table_without_copy( "dolt_head_after, actor, status " "FROM transformation_apply" ).fetchone() == ( - "sqlite", "feature/recode", "abc123", "abc123", "test-agent", + "sqlite", None, None, None, "test-agent", "succeeded", ) def test_public_apply_supports_non_dolt_without_building_undo(catalog) -> None: url, path, dataset_id, table_name = catalog + plan = _plan("RECODE score (1 = 0).") result = openstatspec.apply_spss_in_place( database_url=url, dataset_id=dataset_id, @@ -141,9 +165,184 @@ def test_public_apply_supports_non_dolt_without_building_undo(catalog) -> None: ) assert result["dolt_branch"] is None assert result["dolt_commit_performed"] is False + assert result["plan_hash"] == plan.sha256() assert sqlite3.connect(path).execute( f'SELECT score FROM "{table_name}" ORDER BY __case_ordinal' ).fetchall() == [(0.0,), (2.0,), (3.0,)] + audit = sqlite3.connect(path).execute( + "SELECT source_kind, frontend_contract FROM transformation_apply" + ).fetchone() + assert audit == ( + "spss_syntax", + "openstatspec-spss-syntax-frontend-v0.1", + ) + + +@pytest.mark.parametrize("as_mapping", [False, True]) +def test_public_generic_plan_apply_accepts_object_and_mapping( + catalog, as_mapping, +) -> None: + url, path, dataset_id, table_name = catalog + plan = _plan("RECODE score (1 = 7).") + supplied = plan.as_dict() if as_mapping else plan + result = openstatspec.apply_transformation_plan_in_place( + database_url=url, + dataset_id=dataset_id, + plan=supplied, + actor="test-agent", + ) + assert result["dataset_id"] == dataset_id + assert result["physical_table_name"] == table_name + assert result["source_kind"] == "canonical_plan" + assert result["source_hash"] == result["plan_hash"] == plan.sha256() + connection = sqlite3.connect(path) + assert connection.execute( + f'SELECT score FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(7.0,), (2.0,), (3.0,)] + assert connection.execute( + "SELECT source_kind, source_hash, frontend_contract, plan_hash " + "FROM transformation_apply" + ).fetchone() == ( + "canonical_plan", + plan.sha256(), + None, + plan.sha256(), + ) + + +def test_generic_plan_is_bound_to_live_schema_before_mutation(catalog) -> None: + url, path, dataset_id, table_name = catalog + plan = openstatspec.TransformationPlan(( + openstatspec.SetVariableLabelOperation("missing", "Must fail"), + )) + with pytest.raises(openstatspec.TransformationFrontendError) as caught: + openstatspec.apply_transformation_plan_in_place( + database_url=url, + dataset_id=dataset_id, + plan=plan, + actor="test-agent", + ) + assert caught.value.code == "unknown_variable" + connection = sqlite3.connect(path) + assert connection.execute( + f'SELECT score FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(1.0,), (2.0,), (3.0,)] + assert connection.execute( + "SELECT COUNT(*) FROM transformation_apply" + ).fetchone() == (0,) + + +@pytest.mark.parametrize("as_mapping", [False, True]) +def test_string_create_target_is_rejected_before_any_mutation( + catalog, as_mapping, +) -> None: + url, path, dataset_id, table_name = catalog + plan = openstatspec.TransformationPlan(( + openstatspec.SetVariableLabelOperation("score", "Must roll back"), + openstatspec.RecodeOperation( + source="score", + target="band", + target_mode="create", + rules=( + openstatspec.RecodeRule( + openstatspec.RecodeMatch( + "values", + values=(openstatspec.TypedValue.binary64(1),), + ), + openstatspec.RecodeResult( + "literal", + openstatspec.TypedValue.string("low"), + ), + ), + ), + unmatched=openstatspec.RecodeResult( + "literal", + openstatspec.TypedValue.string("other"), + ), + ), + )) + supplied = plan.as_dict() if as_mapping else plan + with pytest.raises(openstatspec.TransformationError) as caught: + openstatspec.apply_transformation_plan_in_place( + database_url=url, + dataset_id=dataset_id, + plan=supplied, + actor="test-agent", + ) + assert caught.value.code == "in_place_target_type_unsupported" + connection = sqlite3.connect(path) + assert connection.execute( + "SELECT variable_label FROM variable WHERE source_name = 'score'" + ).fetchone() == ("Score",) + assert "band" not in { + row[1] + for row in connection.execute(f'PRAGMA table_info("{table_name}")') + } + assert connection.execute( + "SELECT COUNT(*) FROM transformation_apply" + ).fetchone() == (0,) + + +def test_string_source_can_create_numeric_target(tmp_path) -> None: + path = tmp_path / "string-source.sqlite" + url = f"sqlite:///{path}" + variables = _variables() + variables[0].update({ + "source_name": "color", + "physical_name": "color", + "storage_kind": "string", + "string_width": 8, + "label": "Color", + "format": "A8", + "print_format": "[1, 8, 0]", + "write_format": "[1, 8, 0]", + }) + create_wide_dataset( + database_url=url, + dataset_id="string_source", + source_name="source.sav", + source_format="SAV", + source_sha256="e" * 64, + rows=[{"color": "R"}, {"color": "B"}], + 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() + plan = openstatspec.TransformationPlan(( + openstatspec.RecodeOperation( + source="color", + target="is_red", + target_mode="create", + rules=( + openstatspec.RecodeRule( + openstatspec.RecodeMatch( + "values", + values=(openstatspec.TypedValue.string("R"),), + ), + openstatspec.RecodeResult( + "literal", + openstatspec.TypedValue.binary64(1), + ), + ), + ), + unmatched=openstatspec.RecodeResult( + "literal", + openstatspec.TypedValue.binary64(0), + ), + ), + )) + openstatspec.apply_transformation_plan_in_place( + database_url=url, + dataset_id=dataset_id, + plan=plan, + actor="test-agent", + ) + assert sqlite3.connect(path).execute( + f'SELECT color, is_red FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [("R", 1.0), ("B", 0.0)] def test_missing_audit_schema_fails_before_mutation(catalog) -> None: @@ -172,10 +371,10 @@ def test_nontransactional_ddl_profile_rejects_create_before_mutation( engine = create_engine(url) with pytest.raises(openstatspec.TransformationError) as caught: with engine.begin() as connection: - _apply_on_connection( + _apply_plan_on_connection( connection, dataset_id=dataset_id, - source_text=( + submission=_submission( "VARIABLE LABELS score 'Changed'. " "RECODE score (1 = 0) INTO score_band." ), diff --git a/tests/test_transform_frontend.py b/tests/test_transform_frontend.py index 400f685..016ec79 100644 --- a/tests/test_transform_frontend.py +++ b/tests/test_transform_frontend.py @@ -5,19 +5,28 @@ from pathlib import Path import pytest +from openstatspec.frontends.spss import ( + bind_spss_syntax, + compile_spss_syntax, + normalize_spss_source, + parse_spss_syntax, + spss_source_hash, +) from openstatspec.transform import ( + RecodeMatch, + RecodeOperation, + RecodeResult, + RecodeRule, + SetVariableLabelOperation, + TransformationPlan, TransformationFrontendError, TypedValue, ValueLabel, VariableDefinition, VariableSchema, - bind_spss_syntax, + bind_transformation_plan, canonical_plan_hash, canonical_plan_json, - compile_spss_syntax, - normalize_spss_source, - parse_spss_syntax, - spss_source_hash, transformation_plan_from_dict, ) @@ -328,3 +337,31 @@ def test_custom_nonempty_input_alias_is_canonical() -> None: input_alias="survey", ).plan assert plan.input_alias == "survey" + + +def test_generic_plan_binding_validates_sequential_schema_state() -> None: + plan = TransformationPlan(( + RecodeOperation( + source="q1", + target="q1_binary", + target_mode="create", + rules=( + RecodeRule( + RecodeMatch("values", (TypedValue.binary64(1),)), + RecodeResult("literal", TypedValue.binary64(1)), + ), + ), + unmatched=RecodeResult("literal", TypedValue.binary64(0)), + ), + SetVariableLabelOperation("q1_binary", "Binary response"), + )) + + bound = bind_transformation_plan( + plan, _schema(VariableDefinition("q1", "numeric")) + ) + + assert [variable.name for variable in bound.output_schema.variables] == [ + "q1", + "q1_binary", + ] + assert bound.output_schema.variables[-1].variable_label == "Binary response"