support list[T] columns

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.
-

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]

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())

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,20 +47,37 @@ 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 <path>' 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
 
     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"],
             }
 

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."
     )