fix(client): surface a malformed catalog_contents item as CatalogClientError

contents() decoded the response after leaving the _catalog_connect() block,
so a malformed item (an invalid IPC payload, an unknown enum value) escaped
as a bare ArrowInvalid / KeyError instead of CatalogClientError. load_catalog
then neither fell back to the per-schema RPCs nor raised the client's error
type. Decode inside the block, like every other catalog call.

Found by vgi-polars against vgi-fixture-bad-enum-worker.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: rustyconover

tests/catalog/test_client_load_catalog.py

@@ -12,6 +12,7 @@
 
 from collections.abc import Iterator
 from contextlib import contextmanager
+from types import SimpleNamespace
 from typing import Any
 
 import pytest
@@ -31,7 +32,7 @@
     SchemaInfo,
     TransactionOpaqueData,
 )
-from vgi.client import CatalogClientMixin, CatalogContents, CatalogSnapshot
+from vgi.client import CatalogClientMixin, CatalogContents, CatalogSnapshot, catalog_mixin
 from vgi.client.catalog_mixin import CatalogClientError
 
 _KINDS = (
@@ -333,3 +334,55 @@ def test_absent_supports_catalog_contents_decodes_false() -> None:
     assert CatalogAttachResult.deserialize_from_batch(batch).supports_catalog_contents
     old = batch.drop_columns(["supports_catalog_contents"])
     assert not CatalogAttachResult.deserialize_from_batch(old).supports_catalog_contents
+
+
+class _UndecodableContentsProxy:
+    """A worker proxy whose ``catalog_contents`` item bytes are not a valid ``SchemaInfo``."""
+
+    def catalog_contents(self, **_: Any) -> Any:
+        entry = SimpleNamespace(
+            path=["main"],
+            schema=b"not an arrow ipc stream",
+            tables=[],
+            views=[],
+            scalar_functions=[],
+            aggregate_functions=[],
+            table_functions=[],
+            scalar_macros=[],
+            table_macros=[],
+            indexes=[],
+        )
+        return SimpleNamespace(catalog_version=1, etag=None, not_modified=False, schemas=[entry])
+
+    def catalog_schemas(self, **_: Any) -> Any:
+        return SimpleNamespace(to_infos=list)
+
+
+@pytest.fixture
+def undecodable_contents_client(monkeypatch: pytest.MonkeyPatch) -> CatalogClientMixin:
+    """A client whose real ``_catalog_connect`` hands out `_UndecodableContentsProxy`."""
+
+    @contextmanager
+    def connect(*_: Any, **__: Any) -> Iterator[Any]:
+        yield _UndecodableContentsProxy()
+
+    monkeypatch.setattr(catalog_mixin._catalog_pool, "connect", connect)
+
+    class _Client(CatalogClientMixin):
+        def _worker_argv(self) -> list[str]:
+            return ["unused"]
+
+    return _Client()
+
+
+def test_undecodable_contents_raises_catalog_client_error(undecodable_contents_client: CatalogClientMixin) -> None:
+    """A decode failure surfaces as ``CatalogClientError``, like every other catalog call."""
+    with pytest.raises(CatalogClientError):
+        undecodable_contents_client.contents(attach_opaque_data=_ATTACH)
+
+
+def test_undecodable_contents_falls_back_to_per_schema(undecodable_contents_client: CatalogClientMixin) -> None:
+    """``load_catalog`` treats an undecodable answer as a failure and uses the per-schema RPCs."""
+    snapshot = undecodable_contents_client.load_catalog(attach=_attach_result(True))
+    assert snapshot.source == "per_schema"
+    assert snapshot.fallback_reason

vgi/client/catalog_mixin.py

@@ -541,14 +541,17 @@ def contents(self, *, attach_opaque_data: AttachOpaqueData, if_none_match: str |
             one [`SchemaContentsInfo`][] per schema, parents first.
 
         """
+        # Decode inside the connection block so a malformed item surfaces as
+        # CatalogClientError (which load_catalog falls back on), not a bare
+        # decode error.
         with self._catalog_connect() as proxy:
             response = proxy.catalog_contents(attach_opaque_data=attach_opaque_data, if_none_match=if_none_match)
-        return CatalogContents(
-            catalog_version=response.catalog_version,
-            etag=response.etag,
-            not_modified=response.not_modified,
-            schemas=[_decode_schema_contents(entry) for entry in response.schemas],
-        )
+            return CatalogContents(
+                catalog_version=response.catalog_version,
+                etag=response.etag,
+                not_modified=response.not_modified,
+                schemas=[_decode_schema_contents(entry) for entry in response.schemas],
+            )
 
     def load_catalog(
         self,