diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 41073ce..8caa521 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -138,3 +138,79 @@ jobs: env: OPENSTATSPEC_MARIADB_URL: mysql+pymysql://openstatspec:openstatspec@localhost:3306/openstatspec steps: *sql-test-steps + + dolt-integration: + name: Dolt 2.2.2 integration + runs-on: ubuntu-latest + env: + OPENSTATSPEC_DOLT_URL: mysql+pymysql://openstatspec:openstatspec@127.0.0.1:13308/openstatspec + steps: + - uses: actions/checkout@v7 + - name: Checkout OpenStatSpec specification fixtures + uses: actions/checkout@v7 + with: *specification-checkout + - name: Checkout required SPSS engine + uses: actions/checkout@v7 + with: *engine-checkout + - uses: actions/setup-python@v7 + with: + python-version: "3.12" + - name: Initialize Dolt 2.2.2 + run: | + mkdir -p "$RUNNER_TEMP/openstatspec-dolt" + docker run --rm \ + --volume "$RUNNER_TEMP/openstatspec-dolt:/var/lib/dolt" \ + --workdir /var/lib/dolt \ + dolthub/dolt:2.2.2 init \ + --name "OpenStatSpec CI" --email "ci@openstatspec.org" + docker run --rm \ + --volume "$RUNNER_TEMP/openstatspec-dolt:/var/lib/dolt" \ + --workdir /var/lib/dolt \ + dolthub/dolt:2.2.2 sql \ + -q "CREATE DATABASE openstatspec" + docker run --rm \ + --volume "$RUNNER_TEMP/openstatspec-dolt:/var/lib/dolt" \ + --workdir /var/lib/dolt \ + dolthub/dolt:2.2.2 sql \ + -q "CREATE USER 'openstatspec'@'%' IDENTIFIED BY 'openstatspec'; GRANT ALL PRIVILEGES ON openstatspec.* TO 'openstatspec'@'%'" + docker run --detach --name openstatspec-dolt \ + --publish 127.0.0.1:13308:3306 \ + --volume "$RUNNER_TEMP/openstatspec-dolt:/var/lib/dolt" \ + --workdir /var/lib/dolt \ + dolthub/dolt:2.2.2 sql-server -H 0.0.0.0 -l warning + - run: python -m pip install --upgrade pip + - run: python -m pip install ./openstatspec-pyspssio + - run: python -m pip install -e ".[dev,sql]" + - name: Wait for Dolt + run: | + python - <<'PY' + import time + import pymysql + + for attempt in range(30): + try: + connection = pymysql.connect( + host="127.0.0.1", + port=13308, + user="openstatspec", + password="openstatspec", + database="openstatspec", + ) + except pymysql.MySQLError: + if attempt == 29: + raise + time.sleep(2) + else: + connection.close() + break + PY + - run: python -m pytest -m services + - name: Show Dolt logs on failure and stop server + if: always() + run: | + if docker inspect openstatspec-dolt >/dev/null 2>&1; then + if [ "${{ job.status }}" != "success" ]; then + docker logs openstatspec-dolt + fi + docker rm --force openstatspec-dolt + fi diff --git a/CHANGELOG.md b/CHANGELOG.md index fccbe16..e49d1de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,8 +14,9 @@ SPSS profile. - Export of supported dataset semantics to SAV and ZSAV. - Consistent legacy compatible-name rewriting across type-2, subtype-13, and VLS subtype-14 records, with fail-closed SAV/ZSAV validation. -- SQLite, PostgreSQL, MySQL, and MariaDB profiles, including service-backed - PostgreSQL 17/18, MySQL 8.4/9.7, and MariaDB 11.4/11.8/12.3 CI coverage. +- SQLite, PostgreSQL, MySQL, MariaDB, and independent Dolt profiles, including + service-backed PostgreSQL 17/18, MySQL 8.4/9.7, MariaDB 11.4/11.8/12.3, and + pinned Dolt 2.2.2 CI coverage. Dolt Transformation Workflow support is not claimed. - Preflight checks for target profile limits, atomic imports, validation, a command-line interface, and machine-readable capability and loss reports. diff --git a/README.md b/README.md index 357cf3c..a0ef76a 100644 --- a/README.md +++ b/README.md @@ -62,12 +62,17 @@ engine. Its import module remains `pyspssio`; the exact source commit is recorde in operation metadata. There is no fallback reader or writer. It supports unencrypted SAV and ZSAV import and SAV/ZSAV export for the semantics exposed by that engine. SQLite is the local reference path. -PostgreSQL, MySQL, and MariaDB are each covered by separate service-backed CI -conformance checks. Use these explicit SQLAlchemy URLs: +PostgreSQL, MySQL, MariaDB, and Dolt are each covered by separate service-backed CI +conformance checks. Dolt support is an independent core profile pinned to 2.2.2; +other Dolt versions and unknown MySQL-wire products fail closed. Use these explicit SQLAlchemy URLs: - SQLite: `sqlite:///dataset.sqlite` - PostgreSQL: `postgresql+psycopg://user:password@host/database` - MySQL/MariaDB: `mysql+pymysql://user:password@host/database` +- Dolt 2.2.2: `mysql+pymysql://user:password@host/database` (detected by server identity) + +The Dolt core profile supports strict wide-table import, validation, and export; +the separate Transformation Workflow is unsupported. Run `openstatspec capabilities` before an integration to inspect the machine-readable feature matrix. Export is deliberately strict: if known diff --git a/docs/sav-profile.md b/docs/sav-profile.md index ec81335..387083f 100755 --- a/docs/sav-profile.md +++ b/docs/sav-profile.md @@ -13,12 +13,14 @@ unencrypted .sav and .zsav files, and writes both formats through the same imple ## Verified SQL profiles -SQLite has a local reference fixture. PostgreSQL, MySQL, and MariaDB have +SQLite has a local reference fixture. PostgreSQL, MySQL, MariaDB, and Dolt have separate service-backed conformance checks in GitHub Actions that import, -validate, and export the supported fixture. MySQL 8.4 and MariaDB 11.4 exercise -the shared profile contract; this does not claim coverage for every server -configuration. An import that exceeds a target's strict single-table column -limit fails before it creates a dataset. +validate, and export the supported fixture. Dolt is an independent core profile +pinned to exact server version 2.2.2 and detected over `mysql+pymysql` by active +server identity; other Dolt versions and unknown MySQL-wire products fail closed. +The core Dolt profile does not claim support for the separate Transformation +Workflow. An import that exceeds a target's strict single-table column or row +envelope fails before it creates a dataset. ## Fidelity contract diff --git a/pyproject.toml b/pyproject.toml index df0b670..b2b746f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,7 @@ openstatspec = "openstatspec.cli:main" [tool.pytest.ini_options] testpaths = ["tests"] addopts = "-q" -markers = ["services: requires a configured PostgreSQL or MySQL/MariaDB service"] +markers = ["services: requires a configured PostgreSQL, MySQL, MariaDB, or Dolt service"] [tool.hatch.build.targets.wheel] packages = ["src/openstatspec"] diff --git a/src/openstatspec/spss/sav.py b/src/openstatspec/spss/sav.py index e7d395e..7e51cfa 100644 --- a/src/openstatspec/spss/sav.py +++ b/src/openstatspec/spss/sav.py @@ -38,6 +38,8 @@ variable_attribute_pairs, ) from ..core import UnsupportedOperationError +from ..sql.capabilities import effective_profile +from ..sql.profiles import preflight from ..sql.wide import ( create_wide_dataset, physical_name, @@ -262,7 +264,11 @@ def export_sav_dataset( destination_path = Path(destination) if destination_path.suffix.lower() not in {".sav", ".zsav"}: raise UnsupportedOperationError("Export destinations must use the .sav or .zsav extension.") - dataset, variables, rows = read_wide_dataset(database_url=database_url, dataset_id=dataset_id) + profile, _active = effective_profile(database_url) + dataset, variables, rows = read_wide_dataset( + database_url=database_url, dataset_id=dataset_id, profile=profile, + ) + preflight(profile, variables, rows=rows) validate_spss_catalog( variables, case_weight_variable=dataset.get("case_weight_variable"), diff --git a/src/openstatspec/sql/capabilities.py b/src/openstatspec/sql/capabilities.py index 29851ce..77e0508 100644 --- a/src/openstatspec/sql/capabilities.py +++ b/src/openstatspec/sql/capabilities.py @@ -11,8 +11,8 @@ from sqlalchemy.engine import make_url from .normative import catalog -from .profiles import MYSQL, POSTGRESQL, SQLITE, SqlProfile -from .profiles import profile_for_url +from .profiles import DOLT, MYSQL, POSTGRESQL, SQLITE, SqlProfile +from .profiles import profile_for_url, validate_connection_url from ..core import UnsupportedOperationError SPECIFICATION_COMMIT = "34141dda023d9e0217c37c232e39f436edfb0746" @@ -27,6 +27,10 @@ "claimed": ["MySQL 8.4.x", "MySQL 9.7.x"], "ci": ["MySQL 8.4.x", "MySQL 9.7.x"], }, + "dolt": { + "claimed": ["Dolt 2.2.2"], + "ci": ["Dolt 2.2.2"], + }, "mariadb": { "claimed": ["MariaDB 11.4.x", "MariaDB 11.8.x", "MariaDB 12.3.x"], "ci": ["MariaDB 11.4.x", "MariaDB 11.8.x", "MariaDB 12.3.x"], @@ -45,19 +49,43 @@ def profile_declarations(database_url: str | None = None) -> dict[str, dict[str, "sqlite": _profile("sqlite", SQLITE, active), "mysql": _profile("mysql", MYSQL, active), "mariadb": _profile("mariadb", MYSQL, active), + "dolt": _profile("dolt", DOLT, active), "postgresql": _profile("postgresql", POSTGRESQL, active), } +def _required_text_probe(connection: Any, statement: str, label: str) -> str: + """Return one required identity value without normalizing absence into text.""" + try: + value = connection.execute(text(statement)).scalar_one() + except Exception as error: + raise UnsupportedOperationError( + f"Active SQL server identity probe {label} failed." + ) from error + if value is None or value is False: + raise UnsupportedOperationError( + f"Active SQL server identity probe {label} returned no value." + ) + raw = str(value) + if not raw.strip(): + raise UnsupportedOperationError( + f"Active SQL server identity probe {label} returned no value." + ) + return raw + + def active_connection(database_url: str) -> dict[str, Any]: engine = create_engine(database_url) - configured = make_url(database_url) with engine.connect() as connection: dialect = connection.dialect.name + raw_comment: str | None = None if dialect == "sqlite": profile_name = "sqlite" - raw_version = str(connection.execute(text("select sqlite_version()")).scalar_one()) - identity_source = "select sqlite_version()" + raw_wire_version = _required_text_probe( + connection, "select sqlite_version()", "sqlite_version()", + ) + raw_product_version = raw_wire_version + identity_source = "SELECT sqlite_version()" compile_options = { name: int(value) for option in connection.exec_driver_sql("pragma compile_options").scalars() @@ -68,28 +96,73 @@ def active_connection(database_url: str) -> dict[str, Any]: observed = {"compile_options": compile_options} elif dialect == "postgresql": profile_name = "postgresql" - raw_version = str(connection.execute(text("show server_version")).scalar_one()) + raw_wire_version = _required_text_probe( + connection, "show server_version", "server_version", + ) + raw_product_version = raw_wire_version identity_source = "SHOW server_version" observed = {} elif dialect in {"mysql", "mariadb"}: - raw_version = str(connection.execute(text("select @@version")).scalar_one()) - comment = str(connection.execute(text("select @@version_comment")).scalar_one()) - profile_name = "mariadb" if "mariadb" in (raw_version + " " + comment).lower() else "mysql" - identity_source = "SELECT @@version, @@version_comment" - observed = { - "max_allowed_packet": int( - connection.execute(text("select @@max_allowed_packet")).scalar_one() + raw_wire_version = _required_text_probe( + connection, "select @@version", "@@version", + ) + raw_comment = _required_text_probe( + connection, "select @@version_comment", "@@version_comment", + ) + identity_text = f"{raw_wire_version} {raw_comment}".casefold() + if raw_comment.strip().casefold() == "dolt": + profile_name = "dolt" + raw_product_version = _required_text_probe( + connection, "select DOLT_VERSION()", "DOLT_VERSION()", ) - } + if raw_product_version.strip() != "2.2.2": + raise UnsupportedOperationError( + "The active Dolt product version must be exactly 2.2.2." + ) + identity_source = "SELECT @@version, @@version_comment, DOLT_VERSION()" + elif "mariadb" in identity_text: + profile_name = "mariadb" + raw_product_version = raw_wire_version + identity_source = "SELECT @@version, @@version_comment" + elif "mysql" in raw_comment.casefold(): + profile_name = "mysql" + raw_product_version = raw_wire_version + identity_source = "SELECT @@version, @@version_comment" + else: + raise UnsupportedOperationError( + "The active MySQL-wire server product is unknown or unsupported." + ) + packet_text = _required_text_probe( + connection, "select @@max_allowed_packet", "@@max_allowed_packet", + ) + try: + packet = int(packet_text) + except ValueError as error: + raise UnsupportedOperationError( + "Active SQL server returned an invalid @@max_allowed_packet." + ) from error + if packet <= 0: + raise UnsupportedOperationError( + "Active SQL server returned an invalid @@max_allowed_packet." + ) + observed = {"max_allowed_packet": packet} else: # pragma: no cover - validate_connection_url rejects this first raise ValueError(f"Unsupported active SQL dialect {dialect!r}.") return { + "dialect": dialect, "profile": profile_name, - "server_version": _normalized_version(raw_version), - "raw_server_version": raw_version, + "engine": profile_name, + "product": profile_name, + "transport": "mysql" if dialect in {"mysql", "mariadb"} else dialect, + "driver": engine.dialect.driver, + "server_version": _normalized_version(raw_product_version), + "raw_server_version": raw_product_version, + "raw_wire_version": raw_wire_version, + "raw_product_version": raw_product_version, + "raw_version_comment": raw_comment, "identity_source": identity_source, - "claimed_supported": server_version_supported(profile_name, raw_version), - "matched_claim": _matched_claim(profile_name, raw_version), + "claimed_supported": server_version_supported(profile_name, raw_product_version), + "matched_claim": _matched_claim(profile_name, raw_product_version), "catalog_binding": catalog_binding(database_url), "observed": observed, } @@ -97,21 +170,24 @@ def active_connection(database_url: str) -> dict[str, Any]: def effective_profile(database_url: str) -> tuple[SqlProfile, dict[str, Any]]: """Resolve and enforce the profile used by import preflight.""" - configured = profile_for_url(database_url) + configured = validate_connection_url(database_url) active = active_connection(database_url) if configured is not MYSQL and active["profile"] != configured.name: raise UnsupportedOperationError("The active SQL server does not match the configured profile.") - if configured is MYSQL and active["profile"] not in {"mysql", "mariadb"}: - raise UnsupportedOperationError("The active SQL server is not MySQL or MariaDB.") + if configured is MYSQL and active["profile"] not in {"mysql", "mariadb", "dolt"}: + raise UnsupportedOperationError("The active SQL server is not MySQL, MariaDB, or Dolt.") + if active["profile"] == "dolt" and make_url(database_url).drivername != "mysql+pymysql": + raise UnsupportedOperationError("Dolt requires an explicit mysql+pymysql URL.") if not active["claimed_supported"]: raise UnsupportedOperationError( f"Active {active['profile']} server version {active['server_version']} is not claimed supported." ) - declaration = _profile(active["profile"], configured, active) + selected = DOLT if active["profile"] == "dolt" else configured + declaration = _profile(active["profile"], selected, active) limits = declaration["effective_limits"] assert limits is not None return replace( - configured, + selected, name=active["profile"], max_physical_variables=int(limits["maximum_source_variables"]), max_text_value_bytes=int(limits["maximum_value_bytes"]), @@ -150,29 +226,61 @@ def _profile( "unit": "characters" if name in {"mysql", "mariadb"} else "bytes", "source": ( "MySQL/MariaDB native identifier limit" if name in {"mysql", "mariadb"} + else "Dolt 2.2.2 observed 64-byte ASCII identifier limit" if name == "dolt" else "PostgreSQL NAMEDATALEN minus one native byte limit" if name == "postgresql" else "OpenStatSpec profile boundary; SQLite has no fixed native identifier limit" ), "repertoire": "generated ASCII [a-z0-9_] identifiers", } - theoretical = { + declared = { "maximum_physical_columns": profile.max_physical_variables + 1, "maximum_source_variables": profile.max_physical_variables, "identifier_limit": identifier, "maximum_value_bytes": profile.max_text_value_bytes, "maximum_row_bytes": profile.max_row_bytes, } + theoretical = ( + {"maximum_value_bytes": 4_294_967_295} + if name == "dolt" else declared + ) + proposed = ( + { + "maximum_physical_columns": declared["maximum_physical_columns"], + "maximum_source_variables": declared["maximum_source_variables"], + "maximum_value_bytes": declared["maximum_value_bytes"], + "maximum_row_bytes": declared["maximum_row_bytes"], + } + if name == "dolt" else None + ) + observed_limits = ( + { + "minimum_observed_physical_columns": 307, + "identifier_limit": identifier, + "rejected_identifier_bytes": 65, + } + if name == "dolt" else None + ) effective = None status = "not_connected" if active and active["profile"] == name: - effective = dict(theoretical) - sources = { - "maximum_source_variables": "profile theoretical engine ceiling", - "maximum_physical_columns": "profile theoretical engine ceiling", - "identifier_limit": identifier["source"], - "maximum_value_bytes": "profile theoretical engine ceiling", - "maximum_row_bytes": "profile theoretical engine ceiling", - } + effective = dict(declared) + sources = ( + { + "maximum_source_variables": "proposed Dolt adapter envelope", + "maximum_physical_columns": "proposed Dolt adapter envelope", + "identifier_limit": "observed on exact Dolt 2.2.2", + "maximum_value_bytes": "observed Dolt adapter value envelope", + "maximum_row_bytes": "proposed Dolt adapter envelope", + } + if name == "dolt" + else { + "maximum_source_variables": "profile theoretical engine ceiling", + "maximum_physical_columns": "profile theoretical engine ceiling", + "identifier_limit": identifier["source"], + "maximum_value_bytes": "profile theoretical engine ceiling", + "maximum_row_bytes": "profile theoretical engine ceiling", + } + ) observed = active["observed"] if name == "sqlite": options = observed["compile_options"] @@ -195,11 +303,15 @@ def _profile( sources["maximum_value_bytes"] = "active PRAGMA compile_options MAX_LENGTH" sources["maximum_row_bytes"] = "active PRAGMA compile_options MAX_LENGTH" status = "active_connection_mixed" - elif name in {"mysql", "mariadb"}: + elif name in {"mysql", "mariadb", "dolt"}: packet = int(observed["max_allowed_packet"]) payload = max(0, (packet - 131_072) // 2) effective["maximum_value_bytes"] = min( - theoretical["maximum_value_bytes"], payload, + ( + declared["maximum_value_bytes"] + if name == "dolt" else theoretical["maximum_value_bytes"] + ), + payload, ) effective["maximum_statement_bytes"] = payload sources["maximum_value_bytes"] = "active @@max_allowed_packet worst-case payload" @@ -210,22 +322,76 @@ def _profile( effective["sources"] = sources policy = SERVER_POLICIES[name] return { - "driver": "psycopg" if name == "postgresql" else "PyMySQL" if name in {"mysql", "mariadb"} else "sqlite3", + "profile": name, + "engine": name, + "dialect": "mysql" if name == "dolt" else name, + "transport": "mysql" if name == "dolt" else name, + "specification_commit": SPECIFICATION_COMMIT, + "specification_status": "release_candidate", + "specification_release": SPECIFICATION_RELEASE, + "driver": "psycopg" if name == "postgresql" else "PyMySQL" if name in {"mysql", "mariadb", "dolt"} else "sqlite3", "claimed_server_versions": policy["claimed"], "ci_tested_server_versions": policy["ci"], "theoretical_limits": theoretical, + "proposed_adapter_limits": proposed, + "observed_limits": observed_limits, "effective_limits": effective, "effective_limits_status": status, - "numeric_type": "DOUBLE PRECISION" if name == "postgresql" else "DOUBLE" if name in {"mysql", "mariadb"} else "REAL", - "text_type": "LONGTEXT" if name in {"mysql", "mariadb"} else "TEXT", - "ddl_atomic": name not in {"mysql", "mariadb"}, - "failure_cleanup": "compensating_cleanup" if name in {"mysql", "mariadb"} else "transaction_rollback", + "numeric_type": "DOUBLE PRECISION" if name == "postgresql" else "DOUBLE" if name in {"mysql", "mariadb", "dolt"} else "REAL", + "numeric_value_policy": { + "finite_binary64": "supported", + "nan": "rejected_before_ddl", + "positive_infinity": "rejected_before_ddl", + "negative_infinity": "rejected_before_ddl", + }, + "text_type": "LONGTEXT" if name in {"mysql", "mariadb", "dolt"} else "TEXT", + "ddl_atomic": name not in {"mysql", "mariadb", "dolt"}, + "failure_cleanup": "compensating_cleanup" if name in {"mysql", "mariadb", "dolt"} else "transaction_rollback", + "limit_bases": { + "maximum_physical_columns": ( + "proposed_adapter_envelope" if name == "dolt" else "theoretical_engine_limit" + ), + "maximum_source_variables": ( + "proposed_adapter_envelope" if name == "dolt" else "theoretical_engine_limit" + ), + "identifier_limit": "observed_exact_version" if name == "dolt" else "theoretical_engine_limit", + "maximum_value_bytes": ( + "observed_exact_version" if name == "dolt" else "theoretical_engine_limit" + ), + "maximum_row_bytes": ( + "proposed_adapter_envelope" if name == "dolt" else "theoretical_engine_limit" + ), + "maximum_statement_bytes": "active_connection_observation", + }, + "storage_evidence": ( + { + "binary64": { + "type": "DOUBLE", + "classification": "observed_exact_version", + "source": "Dolt 2.2.2 interoperability verification", + "version": "2.2.2", + "maximum_finite_round_trip_exact": True, + }, + "text": { + "type": "LONGTEXT NOT NULL", + "classification": "observed_exact_version", + "source": "Dolt 2.2.2 interoperability verification", + "version": "2.2.2", + "observed_value_bytes": 65_504, + "unit": "bytes", + }, + } + if name == "dolt" else None + ), + "transformation_workflow": "unsupported" if name == "dolt" else None, "physical_table_mapping": "dataset.physical_table_schema + dataset.physical_table_name", "identifier_policy": "deterministic ASCII mapping; source name remains authoritative", } def server_version_supported(profile: str, raw_version: str) -> bool: + if profile == "dolt": + return raw_version.strip() == "2.2.2" version = _version_tuple(raw_version) if profile == "sqlite": return (3, 24) <= version[:2] < (4, 0) diff --git a/src/openstatspec/sql/normative.py b/src/openstatspec/sql/normative.py index 57ccae0..5af3321 100644 --- a/src/openstatspec/sql/normative.py +++ b/src/openstatspec/sql/normative.py @@ -14,7 +14,7 @@ from sqlalchemy import ( BigInteger, CheckConstraint, Column, DateTime, Float, ForeignKey, Integer, - MetaData, String, Table, Text, UniqueConstraint, insert, inspect, select, update, + MetaData, String, Table, Text, UniqueConstraint, delete, insert, inspect, select, update, ) from sqlalchemy.dialects import mysql, postgresql, sqlite @@ -259,11 +259,10 @@ def create(connection: Any, tables: NormativeTables) -> None: expected = {table.name for table in tables.all()} existing = set(inspect(connection).get_table_names(schema=tables.dataset.schema)) if tables.catalog_identity.name not in existing: - conflicts = sorted(existing & expected) - if conflicts: + if existing: raise RuntimeError( - "The selected catalog namespace contains unowned OpenStatSpec relation names: " - + ", ".join(conflicts) + "The selected catalog namespace is occupied by an unowned OpenStatSpec relation " + "and has no catalog identity: " + ", ".join(sorted(existing)) ) tables.dataset.metadata.create_all(connection, tables=list(tables.all())) connection.execute(insert(tables.catalog_identity).values( @@ -362,8 +361,9 @@ def store_imported_dataset( missing_rules: Iterable[Mapping[str, Any]], attributes: Iterable[Mapping[str, Any]], multiple_response_sets: Iterable[Mapping[str, Any]], source_extensions: Mapping[str, Any], case_weight_variable: str | None = None, + dataset_id: str | None = None, ) -> str: - dataset_id = str(uuid4()) + dataset_id = dataset_id or str(uuid4()) connection.execute(insert(tables.dataset).values( dataset_id=dataset_id, spec_version=SPEC_VERSION, source_format=source_format, physical_table_schema=None, physical_table_name=physical_table_name, @@ -507,6 +507,70 @@ def store_imported_dataset( return dataset_id +def delete_dataset_representation( + connection: Any, tables: NormativeTables, dataset_id: str, +) -> None: + """Delete one partially written dataset while retaining operation audit rows.""" + variable_ids = select(tables.variable.c.variable_id).where( + tables.variable.c.dataset_id == dataset_id + ) + label_set_ids = select(tables.value_label_set.c.value_label_set_id).where( + tables.value_label_set.c.dataset_id == dataset_id + ) + variable_set_ids = select(tables.variable_set.c.variable_set_id).where( + tables.variable_set.c.dataset_id == dataset_id + ) + response_set_ids = select( + tables.multiple_response_set.c.multiple_response_set_id + ).where(tables.multiple_response_set.c.dataset_id == dataset_id) + + connection.execute(delete(tables.fidelity_event).where( + tables.fidelity_event.c.dataset_id == dataset_id + )) + connection.execute(delete(tables.dataset_weight_variable).where( + tables.dataset_weight_variable.c.dataset_id == dataset_id + )) + connection.execute(delete(tables.variable_value_label_set).where( + tables.variable_value_label_set.c.variable_id.in_(variable_ids) + )) + connection.execute(delete(tables.missing_rule).where( + tables.missing_rule.c.variable_id.in_(variable_ids) + )) + connection.execute(delete(tables.variable_attribute).where( + tables.variable_attribute.c.variable_id.in_(variable_ids) + )) + connection.execute(delete(tables.variable_set_member).where( + tables.variable_set_member.c.variable_set_id.in_(variable_set_ids) + )) + connection.execute(delete(tables.multiple_response_member).where( + tables.multiple_response_member.c.multiple_response_set_id.in_(response_set_ids) + )) + connection.execute(delete(tables.value_label).where( + tables.value_label.c.value_label_set_id.in_(label_set_ids) + )) + connection.execute(delete(tables.value_label_set).where( + tables.value_label_set.c.dataset_id == dataset_id + )) + connection.execute(delete(tables.dataset_attribute).where( + tables.dataset_attribute.c.dataset_id == dataset_id + )) + connection.execute(delete(tables.document).where( + tables.document.c.dataset_id == dataset_id + )) + connection.execute(delete(tables.variable_set).where( + tables.variable_set.c.dataset_id == dataset_id + )) + connection.execute(delete(tables.multiple_response_set).where( + tables.multiple_response_set.c.dataset_id == dataset_id + )) + connection.execute(delete(tables.variable).where( + tables.variable.c.dataset_id == dataset_id + )) + connection.execute(delete(tables.dataset).where( + tables.dataset.c.dataset_id == dataset_id + )) + + def dataset_id_for_name(connection: Any, tables: NormativeTables, dataset_name: str) -> str: return str(connection.execute( select(tables.dataset.c.dataset_id).where(tables.dataset.c.dataset_name == dataset_name) diff --git a/src/openstatspec/sql/profiles.py b/src/openstatspec/sql/profiles.py index 2668231..84142c4 100755 --- a/src/openstatspec/sql/profiles.py +++ b/src/openstatspec/sql/profiles.py @@ -4,6 +4,8 @@ Importers use this information for preflight checks before creating a dataset. """ +import math + from collections.abc import Iterable, Mapping from dataclasses import dataclass from typing import Any @@ -53,7 +55,13 @@ def as_dict(self) -> dict[str, object]: "mysql", ("mysql", "mariadb"), 1_016, 64, True, True, 4_294_967_295, 65_535, True, ("PyMySQL",), ) -PROFILES = (SQLITE, POSTGRESQL, MYSQL) +DOLT = SqlProfile( + "dolt", (), 305, 64, True, True, + 65_504, 65_504, True, ("PyMySQL",), +) +# Dolt deliberately has no URL scheme. It is selected only after the server +# reached through the MySQL wire family positively identifies itself as Dolt. +PROFILES = (SQLITE, POSTGRESQL, MYSQL, DOLT) def profile_for_url(database_url: str) -> SqlProfile: @@ -170,6 +178,15 @@ def preflight( row_bytes = 8 for variable in variables: if variable.get("storage_kind") != "string": + value = row.get(str(variable["physical_name"])) + if value is not None and not math.isfinite(float(value)): + raise _exceeded( + "nonfinite_numeric_value", + f"row {row_ordinal} value for {variable['source_name']!r} " + "is not a finite binary64 value.", + row_ordinal=row_ordinal, + source_name=variable["source_name"], + ) row_bytes += 8 continue value = row.get(str(variable["physical_name"]), "") @@ -183,7 +200,7 @@ def preflight( row_ordinal=row_ordinal, source_name=variable["source_name"], encoded_bytes=encoded_bytes, maximum=profile.max_text_value_bytes, ) - row_bytes += 20 if profile.name in {"mysql", "mariadb"} else encoded_bytes + row_bytes += 20 if profile.name in {"mysql", "mariadb", "dolt"} else encoded_bytes if row_bytes > profile.max_row_bytes: raise _exceeded( "row_size_limit", @@ -197,7 +214,7 @@ def preflight( def _row_storage_bytes(profile: SqlProfile, variable: Mapping[str, Any]) -> int: if variable.get("storage_kind") == "numeric": return 8 - if profile.name in {"mysql", "mariadb"}: + if profile.name in {"mysql", "mariadb", "dolt"}: return 20 return int(variable.get("string_width") or 0) diff --git a/src/openstatspec/sql/wide.py b/src/openstatspec/sql/wide.py index e7fdbdc..a9e3bad 100644 --- a/src/openstatspec/sql/wide.py +++ b/src/openstatspec/sql/wide.py @@ -17,6 +17,7 @@ from .normative import ( catalog as normative_catalog, create as create_normative_catalog, + delete_dataset_representation as delete_normative_dataset, dataset_id_for_name as normative_dataset_id_for_name, finish_operation as finish_normative_operation, record_fidelity_events as record_normative_fidelity_events, @@ -40,6 +41,11 @@ def _catalog_error(code: str, detail: str, **details: Any) -> CatalogPreflightEr return CatalogPreflightError(code, detail, details=details) +def string_type(profile: Any) -> Text: + """Use Dolt's tested LONGTEXT storage without changing MySQL/MariaDB DDL.""" + return mysql.LONGTEXT() if profile.name == "dolt" else Text() + + def binary64_type() -> Float: """Return the required IEEE-754 binary64 SQL type for every profile. @@ -202,11 +208,11 @@ def _record_failed_preflight( ) -> None: """Persist a failed preflight without creating any source dataset state.""" with engine.begin() as connection: + create_normative_catalog(connection, normative) metadata.create_all(connection, tables=[ datasets, variable_catalog, multiple_response_catalog, fidelity_event_catalog, operation_catalog, ]) - create_normative_catalog(connection, normative) failed_at = datetime.now(UTC).replace(tzinfo=None) record_normative_operation( connection, normative, operation_id=operation_id, @@ -787,109 +793,246 @@ def create_wide_dataset( data_table = Table( data_table_name(dataset_id), metadata, Column("__case_ordinal", BigInteger, primary_key=True, nullable=False), - *(Column(item["physical_name"], binary64_type() if item["storage_kind"] == "numeric" else Text(), + *(Column(item["physical_name"], binary64_type() if item["storage_kind"] == "numeric" else string_type(profile), nullable=item["storage_kind"] == "numeric") for item in variables), ) - with engine.begin() as connection: - metadata.create_all(connection, tables=[datasets, variable_catalog, multiple_response_catalog, source_extensions_catalog, documents_catalog, value_labels_catalog, missing_rules_catalog, attributes_catalog, fidelity_event_catalog, operation_catalog]) - create_normative_catalog(connection, normative) - _migrate_catalog_columns(connection, datasets, variable_catalog, multiple_response_catalog) - record_normative_operation( - connection, normative, operation_id=operation_id, - operation_kind="import", status="started", source_format=source_format, - ) - connection.execute(insert(operation_catalog).values( - operation_id=operation_id, direction="import", status="running", dataset_id=dataset_id, - source=source_name, created_at=_now(), details=json.dumps({"variable_count": len(variables), **dict(operation_details or {})}, sort_keys=True), - )) - if connection.execute(select(datasets.c.dataset_id).where(datasets.c.dataset_id == dataset_id)).first(): - raise ValueError(f"Dataset {dataset_id!r} already exists; imports never overwrite a dataset.") - if connection.execute(select(datasets.c.dataset_id).where(datasets.c.data_table == data_table.name)).first(): - raise ValueError(f"Dataset ID {dataset_id!r} collides with an existing physical data-table name; import was not started.") - data_table.create(connection) - materialized = [ - {"__case_ordinal": ordinal, **row} - for ordinal, row in enumerate(source_rows, start=1) - ] - connection.execute(insert(datasets).values( - dataset_id=dataset_id, data_table=data_table.name, source_format=source_format, - source_name=source_name, source_encoding=source_encoding, case_count=len(materialized), - source_table_name=source_table_name, - source_sha256=source_sha256, - source_created_at=source_created_at, source_modified_at=source_modified_at, - imported_at=imported_at, - file_label=file_label, documents=documents, - file_attributes=file_attributes, case_weight_variable=case_weight_variable, - multiple_response_sets=multiple_response_sets, - )) - connection.execute(insert(variable_catalog), [dict(dataset_id=dataset_id, **item) for item in variables]) - docs_rows = document_rows(dataset_id, documents) - if docs_rows: - connection.execute(insert(documents_catalog), docs_rows) - labels_rows = value_label_rows(dataset_id, variables) - if labels_rows: - connection.execute(insert(value_labels_catalog), labels_rows) - missing_rows = missing_rule_rows(dataset_id, variables) - if missing_rows: - connection.execute(insert(missing_rules_catalog), missing_rows) - attributes_rows = attribute_rows( - dataset_id, variables, - file_attributes=file_attribute_values, - variable_attributes=variable_attribute_values, - ) - if attributes_rows: - connection.execute(insert(attributes_catalog), attributes_rows) - mrset_rows = multiple_response_set_rows(dataset_id, multiple_response_sets) - if mrset_rows: - connection.execute(insert(multiple_response_catalog), mrset_rows) - extension_rows = source_extension_rows(dataset_id, source_extensions or {}) - if extension_rows: - connection.execute(insert(source_extensions_catalog), extension_rows) - event_rows = _event_rows( - operation_id=operation_id, dataset_id=dataset_id, direction="import", fidelity_events=fidelity_events, - ) - if event_rows: - connection.execute(insert(fidelity_event_catalog), event_rows) - normative_dataset_id = store_normative_dataset( - connection, normative, dataset_name=dataset_id, - source_format=source_format, physical_table_name=data_table.name, - dataset_label=file_label, source_encoding=source_encoding, - source_hash=source_sha256, source_case_count=len(materialized), - imported_at=imported_at or None, variables=variables, - documents=docs_rows, value_labels=labels_rows, - missing_rules=missing_rows, attributes=attributes_rows, - multiple_response_sets=mrset_rows, - source_extensions=source_extensions or {}, - case_weight_variable=case_weight_variable, - ) - record_normative_fidelity_events( - connection, normative, operation_id=operation_id, - dataset_id=normative_dataset_id, direction="import", - events=fidelity_events, - ) - if materialized: - try: + materialized = [ + {"__case_ordinal": ordinal, **row} + for ordinal, row in enumerate(source_rows, start=1) + ] + normative_dataset_id = str(uuid4()) + namespace_owned = False + data_table_was_absent = False + try: + with engine.begin() as connection: + create_normative_catalog(connection, normative) + namespace_owned = True + metadata.create_all(connection, tables=[ + datasets, variable_catalog, multiple_response_catalog, + source_extensions_catalog, documents_catalog, value_labels_catalog, + missing_rules_catalog, attributes_catalog, fidelity_event_catalog, + operation_catalog, + ]) + _migrate_catalog_columns( + connection, datasets, variable_catalog, multiple_response_catalog, + ) + if connection.execute( + select(datasets.c.dataset_id).where( + datasets.c.dataset_id == dataset_id + ) + ).first(): + raise ValueError( + f"Dataset {dataset_id!r} already exists; imports never overwrite a dataset." + ) + if connection.execute( + select(datasets.c.dataset_id).where( + datasets.c.data_table == data_table.name + ) + ).first(): + raise ValueError( + f"Dataset ID {dataset_id!r} collides with an existing physical " + "data-table name; import was not started." + ) + if inspect(connection).has_table(data_table.name): + raise ValueError( + f"Physical data-table name {data_table.name!r} is already occupied." + ) + data_table_was_absent = True + record_normative_operation( + connection, normative, operation_id=operation_id, + operation_kind="import", status="started", + source_format=source_format, + ) + connection.execute(insert(operation_catalog).values( + operation_id=operation_id, direction="import", status="running", + dataset_id=dataset_id, source=source_name, created_at=_now(), + details=json.dumps({ + "variable_count": len(variables), + **dict(operation_details or {}), + }, sort_keys=True), + )) + data_table.create(connection) + connection.execute(insert(datasets).values( + dataset_id=dataset_id, data_table=data_table.name, + source_format=source_format, source_name=source_name, + source_encoding=source_encoding, case_count=len(materialized), + source_table_name=source_table_name, source_sha256=source_sha256, + source_created_at=source_created_at, + source_modified_at=source_modified_at, imported_at=imported_at, + file_label=file_label, documents=documents, + file_attributes=file_attributes, + case_weight_variable=case_weight_variable, + multiple_response_sets=multiple_response_sets, + )) + connection.execute( + insert(variable_catalog), + [dict(dataset_id=dataset_id, **item) for item in variables], + ) + docs_rows = document_rows(dataset_id, documents) + if docs_rows: + connection.execute(insert(documents_catalog), docs_rows) + labels_rows = value_label_rows(dataset_id, variables) + if labels_rows: + connection.execute(insert(value_labels_catalog), labels_rows) + missing_rows = missing_rule_rows(dataset_id, variables) + if missing_rows: + connection.execute(insert(missing_rules_catalog), missing_rows) + attributes_rows = attribute_rows( + dataset_id, variables, + file_attributes=file_attribute_values, + variable_attributes=variable_attribute_values, + ) + if attributes_rows: + connection.execute(insert(attributes_catalog), attributes_rows) + mrset_rows = multiple_response_set_rows( + dataset_id, multiple_response_sets, + ) + if mrset_rows: + connection.execute(insert(multiple_response_catalog), mrset_rows) + extension_rows = source_extension_rows( + dataset_id, source_extensions or {}, + ) + if extension_rows: + connection.execute(insert(source_extensions_catalog), extension_rows) + event_rows = _event_rows( + operation_id=operation_id, dataset_id=dataset_id, + direction="import", fidelity_events=fidelity_events, + ) + if event_rows: + connection.execute(insert(fidelity_event_catalog), event_rows) + store_normative_dataset( + connection, normative, dataset_name=dataset_id, + source_format=source_format, physical_table_name=data_table.name, + dataset_label=file_label, source_encoding=source_encoding, + source_hash=source_sha256, source_case_count=len(materialized), + imported_at=imported_at or None, variables=variables, + documents=docs_rows, value_labels=labels_rows, + missing_rules=missing_rows, attributes=attributes_rows, + multiple_response_sets=mrset_rows, + source_extensions=source_extensions or {}, + case_weight_variable=case_weight_variable, + dataset_id=normative_dataset_id, + ) + record_normative_fidelity_events( + connection, normative, operation_id=operation_id, + dataset_id=normative_dataset_id, direction="import", + events=fidelity_events, + ) + if materialized: connection.execute(insert(data_table), materialized) - except Exception: - data_table.drop(connection, checkfirst=True) - connection.execute(delete(multiple_response_catalog).where(multiple_response_catalog.c.dataset_id == dataset_id)) - connection.execute(delete(source_extensions_catalog).where(source_extensions_catalog.c.dataset_id == dataset_id)) - connection.execute(delete(documents_catalog).where(documents_catalog.c.dataset_id == dataset_id)) - connection.execute(delete(value_labels_catalog).where(value_labels_catalog.c.dataset_id == dataset_id)) - connection.execute(delete(missing_rules_catalog).where(missing_rules_catalog.c.dataset_id == dataset_id)) - connection.execute(delete(attributes_catalog).where(attributes_catalog.c.dataset_id == dataset_id)) - connection.execute(delete(fidelity_event_catalog).where(fidelity_event_catalog.c.dataset_id == dataset_id)) - connection.execute(delete(variable_catalog).where(variable_catalog.c.dataset_id == dataset_id)) - connection.execute(delete(datasets).where(datasets.c.dataset_id == dataset_id)) - connection.commit() - raise - connection.execute(update(operation_catalog).where(operation_catalog.c.operation_id == operation_id).values( - status="succeeded", completed_at=_now(), - )) - finish_normative_operation( - connection, normative, operation_id=operation_id, status="succeeded", - ) - return {"dataset_id": dataset_id, "data_table": data_table.name, "case_count": len(materialized), "operation_id": operation_id} + connection.execute(update(operation_catalog).where( + operation_catalog.c.operation_id == operation_id + ).values(status="succeeded", completed_at=_now())) + finish_normative_operation( + connection, normative, operation_id=operation_id, + status="succeeded", + ) + except Exception as error: + if ( + profile.name in {"mysql", "mariadb", "dolt"} + and namespace_owned + and data_table_was_absent + ): + try: + with engine.begin() as cleanup: + delete_normative_dataset( + cleanup, normative, normative_dataset_id, + ) + cleanup_inspector = inspect(cleanup) + for table in ( + multiple_response_catalog, source_extensions_catalog, + documents_catalog, value_labels_catalog, + missing_rules_catalog, attributes_catalog, + fidelity_event_catalog, variable_catalog, + ): + if cleanup_inspector.has_table(table.name): + cleanup.execute( + delete(table).where(table.c.dataset_id == dataset_id) + ) + if cleanup_inspector.has_table(datasets.name): + cleanup.execute( + delete(datasets).where(datasets.c.dataset_id == dataset_id) + ) + data_table.drop(cleanup, checkfirst=True) + metadata.create_all(cleanup, tables=[ + datasets, variable_catalog, multiple_response_catalog, + source_extensions_catalog, documents_catalog, + value_labels_catalog, missing_rules_catalog, + attributes_catalog, fidelity_event_catalog, + operation_catalog, + ]) + failed_event = ({ + "code": "import_failed", + "detail": str(error), + "severity": "error", + "source_item": source_name, + "details": { + "profile": profile.name, + "phase": "post_ddl", + "cleanup": "complete", + "error_type": type(error).__name__, + }, + },) + normative_operation_exists = cleanup.execute(select( + normative.operation.c.operation_id + ).where( + normative.operation.c.operation_id == operation_id + )).first() + if normative_operation_exists: + finish_normative_operation( + cleanup, normative, operation_id=operation_id, + status="failed", + ) + else: + failed_at = datetime.now(UTC).replace(tzinfo=None) + record_normative_operation( + cleanup, normative, operation_id=operation_id, + operation_kind="import", status="failed", + source_format=source_format, started_at=failed_at, + completed_at=failed_at, + ) + mirror_operation_exists = cleanup.execute(select( + operation_catalog.c.operation_id + ).where( + operation_catalog.c.operation_id == operation_id + )).first() + if mirror_operation_exists: + cleanup.execute(update(operation_catalog).where( + operation_catalog.c.operation_id == operation_id + ).values( + status="failed", dataset_id=None, completed_at=_now(), + )) + else: + cleanup.execute(insert(operation_catalog).values( + operation_id=operation_id, direction="import", + status="failed", dataset_id=None, source=source_name, + created_at=_now(), completed_at=_now(), + details=json.dumps({ + "reason": "post_ddl", + "variable_count": len(variables), + }, sort_keys=True), + )) + cleanup.execute( + insert(fidelity_event_catalog), + _event_rows( + operation_id=operation_id, dataset_id=None, + direction="import", fidelity_events=failed_event, + ), + ) + record_normative_fidelity_events( + cleanup, normative, operation_id=operation_id, + dataset_id=None, direction="import", + events=failed_event, + ) + except Exception as cleanup_error: + raise RuntimeError( + f"OpenStatSpec compensating cleanup failed: {cleanup_error}" + ) from cleanup_error + raise + return { + "dataset_id": dataset_id, "data_table": data_table.name, + "case_count": len(materialized), "operation_id": operation_id, + } def _endpoint_from_row(row: Mapping[str, Any], *, prefix: str) -> Any: @@ -901,8 +1044,12 @@ def _endpoint_from_row(row: Mapping[str, Any], *, prefix: str) -> Any: return row[f"{prefix}_numeric"] if endpoint_type == "numeric" else row[f"{prefix}_text"] -def read_wide_dataset(*, database_url: str, dataset_id: str) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]: - """Read a strict dataset, preferring normalized metadata with JSON compatibility fallback.""" +def read_wide_dataset( + *, database_url: str, dataset_id: str, profile: Any | None = None, +) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]: + """Read a strict dataset only after resolving the active server profile.""" + if profile is None: + profile, _active = effective_profile(database_url) engine = create_engine(database_url) metadata = MetaData() datasets, variable_catalog, _, _ = catalog(metadata) @@ -1051,9 +1198,11 @@ def record_export_operation( return operation_id def validate_wide_dataset(*, database_url: str, dataset_id: str) -> dict[str, Any]: - dataset, variables, rows = read_wide_dataset(database_url=database_url, dataset_id=dataset_id) - profile = validate_connection_url(database_url) - preflight(profile, variables) + profile, _active = effective_profile(database_url) + dataset, variables, rows = read_wide_dataset( + database_url=database_url, dataset_id=dataset_id, profile=profile, + ) + preflight(profile, variables, rows=rows) validate_spss_catalog( variables, case_weight_variable=dataset.get("case_weight_variable"), @@ -1079,6 +1228,11 @@ def validate_wide_dataset(*, database_url: str, dataset_id: str) -> dict[str, An if item["storage_kind"] == "numeric": if not isinstance(column.type, Float) or not column.nullable: raise ValueError(f"Numeric variable {item['source_name']!r} must be a nullable binary64 column.") + elif profile.name == "dolt": + if not isinstance(column.type, mysql.LONGTEXT) or column.nullable: + raise ValueError( + f"String variable {item['source_name']!r} must be a non-null LONGTEXT column." + ) elif not isinstance(column.type, Text) or column.nullable: raise ValueError(f"String variable {item['source_name']!r} must be a non-null text column.") if [row["__case_ordinal"] for row in rows] != list(range(1, len(rows) + 1)): diff --git a/tests/test_atomic_import.py b/tests/test_atomic_import.py index e468f8c..88a98a3 100755 --- a/tests/test_atomic_import.py +++ b/tests/test_atomic_import.py @@ -5,7 +5,7 @@ from sqlalchemy.exc import IntegrityError import openstatspec.sql.wide as wide -from openstatspec.sql.profiles import SQLITE +from openstatspec.sql.profiles import DOLT, MYSQL, SQLITE from openstatspec.sql.wide import create_wide_dataset @@ -28,10 +28,9 @@ def test_failed_row_insert_leaves_no_catalog_or_data_table(tmp_path) -> None: connection = sqlite3.connect(database_path) tables = [row[0] for row in connection.execute("select name from sqlite_master where type = 'table'")] assert "data_broken" not in tables - assert "dataset_catalog" in tables - assert connection.execute("select count(*) from dataset_catalog").fetchone() == (0,) - assert "variable_catalog" in tables - assert connection.execute("select count(*) from variable_catalog").fetchone() == (0,) + assert "catalog_identity" in tables + assert connection.execute("select count(*) from dataset").fetchone() == (0,) + assert connection.execute("select count(*) from variable").fetchone() == (0,) def test_failed_preflight_persists_operation_without_creating_dataset(tmp_path) -> None: @@ -113,3 +112,247 @@ def test_declared_string_width_preflight_is_atomic_and_diagnostic(tmp_path, monk details = connection.execute("select details from fidelity_event_catalog").fetchone()[0] assert '"reason": "declared_string_width_limit"' in details assert '"string_width": 4' in details + +def test_nonatomic_failure_after_normative_write_cleans_both_catalogs_and_data( + tmp_path, monkeypatch, +) -> None: + database_path = tmp_path / "nonatomic-cleanup.sqlite" + database = f"sqlite:///{database_path}" + variables = [{ + "ordinal": 1, "source_name": "name", "physical_name": "name", + "storage_kind": "string", "string_width": 8, "label": "", + "format": "A8", "measure": "nominal", "alignment": "left", + "display_width": 8, "value_labels": "{}", "missing_ranges": "[]", + }] + monkeypatch.setattr( + wide, "effective_profile", + lambda _url: (replace(MYSQL, name="mysql"), {}), + ) + real_store = wide.store_normative_dataset + + def fail_after_normative_write(*args, **kwargs): + real_store(*args, **kwargs) + raise RuntimeError("fault after normative write") + + monkeypatch.setattr(wide, "store_normative_dataset", fail_after_normative_write) + + with pytest.raises(RuntimeError, match="fault after normative write"): + create_wide_dataset( + database_url=database, dataset_id="cleanup", source_name="fixture.sav", + source_format="SAV", rows=[{"name": "ok"}], variables=variables, + ) + + connection = sqlite3.connect(database_path) + assert "data_cleanup" not in { + row[0] for row in connection.execute( + "select name from sqlite_master where type = 'table'" + ) + } + existing_tables = { + row[0] for row in connection.execute( + "select name from sqlite_master where type = 'table'" + ) + } + if "dataset_catalog" in existing_tables: + assert connection.execute( + "select count(*) from dataset_catalog where dataset_id = 'cleanup'" + ).fetchone() == (0,) + if "variable_catalog" in existing_tables: + assert connection.execute( + "select count(*) from variable_catalog where dataset_id = 'cleanup'" + ).fetchone() == (0,) + assert connection.execute( + "select count(*) from dataset where dataset_name = 'cleanup'" + ).fetchone() == (0,) + assert connection.execute( + "select count(*) from variable" + ).fetchone() == (0,) + assert connection.execute( + "select direction, status, dataset_id from operation_catalog" + ).fetchall() == [("import", "failed", None)] + assert connection.execute( + "select direction, severity, code, dataset_id from fidelity_event_catalog" + ).fetchall() == [("import", "error", "import_failed", None)] + assert connection.execute( + "select operation_kind, status from operation" + ).fetchall() == [("import", "failed")] + assert connection.execute( + "select direction, severity, event_code, dataset_id from fidelity_event" + ).fetchall() == [("import", "error", "import_failed", None)] + +def test_occupied_foreign_namespace_fails_without_modification(tmp_path) -> None: + database_path = tmp_path / "foreign.sqlite" + database = f"sqlite:///{database_path}" + connection = sqlite3.connect(database_path) + connection.execute("create table foreign_data (value text not null)") + connection.execute("insert into foreign_data values ('keep')") + connection.commit() + before_schema = connection.execute( + "select name, sql from sqlite_master where type = 'table' order by name" + ).fetchall() + + variables = [{ + "ordinal": 1, "source_name": "name", "physical_name": "name", + "storage_kind": "string", "string_width": 8, "label": "", + "format": "A8", "measure": "nominal", "alignment": "left", + "display_width": 8, "value_labels": "{}", "missing_ranges": "[]", + }] + with pytest.raises(RuntimeError, match="occupied"): + create_wide_dataset( + database_url=database, dataset_id="foreign", source_name="fixture.sav", + source_format="SAV", rows=[{"name": "ok"}], variables=variables, + ) + + assert connection.execute( + "select name, sql from sqlite_master where type = 'table' order by name" + ).fetchall() == before_schema + assert connection.execute("select value from foreign_data").fetchall() == [("keep",)] + + +@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")]) +def test_dolt_nonfinite_preflight_creates_no_dataset_or_physical_table( + tmp_path, monkeypatch, value, +) -> None: + database_path = tmp_path / "dolt-nonfinite.sqlite" + database = f"sqlite:///{database_path}" + monkeypatch.setattr(wide, "effective_profile", lambda _url: (DOLT, {})) + variables = [{ + "ordinal": 1, "source_name": "value", "physical_name": "value", + "storage_kind": "numeric", "string_width": None, "label": "", + "format": "F8.2", "measure": "scale", "alignment": "right", + "display_width": 8, "value_labels": "{}", "missing_ranges": "[]", + }] + + with pytest.raises(Exception, match="Target capability exceeded"): + create_wide_dataset( + database_url=database, dataset_id="nonfinite", + source_name="nonfinite.sav", source_format="SAV", + rows=[{"value": value}], variables=variables, + ) + + connection = sqlite3.connect(database_path) + assert "data_nonfinite" not in { + row[0] for row in connection.execute( + "select name from sqlite_master where type = 'table'" + ) + } + assert connection.execute("select count(*) from dataset").fetchone() == (0,) + assert connection.execute("select count(*) from variable").fetchone() == (0,) + assert connection.execute( + "select status from operation" + ).fetchall() == [("failed",)] + assert connection.execute( + "select dataset_id, event_code from fidelity_event" + ).fetchall() == [(None, "target_capability_exceeded")] + + +def test_empty_namespace_dolt_width_failure_initializes_identity_and_one_audit( + tmp_path, monkeypatch, +) -> None: + database_path = tmp_path / "dolt-preflight.sqlite" + database = f"sqlite:///{database_path}" + monkeypatch.setattr(wide, "effective_profile", lambda _url: (DOLT, {})) + + with pytest.raises(Exception, match="Target capability exceeded"): + create_wide_dataset( + database_url=database, dataset_id="too-wide-dolt", + source_name="too-wide-dolt.sav", source_format="SAV", + rows=(), variables=[{}] * 306, + ) + + connection = sqlite3.connect(database_path) + assert connection.execute( + "select contract_id, schema_version from catalog_identity" + ).fetchall() == [("openstatspec-strict-wide-table-v1", 1)] + assert connection.execute( + "select operation_kind, status from operation" + ).fetchall() == [("import", "failed")] + assert connection.execute( + "select dataset_id, event_code from fidelity_event" + ).fetchall() == [(None, "target_capability_exceeded")] + assert connection.execute( + "select dataset_id, status from operation_catalog" + ).fetchall() == [(None, "failed")] + assert connection.execute( + "select dataset_id, code from fidelity_event_catalog" + ).fetchall() == [(None, "target_capability_exceeded")] + + +@pytest.mark.parametrize("failure_point", ["mirror_completion", "normative_completion"]) +def test_nonatomic_failure_during_final_completion_still_cleans_dataset( + tmp_path, monkeypatch, failure_point, +) -> None: + database_path = tmp_path / f"{failure_point}.sqlite" + database = f"sqlite:///{database_path}" + dataset_id = f"cleanup-{failure_point}" + variables = [{ + "ordinal": 1, "source_name": "name", "physical_name": "name", + "storage_kind": "string", "string_width": 8, "label": "", + "format": "A8", "measure": "nominal", "alignment": "left", + "display_width": 8, "value_labels": "{}", "missing_ranges": "[]", + }] + monkeypatch.setattr( + wide, "effective_profile", + lambda _url: (replace(MYSQL, name="mysql"), {}), + ) + triggered = False + if failure_point == "mirror_completion": + real_update = wide.update + + def fail_first_operation_update(table): + nonlocal triggered + if table.name == "operation_catalog" and not triggered: + triggered = True + raise RuntimeError("fault after data insert") + return real_update(table) + + monkeypatch.setattr(wide, "update", fail_first_operation_update) + else: + real_finish = wide.finish_normative_operation + + def fail_first_normative_finish(*args, **kwargs): + nonlocal triggered + if not triggered: + triggered = True + raise RuntimeError("fault during final completion") + return real_finish(*args, **kwargs) + + monkeypatch.setattr( + wide, "finish_normative_operation", fail_first_normative_finish, + ) + + with pytest.raises(RuntimeError, match="fault"): + create_wide_dataset( + database_url=database, dataset_id=dataset_id, + source_name=f"{dataset_id}.sav", source_format="SAV", + rows=[{"name": "ok"}], variables=variables, + ) + + connection = sqlite3.connect(database_path) + assert triggered is True + assert f"data_{dataset_id}" not in { + row[0] for row in connection.execute( + "select name from sqlite_master where type = 'table'" + ) + } + existing_tables = { + row[0] for row in connection.execute( + "select name from sqlite_master where type = 'table'" + ) + } + if "dataset_catalog" in existing_tables: + assert connection.execute( + "select count(*) from dataset_catalog where dataset_id = ?", (dataset_id,) + ).fetchone() == (0,) + assert connection.execute( + "select count(*) from dataset where dataset_name = ?", (dataset_id,) + ).fetchone() == (0,) + assert connection.execute( + "select status, dataset_id from operation_catalog" + ).fetchall() == [("failed", None)] + assert connection.execute( + "select status from operation" + ).fetchall() == [("failed",)] + assert connection.execute( + "select code, dataset_id from fidelity_event_catalog" + ).fetchall() == [("import_failed", None)] diff --git a/tests/test_official_conformance.py b/tests/test_official_conformance.py index 8fcef21..c2714de 100644 --- a/tests/test_official_conformance.py +++ b/tests/test_official_conformance.py @@ -14,7 +14,7 @@ import openstatspec from conformance import compare_sav_semantics -from openstatspec.sql.profiles import profile_for_url +from openstatspec.sql.capabilities import effective_profile SEMANTIC_EXPECTATIONS = { @@ -276,6 +276,7 @@ def test_official_manifest_round_trips_through_sqlite( ("OPENSTATSPEC_POSTGRES_URL", "postgresql"), ("OPENSTATSPEC_MYSQL_URL", "mysql"), ("OPENSTATSPEC_MARIADB_URL", "mariadb"), + ("OPENSTATSPEC_DOLT_URL", "dolt"), ], ) @pytest.mark.parametrize(("fixture", "source"), _round_trip_fixtures()) @@ -299,7 +300,7 @@ def _assert_official_preflight_failure( ) assert fixture["directions"] == ["import"] assert set(fixture["expects"]) == PREFLIGHT_EXPECTATIONS - maximum = profile_for_url(database_url).max_physical_variables + maximum = effective_profile(database_url)[0].max_physical_variables source = tmp_path / f"preflight-too-wide-{profile}-{uuid4().hex[:8]}.sav" columns = [f"v{ordinal:05d}" for ordinal in range(1, maximum + 2)] pyspssio.write_sav( @@ -358,6 +359,7 @@ def test_official_preflight_failure_is_atomic_and_diagnostic(tmp_path: Path) -> ("OPENSTATSPEC_POSTGRES_URL", "postgresql"), ("OPENSTATSPEC_MYSQL_URL", "mysql"), ("OPENSTATSPEC_MARIADB_URL", "mariadb"), + ("OPENSTATSPEC_DOLT_URL", "dolt"), ], ) def test_official_preflight_failure_through_server_profiles( diff --git a/tests/test_sql_profiles.py b/tests/test_sql_profiles.py index 3c609db..0a46fbc 100755 --- a/tests/test_sql_profiles.py +++ b/tests/test_sql_profiles.py @@ -1,5 +1,6 @@ from dataclasses import replace import os +from types import SimpleNamespace import pytest from sqlalchemy import Column, MetaData, Table @@ -9,9 +10,12 @@ from openstatspec.sql.normative import catalog as normative_catalog from openstatspec.core import UnsupportedOperationError -from openstatspec.sql.profiles import MYSQL, POSTGRESQL, SQLITE, preflight, profile_for_url -from openstatspec.sql.capabilities import active_connection, server_version_supported -from openstatspec.sql.wide import binary64_type +from openstatspec.sql.profiles import DOLT, MYSQL, POSTGRESQL, SQLITE, preflight, profile_for_url +import openstatspec.sql.capabilities as capabilities +import openstatspec.sql.wide as wide +import openstatspec.spss.sav as sav +from openstatspec.sql.capabilities import active_connection, effective_profile, server_version_supported +from openstatspec.sql.wide import binary64_type, string_type def test_profile_detection_tracks_supported_dialect_urls() -> None: @@ -19,6 +23,7 @@ def test_profile_detection_tracks_supported_dialect_urls() -> None: assert profile_for_url("postgresql+psycopg://user@host/database") is POSTGRESQL assert profile_for_url("mysql+pymysql://user@host/database") is MYSQL 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_preflight_fails_without_transforming_a_wide_dataset() -> None: @@ -78,6 +83,9 @@ def test_normative_catalog_compiles_for_every_sql_family() -> None: [ ("mysql", "8.4.6", True), ("mysql", "9.7.0", True), ("mysql", "8.0.44", False), + ("dolt", "2.2.2", True), ("dolt", " 2.2.2 ", True), + ("dolt", "2.2.3", False), ("dolt", "2.2.2-rc1", False), + ("dolt", "2.2.2+build.1", False), ("dolt", "garbage", False), ("mariadb", "11.4.8-MariaDB", True), ("mariadb", "11.8.3-MariaDB", True), ("mariadb", "12.3.1-MariaDB", True), @@ -98,6 +106,7 @@ def test_claimed_server_version_policy_is_explicit( ("OPENSTATSPEC_POSTGRES_URL", "postgresql"), ("OPENSTATSPEC_MYSQL_URL", "mysql"), ("OPENSTATSPEC_MARIADB_URL", "mariadb"), + ("OPENSTATSPEC_DOLT_URL", "dolt"), ], ) def test_active_server_identity_matches_claimed_ci_profile( @@ -110,3 +119,297 @@ def test_active_server_identity_matches_claimed_ci_profile( assert active["profile"] == profile assert active["claimed_supported"] is True assert active["matched_claim"] is not None + +class _ProbeResult: + def __init__(self, value): + self.value = value + + def scalar_one(self): + return self.value + + +class _ProbeConnection: + def __init__(self, responses): + self.responses = responses + self.calls = [] + self.dialect = SimpleNamespace(name="mysql") + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def execute(self, statement): + query = str(statement) + self.calls.append(query) + value = self.responses[query] + if isinstance(value, Exception): + raise value + return _ProbeResult(value) + + +class _ProbeEngine: + def __init__(self, connection): + self.connection = connection + self.dialect = SimpleNamespace(driver="pymysql") + + def connect(self): + return self.connection + + +def _mock_mysql_probes(monkeypatch, **overrides): + responses = { + "select @@version": "8.0.31", + "select @@version_comment": "Dolt", + "select DOLT_VERSION()": "2.2.2", + "select @@max_allowed_packet": 1_073_741_824, + } + responses.update(overrides) + connection = _ProbeConnection(responses) + monkeypatch.setattr(capabilities, "create_engine", lambda _url: _ProbeEngine(connection)) + return connection + + +def test_dolt_identity_is_exact_and_publishes_wire_and_product_versions(monkeypatch) -> None: + connection = _mock_mysql_probes( + monkeypatch, **{"select @@version_comment": " dOlT "}, + ) + + active = active_connection("mysql+pymysql://user@host/database") + + assert active["dialect"] == "mysql" + assert active["profile"] == active["engine"] == active["product"] == "dolt" + assert active["transport"] == "mysql" + assert active["driver"] == "pymysql" + assert active["raw_wire_version"] == "8.0.31" + assert active["raw_product_version"] == active["raw_server_version"] == "2.2.2" + assert active["raw_version_comment"] == " dOlT " + assert active["server_version"] == "2.2.2" + assert active["identity_source"] == "SELECT @@version, @@version_comment, DOLT_VERSION()" + assert active["claimed_supported"] is True + assert connection.calls == [ + "select @@version", "select @@version_comment", + "select DOLT_VERSION()", "select @@max_allowed_packet", + ] + + +@pytest.mark.parametrize("comment", [None, False, "", " "]) +def test_mysql_wire_identity_requires_a_nonempty_version_comment(monkeypatch, comment) -> None: + connection = _mock_mysql_probes( + monkeypatch, **{"select @@version_comment": comment}, + ) + + with pytest.raises(UnsupportedOperationError, match="@@version_comment"): + active_connection("mysql+pymysql://user@host/database") + + assert "select DOLT_VERSION()" not in connection.calls + assert "select @@max_allowed_packet" not in connection.calls + + +def test_mysql_wire_identity_fails_closed_when_comment_probe_raises(monkeypatch) -> None: + connection = _mock_mysql_probes( + monkeypatch, **{"select @@version_comment": RuntimeError("probe unavailable")}, + ) + + with pytest.raises(UnsupportedOperationError, match="@@version_comment"): + active_connection("mysql+pymysql://user@host/database") + + assert "select DOLT_VERSION()" not in connection.calls + + +def test_dolt_comment_requires_a_nonempty_product_version(monkeypatch) -> None: + _mock_mysql_probes(monkeypatch, **{"select DOLT_VERSION()": None}) + + with pytest.raises(UnsupportedOperationError, match=r"DOLT_VERSION\(\)"): + active_connection("mysql+pymysql://user@host/database") + + +@pytest.mark.parametrize( + "product_version", + ["2.2.2-rc1", "2.2.2+build.1", "v2.2.2", "garbage"], +) +def test_dolt_identity_rejects_nonexact_product_versions( + monkeypatch, product_version, +) -> None: + connection = _mock_mysql_probes( + monkeypatch, **{"select DOLT_VERSION()": product_version}, + ) + + with pytest.raises(UnsupportedOperationError, match="exactly 2.2.2"): + active_connection("mysql+pymysql://user@host/database") + + assert connection.calls == [ + "select @@version", "select @@version_comment", "select DOLT_VERSION()", + ] + + +def test_unknown_mysql_wire_product_fails_closed_without_dolt_probe(monkeypatch) -> None: + connection = _mock_mysql_probes( + monkeypatch, + **{ + "select @@version": "8.4.6", + "select @@version_comment": "Percona Server", + }, + ) + + with pytest.raises(UnsupportedOperationError, match="product is unknown"): + active_connection("mysql+pymysql://user@host/database") + + assert "select DOLT_VERSION()" not in connection.calls + assert "select @@max_allowed_packet" not in connection.calls + + +def test_non_dolt_products_never_call_the_dolt_function(monkeypatch) -> None: + connection = _mock_mysql_probes( + monkeypatch, + **{ + "select @@version": "8.4.6", + "select @@version_comment": "MySQL Community Server", + }, + ) + + active = active_connection("mysql+pymysql://user@host/database") + + assert active["profile"] == "mysql" + assert active["claimed_supported"] is True + assert "select DOLT_VERSION()" not in connection.calls + + +def test_effective_profile_selects_dolt_without_changing_url_profile(monkeypatch) -> None: + active = { + "profile": "dolt", "server_version": "2.2.2", "claimed_supported": True, + "observed": {"max_allowed_packet": 1_073_741_824}, + } + monkeypatch.setattr(capabilities, "active_connection", lambda _url: active) + + profile, observed = effective_profile("mysql+pymysql://user@host/database") + + assert profile is not MYSQL + assert profile.name == "dolt" + assert profile.url_schemes == () + assert profile.max_physical_variables == 305 + assert profile.max_text_value_bytes == 65_504 + assert profile.max_row_bytes == 65_504 + assert observed is active + + +def test_dolt_declaration_labels_conservative_envelopes() -> None: + declaration = capabilities.profile_declarations()["dolt"] + + assert declaration["dialect"] == declaration["transport"] == "mysql" + assert declaration["profile"] == declaration["engine"] == "dolt" + assert declaration["claimed_server_versions"] == ["Dolt 2.2.2"] + assert declaration["ci_tested_server_versions"] == ["Dolt 2.2.2"] + assert declaration["proposed_adapter_limits"]["maximum_physical_columns"] == 306 + assert declaration["proposed_adapter_limits"]["maximum_source_variables"] == 305 + assert declaration["proposed_adapter_limits"]["maximum_value_bytes"] == 65_504 + assert declaration["proposed_adapter_limits"]["maximum_row_bytes"] == 65_504 + assert declaration["theoretical_limits"]["maximum_value_bytes"] == 4_294_967_295 + assert declaration["observed_limits"]["minimum_observed_physical_columns"] == 307 + assert declaration["observed_limits"]["identifier_limit"]["value"] == 64 + assert declaration["observed_limits"]["rejected_identifier_bytes"] == 65 + assert set(declaration["proposed_adapter_limits"]) == { + "maximum_physical_columns", "maximum_source_variables", + "maximum_value_bytes", "maximum_row_bytes", + } + assert declaration["limit_bases"]["maximum_physical_columns"] == "proposed_adapter_envelope" + assert declaration["limit_bases"]["identifier_limit"] == "observed_exact_version" + assert declaration["limit_bases"]["maximum_value_bytes"] == "observed_exact_version" + assert declaration["limit_bases"]["maximum_statement_bytes"] == "active_connection_observation" + assert declaration["effective_limits"] is None + assert declaration["text_type"] == "LONGTEXT" + assert capabilities.profile_declarations()["mysql"]["text_type"] == "LONGTEXT" + assert capabilities.profile_declarations()["mariadb"]["text_type"] == "LONGTEXT" + assert declaration["ddl_atomic"] is False + assert declaration["failure_cleanup"] == "compensating_cleanup" + assert declaration["numeric_value_policy"] == { + "finite_binary64": "supported", + "nan": "rejected_before_ddl", + "positive_infinity": "rejected_before_ddl", + "negative_infinity": "rejected_before_ddl", + } + assert declaration["storage_evidence"]["binary64"]["maximum_finite_round_trip_exact"] is True + assert declaration["storage_evidence"]["binary64"]["source"] + assert declaration["storage_evidence"]["binary64"]["version"] == "2.2.2" + assert declaration["storage_evidence"]["text"]["observed_value_bytes"] == 65_504 + assert declaration["storage_evidence"]["text"]["source"] + assert declaration["storage_evidence"]["text"]["version"] == "2.2.2" + assert declaration["storage_evidence"]["text"]["unit"] == "bytes" + assert declaration["transformation_workflow"] == "unsupported" + + +def test_dolt_uses_longtext_without_changing_mysql_storage() -> None: + dolt_table = Table("dolt_text", MetaData(), Column("value", string_type(DOLT))) + mysql_table = Table("mysql_text", MetaData(), Column("value", string_type(MYSQL))) + + assert "VALUE LONGTEXT" in str(CreateTable(dolt_table).compile(dialect=mysql.dialect())).upper() + mysql_ddl = str(CreateTable(mysql_table).compile(dialect=mysql.dialect())).upper() + assert "VALUE TEXT" in mysql_ddl + assert "LONGTEXT" not in mysql_ddl + + +def test_dolt_row_preflight_counts_utf8_values_against_adapter_envelope() -> None: + variables = [{ + "ordinal": 1, "source_name": "value", "physical_name": "value", + "storage_kind": "string", "string_width": 65_504, + }] + preflight(DOLT, variables, rows=[{"value": "x" * 65_504}]) + + with pytest.raises(UnsupportedOperationError) as error: + preflight(DOLT, variables, rows=[{"value": "x" * 65_505}]) + assert error.value.details["reason"] == "text_value_limit" + + +@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")]) +def test_dolt_numeric_preflight_rejects_nonfinite_values(value) -> None: + variables = [{ + "ordinal": 1, "source_name": "value", "physical_name": "value", + "storage_kind": "numeric", "string_width": None, + }] + + with pytest.raises(UnsupportedOperationError) as error: + preflight(DOLT, variables, rows=[{"value": value}]) + + assert error.value.details == { + "reason": "nonfinite_numeric_value", + "row_ordinal": 1, + "source_name": "value", + } + + +def test_validate_identity_failure_happens_before_catalog_access(monkeypatch) -> None: + def fail_identity(_url): + raise UnsupportedOperationError("identity unavailable") + + monkeypatch.setattr(wide, "effective_profile", fail_identity) + monkeypatch.setattr( + wide, "create_engine", + lambda _url: pytest.fail("database access continued after identity failure"), + ) + + with pytest.raises(UnsupportedOperationError, match="identity unavailable"): + wide.validate_wide_dataset( + database_url="mysql+pymysql://user@host/database", dataset_id="fixture", + ) + + +def test_export_identity_failure_happens_before_read_or_destination(monkeypatch, tmp_path) -> None: + destination = tmp_path / "blocked.sav" + + def fail_identity(_url): + raise UnsupportedOperationError("identity unavailable") + + monkeypatch.setattr(sav, "effective_profile", fail_identity) + monkeypatch.setattr( + sav, "read_wide_dataset", + lambda **_kwargs: pytest.fail("catalog read continued after identity failure"), + ) + + with pytest.raises(UnsupportedOperationError, match="identity unavailable"): + sav.export_sav_dataset( + database_url="mysql+pymysql://user@host/database", + dataset_id="fixture", destination=destination, + ) + + assert not destination.exists() diff --git a/tests/test_sql_services.py b/tests/test_sql_services.py index 1141df4..e2e664f 100755 --- a/tests/test_sql_services.py +++ b/tests/test_sql_services.py @@ -1,13 +1,17 @@ -"""Real-service conformance checks for PostgreSQL, MySQL, and MariaDB profiles.""" +"""Real-service conformance checks for PostgreSQL, MySQL, MariaDB, and Dolt.""" import os +from uuid import uuid4 import pandas as pd import pyspssio import pytest -from sqlalchemy import create_engine, text +from sqlalchemy import create_engine, inspect as inspect_database, text +from sqlalchemy.exc import DBAPIError import openstatspec +import openstatspec.sql.wide as wide +from openstatspec.core import UnsupportedOperationError from conformance import compare_sav_semantics, write_supported_semantics_fixture @@ -33,21 +37,26 @@ def source_sav(tmp_path): @pytest.mark.parametrize( ("environment_name", "dataset_id"), - [("OPENSTATSPEC_POSTGRES_URL", "profile_pg"), ("OPENSTATSPEC_MYSQL_URL", "profile_mysql"), ("OPENSTATSPEC_MARIADB_URL", "profile_mariadb")], + [("OPENSTATSPEC_POSTGRES_URL", "profile_pg"), ("OPENSTATSPEC_MYSQL_URL", "profile_mysql"), ("OPENSTATSPEC_MARIADB_URL", "profile_mariadb"), ("OPENSTATSPEC_DOLT_URL", "profile_dolt")], ) def test_live_profile_import_validate_and_export(environment_name, dataset_id, source_sav, tmp_path): database_url = os.environ.get(environment_name) if not database_url: pytest.skip(f"{environment_name} is not configured") - imported = openstatspec.import_sav(source_sav, database_url=database_url, dataset_id=dataset_id) + runtime_dataset_id = f"{dataset_id}_{uuid4().hex[:8]}" + imported = openstatspec.import_sav( + source_sav, database_url=database_url, dataset_id=runtime_dataset_id, + ) assert imported["case_count"] == 2 - assert openstatspec.validate(database_url=database_url, dataset_id=dataset_id)["valid"] is True + assert openstatspec.validate( + database_url=database_url, dataset_id=runtime_dataset_id, + )["valid"] is True engine = create_engine(database_url) with engine.connect() as connection: assert connection.execute(text(f"SELECT COUNT(*) FROM {imported['data_table']} ")).scalar_one() == 2 - assert connection.execute(text("SELECT COUNT(*) FROM variable_catalog WHERE dataset_id = :dataset_id"), {"dataset_id": dataset_id}).scalar_one() == 2 + assert connection.execute(text("SELECT COUNT(*) FROM variable_catalog WHERE dataset_id = :dataset_id"), {"dataset_id": runtime_dataset_id}).scalar_one() == 2 destination = tmp_path / f"{dataset_id}.sav" - exported = openstatspec.export_sav(database_url=database_url, dataset_id=dataset_id, destination=destination, allow_loss=_REQUIRED_ENGINE_LOSS) + exported = openstatspec.export_sav(database_url=database_url, dataset_id=runtime_dataset_id, destination=destination, allow_loss=_REQUIRED_ENGINE_LOSS) assert destination.exists() assert {diagnostic.code for diagnostic in exported.diagnostics} == set(_REQUIRED_ENGINE_LOSS) frame, metadata = pyspssio.read_sav(str(destination), convert_datetimes=False, include_user_missing=True) @@ -60,18 +69,258 @@ def test_live_profile_import_validate_and_export(environment_name, dataset_id, s @pytest.mark.parametrize( ("environment_name", "dataset_id"), - [("OPENSTATSPEC_POSTGRES_URL", "semantics_pg"), ("OPENSTATSPEC_MYSQL_URL", "semantics_mysql"), ("OPENSTATSPEC_MARIADB_URL", "semantics_mariadb")], + [("OPENSTATSPEC_POSTGRES_URL", "semantics_pg"), ("OPENSTATSPEC_MYSQL_URL", "semantics_mysql"), ("OPENSTATSPEC_MARIADB_URL", "semantics_mariadb"), ("OPENSTATSPEC_DOLT_URL", "semantics_dolt")], ) @pytest.mark.parametrize("suffix", [".sav", ".zsav"]) def test_live_profile_preserves_supported_sav_semantics(environment_name, dataset_id, suffix, tmp_path): database_url = os.environ.get(environment_name) if not database_url: pytest.skip(f"{environment_name} is not configured") - source = tmp_path / f"{dataset_id}{suffix}" - destination = tmp_path / f"{dataset_id}-roundtrip{suffix}" + runtime_dataset_id = f"{dataset_id}_{suffix[1:]}_{uuid4().hex[:8]}" + source = tmp_path / f"{runtime_dataset_id}{suffix}" + destination = tmp_path / f"{runtime_dataset_id}-roundtrip{suffix}" write_supported_semantics_fixture(source) - imported = openstatspec.import_sav(source, database_url=database_url, dataset_id=f"{dataset_id}_{suffix[1:]}") + imported = openstatspec.import_sav( + source, database_url=database_url, dataset_id=runtime_dataset_id, + ) assert imported["case_count"] == 4 - assert openstatspec.validate(database_url=database_url, dataset_id=f"{dataset_id}_{suffix[1:]}")["valid"] is True - openstatspec.export_sav(database_url=database_url, dataset_id=f"{dataset_id}_{suffix[1:]}", destination=destination, allow_loss=_COMPAT_NAME_LOSS) - assert compare_sav_semantics(source, destination) == {"equivalent": True, "differences": []} \ No newline at end of file + assert openstatspec.validate( + database_url=database_url, dataset_id=runtime_dataset_id, + )["valid"] is True + openstatspec.export_sav( + database_url=database_url, dataset_id=runtime_dataset_id, + destination=destination, allow_loss=_COMPAT_NAME_LOSS, + ) + assert compare_sav_semantics(source, destination) == {"equivalent": True, "differences": []} + +def test_live_dolt_conservative_source_width_envelope(tmp_path) -> None: + database_url = os.environ.get("OPENSTATSPEC_DOLT_URL") + if not database_url: + pytest.skip("OPENSTATSPEC_DOLT_URL is not configured") + token = uuid4().hex[:8] + accepted_id = f"dolt_width_accepted_{token}" + rejected_id = f"dolt_width_rejected_{token}" + accepted_source = tmp_path / f"{accepted_id}.sav" + rejected_source = tmp_path / f"{rejected_id}.sav" + accepted_columns = [f"v{ordinal:03d}" for ordinal in range(1, 306)] + rejected_columns = [*accepted_columns, "v306"] + pyspssio.write_sav( + str(accepted_source), + pd.DataFrame([[float(ordinal) for ordinal in range(1, 306)]], + columns=accepted_columns), + ) + pyspssio.write_sav( + str(rejected_source), + pd.DataFrame([[float(ordinal) for ordinal in range(1, 307)]], + columns=rejected_columns), + ) + + imported = openstatspec.import_sav( + accepted_source, database_url=database_url, dataset_id=accepted_id, + ) + assert imported["case_count"] == 1 + assert openstatspec.validate( + database_url=database_url, dataset_id=accepted_id, + )["variable_count"] == 305 + + with pytest.raises(UnsupportedOperationError, match="Target capability exceeded"): + openstatspec.import_sav( + rejected_source, database_url=database_url, dataset_id=rejected_id, + ) + + engine = create_engine(database_url) + with engine.connect() as connection: + assert connection.execute(text( + "select count(*) from dataset where dataset_name = :name" + ), {"name": rejected_id}).scalar_one() == 0 + assert connection.execute(text( + "select count(*) from dataset_catalog where dataset_id = :name" + ), {"name": rejected_id}).scalar_one() == 0 + mirror_event = connection.execute(text(""" + select f.dataset_id, f.direction, f.severity, f.code + from fidelity_event_catalog f + join operation_catalog o on o.operation_id = f.operation_id + where o.source = :source + """), {"source": rejected_source.name}).mappings().one() + normative_event = connection.execute(text(""" + select f.dataset_id, f.direction, f.severity, f.event_code + from fidelity_event f + where f.source_item = :source + """), {"source": rejected_source.name}).mappings().one() + assert tuple(mirror_event.values()) == ( + None, "import", "error", "target_capability_exceeded", + ) + assert tuple(normative_event.values()) == ( + None, "import", "error", "target_capability_exceeded", + ) + assert f"data_{rejected_id}" not in inspect_database(engine).get_table_names() + + +def test_live_dolt_post_ddl_fault_has_complete_compensating_cleanup( + tmp_path, monkeypatch, +) -> None: + database_url = os.environ.get("OPENSTATSPEC_DOLT_URL") + if not database_url: + pytest.skip("OPENSTATSPEC_DOLT_URL is not configured") + dataset_id = f"dolt_cleanup_{uuid4().hex[:8]}" + source = tmp_path / f"{dataset_id}.sav" + pyspssio.write_sav(str(source), pd.DataFrame({"answer": [1.0]})) + real_store = wide.store_normative_dataset + + def fail_after_normative_write(*args, **kwargs): + real_store(*args, **kwargs) + raise RuntimeError("injected Dolt post-DDL fault") + + monkeypatch.setattr(wide, "store_normative_dataset", fail_after_normative_write) + + with pytest.raises(RuntimeError, match="injected Dolt post-DDL fault"): + openstatspec.import_sav( + source, database_url=database_url, dataset_id=dataset_id, + ) + + engine = create_engine(database_url) + with engine.connect() as connection: + assert connection.execute(text( + "select count(*) from dataset where dataset_name = :name" + ), {"name": dataset_id}).scalar_one() == 0 + assert connection.execute(text( + "select count(*) from dataset_catalog where dataset_id = :name" + ), {"name": dataset_id}).scalar_one() == 0 + assert connection.execute(text( + "select status, dataset_id from operation_catalog where source = :source" + ), {"source": source.name}).one() == ("failed", None) + assert connection.execute(text(""" + select f.dataset_id, f.direction, f.severity, f.code + from fidelity_event_catalog f + join operation_catalog o on o.operation_id = f.operation_id + where o.source = :source + """), {"source": source.name}).one() == ( + None, "import", "error", "import_failed", + ) + assert f"data_{dataset_id}" not in inspect_database(engine).get_table_names() + + +def test_live_dolt_adapter_value_boundary_is_atomic() -> None: + database_url = os.environ.get("OPENSTATSPEC_DOLT_URL") + if not database_url: + pytest.skip("OPENSTATSPEC_DOLT_URL is not configured") + token = uuid4().hex[:8] + accepted_id = f"dolt_value_accepted_{token}" + rejected_id = f"dolt_value_rejected_{token}" + accepted_value = "é" * 32_752 + rejected_value = accepted_value + "x" + assert len(accepted_value.encode("utf-8")) == 65_504 + assert len(rejected_value.encode("utf-8")) == 65_505 + variables = [{ + "ordinal": 1, "source_name": "value", "physical_name": "value", + "storage_kind": "string", "string_width": 65_504, "label": "", + "format": "A65504", "measure": "nominal", "alignment": "left", + "display_width": 8, "value_labels": "{}", "missing_ranges": "[]", + }] + + imported = wide.create_wide_dataset( + database_url=database_url, dataset_id=accepted_id, + source_name="accepted.sav", source_format="SAV", + rows=[{"value": accepted_value}], variables=variables, + ) + assert imported["case_count"] == 1 + + with pytest.raises(UnsupportedOperationError) as caught: + wide.create_wide_dataset( + database_url=database_url, dataset_id=rejected_id, + source_name="rejected.sav", source_format="SAV", + rows=[{"value": rejected_value}], variables=variables, + ) + assert caught.value.details["reason"] == "text_value_limit" + + engine = create_engine(database_url) + accepted_table = wide.data_table_name(accepted_id) + rejected_table = wide.data_table_name(rejected_id) + quote = engine.dialect.identifier_preparer.quote + with engine.connect() as connection: + assert connection.execute(text( + f"SELECT OCTET_LENGTH(value) FROM {quote(accepted_table)}" + )).scalar_one() == 65_504 + assert connection.execute(text( + "SELECT COUNT(*) FROM dataset_catalog WHERE dataset_id = :dataset_id" + ), {"dataset_id": rejected_id}).scalar_one() == 0 + assert connection.execute(text( + "SELECT COUNT(*) FROM dataset WHERE dataset_name = :dataset_id" + ), {"dataset_id": rejected_id}).scalar_one() == 0 + assert rejected_table not in inspect_database(engine).get_table_names() + engine.dispose() + + +def test_live_dolt_published_storage_and_identifier_evidence() -> None: + database_url = os.environ.get("OPENSTATSPEC_DOLT_URL") + if not database_url: + pytest.skip("OPENSTATSPEC_DOLT_URL is not configured") + engine = create_engine(database_url) + token = uuid4().hex[:8] + identifier_64 = f"evidence_{token}_" + "i" * (64 - len(f"evidence_{token}_")) + identifier_65 = identifier_64 + "i" + assert len(identifier_64.encode("utf-8")) == 64 + assert len(identifier_65.encode("utf-8")) == 65 + storage_table = f"evidence_storage_{token}" + columns_306_table = f"evidence_columns_306_{token}" + columns_307_table = f"evidence_columns_307_{token}" + quote = engine.dialect.identifier_preparer.quote + + try: + with engine.begin() as connection: + connection.execute(text( + f"CREATE TABLE {quote(identifier_64)} (value INTEGER)" + )) + assert identifier_64 in inspect_database(engine).get_table_names() + with pytest.raises(DBAPIError): + with engine.begin() as connection: + connection.execute(text( + f"CREATE TABLE {quote(identifier_65)} (value INTEGER)" + )) + + maximum_finite = float.fromhex("0x1.fffffffffffffp+1023") + utf8_value = "é" * 32_752 + assert len(utf8_value.encode("utf-8")) == 65_504 + with engine.begin() as connection: + connection.execute(text( + f"CREATE TABLE {quote(storage_table)} " + "(binary64_value DOUBLE NOT NULL, text_value LONGTEXT NOT NULL)" + )) + connection.execute( + text( + f"INSERT INTO {quote(storage_table)} " + "(binary64_value, text_value) VALUES (:binary64, :text_value)" + ), + {"binary64": maximum_finite, "text_value": utf8_value}, + ) + with engine.connect() as connection: + row = connection.execute(text( + f"SELECT binary64_value, text_value, OCTET_LENGTH(text_value) " + f"FROM {quote(storage_table)}" + )).one() + assert row[0] == maximum_finite + assert row[1] == utf8_value + assert row[2] == 65_504 + + for column_count, table_name in ( + (306, columns_306_table), (307, columns_307_table), + ): + columns = ", ".join( + f"{quote(f'c{ordinal:03d}')} INTEGER" + for ordinal in range(1, column_count + 1) + ) + with engine.begin() as connection: + connection.execute(text( + f"CREATE TABLE {quote(table_name)} ({columns})" + )) + assert len(inspect_database(engine).get_columns(table_name)) == column_count + finally: + with engine.begin() as connection: + for table_name in ( + identifier_64, identifier_65, storage_table, + columns_306_table, columns_307_table, + ): + connection.execute(text( + f"DROP TABLE IF EXISTS {quote(table_name)}" + )) + engine.dispose() \ No newline at end of file