From 081bc014de8d2fe368b146a1dfc37ac408c0caca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 13:10:40 +0300 Subject: [PATCH 01/12] Add conditional SPSS transformations for Python v0.5.0 --- CHANGELOG.md | 32 ++ README.md | 19 +- docs/release-readiness.md | 29 +- docs/transformations.md | 52 +- pyproject.toml | 2 +- src/openstatspec/__init__.py | 11 +- src/openstatspec/api.py | 10 +- src/openstatspec/cli.py | 2 +- src/openstatspec/frontends/spss/__init__.py | 2 +- src/openstatspec/frontends/spss/binding.py | 171 ++++++- src/openstatspec/frontends/spss/syntax.py | 218 ++++++++- src/openstatspec/sql/capabilities.py | 7 +- src/openstatspec/sql/inplace_transform.py | 498 +++++++++++++++----- src/openstatspec/transform/__init__.py | 9 +- src/openstatspec/transform/plan.py | 305 +++++++++++- src/openstatspec/transform/schema.py | 30 ++ src/openstatspec/transform/validation.py | 142 ++++++ tests/test_cli.py | 6 +- tests/test_conditional_inplace_transform.py | 332 +++++++++++++ tests/test_inplace_transform.py | 3 +- tests/test_sql_profiles.py | 8 +- tests/test_transform_frontend.py | 142 ++++++ 22 files changed, 1836 insertions(+), 194 deletions(-) create mode 100644 tests/test_conditional_inplace_transform.py diff --git a/CHANGELOG.md b/CHANGELOG.md index edb3058..3555b52 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,38 @@ All notable changes to this reference implementation are documented here. +## 0.5.0 — 2026-07-31 + +### Added + +- Added bounded typed expressions with variable references, numeric literals, + parentheses, `=`, `<`, `<=`, `>`, `>=`, `AND`, and `OR`. String comparison + and v0.2 string assignment fail closed pending exact portable semantics. +- Added sequential SPSS-like `COMPUTE` and `IF` assignment operations plus + `FORMATS`, `VARIABLE LEVEL`, and `EXECUTE`. +- Added atomic numeric target creation on SQLite and PostgreSQL. MySQL, + MariaDB, and Dolt fail closed on `target_mode=create`; their targets must be + provisioned physically and in the catalog by a separate versioned stage + before this executor applies assignment and metadata operations. +- Added synthetic exact-program, catalog, failure-boundary, and pre-existing + target regression coverage. + +### Changed + +- Bumped the canonical transformation-plan and SPSS frontend contracts to + `v0.2`; canonical JSON and hashes include every sequential operation. +- In-place apply now records variable label, value labels, `F` print/write + format, and measurement level in both normative and compatibility catalogs. +- Dolt still requires an exact branch, exact HEAD, and clean working set; + successful apply leaves an inspectable diff and never calls `DOLT_COMMIT`. + +### Specification basis + +- Release validation is pinned to the untagged OpenStatSpec specification + release candidate at immutable commit + `e49252c00890aed76dcaabc5d1ab5121b45929db`. Its + `specification_release` remains null until that commit receives a stable tag. + ## 0.4.0 — 2026-07-31 ### Added diff --git a/README.md b/README.md index dead014..5982760 100644 --- a/README.md +++ b/README.md @@ -67,12 +67,19 @@ implemented capability boundary. ## SPSS-like transformation frontend -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`. +The bounded SPSS-like frontend lowers `RECODE`, sequential `COMPUTE` and `IF`, +`VARIABLE LABELS`, `VALUE LABELS`, numeric `FORMATS`, `VARIABLE LEVEL`, and +`EXECUTE` into a language-neutral typed canonical plan. Conditions support +parentheses, variable and numeric literal operands, comparisons, `AND`, and +`OR`. String comparison and v0.2 string assignment currently fail closed. +The in-place path applies the plan to the same logical dataset, physical wide +table, and normative/compatibility metadata without a derived dataset, copied +table, snapshot, or hidden history layer. Numeric targets may be created +atomically on SQLite and PostgreSQL. MySQL, MariaDB, and Dolt fail closed on +`target_mode=create` and require a separately provisioned physical and catalog +target before assignment. Dolt requires the caller's exact clean branch/HEAD, +leaves success as an inspectable working-set diff, and never calls +`DOLT_COMMIT`. See the [dataset transformations manual](docs/transformations.md) for schema installation, Python and CLI surfaces, database invariants, audit provenance, diff --git a/docs/release-readiness.md b/docs/release-readiness.md index 51bca43..362636f 100644 --- a/docs/release-readiness.md +++ b/docs/release-readiness.md @@ -1,4 +1,4 @@ -# 0.4.0 release readiness +# 0.5.0 release readiness This page records the expected release contract, not a publication event. Creating a version tag remains a separate maintainer action. @@ -67,6 +67,17 @@ The gate must prove that: - a TransformationPlan object and its strict JSON mapping produce the same plan hash and in-place result; +- the exact bounded `COMPUTE`/`IF` program compiles to all seven ordered + operations without dropping `FORMATS`, `VARIABLE LEVEL`, or `EXECUTE`; +- boolean data results match the equivalent expression and the target's label, + 0/1 value labels, `F1.0` print/write format, and nominal level exist in both + normative and compatibility catalogs; +- injected schema, data, catalog, and audit failures leave no partial apply; +- compensation tracks only newly created targets and never drops or rewrites a + pre-existing target; +- MySQL, MariaDB, and Dolt reject create-target plans before mutation; their + service evidence covers assignment to a separately provisioned physical and + catalog target without schema DDL; - 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 @@ -78,8 +89,11 @@ The gate must prove that: 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. +- Dolt checks expected branch, HEAD, and a clean working set; success changes + neither HEAD nor branch and never commits or resets; state is rechecked after + the dataset lock and success must leave an inspectable working-set diff; + other supported SQL connections remain allowed; and +- string comparisons and v0.2 string assignments fail closed until exact The built wheel must contain the generic openstatspec.transform modules and the implemented openstatspec.frontends.spss package. Stata and SAS remain empty @@ -98,9 +112,12 @@ capability claim, or implied support. 4. Build with `python -m build` and install the generated wheel in a clean 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.2.0` - at exact commit `79339ec3d8f8aa81789b7e85f6b8afa6f1374e50`. +6. Confirm CI, release fixtures, and capabilities use the untagged OpenStatSpec + specification release candidate at exact commit + `e49252c00890aed76dcaabc5d1ab5121b45929db`, publish + `specification_status=release_candidate`, and leave + `specification_release` null. If that exact commit receives a stable tag + before this package is tagged, update the identity and re-run every gate. 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/docs/transformations.md b/docs/transformations.md index 8f2bb65..037aecd 100644 --- a/docs/transformations.md +++ b/docs/transformations.md @@ -4,8 +4,13 @@ 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. +The implemented bounded SPSS-like frontend accepts `RECODE`, sequential +`COMPUTE` and `IF`, `VARIABLE LABELS`, `VALUE LABELS`, numeric `FORMATS`, +`VARIABLE LEVEL`, and `EXECUTE`. Predicates support typed variable/literal +operands, parentheses, numeric comparisons, `AND`, and `OR`. String comparison +and v0.2 string assignment fail closed until exact profile-independent +collation and explicit-width semantics are available; arbitrary SPSS, Python, +and SQL expressions are rejected. Stata and SAS are not implemented. ## Architecture @@ -106,9 +111,13 @@ result = openstatspec.apply_spss_in_place( 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'. + COMPUTE target = 0. + IF (source_a = 1 AND source_b = 1) target = 1. + VARIABLE LABELS target 'Example label'. + VALUE LABELS target 0 'No' 1 'Yes'. + FORMATS target (F1.0). + VARIABLE LEVEL target (NOMINAL). + EXECUTE. """, ) ``` @@ -130,20 +139,25 @@ On Dolt, also pass `--expected-branch` and `--expected-head`. 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. +Assignments and recodes use ordered `UPDATE` statements; later operations see +earlier results. Label, value-label, format, and measurement-level operations +update both the normative and compatibility catalogs. + +A numeric create target is supported atomically on SQLite and PostgreSQL. +MySQL, MariaDB, and Dolt reject `target_mode=create` before mutation. On those +profiles a separate versioned stage must first provision the nullable numeric +physical column and both catalog representations; the transformation executor +then sees a pre-existing target and performs no schema DDL. +The public operation reports success only after physical data, both metadata +representations, and the compact audit row are mutually complete. + +Before Dolt 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`, `DOLT_RESET`, switch +branches, merge, tag, or create a hidden recovery commit. It rechecks branch, +HEAD, and a clean working set after locking the dataset and immediately before +mutation. The caller reviews a successful `dolt diff` and +separately decides whether to commit or restore. ## Audit and provenance diff --git a/pyproject.toml b/pyproject.toml index d16b5f5..5f6c9a1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "openstatspec" -version = "0.4.0" +version = "0.5.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 d9754f0..57a0a5f 100644 --- a/src/openstatspec/__init__.py +++ b/src/openstatspec/__init__.py @@ -14,17 +14,24 @@ from .sql.workflow import TransformationError from .frontends.spss import SpssFrontendCompilation, compile_spss_syntax from .transform import ( + AssignOperation, BooleanExpression, ComparisonExpression, + ConditionalAssignOperation, ExecuteOperation, Operand, PredicateExpression, RecodeMatch, RecodeOperation, RecodeResult, RecodeRule, - ReplaceValueLabelsOperation, SetVariableLabelOperation, + ReplaceValueLabelsOperation, SetFormatOperation, + SetMeasurementLevelOperation, SetVariableLabelOperation, TransformationFrontendError, TransformationPlan, TypedValue, ValueLabel, VariableDefinition, VariableSchema, transformation_plan_from_dict, ) __all__ = [ + "AssignOperation", "BooleanExpression", "ComparisonExpression", + "ConditionalAssignOperation", "ExecuteOperation", "Operand", + "PredicateExpression", "CapabilityDeclaration", "LossReport", "SpssFrontendCompilation", "TransformationError", "TransformationFrontendError", "RecodeMatch", "RecodeOperation", "RecodeResult", "RecodeRule", - "ReplaceValueLabelsOperation", "SetVariableLabelOperation", + "ReplaceValueLabelsOperation", "SetFormatOperation", + "SetMeasurementLevelOperation", "SetVariableLabelOperation", "TransformationPlan", "TypedValue", "ValueLabel", "VariableDefinition", "VariableSchema", "transformation_plan_from_dict", "UnsupportedOperationError", "capabilities", "capability_matrix", diff --git a/src/openstatspec/api.py b/src/openstatspec/api.py index 9de2e21..cf6220d 100644 --- a/src/openstatspec/api.py +++ b/src/openstatspec/api.py @@ -28,7 +28,8 @@ ) from .transform import TransformationPlan from .sql.capabilities import ( - SPECIFICATION_COMMIT, SPECIFICATION_RELEASE, active_connection, catalog_binding, + SPECIFICATION_COMMIT, SPECIFICATION_RELEASE, SPECIFICATION_STATUS, + active_connection, catalog_binding, ) @@ -44,7 +45,7 @@ def capability_matrix(database_url: str | None = None) -> Mapping[str, Any]: """ declaration = { "specification": "OpenStatSpec", - "specification_status": "released", + "specification_status": SPECIFICATION_STATUS, "specification_release": SPECIFICATION_RELEASE, "specification_commit": SPECIFICATION_COMMIT, "profile": "SPSS SAV/ZSAV 1.0", @@ -145,7 +146,10 @@ def apply_spss_in_place( 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.""" + """Apply bounded sequential SPSS syntax to one SQL dataset/table. + + Supports typed COMPUTE/IF predicates and dictionary metadata operations. + """ return result(_apply_spss_in_place( database_url=str(database_url), dataset_id=dataset_id, diff --git a/src/openstatspec/cli.py b/src/openstatspec/cli.py index cc9d5c7..3997470 100644 --- a/src/openstatspec/cli.py +++ b/src/openstatspec/cli.py @@ -79,7 +79,7 @@ def main(argv: Sequence[str] | None = None) -> int: apply_spss = commands.add_parser( "apply-spss", - help="compile supported SPSS syntax and apply it in-place", + help="compile bounded sequential SPSS syntax and apply it in-place", ) apply_spss.add_argument("--database-url", required=True) apply_spss.add_argument("--dataset-id", required=True) diff --git a/src/openstatspec/frontends/spss/__init__.py b/src/openstatspec/frontends/spss/__init__.py index 0d537df..18a7891 100644 --- a/src/openstatspec/frontends/spss/__init__.py +++ b/src/openstatspec/frontends/spss/__init__.py @@ -11,7 +11,7 @@ ) -SPSS_FRONTEND_CONTRACT = "openstatspec-spss-syntax-frontend-v0.1" +SPSS_FRONTEND_CONTRACT = "openstatspec-spss-syntax-frontend-v0.2" __all__ = [ "SPSS_FRONTEND_CONTRACT", diff --git a/src/openstatspec/frontends/spss/binding.py b/src/openstatspec/frontends/spss/binding.py index 77f4b1b..6ffe3f7 100644 --- a/src/openstatspec/frontends/spss/binding.py +++ b/src/openstatspec/frontends/spss/binding.py @@ -7,17 +7,24 @@ from ...transform.errors import SourceSpan, frontend_error from ...transform.plan import ( - PlanOperation, RecodeMatch, RecodeOperation, RecodeResult, RecodeRule, - ReplaceValueLabelsOperation, SetVariableLabelOperation, TransformationPlan, - TypedValue, ValueLabel, + AssignOperation, BooleanExpression, ComparisonExpression, + ConditionalAssignOperation, ExecuteOperation, Operand, PlanOperation, + PredicateExpression, RecodeMatch, RecodeOperation, RecodeResult, RecodeRule, + ReplaceValueLabelsOperation, SetFormatOperation, + SetMeasurementLevelOperation, SetVariableLabelOperation, + TRANSFORMATION_PLAN_V1_CONTRACT, + 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, + BooleanSyntax, ComparisonSyntax, ComputeCommandSyntax, ExecuteCommandSyntax, + FormatsCommandSyntax, IfCommandSyntax, OperandSyntax, PredicateSyntax, + RecodeCommandSyntax, RecodeMatchSyntax, RecodeResultSyntax, + SpssSyntaxProgram, SyntaxLiteral, ValueLabelsCommandSyntax, + VariableLabelsCommandSyntax, VariableLevelCommandSyntax, ) @@ -48,6 +55,77 @@ def _resolve( return matches[0] +def _bind_operand( + syntax: OperandSyntax, variables: list[VariableDefinition], +) -> tuple[Operand, Literal["binary64", "string"]]: + if syntax.kind == "variable": + assert syntax.variable is not None + _, variable = _resolve(variables, syntax.variable.text, syntax.variable.span) + return ( + Operand.variable_ref(variable.name), + "binary64" if variable.storage_kind == "numeric" else "string", + ) + assert syntax.literal is not None + value = _typed(syntax.literal) + return Operand.literal(value), value.type + + +def _bind_predicate( + syntax: PredicateSyntax, variables: list[VariableDefinition], +) -> PredicateExpression: + if isinstance(syntax, ComparisonSyntax): + left, left_type = _bind_operand(syntax.left, variables) + right, right_type = _bind_operand(syntax.right, variables) + if left_type != right_type: + raise frontend_error( + "type_mismatch", + "Comparison operands must have the same storage kind.", + span=syntax.span, left_type=left_type, right_type=right_type, + ) + if syntax.operator != "=" and left_type != "binary64": + raise frontend_error( + "type_mismatch", + "Ordered comparisons require numeric operands.", + span=syntax.span, operator=syntax.operator, + ) + return ComparisonExpression(left, syntax.operator, right) + assert isinstance(syntax, BooleanSyntax) + return BooleanExpression( + syntax.operator, + tuple(_bind_predicate(operand, variables) for operand in syntax.operands), + ) + + +def _assignment( + target_name: str, target_span: SourceSpan, value_syntax: OperandSyntax, + variables: list[VariableDefinition], +) -> AssignOperation: + value, value_type = _bind_operand(value_syntax, variables) + matches = [ + (index, variable) for index, variable in enumerate(variables) + if variable.name.casefold() == target_name.casefold() + ] + if matches: + _, target = matches[0] + if value_type != _expected_type(target.storage_kind): + raise frontend_error( + "type_mismatch", + "COMPUTE cannot change an existing variable's storage kind.", + span=target_span, variable=target.name, + expected_type=_expected_type(target.storage_kind), + ) + return AssignOperation(target.name, "replace", value) + if target_name.startswith("__"): + raise frontend_error( + "reserved_target_name", f"Target name {target_name!r} is reserved.", + span=target_span, target=target_name, + ) + variables.append(VariableDefinition( + target_name, "numeric" if value_type == "binary64" else "string", + )) + return AssignOperation(target_name, "create", value) + + def _match( syntax: RecodeMatchSyntax, source: VariableDefinition, ) -> RecodeMatch: @@ -211,6 +289,77 @@ def bind_spss_syntax( operations.extend(recodes) spans.extend(recode_spans) continue + if isinstance(command, ComputeCommandSyntax): + operations.append(_assignment( + command.target.text, command.target.span, command.value, variables, + )) + spans.append(command.span) + continue + if isinstance(command, IfCommandSyntax): + condition = _bind_predicate(command.condition, variables) + target_matches = [ + (index, variable) for index, variable in enumerate(variables) + if variable.name.casefold() == command.target.text.casefold() + ] + if len(target_matches) != 1: + raise frontend_error( + "conditional_target_missing", + "IF assignment target must already exist.", + span=command.target.span, variable=command.target.text, + ) + _, target = target_matches[0] + value, value_type = _bind_operand(command.value, variables) + expected = _expected_type(target.storage_kind) + if value_type != expected: + raise frontend_error( + "type_mismatch", + "IF assignment value must match the target storage kind.", + span=command.value.span, variable=target.name, + expected_type=expected, + ) + operations.append(ConditionalAssignOperation( + condition, target.name, value, + )) + spans.append(command.span) + continue + if isinstance(command, FormatsCommandSyntax): + for assignment in command.assignments: + index, variable = _resolve( + variables, assignment.variable.text, assignment.variable.span, + ) + if variable.storage_kind != "numeric": + raise frontend_error( + "type_mismatch", "F formats require a numeric variable.", + span=assignment.span, variable=variable.name, + ) + operations.append(SetFormatOperation( + variable.name, assignment.family, + assignment.width, assignment.decimals, + )) + spans.append(assignment.span) + variables[index] = replace( + variable, format_family=assignment.family, + format_width=assignment.width, + format_decimals=assignment.decimals, + ) + continue + if isinstance(command, VariableLevelCommandSyntax): + for assignment in command.assignments: + index, variable = _resolve( + variables, assignment.variable.text, assignment.variable.span, + ) + operations.append(SetMeasurementLevelOperation( + variable.name, assignment.level, + )) + spans.append(assignment.span) + variables[index] = replace( + variable, measurement_level=assignment.level, + ) + continue + if isinstance(command, ExecuteCommandSyntax): + operations.append(ExecuteOperation()) + spans.append(command.span) + continue if isinstance(command, VariableLabelsCommandSyntax): for assignment in command.assignments: index, variable = _resolve( @@ -258,7 +407,17 @@ def bind_spss_syntax( variables[index] = replace(variable, value_labels=labels) continue raise AssertionError(f"Unknown syntax command: {type(command)!r}") - plan = TransformationPlan(tuple(operations), input_alias=input_alias) + v01_types = ( + RecodeOperation, SetVariableLabelOperation, ReplaceValueLabelsOperation, + ) + contract = ( + TRANSFORMATION_PLAN_V1_CONTRACT + if all(isinstance(operation, v01_types) for operation in operations) + else "openstatspec-transformation-plan-v0.2" + ) + plan = TransformationPlan( + tuple(operations), contract=contract, input_alias=input_alias, + ) return bind_transformation_plan( plan, schema, diff --git a/src/openstatspec/frontends/spss/syntax.py b/src/openstatspec/frontends/spss/syntax.py index e4507f1..2dd5e8d 100644 --- a/src/openstatspec/frontends/spss/syntax.py +++ b/src/openstatspec/frontends/spss/syntax.py @@ -13,7 +13,8 @@ TokenKind = Literal[ "identifier", "number", "string", "left_paren", "right_paren", - "equals", "comma", "slash", "period", "eof", + "equals", "less", "less_equal", "greater", "greater_equal", + "comma", "slash", "period", "eof", ] @@ -62,6 +63,80 @@ class RecodeCommandSyntax: targets: tuple[Token, ...] | None span: SourceSpan +@dataclass(frozen=True) +class OperandSyntax: + kind: Literal["variable", "literal"] + span: SourceSpan + variable: Token | None = None + literal: SyntaxLiteral | None = None + + +@dataclass(frozen=True) +class ComparisonSyntax: + left: OperandSyntax + operator: Literal["=", "<", "<=", ">", ">="] + right: OperandSyntax + span: SourceSpan + + +@dataclass(frozen=True) +class BooleanSyntax: + operator: Literal["and", "or"] + operands: tuple["PredicateSyntax", ...] + span: SourceSpan + + +PredicateSyntax = ComparisonSyntax | BooleanSyntax + + +@dataclass(frozen=True) +class ComputeCommandSyntax: + target: Token + value: OperandSyntax + span: SourceSpan + + +@dataclass(frozen=True) +class IfCommandSyntax: + condition: PredicateSyntax + target: Token + value: OperandSyntax + span: SourceSpan + + +@dataclass(frozen=True) +class FormatAssignmentSyntax: + variable: Token + family: str + width: int + decimals: int + span: SourceSpan + + +@dataclass(frozen=True) +class FormatsCommandSyntax: + assignments: tuple[FormatAssignmentSyntax, ...] + span: SourceSpan + + +@dataclass(frozen=True) +class VariableLevelAssignmentSyntax: + variable: Token + level: Literal["nominal", "ordinal", "scale"] + span: SourceSpan + + +@dataclass(frozen=True) +class VariableLevelCommandSyntax: + assignments: tuple[VariableLevelAssignmentSyntax, ...] + span: SourceSpan + + +@dataclass(frozen=True) +class ExecuteCommandSyntax: + span: SourceSpan + + @dataclass(frozen=True) class VariableLabelSyntax: @@ -97,7 +172,9 @@ class ValueLabelsCommandSyntax: SyntaxCommand = ( - RecodeCommandSyntax | VariableLabelsCommandSyntax | ValueLabelsCommandSyntax + RecodeCommandSyntax | ComputeCommandSyntax | IfCommandSyntax + | VariableLabelsCommandSyntax | ValueLabelsCommandSyntax + | FormatsCommandSyntax | VariableLevelCommandSyntax | ExecuteCommandSyntax ) @@ -108,7 +185,7 @@ class SpssSyntaxProgram: _NUMBER = re.compile( - r"[+-]?(?:(?:[0-9]+(?:\.[0-9]*)?)|(?:\.[0-9]+))(?:[Ee][+-]?[0-9]+)?" + r"[+-]?(?:(?:[0-9]+(?:\.[0-9]+)?)|(?:\.[0-9]+))(?:[Ee][+-]?[0-9]+)?" ) _IDENTIFIER_START = frozenset("_@$#") _IDENTIFIER_CONTINUE = frozenset("_@$#") @@ -151,7 +228,8 @@ def tokenize_spss(source: str) -> tuple[Token, ...]: offset = 0 punctuation: dict[str, TokenKind] = { "(": "left_paren", ")": "right_paren", "=": "equals", - ",": "comma", "/": "slash", ".": "period", + "<": "less", ">": "greater", ",": "comma", "/": "slash", + ".": "period", } while offset < len(source): character = source[offset] @@ -227,6 +305,14 @@ def tokenize_spss(source: str) -> tuple[Token, ...]: "identifier", text, text, _span(source, start, offset), )) continue + if source.startswith("<=", offset) or source.startswith(">=", offset): + text = source[offset:offset + 2] + tokens.append(Token( + "less_equal" if text == "<=" else "greater_equal", + text, text, _span(source, offset, offset + 2), + )) + offset += 2 + continue if character in punctuation: tokens.append(Token( punctuation[character], character, character, @@ -313,6 +399,117 @@ def literal(self) -> SyntaxLiteral: span=token.span, ) + def operand(self) -> OperandSyntax: + if self.current.kind == "identifier": + token = self.advance() + return OperandSyntax("variable", token.span, variable=token) + literal = self.literal() + return OperandSyntax("literal", literal.span, literal=literal) + + def comparison(self) -> PredicateSyntax: + if self.accepts("left_paren") is not None: + expression = self.predicate() + self.expects("right_paren", "Expected ')' after expression.") + return expression + left = self.operand() + operators = { + "equals": "=", "less": "<", "less_equal": "<=", + "greater": ">", "greater_equal": ">=", + } + token = self.current + if token.kind not in operators: + raise frontend_error( + "spss_syntax_error", "Expected a comparison operator.", + span=token.span, + ) + self.advance() + right = self.operand() + return ComparisonSyntax( + left, operators[token.kind], right, _joined_span(left.span, right.span), + ) + + def conjunction(self) -> PredicateSyntax: + operands = [self.comparison()] + while self.accepts_keyword("AND") is not None: + operands.append(self.comparison()) + if len(operands) == 1: + return operands[0] + return BooleanSyntax( + "and", tuple(operands), _joined_span(operands[0].span, operands[-1].span), + ) + + def predicate(self) -> PredicateSyntax: + operands = [self.conjunction()] + while self.accepts_keyword("OR") is not None: + operands.append(self.conjunction()) + if len(operands) == 1: + return operands[0] + return BooleanSyntax( + "or", tuple(operands), _joined_span(operands[0].span, operands[-1].span), + ) + + def compute(self, start: Token) -> ComputeCommandSyntax: + target = self.expects("identifier", "COMPUTE requires a target variable.") + self.expects("equals", "Expected '=' in COMPUTE.") + value = self.operand() + end = self.expects("period", "Expected '.' after COMPUTE.") + return ComputeCommandSyntax(target, value, _joined_span(start.span, end.span)) + + def if_command(self, start: Token) -> IfCommandSyntax: + self.expects("left_paren", "Expected '(' before the IF predicate.") + condition = self.predicate() + self.expects("right_paren", "Expected ')' after the IF predicate.") + target = self.expects("identifier", "IF requires a target variable.") + self.expects("equals", "Expected '=' in IF assignment.") + value = self.operand() + end = self.expects("period", "Expected '.' after IF.") + return IfCommandSyntax(condition, target, value, _joined_span(start.span, end.span)) + + def formats(self, start: Token) -> FormatsCommandSyntax: + assignments: list[FormatAssignmentSyntax] = [] + while self.current.kind not in {"period", "eof"}: + self.accepts("slash") + variable = self.expects("identifier", "FORMATS requires a variable name.") + self.expects("left_paren", "Expected '(' before an SPSS format.") + format_token = self.expects("identifier", "Expected an SPSS format such as F1.0.") + match = re.fullmatch(r"([A-Za-z]+)([0-9]+)(?:[.]([0-9]+))?", format_token.text) + if match is None: + raise frontend_error("invalid_format", "Expected a bounded SPSS format such as F1.0.", span=format_token.span) + right = self.expects("right_paren", "Expected ')' after an SPSS format.") + family, width = match.group(1).upper(), int(match.group(2)) + decimals = int(match.group(3) or 0) + if (family != "F" or width < 1 or width > 40 or decimals > 16 + or (decimals != 0 and width < decimals + 2)): + raise frontend_error("invalid_format", "Only valid numeric F formats are supported.", span=format_token.span, format=format_token.text) + assignments.append(FormatAssignmentSyntax(variable, family, width, decimals, _joined_span(variable.span, right.span))) + if not assignments: + raise frontend_error("spss_syntax_error", "FORMATS requires an assignment.", span=self.current.span) + end = self.expects("period", "Expected '.' after FORMATS.") + return FormatsCommandSyntax(tuple(assignments), _joined_span(start.span, end.span)) + + def variable_level(self, start: Token) -> VariableLevelCommandSyntax: + self.expects_keyword("LEVEL") + assignments: list[VariableLevelAssignmentSyntax] = [] + while self.current.kind not in {"period", "eof"}: + self.accepts("slash") + variable = self.expects("identifier", "VARIABLE LEVEL requires a variable name.") + self.expects("left_paren", "Expected '(' before a measurement level.") + level = self.expects("identifier", "Expected NOMINAL, ORDINAL, or SCALE.") + normalized = level.text.casefold() + if normalized not in {"nominal", "ordinal", "scale"}: + raise frontend_error("invalid_measurement_level", "Expected NOMINAL, ORDINAL, or SCALE.", span=level.span, level=level.text) + right = self.expects("right_paren", "Expected ')' after a measurement level.") + assignments.append(VariableLevelAssignmentSyntax(variable, normalized, _joined_span(variable.span, right.span))) + if not assignments: + raise frontend_error("spss_syntax_error", "VARIABLE LEVEL requires an assignment.", span=self.current.span) + end = self.expects("period", "Expected '.' after VARIABLE LEVEL.") + return VariableLevelCommandSyntax(tuple(assignments), _joined_span(start.span, end.span)) + + def execute(self, start: Token) -> ExecuteCommandSyntax: + end = self.expects("period", "Expected '.' after EXECUTE.") + return ExecuteCommandSyntax(_joined_span(start.span, end.span)) + + def recode_result(self) -> RecodeResultSyntax: if (token := self.accepts_keyword("SYSMIS")) is not None: return RecodeResultSyntax("system_missing", token.span) @@ -447,8 +644,19 @@ def parse(self) -> SpssSyntaxProgram: command = start.text.casefold() if command == "recode": commands.append(self.recode(start)) + elif command == "compute": + commands.append(self.compute(start)) + elif command == "if": + commands.append(self.if_command(start)) + elif command == "formats": + commands.append(self.formats(start)) + elif command == "execute": + commands.append(self.execute(start)) elif command == "variable": - commands.append(self.variable_labels(start)) + if self.current.kind == "identifier" and self.current.text.casefold() == "level": + commands.append(self.variable_level(start)) + else: + commands.append(self.variable_labels(start)) elif command == "value": commands.append(self.value_labels(start)) else: diff --git a/src/openstatspec/sql/capabilities.py b/src/openstatspec/sql/capabilities.py index 2701660..0644b4f 100644 --- a/src/openstatspec/sql/capabilities.py +++ b/src/openstatspec/sql/capabilities.py @@ -15,8 +15,9 @@ from .profiles import profile_for_url, validate_connection_url from ..core import UnsupportedOperationError -SPECIFICATION_COMMIT = "79339ec3d8f8aa81789b7e85f6b8afa6f1374e50" -SPECIFICATION_RELEASE: str | None = "v0.2.0" +SPECIFICATION_COMMIT = "e49252c00890aed76dcaabc5d1ab5121b45929db" +SPECIFICATION_STATUS = "release_candidate" +SPECIFICATION_RELEASE: str | None = None _DOLT_2_2_STABLE_VERSION = re.compile(r"2\.2\.(0|[1-9][0-9]*)") @@ -339,7 +340,7 @@ def _profile( "dialect": "mysql" if name == "dolt" else name, "transport": "mysql_compatible" if name == "dolt" else name, "specification_commit": SPECIFICATION_COMMIT, - "specification_status": "released", + "specification_status": SPECIFICATION_STATUS, "specification_release": SPECIFICATION_RELEASE, "driver": "psycopg" if name == "postgresql" else "PyMySQL" if name in {"mysql", "mariadb", "dolt"} else "sqlite3", "claimed_server_versions": policy["claimed"], diff --git a/src/openstatspec/sql/inplace_transform.py b/src/openstatspec/sql/inplace_transform.py index e763a02..b95e2d4 100644 --- a/src/openstatspec/sql/inplace_transform.py +++ b/src/openstatspec/sql/inplace_transform.py @@ -10,12 +10,16 @@ from uuid import uuid4 from sqlalchemy import ( - Column, DateTime, Float, Integer, MetaData, String, Table, Text, case, + Column, DateTime, Float, Integer, MetaData, String, Table, Text, and_, case, create_engine, delete, inspect, insert, literal, null, select, text, update, + or_, ) from ..transform import ( + AssignOperation, BooleanExpression, ComparisonExpression, + ConditionalAssignOperation, ExecuteOperation, Operand, RecodeOperation, RecodeResult, ReplaceValueLabelsOperation, + SetFormatOperation, SetMeasurementLevelOperation, SetVariableLabelOperation, TransformationPlan, TypedValue, ValueLabel, VariableDefinition, VariableSchema, bind_transformation_plan, transformation_plan_from_dict, @@ -27,7 +31,7 @@ from .workflow import TransformationError -APPLY_CONTRACT = "openstatspec-in-place-transformation-v0.1" +APPLY_CONTRACT = "openstatspec-in-place-transformation-v0.2" @dataclass(frozen=True) @@ -53,14 +57,15 @@ def in_place_transformation_capabilities() -> dict[str, Any]: ], "parent_kinds": ["core"], "mutation": "same_dataset_same_physical_wide_table", - "commands": ["RECODE", "VARIABLE LABELS", "VALUE LABELS"], + "commands": ["RECODE", "COMPUTE", "IF", "VARIABLE LABELS", + "VALUE LABELS", "FORMATS", "VARIABLE LEVEL", "EXECUTE"], "new_target_column": { "sqlite": True, "postgresql": True, "mysql": False, "mariadb": False, "dolt": False, - "reason": "non-transactional DDL must not make apply partially durable", + "reason": "atomic create only on SQLite/PostgreSQL; other profiles require a pre-existing cataloged target", }, "creates_derived_dataset": False, "creates_persistent_data_copy": False, @@ -192,6 +197,21 @@ def _input_schema( str(row["storage_kind"]), variable_label=row["variable_label"], value_labels=tuple(labels_by_variable.get(str(row["variable_id"]), [])), + format_family=( + "F" if str(row["print_format_family"]).upper() in {"F", "5"} + and str(row["storage_kind"]) == "numeric" else None + ), + format_width=( + row["print_format_width"] + if str(row["print_format_family"]).upper() in {"F", "5"} + and str(row["storage_kind"]) == "numeric" else None + ), + format_decimals=( + row["print_format_decimals"] + if str(row["print_format_family"]).upper() in {"F", "5"} + and str(row["storage_kind"]) == "numeric" else None + ), + measurement_level=row["measurement_level"], ) for row in variables )) @@ -313,6 +333,43 @@ def _replace_value_labels( ).values(value_labels=json.dumps(legacy_json, ensure_ascii=False))) +def _failure_boundary(_name: str) -> None: + """Synthetic-test hook for schema/data/catalog/audit failure boundaries.""" + + +def _operand_expression( + operand: Operand, relation: Table, by_name: Mapping[str, dict[str, Any]], +) -> Any: + if operand.kind == "literal": + assert operand.value is not None + return literal(_typed_value(operand.value)) + assert operand.variable is not None + variable = by_name[operand.variable.casefold()] + return relation.c[str(variable["physical_name"])] + + +def _predicate_expression( + expression: ComparisonExpression | BooleanExpression, + relation: Table, + by_name: Mapping[str, dict[str, Any]], +) -> Any: + if isinstance(expression, BooleanExpression): + parts = [ + _predicate_expression(item, relation, by_name) + for item in expression.operands + ] + return and_(*parts) if expression.operator == "and" else or_(*parts) + left = _operand_expression(expression.left, relation, by_name) + right = _operand_expression(expression.right, relation, by_name) + return { + "=": lambda: left == right, + "<": lambda: left < right, + "<=": lambda: left <= right, + ">": lambda: left > right, + ">=": lambda: left >= right, + }[expression.operator]() + + def _apply_plan_on_connection( connection: Any, *, @@ -323,12 +380,9 @@ def _apply_plan_on_connection( allow_schema_change: bool, dolt_branch: str | None, dolt_head: str | None, + mutation_journal: dict[str, Any] | None = None, ) -> dict[str, Any]: - before_identity = _target_identity_state( - connection, - dataset_id, - lock_dataset=True, - ) + before_identity = _target_identity_state(connection, dataset_id, lock_dataset=True) if before_identity[3] != 1: raise TransformationError( "physical_table_missing", @@ -346,107 +400,139 @@ def _apply_plan_on_connection( 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.", + "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): + if not {"source_kind", "frontend_contract"}.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) + create_operations = [ + operation for operation in plan.operations + if isinstance(operation, (RecodeOperation, AssignOperation)) and operation.target_mode == "create" - for operation in plan.operations - ): + ] + if create_operations and not allow_schema_change: raise TransformationError( "schema_change_not_atomic", - "This database profile requires RECODE targets to exist before " - "apply because its DDL is not transaction-atomic.", + "This database profile has no coherent new-target strategy.", ) 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" - ) + operation.target for operation in create_operations + if 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.", + "New string targets require an explicit storage-width operation.", ) + 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"), + 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} + next_ordinal = max(int(row["source_ordinal"]) for row in variables) + 1 + target_rows: list[dict[str, Any]] = [] + quote = connection.dialect.identifier_preparer.quote + qualified_table = connection.dialect.identifier_preparer.format_table(relation) + numeric_type = ( + "DOUBLE PRECISION" if connection.dialect.name == "postgresql" else "DOUBLE" + ) + if mutation_journal is not None: + mutation_journal.update({ + "table_schema": dataset.get("physical_table_schema"), + "legacy_dataset_id": legacy_dataset_id, + "table_name": table_name, + "added_columns": [], + "target_rows": [], + }) + + if database_profile == "dolt": + if dolt_branch is None or dolt_head is None: + raise TransformationError( + "dolt_context_required", + "Dolt mutation requires the preflight branch and HEAD.", + ) + locked_branch, locked_head, locked_dirty = _dolt_state(connection) + if locked_branch != dolt_branch: + raise TransformationError( + "dolt_branch_mismatch", + "The active Dolt branch changed after locking the dataset.", + ) + if locked_head != dolt_head: + raise TransformationError( + "dolt_head_mismatch", + "Dolt HEAD changed after locking the dataset.", + ) + if locked_dirty != 0: + raise TransformationError( + "dolt_working_set_dirty", + "The Dolt working set changed after locking the dataset.", + ) + + # All non-transactional DDL precedes every data/catalog mutation. This makes + # Compensation is journal-bounded to target columns and catalog identities + # created by this apply, without resetting unrelated database state. + for operation in create_operations: + target_physical = physical_name(operation.target, used_physical) + connection.exec_driver_sql( + f"ALTER TABLE {qualified_table} ADD COLUMN " + f"{quote(target_physical)} {numeric_type} NULL" + ) + if mutation_journal is not None: + mutation_journal["added_columns"].append(target_physical) + target_row = { + "variable_id": str(uuid4()), + "dataset_id": dataset_id, + "source_ordinal": next_ordinal, + "source_name": operation.target, + "physical_name": target_physical, + "storage_kind": "numeric", + "variable_label": None, + } + next_ordinal += 1 + target_rows.append(target_row) + if mutation_journal is not None: + mutation_journal["target_rows"].append(dict(target_row)) + if create_operations: + _failure_boundary("schema") + + for target_row in target_rows: + connection.execute(insert(core.variable).values(**target_row)) + connection.execute(insert(legacy_variable).values( + dataset_id=legacy_dataset_id, + ordinal=target_row["source_ordinal"], + source_name=target_row["source_name"], + physical_name=target_row["physical_name"], + storage_kind="numeric", + string_width=None, + label="", + attributes="{}", + value_labels="{}", + missing_ranges="[]", + )) + variables.append(target_row) + by_name[str(target_row["source_name"]).casefold()] = target_row + if target_rows: + _failure_boundary("catalog") + relation = Table( + table_name, MetaData(), schema=dataset.get("physical_table_schema"), + autoload_with=connection, + ) for operation in plan.operations: if isinstance(operation, RecodeOperation): source_variable = by_name[operation.source.casefold()] - if operation.target_mode == "create": - 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"])] @@ -461,6 +547,22 @@ def _apply_plan_on_connection( else_=_result_expression(operation.unmatched, source_column), ) connection.execute(update(relation).values({target_column: expression})) + _failure_boundary("data") + elif isinstance(operation, AssignOperation): + target = by_name[operation.target.casefold()] + target_column = relation.c[str(target["physical_name"])] + value = _operand_expression(operation.value, relation, by_name) + connection.execute(update(relation).values({target_column: value})) + _failure_boundary("data") + elif isinstance(operation, ConditionalAssignOperation): + target = by_name[operation.target.casefold()] + target_column = relation.c[str(target["physical_name"])] + value = _operand_expression(operation.value, relation, by_name) + condition = _predicate_expression(operation.condition, relation, by_name) + connection.execute( + update(relation).where(condition).values({target_column: value}) + ) + _failure_boundary("data") elif isinstance(operation, SetVariableLabelOperation): variable = by_name[operation.variable.casefold()] connection.execute(update(core.variable).where( @@ -470,6 +572,7 @@ def _apply_plan_on_connection( legacy_variable.c.dataset_id == legacy_dataset_id, legacy_variable.c.ordinal == variable["source_ordinal"], ).values(label=operation.label)) + _failure_boundary("catalog") elif isinstance(operation, ReplaceValueLabelsOperation): _replace_value_labels( connection, @@ -480,7 +583,42 @@ def _apply_plan_on_connection( variable=by_name[operation.variable.casefold()], labels=operation.labels, ) - else: # pragma: no cover - canonical plan type is closed + _failure_boundary("catalog") + elif isinstance(operation, SetFormatOperation): + variable = by_name[operation.variable.casefold()] + connection.execute(update(core.variable).where( + core.variable.c.variable_id == variable["variable_id"] + ).values( + print_format_family=operation.family, + print_format_width=operation.width, + print_format_decimals=operation.decimals, + write_format_family=operation.family, + write_format_width=operation.width, + write_format_decimals=operation.decimals, + )) + encoded = json.dumps([5, operation.width, operation.decimals]) + connection.execute(update(legacy_variable).where( + legacy_variable.c.dataset_id == legacy_dataset_id, + legacy_variable.c.ordinal == variable["source_ordinal"], + ).values( + format=f"F{operation.width}.{operation.decimals}", + print_format=encoded, + write_format=encoded, + )) + _failure_boundary("catalog") + elif isinstance(operation, SetMeasurementLevelOperation): + variable = by_name[operation.variable.casefold()] + connection.execute(update(core.variable).where( + core.variable.c.variable_id == variable["variable_id"] + ).values(measurement_level=operation.level)) + connection.execute(update(legacy_variable).where( + legacy_variable.c.dataset_id == legacy_dataset_id, + legacy_variable.c.ordinal == variable["source_ordinal"], + ).values(measure=operation.level)) + _failure_boundary("catalog") + elif isinstance(operation, ExecuteOperation): + continue + else: # pragma: no cover raise TransformationError( "operation_not_supported", "Unsupported in-place plan operation." ) @@ -489,11 +627,12 @@ def _apply_plan_on_connection( if after_identity != before_identity: raise TransformationError( "dataset_identity_changed", - "In-place apply changed the target dataset or its physical " - "data-table identity.", + "In-place apply changed the target dataset or its physical data-table identity.", ) apply_id = str(uuid4()) started = _now() + if mutation_journal is not None: + mutation_journal["apply_id"] = apply_id connection.execute(insert(audit).values( apply_id=apply_id, contract_id=APPLY_CONTRACT, @@ -515,6 +654,7 @@ def _apply_plan_on_connection( started_at=started, completed_at=_now(), )) + _failure_boundary("audit") forbidden = { name for name in inspect(connection).get_table_names() if name.startswith("derived_plan_") @@ -532,6 +672,7 @@ def _apply_plan_on_connection( "status": "succeeded", "dataset_id": dataset_id, "database_profile": database_profile, + "physical_table_schema": dataset.get("physical_table_schema"), "physical_table_name": table_name, "source_kind": submission.source_kind, @@ -584,6 +725,90 @@ def load_transformation_schema(connection: Any, dataset_id: str) -> VariableSche return _input_schema(connection, dataset_id)[2] +def _compensate_failed_apply( + engine: Any, + *, + journal: Mapping[str, Any], +) -> None: + columns = list(journal.get("added_columns") or ()) + if not columns: + return + with engine.begin() as connection: + + core = core_catalog(MetaData()) + legacy_metadata = MetaData() + _, legacy_variable, _, _ = legacy_catalog(legacy_metadata) + _, legacy_labels, _, _ = normalized_metadata_tables(legacy_metadata) + target_rows = list(journal.get("target_rows") or ()) + variable_ids = [str(row["variable_id"]) for row in target_rows] + ordinals = [int(row["source_ordinal"]) for row in target_rows] + if variable_ids: + label_set_ids = list(connection.execute( + select(core.variable_value_label_set.c.value_label_set_id).where( + core.variable_value_label_set.c.variable_id.in_(variable_ids) + ) + ).scalars()) + if label_set_ids: + connection.execute(delete(core.value_label).where( + core.value_label.c.value_label_set_id.in_(label_set_ids) + )) + connection.execute(delete(core.variable_value_label_set).where( + core.variable_value_label_set.c.variable_id.in_(variable_ids) + )) + connection.execute(delete(core.value_label_set).where( + core.value_label_set.c.value_label_set_id.in_(label_set_ids) + )) + connection.execute(delete(core.variable).where( + core.variable.c.variable_id.in_(variable_ids) + )) + legacy_dataset_id = journal.get("legacy_dataset_id") + if legacy_dataset_id is not None and ordinals: + connection.execute(delete(legacy_labels).where( + legacy_labels.c.dataset_id == legacy_dataset_id, + legacy_labels.c.variable_ordinal.in_(ordinals), + )) + connection.execute(delete(legacy_variable).where( + legacy_variable.c.dataset_id == legacy_dataset_id, + legacy_variable.c.ordinal.in_(ordinals), + )) + apply_id = journal.get("apply_id") + if apply_id: + audit = apply_audit_catalog(MetaData()) + connection.execute(delete(audit).where( + audit.c.apply_id == apply_id + )) + existing = { + str(item["name"]).casefold() + for item in inspect(connection).get_columns( + str(journal["table_name"]), schema=journal.get("table_schema") + ) + } + columns = [column for column in columns if str(column).casefold() in existing] + if not columns: + return + relation = Table( + str(journal["table_name"]), MetaData(), + schema=journal.get("table_schema"), autoload_with=connection, + ) + quote = connection.dialect.identifier_preparer.quote + qualified = connection.dialect.identifier_preparer.format_table(relation) + for column in reversed(columns): + connection.exec_driver_sql( + f"ALTER TABLE {qualified} DROP COLUMN {quote(str(column))}" + ) + remaining = { + str(item["name"]).casefold() + for item in inspect(connection).get_columns( + str(journal["table_name"]), schema=journal.get("table_schema") + ) + } + if any(str(column).casefold() in remaining for column in columns): + raise TransformationError( + "schema_compensation_incomplete", + "New target columns remain after compensating cleanup.", + ) + + def _run_in_place_submission( *, database_url: str, @@ -593,60 +818,81 @@ def _run_in_place_submission( expected_branch: str | None = None, expected_head: str | None = None, ) -> dict[str, Any]: - """Prepare and apply one canonical plan in the same controlled transaction.""" + """Prepare and apply one canonical plan in one controlled operation.""" 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) + journal: dict[str, Any] = {} + branch: str | None = None + head: str | None = None 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.", + try: + with engine.begin() as connection: + 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.", + ) + 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, + submission=submission, + actor=actor, + database_profile=profile.name, + allow_schema_change=profile.name in {"sqlite", "postgresql"}, + dolt_branch=branch, + dolt_head=head, + mutation_journal=journal, + ) + 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.", + ) + if dirty_after <= 0: + raise TransformationError( + "dolt_expected_working_set_diff_missing", + "A successful Dolt apply must leave an inspectable working-set diff.", + ) + return result + except Exception: + if journal.get("added_columns"): + try: + _compensate_failed_apply( + engine, + journal=journal, ) - 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, - submission=submission, - 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: + except Exception as cleanup_error: raise TransformationError( - "dolt_context_changed", - "Apply must not switch branches or create a Dolt commit.", - ) - return result + "in_place_compensation_failed", + "Apply failed and compensating cleanup did not complete.", + ) from cleanup_error + raise finally: engine.dispose() diff --git a/src/openstatspec/transform/__init__.py b/src/openstatspec/transform/__init__.py index 6ef1124..bf89591 100644 --- a/src/openstatspec/transform/__init__.py +++ b/src/openstatspec/transform/__init__.py @@ -3,8 +3,11 @@ from .errors import SourcePosition, SourceSpan, TransformationFrontendError from .plan import ( TRANSFORMATION_PLAN_CONTRACT, + AssignOperation, BooleanExpression, ComparisonExpression, + ConditionalAssignOperation, ExecuteOperation, Operand, PredicateExpression, RecodeMatch, RecodeOperation, RecodeResult, RecodeRule, - ReplaceValueLabelsOperation, SetVariableLabelOperation, TransformationPlan, + ReplaceValueLabelsOperation, SetFormatOperation, + SetMeasurementLevelOperation, SetVariableLabelOperation, TransformationPlan, TypedValue, ValueLabel, canonical_plan_hash, canonical_plan_json, transformation_plan_from_dict, ) @@ -13,9 +16,13 @@ ) from .validation import bind_transformation_plan __all__ = [ + "AssignOperation", "BooleanExpression", "ComparisonExpression", + "ConditionalAssignOperation", "ExecuteOperation", "Operand", + "PredicateExpression", "BoundTransformation", "RecodeMatch", "RecodeOperation", "RecodeResult", "RecodeRule", "ReplaceValueLabelsOperation", "SPSS_FRONTEND_CONTRACT", + "SetFormatOperation", "SetMeasurementLevelOperation", "SetVariableLabelOperation", "SourcePosition", "SourceSpan", "StorageKind", "SpssFrontendCompilation", "SpssSyntaxProgram", "TRANSFORMATION_PLAN_CONTRACT", "TransformationFrontendError", diff --git a/src/openstatspec/transform/plan.py b/src/openstatspec/transform/plan.py index 6305ff8..e0b1479 100644 --- a/src/openstatspec/transform/plan.py +++ b/src/openstatspec/transform/plan.py @@ -14,7 +14,9 @@ from .errors import frontend_error -TRANSFORMATION_PLAN_CONTRACT = "openstatspec-transformation-plan-v0.1" +TRANSFORMATION_PLAN_V1_CONTRACT = "openstatspec-transformation-plan-v0.1" +TRANSFORMATION_PLAN_CONTRACT = "openstatspec-transformation-plan-v0.2" +_TRANSFORMATION_PLAN_CONTRACTS = {TRANSFORMATION_PLAN_V1_CONTRACT, TRANSFORMATION_PLAN_CONTRACT} _BINARY64 = re.compile(r"[0-9a-f]{16}") @@ -256,7 +258,209 @@ def as_dict(self) -> dict[str, Any]: } -PlanOperation = RecodeOperation | SetVariableLabelOperation | ReplaceValueLabelsOperation +@dataclass(frozen=True) +class Operand: + """One typed literal or one variable reference in a bounded expression.""" + + kind: Literal["variable", "literal"] + variable: str | None = None + value: TypedValue | None = None + + def __post_init__(self) -> None: + if self.kind == "variable": + if not isinstance(self.variable, str) or not self.variable or self.value is not None: + _invalid("A variable operand requires a non-empty variable only.") + elif self.kind == "literal": + if self.variable is not None or not isinstance(self.value, TypedValue): + _invalid("A literal operand requires a typed value only.") + else: # pragma: no cover + _invalid("Operand kind must be variable or literal.") + + @classmethod + def variable_ref(cls, variable: str) -> "Operand": + return cls("variable", variable=variable) + + @classmethod + def literal(cls, value: TypedValue) -> "Operand": + return cls("literal", value=value) + + def as_dict(self) -> dict[str, Any]: + if self.kind == "variable": + assert self.variable is not None + return {"kind": "variable", "variable": self.variable} + assert self.value is not None + return {"kind": "literal", "value": self.value.as_dict()} + + +@dataclass(frozen=True) +class ComparisonExpression: + left: Operand + operator: Literal["=", "<", "<=", ">", ">="] + right: Operand + expression: Literal["comparison"] = "comparison" + + def __post_init__(self) -> None: + if self.expression != "comparison": + _invalid("Comparison expression discriminator is invalid.") + if not isinstance(self.left, Operand) or not isinstance(self.right, Operand): + _invalid("A comparison requires two typed operands.") + if self.operator not in {"=", "<", "<=", ">", ">="}: + _invalid("Comparison operator is outside the bounded expression profile.") + + def as_dict(self) -> dict[str, Any]: + return { + "expression": self.expression, "left": self.left.as_dict(), + "operator": self.operator, "right": self.right.as_dict(), + } + + +@dataclass(frozen=True) +class BooleanExpression: + operator: Literal["and", "or"] + operands: tuple["PredicateExpression", ...] + expression: Literal["boolean"] = "boolean" + + def __post_init__(self) -> None: + if self.expression != "boolean": + _invalid("Boolean expression discriminator is invalid.") + if self.operator not in {"and", "or"}: + _invalid("Boolean operator must be and or or.") + if not isinstance(self.operands, tuple) or len(self.operands) < 2: + _invalid("A boolean expression requires at least two operands.") + if not all(isinstance(item, (ComparisonExpression, BooleanExpression)) for item in self.operands): + _invalid("Boolean operands must be predicate expressions.") + + def as_dict(self) -> dict[str, Any]: + return { + "expression": self.expression, "operator": self.operator, + "operands": [item.as_dict() for item in self.operands], + } + + +PredicateExpression = ComparisonExpression | BooleanExpression + + +@dataclass(frozen=True) +class AssignOperation: + target: str + target_mode: Literal["create", "replace"] + value: Operand + op: Literal["assign"] = "assign" + + def __post_init__(self) -> None: + if self.op != "assign": + _invalid("Assign operation discriminator is invalid.") + if not isinstance(self.target, str) or not self.target: + _invalid("Assign target must be non-empty text.") + if self.target_mode not in {"create", "replace"}: + _invalid("Assign target_mode must be create or replace.") + if not isinstance(self.value, Operand): + _invalid("Assign value must be a typed operand.") + + def as_dict(self) -> dict[str, Any]: + return { + "op": self.op, "target": self.target, + "target_mode": self.target_mode, "value": self.value.as_dict(), + } + + +@dataclass(frozen=True) +class ConditionalAssignOperation: + condition: PredicateExpression + target: str + value: Operand + op: Literal["conditional_assign"] = "conditional_assign" + + def __post_init__(self) -> None: + if self.op != "conditional_assign": + _invalid("Conditional-assign operation discriminator is invalid.") + if not isinstance(self.condition, (ComparisonExpression, BooleanExpression)): + _invalid("Conditional assignment requires a predicate expression.") + if not isinstance(self.target, str) or not self.target: + _invalid("Conditional-assign target must be non-empty text.") + if not isinstance(self.value, Operand): + _invalid("Conditional-assign value must be a typed operand.") + + def as_dict(self) -> dict[str, Any]: + return { + "op": self.op, "condition": self.condition.as_dict(), + "target": self.target, "value": self.value.as_dict(), + } + + +@dataclass(frozen=True) +class SetFormatOperation: + variable: str + family: str + width: int + decimals: int + op: Literal["set_format"] = "set_format" + + def __post_init__(self) -> None: + if self.op != "set_format": + _invalid("Format operation discriminator is invalid.") + if not isinstance(self.variable, str) or not self.variable: + _invalid("Format operation requires a variable.") + if self.family != "F": + _invalid("The bounded format profile supports numeric F formats only.") + if not isinstance(self.width, int) or isinstance(self.width, bool) or not 1 <= self.width <= 40: + _invalid("F format width must be an integer from 1 through 40.") + if ( + not isinstance(self.decimals, int) + or isinstance(self.decimals, bool) + or not 0 <= self.decimals <= 16 + or (self.decimals != 0 and self.width < self.decimals + 2) + ): + raise frontend_error( + "invalid_format", + "F format decimals must be zero through 16 and fit the width.", + width=self.width, decimals=self.decimals, + ) + + def as_dict(self) -> dict[str, Any]: + return { + "op": self.op, "variable": self.variable, "family": self.family, + "width": self.width, "decimals": self.decimals, + } + + +@dataclass(frozen=True) +class SetMeasurementLevelOperation: + variable: str + level: Literal["nominal", "ordinal", "scale"] + op: Literal["set_measurement_level"] = "set_measurement_level" + + def __post_init__(self) -> None: + if self.op != "set_measurement_level": + _invalid("Measurement-level operation discriminator is invalid.") + if not isinstance(self.variable, str) or not self.variable: + _invalid("Measurement-level operation requires a variable.") + if self.level not in {"nominal", "ordinal", "scale"}: + _invalid("Measurement level must be nominal, ordinal, or scale.") + + def as_dict(self) -> dict[str, Any]: + return {"op": self.op, "variable": self.variable, "level": self.level} + + +@dataclass(frozen=True) +class ExecuteOperation: + """An explicit SPSS procedure boundary; a deterministic data no-op.""" + + op: Literal["execute"] = "execute" + + def __post_init__(self) -> None: + if self.op != "execute": + _invalid("Execute operation discriminator is invalid.") + + def as_dict(self) -> dict[str, str]: + return {"op": self.op} + + +PlanOperation = ( + RecodeOperation | AssignOperation | ConditionalAssignOperation + | SetVariableLabelOperation | ReplaceValueLabelsOperation + | SetFormatOperation | SetMeasurementLevelOperation | ExecuteOperation +) @dataclass(frozen=True) @@ -266,8 +470,16 @@ class TransformationPlan: 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 self.contract not in _TRANSFORMATION_PLAN_CONTRACTS: + _invalid("Plan contract is not a supported transformation-plan contract.") + if self.contract == TRANSFORMATION_PLAN_V1_CONTRACT and any( + isinstance(operation, ( + AssignOperation, ConditionalAssignOperation, SetFormatOperation, + SetMeasurementLevelOperation, ExecuteOperation, + )) + for operation in self.operations + ): + _invalid("Transformation-plan v0.1 cannot contain v0.2 operations.") 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: @@ -275,7 +487,11 @@ def __post_init__(self) -> None: if not all( isinstance( operation, - (RecodeOperation, SetVariableLabelOperation, ReplaceValueLabelsOperation), + ( + RecodeOperation, AssignOperation, ConditionalAssignOperation, + SetVariableLabelOperation, ReplaceValueLabelsOperation, + SetFormatOperation, SetMeasurementLevelOperation, ExecuteOperation, + ), ) for operation in self.operations ): @@ -308,6 +524,42 @@ def _typed(raw: Any) -> TypedValue: return TypedValue.from_dict(raw) +def _operand(raw: Any) -> Operand: + if not isinstance(raw, Mapping) or not isinstance(raw.get("kind"), str): + _invalid("Operand must be an object with a kind.") + if raw["kind"] == "variable": + _exact(raw, {"kind", "variable"}, "Variable operand") + return Operand.variable_ref(raw["variable"]) + if raw["kind"] == "literal": + _exact(raw, {"kind", "value"}, "Literal operand") + return Operand.literal(_typed(raw["value"])) + _invalid("Unknown operand kind.") + + +def _predicate(raw: Any) -> PredicateExpression: + if not isinstance(raw, Mapping) or not isinstance(raw.get("expression"), str): + _invalid("Predicate expression must be an object with an expression discriminator.") + if raw["expression"] == "comparison": + _exact( + raw, {"expression", "left", "operator", "right"}, + "Comparison expression", + ) + return ComparisonExpression( + _operand(raw["left"]), raw["operator"], _operand(raw["right"]), + ) + if raw["expression"] == "boolean": + _exact(raw, {"expression", "operator", "operands"}, "Boolean expression") + operands_raw = raw["operands"] + if not isinstance(operands_raw, Sequence) or isinstance( + operands_raw, (str, bytes) + ): + _invalid("Boolean operands must be an array.") + return BooleanExpression( + raw["operator"], tuple(_predicate(item) for item in operands_raw), + ) + _invalid("Unknown predicate expression kind.") + + 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.") @@ -341,7 +593,7 @@ def _match(raw: Any) -> RecodeMatch: def transformation_plan_from_dict(raw: Mapping[str, Any]) -> TransformationPlan: - """Strictly validate and construct the canonical v0.1 plan document.""" + """Strictly validate canonical v0.1 or additive v0.2 plan documents.""" if not isinstance(raw, Mapping): _invalid("Transformation plan must be an object.") _exact(raw, {"contract", "input_alias", "operations"}, "Transformation plan") @@ -373,6 +625,47 @@ def transformation_plan_from_dict(raw: Mapping[str, Any]) -> TransformationPlan: target_mode=raw_operation["target_mode"], rules=tuple(rules), unmatched=_result(raw_operation["unmatched"]), )) + elif operation == "assign": + _exact( + raw_operation, {"op", "target", "target_mode", "value"}, + "Assign operation", + ) + operations.append(AssignOperation( + target=raw_operation["target"], + target_mode=raw_operation["target_mode"], + value=_operand(raw_operation["value"]), + )) + elif operation == "conditional_assign": + _exact( + raw_operation, {"op", "condition", "target", "value"}, + "Conditional-assign operation", + ) + operations.append(ConditionalAssignOperation( + condition=_predicate(raw_operation["condition"]), + target=raw_operation["target"], + value=_operand(raw_operation["value"]), + )) + elif operation == "set_format": + _exact( + raw_operation, + {"op", "variable", "family", "width", "decimals"}, + "Format operation", + ) + operations.append(SetFormatOperation( + variable=raw_operation["variable"], family=raw_operation["family"], + width=raw_operation["width"], decimals=raw_operation["decimals"], + )) + elif operation == "set_measurement_level": + _exact( + raw_operation, {"op", "variable", "level"}, + "Measurement-level operation", + ) + operations.append(SetMeasurementLevelOperation( + variable=raw_operation["variable"], level=raw_operation["level"], + )) + elif operation == "execute": + _exact(raw_operation, {"op"}, "Execute operation") + operations.append(ExecuteOperation()) elif operation == "set_variable_label": _exact(raw_operation, {"op", "variable", "label"}, "Variable-label operation") operations.append(SetVariableLabelOperation( diff --git a/src/openstatspec/transform/schema.py b/src/openstatspec/transform/schema.py index 15cc638..5dace1d 100644 --- a/src/openstatspec/transform/schema.py +++ b/src/openstatspec/transform/schema.py @@ -17,6 +17,10 @@ class VariableDefinition: storage_kind: StorageKind variable_label: str | None = None value_labels: tuple[ValueLabel, ...] = () + format_family: str | None = None + format_width: int | None = None + format_decimals: int | None = None + measurement_level: Literal["nominal", "ordinal", "scale"] | None = None def __post_init__(self) -> None: if not isinstance(self.name, str) or not self.name: @@ -29,6 +33,32 @@ def __post_init__(self) -> None: "Value-label types must match their variable storage kind." ) + format_parts = (self.format_family, self.format_width, self.format_decimals) + if any(part is not None for part in format_parts): + if any(part is None for part in format_parts): + raise ValueError("Format family, width, and decimals must be set together.") + if self.storage_kind != "numeric" or self.format_family != "F": + raise ValueError("The bounded schema supports F formats on numeric variables only.") + if ( + not isinstance(self.format_width, int) + or isinstance(self.format_width, bool) + or not 1 <= self.format_width <= 40 + ): + raise ValueError("F format width must be an integer from 1 through 40.") + if ( + not isinstance(self.format_decimals, int) + or isinstance(self.format_decimals, bool) + or not 0 <= self.format_decimals <= 16 + or ( + self.format_decimals != 0 + and self.format_width < self.format_decimals + 2 + ) + ): + raise ValueError( + "F format decimals must be zero through 16 and fit the width." + ) + if self.measurement_level not in {None, "nominal", "ordinal", "scale"}: + raise ValueError("measurement_level must be nominal, ordinal, scale, or None.") @dataclass(frozen=True) class VariableSchema: diff --git a/src/openstatspec/transform/validation.py b/src/openstatspec/transform/validation.py index 79338b2..57b605b 100644 --- a/src/openstatspec/transform/validation.py +++ b/src/openstatspec/transform/validation.py @@ -7,10 +7,18 @@ from .errors import frontend_error from .plan import ( + AssignOperation, + BooleanExpression, + ComparisonExpression, + ConditionalAssignOperation, + ExecuteOperation, + Operand, RecodeMatch, RecodeOperation, RecodeResult, ReplaceValueLabelsOperation, + SetFormatOperation, + SetMeasurementLevelOperation, SetVariableLabelOperation, TransformationPlan, ) @@ -145,6 +153,113 @@ def _bind_recode( ) +def _operand_type( + operand: Operand, variables: list[VariableDefinition] +) -> ValueType: + if operand.kind == "literal": + assert operand.value is not None + return operand.value.type + assert operand.variable is not None + _, variable = _resolve(variables, operand.variable) + return _expected_type(variable.storage_kind) + + +def _validate_predicate( + predicate: ComparisonExpression | BooleanExpression, + variables: list[VariableDefinition], +) -> None: + if isinstance(predicate, BooleanExpression): + for item in predicate.operands: + _validate_predicate(item, variables) + return + left_type = _operand_type(predicate.left, variables) + right_type = _operand_type(predicate.right, variables) + if left_type != right_type: + raise frontend_error( + "type_mismatch", + "Comparison operands must have the same storage type.", + operator=predicate.operator, + left_type=left_type, + right_type=right_type, + ) + if left_type == "string": + raise frontend_error( + "expression_type_unsupported", + "String comparisons are not supported until exact, " + "profile-independent collation semantics are available.", + operator=predicate.operator, + ) + if predicate.operator != "=" and left_type != "binary64": + raise frontend_error( + "type_mismatch", + "Ordered comparisons require numeric operands.", + operator=predicate.operator, + ) + + +def _bind_assign( + operation: AssignOperation, variables: list[VariableDefinition] +) -> None: + output_type = _operand_type(operation.value, variables) + if output_type == "string": + raise frontend_error( + "expression_type_unsupported", + "String assignment is not supported until explicit width and " + "profile-independent semantics are available.", + variable=operation.target, + ) + 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, + ) + variables.append(VariableDefinition( + operation.target, + "numeric" if output_type == "binary64" else "string", + )) + return + _, target = _resolve(variables, operation.target) + if output_type != _expected_type(target.storage_kind): + raise frontend_error( + "type_mismatch", + "Assignment cannot change the target storage kind.", + variable=target.name, + ) + + +def _bind_conditional_assign( + operation: ConditionalAssignOperation, + variables: list[VariableDefinition], +) -> None: + _validate_predicate(operation.condition, variables) + _, target = _resolve(variables, operation.target) + output_type = _operand_type(operation.value, variables) + if output_type != _expected_type(target.storage_kind): + raise frontend_error( + "type_mismatch", + "Conditional assignment value must match the target storage kind.", + variable=target.name, + ) + if output_type == "string": + raise frontend_error( + "expression_type_unsupported", + "String assignment is not supported until explicit width and " + "profile-independent semantics are available.", + variable=operation.target, + ) + + def bind_transformation_plan( plan: TransformationPlan, schema: VariableSchema ) -> BoundTransformation: @@ -158,6 +273,12 @@ def bind_transformation_plan( if isinstance(operation, RecodeOperation): _bind_recode(operation, variables) continue + if isinstance(operation, AssignOperation): + _bind_assign(operation, variables) + continue + if isinstance(operation, ConditionalAssignOperation): + _bind_conditional_assign(operation, variables) + continue if isinstance(operation, SetVariableLabelOperation): index, variable = _resolve(variables, operation.variable) variables[index] = replace(variable, variable_label=operation.label) @@ -174,5 +295,26 @@ def bind_transformation_plan( ) variables[index] = replace(variable, value_labels=operation.labels) continue + if isinstance(operation, SetFormatOperation): + index, variable = _resolve(variables, operation.variable) + if variable.storage_kind != "numeric": + raise frontend_error( + "type_mismatch", + "Numeric F formats require a numeric variable.", + variable=variable.name, + ) + variables[index] = replace( + variable, + format_family=operation.family, + format_width=operation.width, + format_decimals=operation.decimals, + ) + continue + if isinstance(operation, SetMeasurementLevelOperation): + index, variable = _resolve(variables, operation.variable) + variables[index] = replace(variable, measurement_level=operation.level) + continue + if isinstance(operation, ExecuteOperation): + 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 26d3ff8..2a39943 100755 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -39,9 +39,9 @@ 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.2.0" - assert matrix["specification_commit"] == "79339ec3d8f8aa81789b7e85f6b8afa6f1374e50" + assert matrix["specification_status"] == "release_candidate" + assert matrix["specification_release"] is None + assert matrix["specification_commit"] == "e49252c00890aed76dcaabc5d1ab5121b45929db" 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_conditional_inplace_transform.py b/tests/test_conditional_inplace_transform.py new file mode 100644 index 0000000..3bfb1e3 --- /dev/null +++ b/tests/test_conditional_inplace_transform.py @@ -0,0 +1,332 @@ +from __future__ import annotations + +import sqlite3 + +from types import SimpleNamespace +import pytest + +import openstatspec +import openstatspec.sql.inplace_transform as inplace_transform +from openstatspec.sql.wide import create_wide_dataset + + +SYNTAX = """COMPUTE target = 0. +IF (source_a = 1 AND source_b = 1) target = 1. +VARIABLE LABELS target 'Example label'. +VALUE LABELS target 0 'No' 1 'Yes'. +FORMATS target (F1.0). +VARIABLE LEVEL target (NOMINAL). +EXECUTE.""" + + +def _variable(ordinal: int, name: str) -> dict[str, object]: + return { + "ordinal": ordinal, + "source_name": name, + "physical_name": name, + "storage_kind": "numeric", + "string_width": None, + "label": name, + "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 conditional_catalog(tmp_path): + path = tmp_path / "conditional.sqlite" + url = f"sqlite:///{path}" + create_wide_dataset( + database_url=url, + dataset_id="conditional_source", + source_name="synthetic.sav", + source_format="SAV", + source_sha256="e" * 64, + rows=[ + {"source_a": 1.0, "source_b": 1.0}, + {"source_a": 1.0, "source_b": 0.0}, + {"source_a": 0.0, "source_b": 1.0}, + {"source_a": 2.0, "source_b": 2.0}, + ], + variables=[_variable(1, "source_a"), _variable(2, "source_b")], + ) + 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() + connection.close() + return url, path, dataset_id, table_name + + +def test_exact_bounded_program_compiles_to_stable_v02_plan() -> None: + schema = openstatspec.VariableSchema(( + openstatspec.VariableDefinition("source_a", "numeric"), + openstatspec.VariableDefinition("source_b", "numeric"), + )) + compilation = openstatspec.compile_spss_syntax(SYNTAX, schema) + assert compilation.plan.contract == "openstatspec-transformation-plan-v0.2" + assert [operation.op for operation in compilation.plan.operations] == [ + "assign", + "conditional_assign", + "set_variable_label", + "replace_value_labels", + "set_format", + "set_measurement_level", + "execute", + ] + restored = openstatspec.transformation_plan_from_dict( + compilation.plan.as_dict() + ) + assert restored.canonical_json() == compilation.plan.canonical_json() + assert restored.sha256() == compilation.plan_hash + assert compilation.plan_hash == "f57b176eb86027eeccd5fc2da5c421444f55b0f6cc70c93aa3e673f8cdbb2e90" + + +def test_exact_bounded_program_applies_data_and_both_catalogs( + conditional_catalog, +) -> None: + url, path, dataset_id, table_name = conditional_catalog + result = openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text=SYNTAX, + actor="synthetic-test", + ) + assert result["status"] == "succeeded" + assert result["dolt_commit_performed"] is False + connection = sqlite3.connect(path) + assert connection.execute( + f'SELECT target FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(1.0,), (0.0,), (0.0,), (0.0,)] + assert connection.execute( + "SELECT variable_label, print_format_family, print_format_width, " + "print_format_decimals, write_format_family, write_format_width, " + "write_format_decimals, measurement_level FROM variable " + "WHERE source_name = 'target'" + ).fetchone() == ("Example label", "F", 1, 0, "F", 1, 0, "nominal") + assert connection.execute( + "SELECT label, format, print_format, write_format, measure " + "FROM variable_catalog WHERE source_name = 'target'" + ).fetchone() == ("Example label", "F1.0", "[5, 1, 0]", "[5, 1, 0]", "nominal") + assert connection.execute( + "SELECT numeric_code, label FROM value_label ORDER BY ordinal" + ).fetchall() == [(0.0, "No"), (1.0, "Yes")] + assert connection.execute( + "SELECT contract_id, source_kind, operation_count, status " + "FROM transformation_apply" + ).fetchone() == ( + "openstatspec-in-place-transformation-v0.2", + "spss_syntax", + 7, + "succeeded", + ) + connection.close() + + +@pytest.mark.parametrize("boundary", ["schema", "data", "catalog", "audit"]) +def test_injected_boundary_failure_leaves_no_partial_apply( + conditional_catalog, monkeypatch, boundary, +) -> None: + url, path, dataset_id, table_name = conditional_catalog + raised = False + connection = sqlite3.connect(path) + connection.execute(f'ALTER TABLE "{table_name}" ADD COLUMN unrelated REAL') + connection.execute(f'UPDATE "{table_name}" SET unrelated = 42') + connection.commit() + connection.close() + + def fail(selected: str) -> None: + nonlocal raised + if selected == boundary and not raised: + raised = True + raise RuntimeError(f"synthetic {boundary} failure") + + monkeypatch.setattr(inplace_transform, "_failure_boundary", fail) + with pytest.raises(RuntimeError, match=f"synthetic {boundary} failure"): + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text=SYNTAX, + actor="synthetic-test", + ) + connection = sqlite3.connect(path) + columns = { + row[1] for row in connection.execute(f'PRAGMA table_info("{table_name}")') + } + assert "target" not in columns + assert connection.execute( + "SELECT COUNT(*) FROM variable WHERE source_name = 'target'" + ).fetchone() == (0,) + assert connection.execute( + "SELECT COUNT(*) FROM variable_catalog WHERE source_name = 'target'" + ).fetchone() == (0,) + assert connection.execute( + "SELECT COUNT(*) FROM transformation_apply" + ).fetchone() == (0,) + assert connection.execute( + f'SELECT source_a, source_b FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(1.0, 1.0), (1.0, 0.0), (0.0, 1.0), (2.0, 2.0)] + + assert connection.execute( + f'SELECT unrelated FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(42.0,), (42.0,), (42.0,), (42.0,)] + connection.close() + +def test_failure_never_drops_a_preexisting_target( + conditional_catalog, monkeypatch, +) -> None: + url, path, dataset_id, table_name = conditional_catalog + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text=SYNTAX, + actor="synthetic-test", + ) + + def fail(selected: str) -> None: + if selected == "data": + raise RuntimeError("synthetic replace failure") + + monkeypatch.setattr(inplace_transform, "_failure_boundary", fail) + with pytest.raises(RuntimeError, match="synthetic replace failure"): + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text="COMPUTE target = 9. EXECUTE.", + actor="synthetic-test", + ) + connection = sqlite3.connect(path) + assert connection.execute( + f'SELECT target FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(1.0,), (0.0,), (0.0,), (0.0,)] + assert connection.execute( + "SELECT COUNT(*) FROM variable WHERE source_name = 'target'" + ).fetchone() == (1,) + assert connection.execute( + "SELECT COUNT(*) FROM transformation_apply" + ).fetchone() == (1,) + connection.close() + + + +def test_numeric_predicates_use_sql_three_valued_logic_and_ordered_execution( + conditional_catalog, +) -> None: + url, path, dataset_id, table_name = conditional_catalog + connection = sqlite3.connect(path) + connection.execute( + f'UPDATE "{table_name}" SET source_a = NULL WHERE __case_ordinal = 2' + ) + connection.commit() + connection.close() + source = ( + "COMPUTE target = 0. " + "IF ((source_a < 1 OR source_a >= 2) AND source_b <= 2) target = 1. " + "IF (target = 1 AND source_b > 1) target = 2. EXECUTE." + ) + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text=source, + actor="synthetic-test", + ) + connection = sqlite3.connect(path) + assert connection.execute( + f'SELECT target FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == [(0.0,), (0.0,), (1.0,), (2.0,)] + connection.close() + + +def test_dolt_mock_applies_exact_program_to_preexisting_target_without_schema_ddl( + conditional_catalog, monkeypatch, +) -> None: + url, path, dataset_id, table_name = conditional_catalog + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text=SYNTAX, + actor="provisioning-stage", + ) + monkeypatch.setattr( + inplace_transform, + "effective_profile", + lambda _url: (SimpleNamespace(name="dolt"), {}), + ) + states = iter([ + ("main", "abc123", 0), + ("main", "abc123", 0), + ("main", "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=SYNTAX, + actor="synthetic-test", + expected_branch="main", + expected_head="abc123", + ) + assert result["dolt_commit_performed"] is False + connection = sqlite3.connect(path) + columns = [ + row[1] for row in connection.execute(f'PRAGMA table_info("{table_name}")') + ] + assert columns.count("target") == 1 + assert connection.execute( + "SELECT COUNT(*) FROM variable WHERE source_name = 'target'" + ).fetchone() == (1,) + connection.close() + + +def test_dolt_mock_rejects_create_target_before_schema_mutation( + conditional_catalog, monkeypatch, +) -> None: + url, path, dataset_id, table_name = conditional_catalog + monkeypatch.setattr( + inplace_transform, + "effective_profile", + lambda _url: (SimpleNamespace(name="dolt"), {}), + ) + monkeypatch.setattr( + inplace_transform, + "_dolt_state", + lambda _connection: ("main", "abc123", 0), + ) + with pytest.raises(openstatspec.TransformationError) as caught: + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text=SYNTAX, + actor="synthetic-test", + expected_branch="main", + expected_head="abc123", + ) + assert caught.value.code == "schema_change_not_atomic" + assert "target" not in { + row[1] for row in sqlite3.connect(path).execute( + f'PRAGMA table_info("{table_name}")' + ) + } + + +def test_dolt_mock_rechecks_clean_state_after_dataset_lock( + conditional_catalog, monkeypatch, +) -> None: + url, path, dataset_id, table_name = conditional_catalog + monkeypatch.setattr( + inplace_transform, + "effective_profile", + lambda _url: (SimpleNamespace(name="dolt"), {}), + ) diff --git a/tests/test_inplace_transform.py b/tests/test_inplace_transform.py index 3c4cda4..7cdf69f 100644 --- a/tests/test_inplace_transform.py +++ b/tests/test_inplace_transform.py @@ -174,7 +174,7 @@ def test_public_apply_supports_non_dolt_without_building_undo(catalog) -> None: ).fetchone() assert audit == ( "spss_syntax", - "openstatspec-spss-syntax-frontend-v0.1", + "openstatspec-spss-syntax-frontend-v0.2", ) @@ -407,6 +407,7 @@ def test_public_apply_binds_expected_dolt_branch_and_head( lambda _url: (SimpleNamespace(name="dolt"), {}), ) states = iter([ + ("feature/recode", "abc123", 0), ("feature/recode", "abc123", 0), ("feature/recode", "abc123", 4), ]) diff --git a/tests/test_sql_profiles.py b/tests/test_sql_profiles.py index 5988265..3197ad4 100755 --- a/tests/test_sql_profiles.py +++ b/tests/test_sql_profiles.py @@ -25,13 +25,13 @@ def test_profile_detection_tracks_supported_dialect_urls() -> None: assert profile_for_url("mariadb+mariadbconnector://user@host/database") is MYSQL assert profile_for_url("mysql+pymysql://user@host/dolt_database") is MYSQL -def test_profile_declarations_publish_released_specification_provenance() -> None: +def test_profile_declarations_publish_specification_candidate_provenance() -> None: for declaration in capabilities.profile_declarations().values(): - assert declaration["specification_status"] == "released" - assert declaration["specification_release"] == "v0.2.0" + assert declaration["specification_status"] == "release_candidate" + assert declaration["specification_release"] is None assert ( declaration["specification_commit"] - == "79339ec3d8f8aa81789b7e85f6b8afa6f1374e50" + == "e49252c00890aed76dcaabc5d1ab5121b45929db" ) 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 index 016ec79..9f3f376 100644 --- a/tests/test_transform_frontend.py +++ b/tests/test_transform_frontend.py @@ -133,6 +133,73 @@ def test_official_spss_frontend_conformance_manifest() -> None: assert actual_metadata == case["expected_output_metadata"], case["id"] + +def _frontend_conformance_manifest_v02() -> Path: + configured = os.environ.get("OPENSTATSPEC_SPECIFICATION_DIR") + candidates = [ + ( + Path(configured) / "conformance/spss-syntax-frontend-0.2.json" + if configured + else None + ), + Path(__file__).resolve().parents[1] + / "openstatspec-specification/conformance/spss-syntax-frontend-0.2.json", + Path(__file__).resolve().parents[2] + / "specification/conformance/spss-syntax-frontend-0.2.json", + ] + for candidate in candidates: + if candidate and candidate.is_file(): + return candidate + raise RuntimeError( + "The SPSS frontend v0.2 conformance fixture is required; " + "set OPENSTATSPEC_SPECIFICATION_DIR." + ) + + +def test_official_spss_frontend_v02_conformance_manifest() -> None: + manifest_path = _frontend_conformance_manifest_v02() + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + plan_manifest = json.loads( + (manifest_path.parent / "transformation-plan-0.2.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"), + format_family=variable.get("format_family"), + format_width=variable.get("format_width"), + format_decimals=variable.get("format_decimals"), + measurement_level=variable.get("measurement_level"), + ) + 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"], + ) + assert compilation.plan_hash == case["expected_plan_hash"], case["id"] + if "expected_plan_case" in case: + expected = plan_cases[case["expected_plan_case"]] + assert compilation.plan.as_dict() == expected["plan"], case["id"] + assert compilation.plan_hash == expected["expected_plan_hash"], case["id"] + else: + assert compilation.plan.contract == case["expected_plan_contract"], 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) " @@ -276,6 +343,47 @@ def test_source_normalization_hash_and_positions_are_stable() -> None: assert compilation.source_hash == spss_source_hash(lf) assert compilation.plan_hash == compilation.plan.sha256() +def test_string_comparison_fails_closed_until_exact_collation_is_supported() -> None: + error = _error( + "COMPUTE flag = 0. IF (Name = 'alice') flag = 1.", + _schema(VariableDefinition("Name", "string")), + ) + assert error.code == "expression_type_unsupported" + + +def test_string_assignment_fails_closed_until_width_semantics_are_supported() -> None: + error = _error( + "COMPUTE Copy = Name.", + _schema(VariableDefinition("Name", "string")), + ) + assert error.code == "expression_type_unsupported" + + +def test_v02_frontend_stable_diagnostics_match_the_normative_profile() -> None: + schema = _schema(VariableDefinition("q1", "numeric")) + assert _error( + "COMPUTE flag = 0. IF q1 = 1 flag = 1.", schema, + ).code == "spss_syntax_error" + assert _error( + "IF (q1 = 1) flag = 1.", schema, + ).code == "conditional_target_missing" + assert _error( + "FORMATS q1 (F2.1).", schema, + ).code == "invalid_format" + assert _error( + "FORMATS q1 (F8.17).", schema, + ).code == "invalid_format" + assert _error( + "FORMATS q1 (A8).", schema, + ).code == "invalid_format" + + +def test_v01_plan_loader_preserves_official_contract_and_hash() -> None: + plan = _compile("RECODE score (1 = 2).", _schema(VariableDefinition("score", "numeric"))).plan + assert plan.contract == "openstatspec-transformation-plan-v0.1" + assert transformation_plan_from_dict(plan.as_dict()).sha256() == plan.sha256() + + @pytest.mark.parametrize( ("source", "code"), @@ -296,6 +404,16 @@ def test_stable_failures(source: str, code: str) -> None: ).code == code +def test_arbitrary_sql_and_python_sources_fail_closed() -> None: + schema = _schema(VariableDefinition("q1", "numeric")) + assert _error( + "SELECT * FROM q1.", schema, + ).code == "spss_syntax_error" + assert _error( + "PYTHON PROGRAM.", schema, + ).code == "unsupported_spss_command" + + def test_string_create_requires_declaration_before_mixed_type_diagnostic() -> None: error = _error( "RECODE color ('R' = 'red') INTO normalized.", @@ -330,6 +448,30 @@ def test_strict_plan_loader_rejects_runtime_type_confusion() -> None: assert caught.value.code == "invalid_transformation_plan" +def test_v02_plan_and_schema_reject_decimal_format_that_cannot_fit() -> None: + raw = { + "contract": "openstatspec-transformation-plan-v0.2", + "input_alias": "parent", + "operations": [ + { + "op": "set_format", + "variable": "target", + "family": "F", + "width": 2, + "decimals": 2, + } + ], + } + with pytest.raises(TransformationFrontendError) as caught: + transformation_plan_from_dict(raw) + assert caught.value.code == "invalid_format" + with pytest.raises(ValueError, match="fit the width"): + VariableDefinition( + "target", "numeric", format_family="F", + format_width=2, format_decimals=2, + ) + + def test_custom_nonempty_input_alias_is_canonical() -> None: plan = bind_spss_syntax( parse_spss_syntax("VARIABLE LABELS q1 'One'."), From a2a2644de5cd70d467298c8d411561010c3d7c18 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 17:11:04 +0300 Subject: [PATCH 02/12] Test against the conditional specification profile --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e31ca4e..7f3f773 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: 79339ec3d8f8aa81789b7e85f6b8afa6f1374e50 + ref: agent/conditional-transformation-0.2 path: openstatspec-specification - name: Checkout required SPSS engine uses: actions/checkout@v7 From 54c9eb873c6e3f9f9cbad687c2627336ea9d1ef1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 19:16:41 +0300 Subject: [PATCH 03/12] Keep conditional adapter work on the unreleased 0.5.0 line --- CHANGELOG.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3555b52..8daa85b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,9 @@ All notable changes to this reference implementation are documented here. -## 0.5.0 — 2026-07-31 +## Unreleased + +Planned adapter release: `0.5.0`, after lifecycle integration and final specification conformance. ### Added From e5a2b8d2581012f2e24780b259561465154c4653 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 20:03:28 +0300 Subject: [PATCH 04/12] ci: pin conditional specification fixtures --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7f3f773..219aeb8 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: agent/conditional-transformation-0.2 + ref: e49252c00890aed76dcaabc5d1ab5121b45929db path: openstatspec-specification - name: Checkout required SPSS engine uses: actions/checkout@v7 From 9da253d39f2fccb9f2cb96b4dfae80fc19bea177 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 20:17:01 +0300 Subject: [PATCH 05/12] fix: keep PostgreSQL rollback atomic --- src/openstatspec/sql/inplace_transform.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/openstatspec/sql/inplace_transform.py b/src/openstatspec/sql/inplace_transform.py index b95e2d4..0a24ad4 100644 --- a/src/openstatspec/sql/inplace_transform.py +++ b/src/openstatspec/sql/inplace_transform.py @@ -881,7 +881,13 @@ def _run_in_place_submission( ) return result except Exception: - if journal.get("added_columns"): + # PostgreSQL rolls back the DDL and catalog writes atomically. Once + # that rollback releases the dataset lock, a separate compensation + # transaction could delete artifacts created by a waiting apply. + if ( + journal.get("added_columns") + and profile.name != "postgresql" + ): try: _compensate_failed_apply( engine, From a07c6b44a936f0d08043672ee7f9494cdeb9034c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 20:17:18 +0300 Subject: [PATCH 06/12] test: cover PostgreSQL rollback compensation boundary --- tests/test_inplace_transform.py | 38 +++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/tests/test_inplace_transform.py b/tests/test_inplace_transform.py index 7cdf69f..95ddb93 100644 --- a/tests/test_inplace_transform.py +++ b/tests/test_inplace_transform.py @@ -507,3 +507,41 @@ def test_capability_declares_dolt_owned_versioning() -> None: assert declaration["creates_persistent_data_copy"] is False assert declaration["openstatspec_rollback_or_version_history"] is False assert declaration["performs_dolt_commit"] is False + +def test_postgresql_rollback_skips_unlocked_compensation( + tmp_path, monkeypatch, +) -> None: + database_url = f"sqlite:///{tmp_path / 'postgres-rollback.sqlite'}" + monkeypatch.setattr( + inplace_transform, + "effective_profile", + lambda _url: (SimpleNamespace(name="postgresql"), {}), + ) + + def fail_after_schema_change( + _connection, *, mutation_journal, **_kwargs, + ): + mutation_journal["added_columns"] = ["score_band"] + raise RuntimeError("simulated transactional failure") + + monkeypatch.setattr( + inplace_transform, "_apply_plan_on_connection", fail_after_schema_change, + ) + monkeypatch.setattr( + inplace_transform, + "_compensate_failed_apply", + lambda *_args, **_kwargs: pytest.fail( + "PostgreSQL rollback must not run unlocked compensation" + ), + ) + + with pytest.raises(RuntimeError, match="simulated transactional failure"): + inplace_transform._run_in_place_submission( + database_url=database_url, + dataset_id="dataset", + actor="test-agent", + prepare=lambda _connection, _dataset_id: _submission( + "RECODE score (1 = 0) INTO score_band." + ), + ) + From bb6221685201a62dc9a87a5fe17d2e0c7a29400b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 20:30:45 +0300 Subject: [PATCH 07/12] fix: validate plan enums before membership checks --- src/openstatspec/transform/plan.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/openstatspec/transform/plan.py b/src/openstatspec/transform/plan.py index e0b1479..55b689a 100644 --- a/src/openstatspec/transform/plan.py +++ b/src/openstatspec/transform/plan.py @@ -183,7 +183,7 @@ def __post_init__(self) -> None: _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"}: + if not isinstance(self.target_mode, str) or 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.") @@ -304,7 +304,7 @@ def __post_init__(self) -> None: _invalid("Comparison expression discriminator is invalid.") if not isinstance(self.left, Operand) or not isinstance(self.right, Operand): _invalid("A comparison requires two typed operands.") - if self.operator not in {"=", "<", "<=", ">", ">="}: + if not isinstance(self.operator, str) or self.operator not in {"=", "<", "<=", ">", ">="}: _invalid("Comparison operator is outside the bounded expression profile.") def as_dict(self) -> dict[str, Any]: @@ -323,7 +323,7 @@ class BooleanExpression: def __post_init__(self) -> None: if self.expression != "boolean": _invalid("Boolean expression discriminator is invalid.") - if self.operator not in {"and", "or"}: + if not isinstance(self.operator, str) or self.operator not in {"and", "or"}: _invalid("Boolean operator must be and or or.") if not isinstance(self.operands, tuple) or len(self.operands) < 2: _invalid("A boolean expression requires at least two operands.") @@ -352,7 +352,7 @@ def __post_init__(self) -> None: _invalid("Assign operation discriminator is invalid.") if not isinstance(self.target, str) or not self.target: _invalid("Assign target must be non-empty text.") - if self.target_mode not in {"create", "replace"}: + if not isinstance(self.target_mode, str) or self.target_mode not in {"create", "replace"}: _invalid("Assign target_mode must be create or replace.") if not isinstance(self.value, Operand): _invalid("Assign value must be a typed operand.") @@ -435,7 +435,7 @@ def __post_init__(self) -> None: _invalid("Measurement-level operation discriminator is invalid.") if not isinstance(self.variable, str) or not self.variable: _invalid("Measurement-level operation requires a variable.") - if self.level not in {"nominal", "ordinal", "scale"}: + if not isinstance(self.level, str) or self.level not in {"nominal", "ordinal", "scale"}: _invalid("Measurement level must be nominal, ordinal, or scale.") def as_dict(self) -> dict[str, Any]: @@ -470,7 +470,7 @@ class TransformationPlan: input_alias: str = "parent" def __post_init__(self) -> None: - if self.contract not in _TRANSFORMATION_PLAN_CONTRACTS: + if not isinstance(self.contract, str) or self.contract not in _TRANSFORMATION_PLAN_CONTRACTS: _invalid("Plan contract is not a supported transformation-plan contract.") if self.contract == TRANSFORMATION_PLAN_V1_CONTRACT and any( isinstance(operation, ( From c28fd2aa3d3a98187f0c805af282c6bb1d7cd4b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 20:31:10 +0300 Subject: [PATCH 08/12] test: reject unhashable canonical plan enums --- tests/test_transform_frontend.py | 71 ++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/tests/test_transform_frontend.py b/tests/test_transform_frontend.py index 9f3f376..f858cf1 100644 --- a/tests/test_transform_frontend.py +++ b/tests/test_transform_frontend.py @@ -448,6 +448,77 @@ def test_strict_plan_loader_rejects_runtime_type_confusion() -> None: assert caught.value.code == "invalid_transformation_plan" +def test_strict_plan_loader_rejects_unhashable_enum_values() -> None: + typed_value = TypedValue.binary64(1).as_dict() + literal = {"kind": "literal", "value": typed_value} + variable = {"kind": "variable", "variable": "q1"} + comparison = { + "expression": "comparison", + "left": variable, + "operator": "=", + "right": literal, + } + invalid_operations = [ + { + "op": "conditional_assign", + "condition": {**comparison, "operator": []}, + "target": "q1", + "value": literal, + }, + { + "op": "conditional_assign", + "condition": { + "expression": "boolean", + "operator": {}, + "operands": [comparison, comparison], + }, + "target": "q1", + "value": literal, + }, + { + "op": "assign", + "target": "q2", + "target_mode": [], + "value": literal, + }, + { + "op": "set_measurement_level", + "variable": "q1", + "level": {}, + }, + { + "op": "recode", + "source": "q1", + "target": "q2", + "target_mode": [], + "rules": [{ + "match": {"kind": "values", "values": [typed_value]}, + "result": {"kind": "literal", "value": typed_value}, + }], + "unmatched": {"kind": "copy"}, + }, + ] + + raw_plans = [ + { + "contract": "openstatspec-transformation-plan-v0.2", + "input_alias": "parent", + "operations": [operation], + } + for operation in invalid_operations + ] + raw_plans.append({ + "contract": [], + "input_alias": "parent", + "operations": [{"op": "execute"}], + }) + + for raw in raw_plans: + with pytest.raises(TransformationFrontendError) as caught: + transformation_plan_from_dict(raw) + assert caught.value.code == "invalid_transformation_plan" + + def test_v02_plan_and_schema_reject_decimal_format_that_cannot_fit() -> None: raw = { "contract": "openstatspec-transformation-plan-v0.2", From cdb2ebd8e5e413965d5cc076ce3f7227fca9224a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 20:40:47 +0300 Subject: [PATCH 09/12] fix: preserve transactional DDL rollback ownership --- src/openstatspec/sql/inplace_transform.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/openstatspec/sql/inplace_transform.py b/src/openstatspec/sql/inplace_transform.py index 0a24ad4..b2fdb45 100644 --- a/src/openstatspec/sql/inplace_transform.py +++ b/src/openstatspec/sql/inplace_transform.py @@ -881,12 +881,13 @@ def _run_in_place_submission( ) return result except Exception: - # PostgreSQL rolls back the DDL and catalog writes atomically. Once - # that rollback releases the dataset lock, a separate compensation - # transaction could delete artifacts created by a waiting apply. + # Transactional-DDL profiles roll back schema and catalog writes + # atomically. Once that rollback releases the dataset lock, a + # separate compensation transaction could delete artifacts created + # by a waiting apply. if ( journal.get("added_columns") - and profile.name != "postgresql" + and profile.name not in {"sqlite", "postgresql"} ): try: _compensate_failed_apply( From 4a62a6fbfa90b0d80bf4d73cffb1623f3f23302d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 20:41:02 +0300 Subject: [PATCH 10/12] test: cover SQLite transactional rollback boundary --- tests/test_inplace_transform.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/tests/test_inplace_transform.py b/tests/test_inplace_transform.py index 95ddb93..e3ac001 100644 --- a/tests/test_inplace_transform.py +++ b/tests/test_inplace_transform.py @@ -508,14 +508,15 @@ def test_capability_declares_dolt_owned_versioning() -> None: assert declaration["openstatspec_rollback_or_version_history"] is False assert declaration["performs_dolt_commit"] is False -def test_postgresql_rollback_skips_unlocked_compensation( - tmp_path, monkeypatch, +@pytest.mark.parametrize("profile_name", ["sqlite", "postgresql"]) +def test_transactional_ddl_rollback_skips_unlocked_compensation( + tmp_path, monkeypatch, profile_name, ) -> None: - database_url = f"sqlite:///{tmp_path / 'postgres-rollback.sqlite'}" + database_url = f"sqlite:///{tmp_path / 'transactional-rollback.sqlite'}" monkeypatch.setattr( inplace_transform, "effective_profile", - lambda _url: (SimpleNamespace(name="postgresql"), {}), + lambda _url: (SimpleNamespace(name=profile_name), {}), ) def fail_after_schema_change( @@ -531,7 +532,7 @@ def fail_after_schema_change( inplace_transform, "_compensate_failed_apply", lambda *_args, **_kwargs: pytest.fail( - "PostgreSQL rollback must not run unlocked compensation" + "Transactional rollback must not run unlocked compensation" ), ) From ee5a73e452e04679a5cce3b7fb4931247ca97a4c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 20:45:00 +0300 Subject: [PATCH 11/12] fix: begin SQLite DDL transaction explicitly --- src/openstatspec/sql/inplace_transform.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/openstatspec/sql/inplace_transform.py b/src/openstatspec/sql/inplace_transform.py index b2fdb45..090626e 100644 --- a/src/openstatspec/sql/inplace_transform.py +++ b/src/openstatspec/sql/inplace_transform.py @@ -831,6 +831,12 @@ def _run_in_place_submission( try: try: with engine.begin() as connection: + if profile.name == "sqlite": + # Python's sqlite3 legacy transaction mode does not begin a + # transaction for DDL. Start one explicitly so schema and + # catalog mutations roll back together before the write lock + # is released. + connection.exec_driver_sql("BEGIN") if profile.name == "dolt": if not expected_branch or not expected_head: raise TransformationError( From d54c5298a5a44266993a996114b5db161b9f046b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?T=C3=B5nis=20Ormisson?= Date: Mon, 3 Aug 2026 20:56:26 +0300 Subject: [PATCH 12/12] test: exercise Dolt post-lock dirty-state gate --- tests/test_conditional_inplace_transform.py | 49 +++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/tests/test_conditional_inplace_transform.py b/tests/test_conditional_inplace_transform.py index 3bfb1e3..627161f 100644 --- a/tests/test_conditional_inplace_transform.py +++ b/tests/test_conditional_inplace_transform.py @@ -325,8 +325,57 @@ def test_dolt_mock_rechecks_clean_state_after_dataset_lock( conditional_catalog, monkeypatch, ) -> None: url, path, dataset_id, table_name = conditional_catalog + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text=SYNTAX, + actor="provisioning-stage", + ) + connection = sqlite3.connect(path) + before = connection.execute( + f'SELECT target FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() + connection.close() + monkeypatch.setattr( inplace_transform, "effective_profile", lambda _url: (SimpleNamespace(name="dolt"), {}), ) + states = iter([ + ("main", "abc123", 0), + ("main", "abc123", 1), + ]) + observed_states = [] + + def next_state(_connection): + state = next(states) + observed_states.append(state) + return state + + monkeypatch.setattr(inplace_transform, "_dolt_state", next_state) + + with pytest.raises(openstatspec.TransformationError) as caught: + openstatspec.apply_spss_in_place( + database_url=url, + dataset_id=dataset_id, + source_text=SYNTAX, + actor="synthetic-test", + expected_branch="main", + expected_head="abc123", + ) + + assert caught.value.code == "dolt_working_set_dirty" + assert observed_states == [ + ("main", "abc123", 0), + ("main", "abc123", 1), + ] + connection = sqlite3.connect(path) + assert connection.execute( + f'SELECT target FROM "{table_name}" ORDER BY __case_ordinal' + ).fetchall() == before + assert connection.execute( + "SELECT COUNT(*) FROM transformation_apply" + ).fetchone() == (1,) + connection.close() +