Add catalog_output_type/catalog_output_schema to ScalarFunction

Scalar functions now expose their output type for catalog introspection:
- Added abstract classmethod catalog_output_type() returning pa.DataType or AnyArrow
- Added final classmethod catalog_output_schema() wrapping type in schema
- Default output_type property uses catalog_output_type() (DRY)
- Functions with AnyArrow output produce schema with pa.null() + vgi:any metadata
- CLI shows output_schema only for scalar functions, displays "any" for dynamic types

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

Author: rustyconover

tests/catalog/test_example_worker_catalog.py

@@ -169,3 +169,68 @@ def test_function_info_schema_name(self) -> None:
         # All functions should be in 'main' schema
         for item in contents:
             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 = _get_functions(contents)
+
+        # Create lookup by name
+        by_name = {fn.name: fn for fn in functions}
+
+        # upper_case has static output type (string)
+        upper_info = by_name["upper_case"]
+        output_schema = pa.ipc.read_schema(pa.py_buffer(upper_info.output_schema))
+
+        # Should have a single column named "result" with string type
+        assert len(output_schema) == 1
+        assert output_schema.field(0).name == "result"
+        assert output_schema.field(0).type == pa.string()
+
+    def test_scalar_function_with_dynamic_output_has_any_type(self) -> None:
+        """Scalar functions with AnyArrow output type have 'any' output_schema."""
+        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)
+
+        # Create lookup by name
+        by_name = {fn.name: fn for fn in functions}
+
+        # double_column returns AnyArrow (output depends on input)
+        double_info = by_name["double_column"]
+        output_schema = pa.ipc.read_schema(pa.py_buffer(double_info.output_schema))
+
+        # Should have a single "result" field with null type and vgi:any metadata
+        assert len(output_schema) == 1
+        assert output_schema.field(0).name == "result"
+        assert output_schema.field(0).type == pa.null()
+        assert output_schema.field(0).metadata == {b"vgi:any": b"true"}
+
+    def test_table_function_has_empty_output_schema(self) -> None:
+        """Table functions have empty output_schema (can't determine without input)."""
+        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)
+
+        # Create lookup by name
+        by_name = {fn.name: fn for fn in functions}
+
+        # echo is a table function
+        echo_info = by_name["echo"]
+        output_schema = pa.ipc.read_schema(pa.py_buffer(echo_info.output_schema))
+
+        # Table functions don't have catalog_output_schema, so it's empty
+        assert len(output_schema) == 0

tests/scalar/test_function.py

@@ -9,7 +9,7 @@
 import structlog
 
 from tests.conftest import make_scalar_invocation
-from vgi.arguments import Arg, Arguments
+from vgi.arguments import AnyArrow, Arg, Arguments
 from vgi.exceptions import SchemaValidationError
 from vgi.invocation import Invocation, InvocationType
 from vgi.log import Level, Message
@@ -158,8 +158,8 @@ def test_basic_compute(self) -> None:
         class DoubleColumn(ScalarFunction):
             column = Arg[str](0)
 
-            @property
-            def output_type(self) -> pa.DataType:
+            @classmethod
+            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
                 return pa.int64()
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
@@ -194,8 +194,8 @@ def test_log_method(self) -> None:
         """Test self.log() method."""
 
         class LoggingFunc(ScalarFunction):
-            @property
-            def output_type(self) -> pa.DataType:
+            @classmethod
+            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
                 return pa.int64()
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
@@ -228,8 +228,8 @@ def test_row_count_validation(self) -> None:
         """Test that row count mismatch raises error."""
 
         class WrongRowCount(ScalarFunction):
-            @property
-            def output_type(self) -> pa.DataType:
+            @classmethod
+            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
                 return pa.int64()
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
@@ -255,8 +255,8 @@ def test_row_count_exceeds_input(self) -> None:
         """Test that output with more rows than input raises error (lines 134-142)."""
 
         class TooManyRows(ScalarFunction):
-            @property
-            def output_type(self) -> pa.DataType:
+            @classmethod
+            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
                 return pa.int64()
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
@@ -284,8 +284,8 @@ def test_empty_batch(self) -> None:
         """Test handling of empty batches."""
 
         class DoubleFunc(ScalarFunction):
-            @property
-            def output_type(self) -> pa.DataType:
+            @classmethod
+            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
                 return pa.int64()
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:

vgi/catalog/catalog_interface.py

@@ -1043,8 +1043,12 @@ def _function_to_info(self, func_cls: type, schema_name: str) -> FunctionInfo:
         args_schema = argument_specs_to_schema(arg_specs)
         args_bytes = SerializedSchema(args_schema.serialize().to_pybytes())
 
-        # Output schema placeholder (not available without instantiation)
-        output_schema = pa.schema([])
+        # Get output schema from catalog introspection methods if available
+        output_schema: pa.Schema = pa.schema([])
+        has_catalog_schema = hasattr(func_cls, "catalog_output_schema")
+        if func_type == FunctionType.SCALAR and has_catalog_schema:
+            # ScalarFunction has catalog_output_schema() classmethod
+            output_schema = func_cls.catalog_output_schema()  # type: ignore[attr-defined]
         output_bytes = SerializedSchema(output_schema.serialize().to_pybytes())
 
         return FunctionInfo(

vgi/client/cli_utils.py

@@ -177,7 +177,20 @@ def arrow_schema_to_json(serialized: bytes) -> list[dict[str, str]]:
     """
     reader = pa.BufferReader(serialized)
     schema = pa.ipc.read_schema(reader)  # type: ignore[arg-type]
-    return [{"name": f.name, "type": str(f.type)} for f in schema]
+    result = []
+    for f in schema:
+        type_str = str(f.type)
+        if f.metadata:
+            # Check for vgi:any metadata (output schema)
+            if f.metadata.get(b"vgi:any") == b"true":
+                type_str = "any"
+            # Check for vgi_type metadata (argument schema)
+            elif f.metadata.get(b"vgi_type") == b"table":
+                type_str = "table"
+            elif f.metadata.get(b"vgi_type") == b"any":
+                type_str = "any"
+        result.append({"name": f.name, "type": type_str})
+    return result
 
 
 def output_json(data: Any) -> None:
@@ -279,14 +292,18 @@ def function_info_to_dict(function_info: Any) -> dict[str, Any]:
         Dictionary representation
 
     """
-    return {
+    result: dict[str, Any] = {
         "name": function_info.name,
         "schema_name": function_info.schema_name,
         "function_type": function_info.function_type.value,
         "arguments": arrow_schema_to_json(function_info.arguments),
         "comment": function_info.comment,
         "tags": function_info.tags,
     }
+    # Only include output_schema for scalar functions
+    if function_info.function_type.value == "scalar":
+        result["output_schema"] = arrow_schema_to_json(function_info.output_schema)
+    return result
 
 
 def catalog_attach_result_to_dict(result: Any) -> dict[str, Any]:

vgi/examples/scalar.py

@@ -17,7 +17,7 @@
 import pyarrow as pa
 import pyarrow.compute as pc
 
-from vgi.arguments import Arg
+from vgi.arguments import AnyArrow, Arg
 from vgi.scalar_function import ScalarFunction
 
 __all__ = [
@@ -46,6 +46,11 @@ class Meta:
     # Explicit arrow_type demonstrates type specification
     column = Arg[str](0, doc="Column name to double", arrow_type=pa.utf8())
 
+    @classmethod
+    def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
+        """Output type depends on input column type."""
+        return AnyArrow
+
     @property
     def output_type(self) -> pa.DataType:
         """Return the type of the doubled column."""
@@ -75,6 +80,11 @@ class Meta:
     col1 = Arg[str](0, doc="First column name")
     col2 = Arg[str](1, doc="Second column name")
 
+    @classmethod
+    def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
+        """Output type depends on input column type."""
+        return AnyArrow
+
     @property
     def output_type(self) -> pa.DataType:
         """Return the type of the first column."""
@@ -103,11 +113,13 @@ class Meta:
 
     column = Arg[str](0, doc="Column name to uppercase")
 
-    @property
-    def output_type(self) -> pa.DataType:
-        """Return string type."""
+    @classmethod
+    def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
+        """Return string type (static output)."""
         return pa.string()
 
+    # Note: No need to override output_type - default uses catalog_output_type()
+
     def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
         """Convert the column values to uppercase."""
         return pc.utf8_upper(batch.column(self.column))  # type: ignore[no-matching-overload]