Address strict catalog review findings

Author: TonisOrmisson

CHANGELOG.md

@@ -10,6 +10,9 @@
   use the normative UUID-keyed OpenStatSpec catalog exclusively.
 - Existing databases that contain only the former compatibility catalog must be
   remediated manually before export.
+- Imports now reject dataset names that collide with normative UUID identifiers,
+  cleanup drops only a physical table actually created by the failing import,
+  and non-preserved SPSS compatible variable names are reported as an explicit
 All notable changes to this reference implementation are documented here.
 
 ## 0.4.0 — 2026-07-31

src/openstatspec/api.py

@@ -73,7 +73,7 @@ def capability_matrix(database_url: str | None = None) -> Mapping[str, Any]:
             "multiple_response_sets": "supported",
             "variable_alignment": "supported",
             "variable_sets": "supported",
-            "compatible_variable_names": "supported",
+            "compatible_variable_names": "requires-explicit-loss-consent",
             "custom_attributes": {
                 "scalar_values": "supported",
                 "ordered_value_arrays": "supported",

src/openstatspec/spss/sav.py

@@ -494,6 +494,21 @@ def _engine_loss_report(metadata: dict[str, Any]) -> dict[str, dict[str, Any]]:
             "detail": "pyspssio could not inspect source variable sets; they cannot be preserved silently.",
             "details": {"engine_error": metadata.get("_var_sets_error", "unknown")},
         }
+    compatible_names = {
+        str(source_name): str(compatible_name)
+        for source_name, compatible_name
+        in dict(metadata.get("var_compat_names") or {}).items()
+        if compatible_name
+        and str(source_name).casefold() != str(compatible_name).casefold()
+    }
+    if compatible_names:
+        events["compatible-variable-names-not-preserved"] = {
+            "code": "compatible-variable-names-not-preserved",
+            "detail": (
+                "The normative catalog does not store SPSS compatible variable names."
+            ),
+            "details": {"variable_names": sorted(compatible_names)},
+        }
     if _is_non_utf8_encoding(metadata.get("encoding")):
         events["source-encoding-not-preserved"] = {
             "code": "source-encoding-not-preserved",

src/openstatspec/sql/wide.py

@@ -11,7 +11,7 @@
 
 from sqlalchemy import (
     BigInteger, Column, Float, MetaData, Table, Text, create_engine, insert,
-    inspect, select,
+    inspect, or_, select,
 )
 from sqlalchemy.dialects import mysql, postgresql, sqlite
 from ..core import UnsupportedOperationError
@@ -509,16 +509,18 @@ def create_wide_dataset(
         normative_dataset_id, multiple_response_sets,
     )
     namespace_owned = False
-    data_table_was_absent = False
+    data_table_created = False
     try:
         with engine.begin() as setup:
             create_normative_catalog(setup, normative)
         namespace_owned = True
         with engine.begin() as connection:
             if connection.execute(
-                select(normative.dataset.c.dataset_id).where(
-                    normative.dataset.c.dataset_name == dataset_id
-                )
+                select(normative.dataset.c.dataset_id).where(or_(
+                    normative.dataset.c.dataset_name == dataset_id,
+                    normative.dataset.c.dataset_id == dataset_id,
+                    normative.dataset.c.dataset_name == normative_dataset_id,
+                ))
             ).first():
                 raise ValueError(
                     f"Dataset {dataset_id!r} already exists; imports never overwrite a dataset."
@@ -542,8 +544,8 @@ def create_wide_dataset(
                 operation_kind="import", status="started",
                 source_format=source_format,
             )
-            data_table_was_absent = True
             data_table.create(connection)
+            data_table_created = True
             store_normative_dataset(
                 connection, normative, dataset_name=dataset_id,
                 source_format=source_format, physical_table_name=data_table.name,
@@ -582,7 +584,7 @@ def create_wide_dataset(
                             cleanup, normative, normative_dataset_id,
                         )
                     if (
-                        data_table_was_absent
+                        data_table_created
                         and inspect(cleanup).has_table(data_table.name)
                     ):
                         data_table.drop(cleanup, checkfirst=True)

tests/test_atomic_import.py

@@ -32,6 +32,80 @@ def test_failed_row_insert_leaves_no_catalog_or_data_table(tmp_path) -> None:
     assert connection.execute("select count(*) from dataset").fetchone() == (0,)
     assert connection.execute("select count(*) from variable").fetchone() == (0,)
 
+def test_failed_create_does_not_drop_a_concurrently_created_table(
+    tmp_path, monkeypatch,
+) -> None:
+    database_path = tmp_path / "race.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": "[]",
+    }]
+    real_inspect = wide.inspect
+    inspected_data_table = 0
+
+    def inspect_with_concurrent_table(bind):
+        actual = real_inspect(bind)
+
+        class Inspector:
+            def has_table(self, table_name):
+                nonlocal inspected_data_table
+                if table_name == "data_race":
+                    inspected_data_table += 1
+                    return inspected_data_table > 1
+                return actual.has_table(table_name)
+
+        return Inspector()
+
+    real_create = wide.Table.create
+    real_drop = wide.Table.drop
+    dropped = []
+
+    def fail_data_table_create(table, bind, **kwargs):
+        if table.name == "data_race":
+            raise RuntimeError("concurrent create won")
+        return real_create(table, bind, **kwargs)
+
+    def observe_drop(table, bind, **kwargs):
+        if table.name == "data_race":
+            dropped.append(table.name)
+        return real_drop(table, bind, **kwargs)
+
+    monkeypatch.setattr(wide, "inspect", inspect_with_concurrent_table)
+    monkeypatch.setattr(wide.Table, "create", fail_data_table_create)
+    monkeypatch.setattr(wide.Table, "drop", observe_drop)
+
+    with pytest.raises(RuntimeError, match="concurrent create won"):
+        create_wide_dataset(
+            database_url=database, dataset_id="race", source_name="race.sav",
+            source_format="SAV", rows=[{"name": "ok"}], variables=variables,
+        )
+
+    assert inspected_data_table == 1
+    assert dropped == []
+
+
+def test_dataset_name_cannot_equal_an_existing_normative_uuid(tmp_path) -> None:
+    database = f"sqlite:///{tmp_path / 'namespace.sqlite'}"
+    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": "[]",
+    }]
+    first = create_wide_dataset(
+        database_url=database, dataset_id="first", source_name="first.sav",
+        source_format="SAV", rows=[{"name": "first"}], variables=variables,
+    )
+
+    with pytest.raises(ValueError, match="already exists"):
+        create_wide_dataset(
+            database_url=database, dataset_id=first["dataset_id"],
+            source_name="second.sav", source_format="SAV",
+            rows=[{"name": "second"}], variables=variables,
+        )
 
 def test_failed_preflight_persists_operation_without_creating_dataset(tmp_path) -> None:
     database_path = tmp_path / "preflight.sqlite"

tests/test_cli.py

@@ -64,7 +64,7 @@ def test_capability_matrix_is_public_and_cli_matches_engine_boundary(capsys) ->
         "multiple_response_sets": "supported",
         "variable_alignment": "supported",
         "variable_sets": "supported",
-        "compatible_variable_names": "supported",
+        "compatible_variable_names": "requires-explicit-loss-consent",
         "custom_attributes": {
             "scalar_values": "supported",
             "ordered_value_arrays": "supported",

tests/test_document_round_trip.py

@@ -81,7 +81,10 @@ def test_document_and_long_variable_name_round_trip_to_zsav(tmp_path: Path) -> N
     write_document_lines(source, ["Combined dictionary fixture."], encoding="UTF-8")
 
     openstatspec.import_sav(source, database_url=database, dataset_id="combined")
-    openstatspec.export_sav(database_url=database, dataset_id="combined", destination=destination)
+    openstatspec.export_sav(
+        database_url=database, dataset_id="combined", destination=destination,
+        allow_loss=["compatible-variable-names-not-preserved"],
+    )
 
     assert read_document_lines(destination, encoding="UTF-8") == ["Combined dictionary fixture."]
     assert pyspssio.read_sav(str(destination))[0][source_name].tolist() == [7.0]

tests/test_loss_reports.py

@@ -8,6 +8,7 @@
 from openstatspec.core import UnsupportedOperationError
 from openstatspec.sql.wide import create_wide_dataset
 from openstatspec.spss import sav as sav_module
+from openstatspec.spss.raw_dictionary import write_compatible_names
 
 
 _REQUIRED_ENGINE_LOSS = []
@@ -118,3 +119,36 @@ def test_legacy_locale_must_emit_the_exact_source_encoding() -> None:
     sav_module._require_matching_legacy_encoding("WINDOWS-1252", "CP1252", True)
     with pytest.raises(UnsupportedOperationError, match="instead of required"):
         sav_module._require_matching_legacy_encoding("WINDOWS-1252", "UTF-8", True)
+
+
+def test_compatible_variable_names_are_explicit_imported_loss(tmp_path) -> None:
+    source = tmp_path / "compatible-name.sav"
+    destination = tmp_path / "compatible-name-out.sav"
+    database = f"sqlite:///{tmp_path / 'compatible-name.sqlite'}"
+    source_name = "long_variable_name"
+    loss_code = "compatible-variable-names-not-preserved"
+
+    pyspssio.write_sav(str(source), pd.DataFrame({source_name: [1.0]}))
+    write_compatible_names(source, {source_name: "ANSWER"}, encoding="UTF-8")
+
+    imported = openstatspec.import_sav(
+        source, database_url=database, dataset_id="compatible-name",
+    )
+    assert {diagnostic.code for diagnostic in imported.diagnostics} == {loss_code}
+    assert imported.diagnostics[0].details == {"variable_names": [source_name]}
+
+    with pytest.raises(UnsupportedOperationError, match=loss_code):
+        openstatspec.export_sav(
+            database_url=database, dataset_id="compatible-name",
+            destination=destination,
+        )
+
+    exported = openstatspec.export_sav(
+        database_url=database, dataset_id="compatible-name",
+        destination=destination, allow_loss=[loss_code],
+    )
+    assert {diagnostic.code for diagnostic in exported.diagnostics} == {loss_code}
+    assert (
+        pyspssio.read_metadata(str(destination))["var_compat_names"][source_name]
+        != "ANSWER"
+    )

tests/test_official_conformance.py

@@ -235,16 +235,21 @@ def _assert_round_trip(
     imported = openstatspec.import_sav(
         source, database_url=database_url, dataset_id=dataset_id,
     )
-    assert imported.diagnostics == ()
+    assert {diagnostic.code for diagnostic in imported.diagnostics} <= {
+        "compatible-variable-names-not-preserved",
+    }
     assert openstatspec.validate(
         database_url=database_url, dataset_id=dataset_id,
     )["valid"] is True
     _assert_expected_catalog(database_url, dataset_id, fixture.get("expected_catalog", {}))
 
     exported = openstatspec.export_sav(
         database_url=database_url, dataset_id=dataset_id, destination=destination,
+        allow_loss=["compatible-variable-names-not-preserved"],
     )
-    assert exported.diagnostics == ()
+    assert {diagnostic.code for diagnostic in exported.diagnostics} <= {
+        "compatible-variable-names-not-preserved",
+    }
     assert compare_sav_semantics(source, destination) == {
         "equivalent": True,
         "differences": [],

tests/test_sav_sqlite.py

@@ -12,7 +12,7 @@
 
 _REQUIRED_ENGINE_LOSS = []
 
-_COMPAT_NAME_LOSS = _REQUIRED_ENGINE_LOSS
+_COMPAT_NAME_LOSS = [*_REQUIRED_ENGINE_LOSS, "compatible-variable-names-not-preserved"]
 
 def test_pyspssio_round_trip_uses_one_wide_table_and_catalog(tmp_path) -> None:
     source = tmp_path / "tiny.sav"