diff --git a/README.md b/README.md index 8b57cae..b379a9b 100644 --- a/README.md +++ b/README.md @@ -128,6 +128,25 @@ filters = { } ``` +### Migrating a legacy registry + +The schema registry (`_registry/tables.json`) stores each table's schema as a +base64 Arrow-IPC blob under a `schema_ipc` key. Registries written by older +versions instead used a `schema_fields` list of `{name, type, nullable}` dicts, +which is no longer read on load — such a store fails to open with a `ValueError` +pointing you here. + +Upgrade the sidecar in place with the bundled console script: + +```bash +amplify-db-migrate path/to/store/_registry/tables.json +``` + +It rewrites each legacy entry to carry `schema_ipc` plus a human-readable +`columns` list, leaving `partition_by` untouched. The command is idempotent — +re-running it on an already-migrated file is a no-op — so it is safe to run +defensively before opening a store of unknown age. + --- ## Design notes @@ -141,4 +160,3 @@ filters = { **Schema registry.** Per-table schema and `partition_by` metadata are persisted as `_registry/tables.json` at the store root, readable and writable via PyArrow's filesystem abstraction (local or S3). **Backend independence.** `ColumnarStore` is an abstract base class. The DuckDB+Parquet backend is intended for local development, laptops, and single-process workflows. A VAST DB backend can be added as a drop-in for production-scale concurrent access without changes to the consuming service. - diff --git a/pyproject.toml b/pyproject.toml index 464e807..1cb0f68 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,12 +13,16 @@ dependencies = [ "pydantic>=2.0", ] +[project.scripts] +amplify-db-migrate = "amplify_db_utils.migrate:main" + [project.optional-dependencies] pandas = ["pandas>=2.0"] dev = [ "pytest>=7.0", "pytest-cov", "pandas>=2.0", + "pytz", ] [tool.hatch.build.targets.wheel] diff --git a/src/amplify_db_utils/migrate.py b/src/amplify_db_utils/migrate.py new file mode 100644 index 0000000..76be627 --- /dev/null +++ b/src/amplify_db_utils/migrate.py @@ -0,0 +1,164 @@ +"""In-place migration of legacy registry sidecar files to the IPC format. + +Legacy registries (written before the schema was stored as an Arrow-IPC blob) +recorded each table's schema as a ``schema_fields`` list of +``{name, type, nullable}`` dicts, where ``type`` was the stringified PyArrow +type. That per-field JSON form is no longer read by +:mod:`amplify_db_utils.registry`; the only authoritative representation is now +the base64 ``schema_ipc`` blob. + +This module reads such a legacy ``tables.json``, reconstructs each schema, and +rewrites the file in place so every entry carries: + +* ``schema_ipc`` — base64 Arrow-IPC schema (authoritative on load) +* ``columns`` — human-readable list of column names +* ``partition_by`` — carried through unchanged + +The legacy string→PyArrow parser lives here (not in ``registry.py``) so the +registry stays free of the format we are retiring. +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import re +import sys +from pathlib import Path + +import pyarrow as pa + + +def _arrow_type_from_str(s: str) -> pa.DataType: + """Deserialize a PyArrow type from its legacy string representation. + + Covers the scalar + timestamp types that legacy registries could emit. + """ + simple: dict[str, pa.DataType] = { + "string": pa.utf8(), + "utf8": pa.utf8(), + "large_string": pa.large_utf8(), + "large_utf8": pa.large_utf8(), + "int8": pa.int8(), + "int16": pa.int16(), + "int32": pa.int32(), + "int64": pa.int64(), + "uint8": pa.uint8(), + "uint16": pa.uint16(), + "uint32": pa.uint32(), + "uint64": pa.uint64(), + "float": pa.float32(), + "float32": pa.float32(), + "float64": pa.float64(), + "double": pa.float64(), + "halffloat": pa.float16(), + "bool": pa.bool_(), + "date32[day]": pa.date32(), + "date64[ms]": pa.date64(), + } + if s in simple: + return simple[s] + + # Timestamp: "timestamp[us, tz=UTC]" or "timestamp[us]" + m = re.match(r"timestamp\[(\w+)(?:,\s*tz=(.+))?\]", s) + if m: + unit, tz = m.group(1), m.group(2) + return pa.timestamp(unit, tz=tz) + + raise ValueError(f"Cannot deserialize legacy PyArrow type: {s!r}") + + +def _schema_from_legacy_fields(fields: list[dict]) -> pa.Schema: + """Reconstruct a schema from the legacy ``schema_fields`` form.""" + return pa.schema([ + pa.field(f["name"], _arrow_type_from_str(f["type"]), nullable=f["nullable"]) + for f in fields + ]) + + +def _schema_to_ipc_b64(schema: pa.Schema) -> str: + """Serialize a schema to a base64 Arrow-IPC string.""" + return base64.b64encode(schema.serialize().to_pybytes()).decode("ascii") + + +def migrate_file(path: str | Path) -> int: + """Migrate a legacy registry ``tables.json`` in place. + + Each entry with a ``schema_fields`` list (and no ``schema_ipc``) is rewritten + to carry ``schema_ipc`` + ``columns``; ``schema_fields`` is dropped. Entries + that already have ``schema_ipc`` are left untouched (idempotent). + + Args: + path: Path to the ``tables.json`` sidecar file. + + Returns: + Number of table entries that were migrated. + + Raises: + FileNotFoundError: If ``path`` does not exist. + ValueError: If an entry has neither ``schema_ipc`` nor ``schema_fields``. + """ + path = Path(path) + data = json.loads(path.read_text(encoding="utf-8")) + + migrated = 0 + for table_name, entry in data.items(): + if "schema_ipc" in entry: + # Already IPC-backed; just drop any stale legacy field. + entry.pop("schema_fields", None) + entry.setdefault( + "columns", _schema_from_ipc_columns(entry["schema_ipc"]) + ) + continue + + if "schema_fields" not in entry: + raise ValueError( + f"Registry entry for table '{table_name}' has neither " + f"'schema_ipc' nor 'schema_fields'; cannot migrate." + ) + + schema = _schema_from_legacy_fields(entry.pop("schema_fields")) + entry["schema_ipc"] = _schema_to_ipc_b64(schema) + entry["columns"] = schema.names + migrated += 1 + + path.write_text(json.dumps(data, indent=2), encoding="utf-8") + return migrated + + +def _schema_from_ipc_columns(ipc_b64: str) -> list[str]: + """Column names from an existing base64 IPC blob (for idempotent runs).""" + schema = pa.ipc.read_schema(pa.py_buffer(base64.b64decode(ipc_b64))) + return schema.names + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="amplify-db-migrate", + description=( + "Migrate a legacy registry tables.json to the Arrow-IPC schema " + "format in place." + ), + ) + parser.add_argument( + "path", + help="Path to the registry tables.json file to migrate.", + ) + args = parser.parse_args(argv) + + try: + count = migrate_file(args.path) + except FileNotFoundError: + print(f"error: no such file: {args.path}", file=sys.stderr) + return 1 + except ValueError as e: + print(f"error: {e}", file=sys.stderr) + return 1 + + print(f"Migrated {count} table entr{'y' if count == 1 else 'ies'} in {args.path}") + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/src/amplify_db_utils/registry.py b/src/amplify_db_utils/registry.py index e78bb93..9599dd0 100644 --- a/src/amplify_db_utils/registry.py +++ b/src/amplify_db_utils/registry.py @@ -2,66 +2,26 @@ from __future__ import annotations +import base64 import json -import re import pyarrow as pa import pyarrow.fs as pa_fs -def _arrow_type_to_str(t: pa.DataType) -> str: - """Serialize a PyArrow type to a JSON-safe string.""" - return str(t) - - -def _arrow_type_from_str(s: str) -> pa.DataType: - """Deserialize a PyArrow type from its string representation.""" - simple: dict[str, pa.DataType] = { - "string": pa.utf8(), - "utf8": pa.utf8(), - "large_string": pa.large_utf8(), - "large_utf8": pa.large_utf8(), - "int8": pa.int8(), - "int16": pa.int16(), - "int32": pa.int32(), - "int64": pa.int64(), - "uint8": pa.uint8(), - "uint16": pa.uint16(), - "uint32": pa.uint32(), - "uint64": pa.uint64(), - "float": pa.float32(), - "float32": pa.float32(), - "float64": pa.float64(), - "double": pa.float64(), - "halffloat": pa.float16(), - "bool": pa.bool_(), - "date32[day]": pa.date32(), - "date64[ms]": pa.date64(), - } - if s in simple: - return simple[s] - - # Timestamp: "timestamp[us, tz=UTC]" or "timestamp[us]" - m = re.match(r"timestamp\[(\w+)(?:,\s*tz=(.+))?\]", s) - if m: - unit, tz = m.group(1), m.group(2) - return pa.timestamp(unit, tz=tz) - - raise ValueError(f"Cannot deserialize PyArrow type: {s!r}") - - -def _schema_to_json(schema: pa.Schema) -> list[dict]: - return [ - {"name": f.name, "type": _arrow_type_to_str(f.type), "nullable": f.nullable} - for f in schema - ] - - -def _schema_from_json(fields: list[dict]) -> pa.Schema: - return pa.schema([ - pa.field(f["name"], _arrow_type_from_str(f["type"]), nullable=f["nullable"]) - for f in fields - ]) +def _schema_to_ipc_b64(schema: pa.Schema) -> str: + """Serialize a full schema to a base64 Arrow-IPC string. + + Round-trips losslessly for every Arrow type — including nested types like + ``list``, ``large_list``, ``struct``, and ``map`` — plus field nullability + and metadata, with no hand-maintained type table to drift behind PyArrow. + """ + return base64.b64encode(schema.serialize().to_pybytes()).decode("ascii") + + +def _schema_from_ipc_b64(s: str) -> pa.Schema: + """Deserialize a schema from its base64 Arrow-IPC representation.""" + return pa.ipc.read_schema(pa.py_buffer(base64.b64decode(s))) class SchemaRegistry: @@ -87,11 +47,23 @@ def load(cls, fs: pa_fs.FileSystem, fs_root: str) -> "SchemaRegistry": with fs.open_input_stream(registry_path) as f: data = json.loads(f.read().decode("utf-8")) for table_name, entry in data.items(): + if "schema_ipc" not in entry: + if "schema_fields" in entry: + raise ValueError( + f"Registry entry for table '{table_name}' has no " + f"'schema_ipc' key but carries a legacy " + f"'schema_fields' entry — run " + f"'amplify-db-migrate ' to upgrade it." + ) + raise ValueError( + f"Malformed registry: entry for table '{table_name}' " + f"has no 'schema_ipc' key." + ) registry._tables[table_name] = { - "schema": _schema_from_json(entry["schema_fields"]), + "schema": _schema_from_ipc_b64(entry["schema_ipc"]), "partition_by": entry.get("partition_by"), } - except (FileNotFoundError, pa.ArrowIOError, KeyError): + except (FileNotFoundError, pa.ArrowIOError): pass # Empty registry — first use return registry @@ -99,8 +71,13 @@ def save(self, fs: pa_fs.FileSystem, fs_root: str) -> None: """Persist registry to ``{fs_root}/_registry/tables.json``.""" data = {} for table_name, entry in self._tables.items(): + schema: pa.Schema = entry["schema"] data[table_name] = { - "schema_fields": _schema_to_json(entry["schema"]), + # schema_ipc is the only authoritative representation on load. + "schema_ipc": _schema_to_ipc_b64(schema), + # columns is a human-readable convenience listing the column + # names only; it is never read back on load. + "columns": schema.names, "partition_by": entry["partition_by"], } diff --git a/src/amplify_db_utils/schema.py b/src/amplify_db_utils/schema.py index b56bbbc..3329919 100644 --- a/src/amplify_db_utils/schema.py +++ b/src/amplify_db_utils/schema.py @@ -20,13 +20,39 @@ dict: pa.large_utf8(), } +# Scalar types allowed as list element types (dict excluded — list[dict] is not supported). +_LIST_ELEM_TYPES: dict[type, pa.DataType] = { + k: v for k, v in _PYTHON_TO_ARROW.items() if k is not dict +} + def _annotation_to_arrow(annotation: Any) -> tuple[pa.DataType, bool]: """Convert a Python type annotation to ``(arrow_type, nullable)``. Handles ``Optional[T]`` (``Union[T, None]``) and the ``T | None`` pipe syntax (Python 3.10+). Returns ``nullable=True`` for Optional types. + + Handles ``list[T]`` annotations, mapping them to ``pa.list_(arrow_type)``. + Supported element types are the same scalar types as the base mapping + (str, int, float, bool, datetime). Bare ``list`` without an element type + is not supported. """ + origin = get_origin(annotation) + + if origin is list: + args = get_args(annotation) + if not args: + raise TypeError( + "Bare list is not supported. Use list[str], list[int], list[float], etc." + ) + elem_type = args[0] + if elem_type not in _LIST_ELEM_TYPES: + raise TypeError( + f"Unsupported list element type {elem_type!r}. " + f"Supported element types: {list(_LIST_ELEM_TYPES.keys())}" + ) + return pa.list_(_LIST_ELEM_TYPES[elem_type]), False + args = get_args(annotation) if args: @@ -45,7 +71,8 @@ def _annotation_to_arrow(annotation: Any) -> tuple[pa.DataType, bool]: raise TypeError( f"Unsupported field type {annotation!r}. " - f"Supported types: {list(_PYTHON_TO_ARROW.keys())}" + f"Supported types: {list(_PYTHON_TO_ARROW.keys())}. " + f"For lists use list[str], list[int], list[float], etc." ) diff --git a/tests/test_migrate.py b/tests/test_migrate.py new file mode 100644 index 0000000..e27c079 --- /dev/null +++ b/tests/test_migrate.py @@ -0,0 +1,93 @@ +"""Tests for the legacy registry migration script.""" + +from __future__ import annotations + +import json + +import pyarrow as pa +import pyarrow.fs as pa_fs +import pytest + +from amplify_db_utils.migrate import main, migrate_file +from amplify_db_utils.registry import SchemaRegistry + + +def _write_legacy(tmp_path): + """Write a true-legacy tables.json (schema_fields only) and return its path.""" + registry_dir = tmp_path / "_registry" + registry_dir.mkdir() + path = registry_dir / "tables.json" + path.write_text(json.dumps({ + "t": { + "schema_fields": [ + {"name": "id", "type": "string", "nullable": False}, + {"name": "year", "type": "int64", "nullable": False}, + {"name": "ts", "type": "timestamp[us, tz=UTC]", "nullable": True}, + ], + "partition_by": ["year"], + } + })) + return path + + +def test_migrate_file_upgrades_legacy_in_place(tmp_path): + path = _write_legacy(tmp_path) + + count = migrate_file(path) + assert count == 1 + + data = json.loads(path.read_text()) + entry = data["t"] + assert "schema_ipc" in entry + assert "schema_fields" not in entry # legacy form dropped + assert entry["columns"] == ["id", "year", "ts"] + assert entry["partition_by"] == ["year"] + + +def test_migrated_file_loads_with_correct_schema(tmp_path): + path = _write_legacy(tmp_path) + migrate_file(path) + + loaded = SchemaRegistry.load(pa_fs.LocalFileSystem(), str(tmp_path)) + schema, partition_by = loaded.get("t") + + assert schema.field("id").type == pa.utf8() + assert schema.field("year").type == pa.int64() + assert schema.field("ts").type == pa.timestamp("us", tz="UTC") + assert partition_by == ["year"] + + +def test_migrate_is_idempotent(tmp_path): + path = _write_legacy(tmp_path) + + assert migrate_file(path) == 1 + first = path.read_text() + + # Second run finds nothing to migrate and leaves the file unchanged. + assert migrate_file(path) == 0 + assert path.read_text() == first + + +def test_migrate_raises_on_unmigratable_entry(tmp_path): + registry_dir = tmp_path / "_registry" + registry_dir.mkdir() + path = registry_dir / "tables.json" + path.write_text(json.dumps({"t": {"partition_by": None}})) + + with pytest.raises(ValueError, match="neither"): + migrate_file(path) + + +def test_main_cli_returns_zero_and_migrates(tmp_path, capsys): + path = _write_legacy(tmp_path) + + rc = main([str(path)]) + assert rc == 0 + assert "Migrated 1 table entry" in capsys.readouterr().out + assert "schema_ipc" in json.loads(path.read_text())["t"] + + +def test_main_cli_missing_file_returns_one(tmp_path, capsys): + rc = main([str(tmp_path / "nope.json")]) + assert rc == 1 + assert "no such file" in capsys.readouterr().err diff --git a/tests/test_schema.py b/tests/test_schema.py index e3f67cb..8cb1272 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -83,10 +83,98 @@ def test_to_arrow_schema_unsupported_type(): class Bad(BaseModel): x: list # not supported - with pytest.raises(TypeError, match="Unsupported"): + with pytest.raises(TypeError, match="Unsupported|Bare list"): to_arrow_schema(Bad) +def test_to_arrow_schema_bare_list_suggests_typed(): + class Bad(BaseModel): + x: list + + with pytest.raises(TypeError, match="list\\["): + to_arrow_schema(Bad) + + +# --------------------------------------------------------------------------- +# list[T] support +# --------------------------------------------------------------------------- + + +class WithListStr(BaseModel): + tags: list[str] + + +class WithListFloat(BaseModel): + scores: list[float] + + +class WithListInt(BaseModel): + counts: list[int] + + +class WithOptionalList(BaseModel): + image_id: str + embeddings: Optional[list[float]] = None + + +def test_list_str_maps_to_arrow_list(): + schema = to_arrow_schema(WithListStr) + assert schema.field("tags").type == pa.list_(pa.utf8()) + assert not schema.field("tags").nullable + + +def test_list_float_maps_to_arrow_list(): + schema = to_arrow_schema(WithListFloat) + assert schema.field("scores").type == pa.list_(pa.float64()) + + +def test_list_int_maps_to_arrow_list(): + schema = to_arrow_schema(WithListInt) + assert schema.field("counts").type == pa.list_(pa.int64()) + + +def test_optional_list_is_nullable(): + schema = to_arrow_schema(WithOptionalList) + assert schema.field("embeddings").nullable + assert schema.field("embeddings").type == pa.list_(pa.float64()) + + +def test_list_unsupported_element_type(): + class Bad(BaseModel): + x: list[dict] + + with pytest.raises(TypeError, match="Unsupported list element type"): + to_arrow_schema(Bad) + + +def test_validate_list_column_round_trip(): + schema = to_arrow_schema(WithListFloat) + records = [ + {"scores": [0.1, 0.2, 0.9]}, + {"scores": [0.5]}, + ] + table = validate_records(records, schema) + assert table.column("scores")[0].as_py() == [0.1, 0.2, 0.9] + assert table.column("scores")[1].as_py() == [0.5] + + +def test_validate_empty_list(): + schema = to_arrow_schema(WithListFloat) + records = [{"scores": []}] + table = validate_records(records, schema) + assert table.column("scores")[0].as_py() == [] + + +def test_validate_list_not_json_encoded(): + """list[T] columns must NOT be serialized to JSON strings.""" + schema = to_arrow_schema(WithListStr) + records = [{"tags": ["a", "b", "c"]}] + table = validate_records(records, schema) + val = table.column("tags")[0].as_py() + assert isinstance(val, list) + assert val == ["a", "b", "c"] + + def test_to_arrow_schema_not_a_model(): with pytest.raises(TypeError): to_arrow_schema("not a model") diff --git a/tests/test_schema_evolution.py b/tests/test_schema_evolution.py index a1a2b29..a99134e 100644 --- a/tests/test_schema_evolution.py +++ b/tests/test_schema_evolution.py @@ -51,6 +51,15 @@ class ImageRecordChangeType(BaseModel): month: int +class ImageRecordWithEmbedding(BaseModel): + """V1 + a nullable list[float] column (e.g. an embedding vector).""" + image_id: str + instrument: str + year: int + month: int + embedding: Optional[list[float]] = None + + # --------------------------------------------------------------------------- # Idempotency # --------------------------------------------------------------------------- @@ -155,3 +164,135 @@ def test_registry_prevents_breaking_change_after_reload(tmp_path): store2 = DuckDBParquetStore(config) with pytest.raises(ValueError): store2.create_table("t", ImageRecordRemoveColumn, partition_by=["instrument", "year", "month"]) + + +# --------------------------------------------------------------------------- +# list[T] columns across the full db lifecycle +# --------------------------------------------------------------------------- + + +def test_list_column_persists_across_instances(tmp_path): + """A list[T] column survives create → write → reopen → read. + + Regression: before the registry round-trip fix, opening a fresh store on a + directory whose registry held a ``list<…>`` type crashed in + ``SchemaRegistry.load()`` with "Cannot deserialize PyArrow type". + """ + from amplify_db_utils import DuckDBParquetConfig, DuckDBParquetStore + + config = DuckDBParquetConfig(root=str(tmp_path)) + + store1 = DuckDBParquetStore(config) + store1.create_table("t", ImageRecordWithEmbedding, partition_by=["instrument", "year", "month"]) + store1.write("t", [{ + "image_id": "a", + "instrument": "IFCB107", + "year": 2024, + "month": 1, + "embedding": [0.1, 0.2, 0.3], + }]) + + # Fresh instance on the same root must load the list type without crashing. + store2 = DuckDBParquetStore(config) + assert store2.count("t") == 1 + rows = list(store2.read("t", filters={"instrument": "IFCB107", "year": 2024, "month": 1})) + assert len(rows) == 1 + assert rows[0]["embedding"] == [0.1, 0.2, 0.3] + + +def test_list_column_idempotent_after_reload(tmp_path): + """Re-registering a list[T] schema after reload from disk is a no-op.""" + from amplify_db_utils import DuckDBParquetConfig, DuckDBParquetStore + + config = DuckDBParquetConfig(root=str(tmp_path)) + + store1 = DuckDBParquetStore(config) + store1.create_table("t", ImageRecordWithEmbedding, partition_by=["instrument", "year", "month"]) + + store2 = DuckDBParquetStore(config) + # Round-tripped schema must compare equal, so this should not raise. + store2.create_table("t", ImageRecordWithEmbedding, partition_by=["instrument", "year", "month"]) + + +# --------------------------------------------------------------------------- +# SchemaRegistry serialization round-trip (unit level) +# --------------------------------------------------------------------------- + + +def test_registry_roundtrips_list_type(tmp_path): + import pyarrow as pa + import pyarrow.fs as pa_fs + + from amplify_db_utils.registry import SchemaRegistry + + schema = pa.schema([ + pa.field("id", pa.string(), nullable=False), + pa.field("vec", pa.list_(pa.float32()), nullable=True), + ]) + + registry = SchemaRegistry() + registry.register("t", schema, partition_by=None) + + fs = pa_fs.LocalFileSystem() + registry.save(fs, str(tmp_path)) + loaded = SchemaRegistry.load(fs, str(tmp_path)) + + got, _ = loaded.get("t") + assert got == schema + assert got.field("vec").type == pa.list_(pa.float32()) + + +def test_registry_roundtrips_nested_types(tmp_path): + import pyarrow as pa + import pyarrow.fs as pa_fs + + from amplify_db_utils.registry import SchemaRegistry + + schema = pa.schema([ + pa.field("tags", pa.large_list(pa.utf8()), nullable=True), + pa.field("meta", pa.struct([ + pa.field("x", pa.int32()), + pa.field("y", pa.utf8()), + ]), nullable=True), + ]) + + registry = SchemaRegistry() + registry.register("t", schema, partition_by=None) + + fs = pa_fs.LocalFileSystem() + registry.save(fs, str(tmp_path)) + loaded = SchemaRegistry.load(fs, str(tmp_path)) + + got, _ = loaded.get("t") + assert got == schema + + +def test_registry_rejects_legacy_schema_fields_only(tmp_path): + """Legacy registries (schema_fields, no schema_ipc) no longer load. + + IPC is now the only authoritative representation; a legacy sidecar must be + upgraded via ``amplify-db-migrate`` before it will load. + """ + import json + + import pyarrow.fs as pa_fs + import pytest + + from amplify_db_utils.registry import SchemaRegistry + + registry_dir = tmp_path / "_registry" + registry_dir.mkdir() + (registry_dir / "tables.json").write_text(json.dumps({ + "t": { + "schema_fields": [ + {"name": "id", "type": "string", "nullable": False}, + {"name": "year", "type": "int64", "nullable": False}, + {"name": "ts", "type": "timestamp[us, tz=UTC]", "nullable": True}, + ], + "partition_by": ["year"], + } + })) + + fs = pa_fs.LocalFileSystem() + with pytest.raises(ValueError, match="amplify-db-migrate"): + SchemaRegistry.load(fs, str(tmp_path))