Improve catalog streaming and refactor CatalogClientMixin

Catalog streaming improvements:
- Add to_row_dict() methods to SchemaInfo, TableInfo, ViewInfo, FunctionInfo
- Use RecordBatchStreamWriter in worker for list-returning methods
- Produces 1 schema + 1 batch (N rows) + 1 EOS instead of N complete streams

Type fixes:
- Change overload return types from list to Sequence (covariant)
- Fix tags parameter type: set[str] -> dict[str, str]
- Fix list_schemas() generator to return actual list

CatalogClientMixin refactoring:
- Extract shared subprocess logic into _send_catalog_invocation helper
- Remove duplicate transaction_begin/commit/rollback methods
- Remove unused TransactionBeginResult class
- Improve docstrings with complete examples and error handling docs
- Update cli_transaction.py to use catalog_transaction_* methods

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: rustyconover

tests/catalog/test_catalog_interface.py

@@ -1,6 +1,6 @@
 """Tests for CatalogInterface ABC and default implementations."""
 
-from collections.abc import Callable, Iterable
+from collections.abc import Callable
 from typing import Any
 
 import pyarrow as pa
@@ -61,7 +61,7 @@ def readonly_catalog() -> "MinimalReadOnlyCatalog":
 class MinimalCatalog(CatalogInterface):
     """Minimal implementation for testing abstract method requirements."""
 
-    def catalogs(self) -> Iterable[str]:
+    def catalogs(self) -> list[str]:
         """Return list of catalogs."""
         return ["test"]
 
@@ -227,7 +227,10 @@ def _not_implemented_test_cases() -> list[
             "schema_contents",
             "Schema contents not implemented",
             lambda c: c.schema_contents(
-                attach_id=TEST_ATTACH_ID, transaction_id=None, name="main"
+                attach_id=TEST_ATTACH_ID,
+                transaction_id=None,
+                name="main",
+                type=SchemaObjectType.TABLE,
             ),
         ),
         (
@@ -283,7 +286,7 @@ def test_not_implemented(
 class MinimalReadOnlyCatalog(ReadOnlyCatalogInterface):
     """Minimal read-only implementation for testing."""
 
-    def catalogs(self) -> Iterable[str]:
+    def catalogs(self) -> list[str]:
         """Return list of catalogs."""
         return ["readonly"]
 
@@ -774,20 +777,36 @@ class TestCatalog(ReadOnlyCatalogInterface):
 
         return TestCatalog()
 
-    def test_no_filter_returns_all(
+    def test_fetch_all_function_types(
         self, catalog_with_functions: ReadOnlyCatalogInterface
     ) -> None:
-        """schema_contents with no type filter returns all functions."""
+        """Can fetch both scalar and table functions with separate calls."""
         attach_result = catalog_with_functions.catalog_attach(name="test", options={})
-        contents = list(
+
+        # Get scalar functions
+        scalar_contents = list(
+            catalog_with_functions.schema_contents(
+                attach_id=attach_result.attach_id,
+                transaction_id=None,
+                name="main",
+                type=SchemaObjectType.SCALAR_FUNCTION,
+            )
+        )
+
+        # Get table functions
+        table_contents = list(
             catalog_with_functions.schema_contents(
                 attach_id=attach_result.attach_id,
                 transaction_id=None,
                 name="main",
+                type=SchemaObjectType.TABLE_FUNCTION,
             )
         )
-        assert len(contents) == 2
-        names = {obj.name for obj in contents}
+
+        # Combined should have both functions
+        all_contents = scalar_contents + table_contents
+        assert len(all_contents) == 2
+        names = {obj.name for obj in all_contents}
         # Names are derived from class names: MyScalarFunction -> my_scalar
         assert "my_scalar" in names
         assert "my_table" in names
@@ -870,21 +889,7 @@ def test_wrong_schema_returns_empty(
                 attach_id=attach_result.attach_id,
                 transaction_id=None,
                 name="nonexistent",
+                type=SchemaObjectType.TABLE_FUNCTION,
             )
         )
         assert len(contents) == 0
-
-    def test_type_filter_with_none_explicitly(
-        self, catalog_with_functions: ReadOnlyCatalogInterface
-    ) -> None:
-        """schema_contents with type=None explicitly returns all functions."""
-        attach_result = catalog_with_functions.catalog_attach(name="test", options={})
-        contents = list(
-            catalog_with_functions.schema_contents(
-                attach_id=attach_result.attach_id,
-                transaction_id=None,
-                name="main",
-                type=None,
-            )
-        )
-        assert len(contents) == 2

tests/catalog/test_example_worker_catalog.py

@@ -6,7 +6,14 @@
 
 import pyarrow as pa
 
-from vgi.catalog import FunctionInfo, FunctionType, TableInfo, ViewInfo
+from vgi.catalog import (
+    AttachId,
+    FunctionInfo,
+    FunctionType,
+    SchemaObjectType,
+    TableInfo,
+    ViewInfo,
+)
 from vgi.client import Client
 from vgi.examples.worker import ExampleWorker
 
@@ -30,6 +37,26 @@ def _get_functions(
     return [item for item in contents if isinstance(item, FunctionInfo)]
 
 
+def _get_all_functions(client: Client, attach_id: AttachId) -> list[FunctionInfo]:
+    """Get both table and scalar functions from the catalog."""
+    table_funcs = list(
+        client.schema_contents(
+            attach_id=attach_id,
+            name="main",
+            type=SchemaObjectType.TABLE_FUNCTION,
+        )
+    )
+    scalar_funcs = list(
+        client.schema_contents(
+            attach_id=attach_id,
+            name="main",
+            type=SchemaObjectType.SCALAR_FUNCTION,
+        )
+    )
+    # Combine to list - the overloads guarantee FunctionInfo for function types
+    return list(table_funcs) + list(scalar_funcs)
+
+
 class TestExampleWorkerCatalog:
     """Test ExampleWorker's catalog interface."""
 
@@ -49,15 +76,19 @@ def test_catalog_attach_works(self) -> None:
         assert result.catalog_version_frozen is True
 
     def test_schema_contents_returns_functions(self) -> None:
-        """schema_contents() returns FunctionInfo for all example functions."""
+        """schema_contents() returns FunctionInfo for table functions."""
         client = Client(EXAMPLE_WORKER)
 
         # Attach to catalog
         attach_result = client.catalog_attach(name="example", options={})
         attach_id = attach_result.attach_id
 
-        # Get schema contents
-        contents = list(client.schema_contents(attach_id=attach_id, name="main"))
+        # Get table functions
+        contents = list(
+            client.schema_contents(
+                attach_id=attach_id, name="main", type=SchemaObjectType.TABLE_FUNCTION
+            )
+        )
 
         # Should have functions
         assert len(contents) > 0
@@ -70,12 +101,30 @@ def test_all_example_functions_listed(self) -> None:
         """All example functions are listed in the catalog."""
         client = Client(EXAMPLE_WORKER)
 
-        # Attach and get contents
+        # Attach and get both table and scalar functions
         attach_result = client.catalog_attach(name="example", options={})
-        contents = list(
-            client.schema_contents(attach_id=attach_result.attach_id, name="main")
+
+        # Get table functions
+        table_funcs = list(
+            client.schema_contents(
+                attach_id=attach_result.attach_id,
+                name="main",
+                type=SchemaObjectType.TABLE_FUNCTION,
+            )
         )
 
+        # Get scalar functions
+        scalar_funcs = list(
+            client.schema_contents(
+                attach_id=attach_result.attach_id,
+                name="main",
+                type=SchemaObjectType.SCALAR_FUNCTION,
+            )
+        )
+
+        # Combine all functions
+        contents = table_funcs + scalar_funcs
+
         # Get function names
         function_names = {item.name for item in contents}
 
@@ -95,10 +144,7 @@ def test_function_info_has_correct_types(self) -> None:
         client = Client(EXAMPLE_WORKER)
 
         attach_result = client.catalog_attach(name="example", options={})
-        contents = list(
-            client.schema_contents(attach_id=attach_result.attach_id, name="main")
-        )
-        functions = _get_functions(contents)
+        functions = _get_all_functions(client, attach_result.attach_id)
 
         # Create lookup by name
         by_name = {fn.name: fn for fn in functions}
@@ -117,10 +163,7 @@ def test_function_info_has_arguments(self) -> None:
         client = Client(EXAMPLE_WORKER)
 
         attach_result = client.catalog_attach(name="example", options={})
-        contents = list(
-            client.schema_contents(attach_id=attach_result.attach_id, name="main")
-        )
-        functions = _get_functions(contents)
+        functions = _get_all_functions(client, attach_result.attach_id)
 
         # Create lookup by name
         by_name = {fn.name: fn for fn in functions}
@@ -142,10 +185,13 @@ def test_function_info_has_description(self) -> None:
         client = Client(EXAMPLE_WORKER)
 
         attach_result = client.catalog_attach(name="example", options={})
-        contents = list(
-            client.schema_contents(attach_id=attach_result.attach_id, name="main")
+        functions = list(
+            client.schema_contents(
+                attach_id=attach_result.attach_id,
+                name="main",
+                type=SchemaObjectType.TABLE_FUNCTION,
+            )
         )
-        functions = _get_functions(contents)
 
         # Create lookup by name
         by_name = {fn.name: fn for fn in functions}
@@ -160,23 +206,24 @@ def test_function_info_schema_name(self) -> None:
         client = Client(EXAMPLE_WORKER)
 
         attach_result = client.catalog_attach(name="example", options={})
-        contents = list(
-            client.schema_contents(attach_id=attach_result.attach_id, name="main")
-        )
+        functions = _get_all_functions(client, attach_result.attach_id)
 
         # All functions should be in 'main' schema
-        for item in contents:
+        for item in functions:
             assert item.schema_name == "main"
 
     def test_scalar_function_has_output_schema(self) -> None:
         """Scalar functions with static output types have output_schema populated."""
         client = Client(EXAMPLE_WORKER)
 
         attach_result = client.catalog_attach(name="example", options={})
-        contents = list(
-            client.schema_contents(attach_id=attach_result.attach_id, name="main")
+        functions = list(
+            client.schema_contents(
+                attach_id=attach_result.attach_id,
+                name="main",
+                type=SchemaObjectType.SCALAR_FUNCTION,
+            )
         )
-        functions = _get_functions(contents)
 
         # Create lookup by name
         by_name = {fn.name: fn for fn in functions}
@@ -195,10 +242,13 @@ def test_scalar_function_with_dynamic_output_has_any_type(self) -> None:
         client = Client(EXAMPLE_WORKER)
 
         attach_result = client.catalog_attach(name="example", options={})
-        contents = list(
-            client.schema_contents(attach_id=attach_result.attach_id, name="main")
+        functions = list(
+            client.schema_contents(
+                attach_id=attach_result.attach_id,
+                name="main",
+                type=SchemaObjectType.SCALAR_FUNCTION,
+            )
         )
-        functions = _get_functions(contents)
 
         # Create lookup by name
         by_name = {fn.name: fn for fn in functions}
@@ -218,10 +268,13 @@ def test_table_function_has_empty_output_schema(self) -> None:
         client = Client(EXAMPLE_WORKER)
 
         attach_result = client.catalog_attach(name="example", options={})
-        contents = list(
-            client.schema_contents(attach_id=attach_result.attach_id, name="main")
+        functions = list(
+            client.schema_contents(
+                attach_id=attach_result.attach_id,
+                name="main",
+                type=SchemaObjectType.TABLE_FUNCTION,
+            )
         )
-        functions = _get_functions(contents)
 
         # Create lookup by name
         by_name = {fn.name: fn for fn in functions}

tests/catalog/test_integration.py

@@ -476,7 +476,9 @@ class TestInMemoryCatalogSchemaContents:
     """Test schema_contents operation."""
 
     def test_schema_contents_lists_tables_and_views(self) -> None:
-        """schema_contents returns tables and views."""
+        """schema_contents returns tables and views when queried by type."""
+        from vgi.catalog import SchemaObjectType
+
         catalog = InMemoryCatalog()
         result = catalog.catalog_attach(name="memory", options={})
 
@@ -502,15 +504,30 @@ def test_schema_contents_lists_tables_and_views(self) -> None:
             on_conflict=OnConflict.ERROR,
         )
 
-        contents = list(
+        # Get tables
+        tables = list(
             catalog.schema_contents(
-                attach_id=result.attach_id, transaction_id=None, name="main"
+                attach_id=result.attach_id,
+                transaction_id=None,
+                name="main",
+                type=SchemaObjectType.TABLE,
+            )
+        )
+
+        # Get views
+        views = list(
+            catalog.schema_contents(
+                attach_id=result.attach_id,
+                transaction_id=None,
+                name="main",
+                type=SchemaObjectType.VIEW,
             )
         )
 
-        names = [c.name for c in contents]
-        assert "users" in names
-        assert "user_view" in names
+        table_names = [t.name for t in tables]
+        view_names = [v.name for v in views]
+        assert "users" in table_names
+        assert "user_view" in view_names
 
 
 class TestInMemoryCatalogOnConflict:

tests/ci/test_catalog_ddl.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from vgi.catalog import AttachId, OnConflict, SerializedSchema
+from vgi.catalog import AttachId, OnConflict, SchemaObjectType, SerializedSchema
 from vgi.ci.catalog import CICatalog
 
 
@@ -188,7 +188,10 @@ def test_contents_empty(self, attached_catalog: tuple[CICatalog, AttachId]) -> N
 
         contents = list(
             catalog.schema_contents(
-                attach_id=attach_id, transaction_id=None, name="main"
+                attach_id=attach_id,
+                transaction_id=None,
+                name="main",
+                type=SchemaObjectType.TABLE,
             )
         )
         assert contents == []
@@ -214,7 +217,10 @@ def test_contents_with_table(
 
         contents = list(
             catalog.schema_contents(
-                attach_id=attach_id, transaction_id=None, name="main"
+                attach_id=attach_id,
+                transaction_id=None,
+                name="main",
+                type=SchemaObjectType.TABLE,
             )
         )
         assert len(contents) == 1