Fix catalog wire 2.2 fixture and generator CI checks

Author: rustyconover

tests/catalog/test_catalog_publication.py

@@ -18,7 +18,7 @@
 from vgi_rpc.external import ClientExternalConfig, Compression, ExternalRef
 
 from vgi._test_fixtures.catalog_contents import CATALOG_PROBE, ContentsProbeCatalog, ContentsProbeWorker
-from vgi.catalog import SchemaInfo
+from vgi.catalog import AttachOpaqueData, SchemaInfo
 from vgi.catalog.contents_codec import decode_collection, decode_inventory, encode_collection
 from vgi.catalog.snapshot import (
     CatalogSnapshotManager,
@@ -103,6 +103,11 @@ def _policy(repository: Any, **kwargs: Any) -> CatalogSnapshotPolicy:
     )
 
 
+def _close_policy(policy: CatalogSnapshotPolicy) -> None:
+    assert policy.fetch_config is not None
+    policy.fetch_config.fetch_config.close()
+
+
 @pytest.mark.parametrize("compression", [None, Compression(algorithm="gzip"), Compression()])
 def test_recovery_and_warm_hits_skip_inventory_and_upload(storage: _Storage, compression: Compression | None) -> None:
     """Restart recovery does one validated GET; subsequent hits do no catalog work."""
@@ -116,15 +121,16 @@ def test_recovery_and_warm_hits_skip_inventory_and_upload(storage: _Storage, com
         assert isinstance(ref, ExternalRef)
         assert first.statistics == {"builds": 1, "uploads": 1, "fetches": 1}
         assert first.resolve(request, lambda: pytest.fail("warm build")) == ref
-        assert first.resolve(request, lambda: pytest.fail("conditional build"), "generation-1").not_modified
+        unchanged = first.resolve(request, lambda: pytest.fail("conditional build"), "generation-1")
+        assert isinstance(unchanged, CatalogContentsResponse) and unchanged.not_modified
         assert storage.uploads == 1 and storage.gets == 1
         recovered = CatalogSnapshotManager(policy)
         assert recovered.resolve(request, lambda: pytest.fail("recovery build")) == ref
         assert recovered.statistics == {"builds": 0, "uploads": 0, "fetches": 1}
         assert recovered.resolve(request, lambda: pytest.fail("second recovery build")) == ref
         assert storage.uploads == 1 and storage.gets == 2
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_scopes_singleflight_and_conditional_invalidation(storage: _Storage) -> None:
@@ -150,7 +156,7 @@ def test_scopes_singleflight_and_conditional_invalidation(storage: _Storage) ->
         manager.resolve(request, lambda: response)
         assert storage.uploads == 3
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_singleflight_allows_independent_keys_to_build_concurrently(storage: _Storage) -> None:
@@ -174,7 +180,7 @@ def build() -> CatalogContentsResponse:
             assert all(isinstance(future.result(), ExternalRef) for future in futures)
         assert manager.statistics == {"builds": 2, "uploads": 2, "fetches": 2}
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_failed_readback_never_commits_and_caps_prevent_upload(storage: _Storage) -> None:
@@ -195,7 +201,7 @@ def test_failed_readback_never_commits_and_caps_prevent_upload(storage: _Storage
             limited.resolve(request, lambda: response)
         assert storage.uploads == 1
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_import_validates_nested_catalog_and_contract(storage: _Storage) -> None:
@@ -217,7 +223,7 @@ def test_import_validates_nested_catalog_and_contract(storage: _Storage) -> None
             manager.resolve(other, lambda: malformed)
         assert storage.uploads == 1
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_cheap_validator_avoids_all_publication_work(storage: _Storage) -> None:
@@ -227,10 +233,10 @@ def test_cheap_validator_avoids_all_publication_work(storage: _Storage) -> None:
     manager = CatalogSnapshotManager(policy)
     try:
         result = manager.resolve(_request(), lambda: pytest.fail("build"), "fresh")
-        assert result.not_modified
+        assert isinstance(result, CatalogContentsResponse) and result.not_modified
         assert manager.statistics == {"builds": 0, "uploads": 0, "fetches": 0}
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_worker_auth_precedes_cache_and_fallback_is_explicit(
@@ -263,7 +269,7 @@ def test_worker_auth_precedes_cache_and_fallback_is_explicit(
         assert isinstance(response, CatalogContentsResponse)
         assert decode_inventory(response)
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_worker_fallback_rejects_generation_change(storage: _Storage, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -294,15 +300,17 @@ def changing_contents(*args: Any, **kwargs: Any) -> Any:
             worker.catalog_contents(attach.attach_opaque_data)
         assert storage.uploads == 0
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_collections_consume_multiple_batches_and_reject_wrong_schema() -> None:
     """Later rows and batches survive; legacy schema attachment bytes disappear."""
     records = [
-        SchemaInfo(path=[str(i)], comment=None, tags={"x": str(i)}, attach_opaque_data=b"secret") for i in range(7)
+        SchemaInfo(path=[str(i)], comment=None, tags={"x": str(i)}, attach_opaque_data=AttachOpaqueData(b"secret"))
+        for i in range(7)
     ]
     encoded = encode_collection(SchemaInfo, records, batch_size=2)
+    assert encoded is not None
     assert decode_collection(SchemaInfo, encoded) == records
     assert len(list(pa.ipc.open_stream(encoded))) == 4
     assert b"secret" not in encoded
@@ -337,7 +345,7 @@ def put_object(self, *, Bucket: str, Key: str, Body: bytes, **kwargs: Any) -> No
         self.store.objects["/" + Key] = (Body, kwargs.get("ContentEncoding"))
         self.etags[Key] = str(int(self.etags.get(Key, "0")) + 1)
 
-    def generate_presigned_url(self, operation: str, *, Params: Any, ExpiresIn: int) -> str:
+    def generate_presigned_url(self, operation: str, *, Params: dict[str, str], ExpiresIn: int) -> str:
         del operation, ExpiresIn
         return self.store.base + "/" + Params["Key"]
 
@@ -364,7 +372,7 @@ def test_durable_s3_recovery_and_conditional_manifest_tombstone(storage: _Storag
         assert second.commit_if_absent(manifest) == manifest
         assert second.lookup(key) == manifest
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_s3_renewal_is_persisted_and_preserves_a_concurrent_replacement(
@@ -415,7 +423,7 @@ def concurrent_put(**kwargs: Any) -> None:
         assert repository.renew(renewed) is None
         assert restarted.lookup(key) == replacement
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_subprocess_client_resolves_published_catalog() -> None:
@@ -441,6 +449,7 @@ class Entry(ArrowSerializableDataclass):
         value: Annotated[str, ArrowType(pa.dictionary(pa.int16(), pa.string()))]
 
     encoded = encode_collection(Entry, [Entry("a"), Entry("b"), Entry("c")], batch_size=1)
+    assert encoded is not None
     assert decode_collection(Entry, encoded) == [Entry("a"), Entry("b"), Entry("c")]
     with pytest.raises(ValueError, match="trailing bytes"):
         decode_collection(Entry, encoded + encoded)
@@ -469,7 +478,7 @@ def test_import_clamps_supplied_fetch_limits(storage: _Storage) -> None:
         with pytest.raises((ValueError, RuntimeError), match="decompress|bytes|limit"):
             limited.import_publication(request, replace(manifest, raw_size=999))
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_import_checks_actual_outer_size_digest_and_stream_end(storage: _Storage) -> None:
@@ -503,7 +512,7 @@ def test_import_checks_actual_outer_size_digest_and_stream_end(storage: _Storage
         with pytest.raises(ValueError, match="trailing bytes"):
             CatalogSnapshotManager(policy).import_publication(request, invalid)
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_warm_renewal_rejects_changed_immutable_metadata(storage: _Storage) -> None:
@@ -527,7 +536,7 @@ def renew(self, manifest: Any) -> Any:
         with pytest.raises(ValueError, match="immutable"):
             manager.resolve(request, lambda: pytest.fail("renewal must not rebuild"))
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 def test_generation_change_does_not_commit(storage: _Storage) -> None:
@@ -555,7 +564,7 @@ def build() -> CatalogContentsResponse:
         assert storage.uploads == 0
         assert repository.lookup(key) is None
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 @pytest.mark.parametrize("phase", ["lookup", "readback", "commit"])
@@ -593,7 +602,7 @@ def change_generation(*args: Any, **kwargs: Any) -> Any:
         with pytest.raises(ValueError, match="generation changed"):
             manager.resolve(request, lambda: response, if_none_match=response.etag)
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)
 
 
 @pytest.mark.parametrize("phase", ["readback", "commit"])
@@ -631,4 +640,4 @@ def change_generation(*args: Any, **kwargs: Any) -> Any:
         if phase == "readback":
             assert repository.lookup(manifest.cache_key) is None
     finally:
-        policy.fetch_config.fetch_config.close()
+        _close_policy(policy)

tests/catalog/test_declarative.py

@@ -1234,7 +1234,7 @@ def test_schema_to_schema_info(self) -> None:
         info = schema.to_schema_info(attach_opaque_data)
         assert isinstance(info, SchemaInfo)
         assert info.path == ["main"]
-        assert info.attach_opaque_data == attach_opaque_data
+        assert "attach_opaque_data" not in pa.ipc.open_stream(info.serialize_to_bytes()).schema.names
         assert info.comment == "Main schema"
         assert info.tags == {"type": "core"}
 

tests/catalog/test_function_info_binary_dictionary.py

@@ -1,6 +1,9 @@
 """Binary metadata dictionaries preserve bytes, typed defaults and batch boundaries."""
 
+from __future__ import annotations
+
 from dataclasses import replace
+from typing import Any
 
 import pyarrow as pa
 import pytest
@@ -14,6 +17,12 @@
 _BINARY_DICTIONARY = pa.dictionary(pa.int32(), pa.binary())
 
 
+def _dictionary_column(batch: pa.RecordBatch, name: str) -> pa.DictionaryArray[Any, Any]:
+    column = batch.column(name)
+    assert isinstance(column, pa.DictionaryArray)
+    return column
+
+
 def _function(name: str, arguments: bytes, defaults: pa.RecordBatch | None = None) -> FunctionInfo:
     return FunctionInfo(
         name=name,
@@ -56,7 +65,7 @@ def test_collection_dictionary_replacements_preserve_values_and_nulls() -> None:
     assert encoded is not None
     batches = list(pa.ipc.open_stream(encoded))
     assert len(batches) == 3
-    assert [batch.column("arguments").dictionary.to_pylist() for batch in batches] == [
+    assert [_dictionary_column(batch, "arguments").dictionary.to_pylist() for batch in batches] == [
         [b"a"],
         [b"\xff"],
         [b"a", b""],
@@ -73,16 +82,16 @@ def test_binary_dictionary_deduplicates_equal_content() -> None:
     assert encoded is not None
     batch = pa.ipc.open_stream(encoded).read_next_batch()
     for name in _FIELDS:
-        column = batch.column(name)
+        column = _dictionary_column(batch, name)
         assert len(column.dictionary) == 1
         assert column.indices.to_pylist() == [0] * 20
     plain_schema = pa.schema(
         [field.with_type(pa.binary()) if field.name in _FIELDS else field for field in batch.schema]
     )
     plain = pa.RecordBatch.from_arrays(
         [
-            column.dictionary_decode() if field.name in _FIELDS else column
-            for column, field in zip(batch, batch.schema, strict=True)
+            column.cast(pa.binary()) if field.name in _FIELDS else column
+            for column, field in zip(batch.columns, batch.schema, strict=True)
         ],
         schema=plain_schema,
     )

vgi/_test_fixtures/narrow_bind/worker.py

@@ -168,7 +168,6 @@ def schemas(
         for i, info in enumerate(infos):
             if info.path == ["main"]:
                 infos[i] = SchemaInfo(
-                    attach_opaque_data=info.attach_opaque_data,
                     path=list(info.path),
                     comment=info.comment,
                     tags=info.tags,

vgi/_test_fixtures/projection_repro/worker.py

@@ -381,7 +381,6 @@ def schemas(
         for i, info in enumerate(infos):
             if info.path == ["main"]:
                 infos[i] = SchemaInfo(
-                    attach_opaque_data=info.attach_opaque_data,
                     path=list(info.path),
                     comment=info.comment,
                     tags=info.tags,

vgi/_test_fixtures/schema_reconcile/worker.py

@@ -544,7 +544,6 @@ def schemas(
         for i, info in enumerate(infos):
             if info.path == [_SCHEMA_NAME]:
                 infos[i] = SchemaInfo(
-                    attach_opaque_data=info.attach_opaque_data,
                     path=list(info.path),
                     comment=info.comment,
                     tags=info.tags,

vgi/_test_fixtures/writable/worker.py

@@ -566,7 +566,6 @@ def schemas(
         # default of 1 (unknown / eager-load), which restores correct lookups.
         static_schemas = [
             SchemaInfo(
-                attach_opaque_data=s.attach_opaque_data,
                 path=list(s.path),
                 comment=s.comment,
                 tags=s.tags,

vgi/catalog/catalog_interface.py

@@ -363,6 +363,9 @@ class SchemaInfo(CatalogObject, ArrowSerializableDataclass):
 
     Attributes:
         path: Raw identifier components from the outermost schema to this schema.
+        attach_opaque_data: Deprecated constructor-only compatibility argument.
+            Discarded and never serialized; attachment state belongs to
+            ``CatalogAttachResult`` and subsequent request envelopes.
         estimated_object_count: Approximate population per object kind, keyed by
             the same names the C++ extension uses for its set-cache
             instrumentation: ``"table"``, ``"view"``, ``"scalar_function"``,
@@ -805,9 +808,7 @@ class FunctionInfo(CatalogSchemaObject, ArrowSerializableDataclass):
 
     output_schema: Annotated[SerializedSchema, ArrowType(pa.dictionary(pa.int32(), pa.binary()))]
 
-    parameter_default_values: Annotated[
-        pa.RecordBatch | None, ArrowType(pa.dictionary(pa.int32(), pa.binary()))
-    ] = None
+    parameter_default_values: Annotated[pa.RecordBatch | None, ArrowType(pa.dictionary(pa.int32(), pa.binary()))] = None
 
     stability: FunctionStability | None = None
     null_handling: NullHandling | None = None

vgi/catalog/descriptors.py

@@ -899,6 +899,9 @@ class Schema:
     def to_schema_info(self, attach_opaque_data: AttachOpaqueData) -> SchemaInfo:
         """Convert to [`SchemaInfo`][] for catalog response.
 
+        ``attach_opaque_data`` is accepted for compatibility and discarded;
+        schema metadata does not contain attachment state.
+
         Populates ``estimated_object_count`` from the declared population so
         the C++ extension's eager-load gate can choose between bulk
         ``LoadEntries`` and per-name single-entry RPCs without an extra round
@@ -923,7 +926,6 @@ def to_schema_info(self, attach_opaque_data: AttachOpaqueData) -> SchemaInfo:
         for func in self.functions:
             function_counts[func.get_metadata().function_type] += 1
         return SchemaInfo(
-            attach_opaque_data=attach_opaque_data,
             path=list(self.path),
             comment=self.comment,
             tags=dict(self.tags),

vgi/catalog/snapshot.py

@@ -216,7 +216,7 @@ def __init__(
             raise ValueError("retention must exceed a positive URL lifetime")
         self.client, self.bucket, self.prefix = client, bucket, prefix.rstrip("/") + "/"
         self.profile, self.url_ttl_seconds, self.retention_seconds = profile, url_ttl_seconds, retention_seconds
-        self.storage = self
+        self.storage: ExternalStorage = self
         self._locations: dict[str, PublicationLocation] = {}
         self._lock = threading.Lock()