Replace catalog_output_type() with Meta.output_type for scalar functions

This is a breaking change that simplifies scalar function definitions:
- Add ScalarOutputType type alias for pa.DataType | type[AnyArrow]
- Add output_type to valid Meta attributes in metadata.py
- Add _get_meta_output_type() helper to read output_type from Meta class
- Remove abstract catalog_output_type() classmethod
- Update catalog_output_schema() and output_type property to use Meta
- Update all example functions and tests to use Meta.output_type
- Update CLAUDE.md documentation with new pattern

Before:
  @classmethod
  def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
      return pa.string()

After:
  class Meta:
      output_type = pa.string()

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

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

Author: rustyconover

CLAUDE.md

@@ -113,10 +113,14 @@ vgi-client --input data.parquet --function sum_all_columns --server vgi-example-
 import pyarrow as pa
 import pyarrow.compute as pc
 from vgi import ScalarFunction, Arg
+from vgi.arguments import AnyArrow
 
 class DoubleColumn(ScalarFunction):
     """Double the value in a specified column."""
 
+    class Meta:
+        output_type = AnyArrow  # Output type depends on input column
+
     column = Arg[str](0, doc="Column to double")
 
     @property
@@ -329,17 +333,16 @@ At bind time:
 class AddColumns(ScalarFunction):
     """Add two numeric columns with dynamic output type."""
 
+    class Meta:
+        output_type = AnyArrow  # Output type depends on input columns
+
     col1 = Arg[AnyArrow](0, type_bound=pa.types.is_numeric)
     col2 = Arg[AnyArrow](1, type_bound=pa.types.is_numeric)
 
     def bind(self) -> None:
         """Compute output type from input columns."""
         self._output_type = self.input_schema.field(self.col1.value).type
 
-    @classmethod
-    def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
-        return AnyArrow
-
     @property
     def output_type(self) -> pa.DataType:
         return self._output_type
@@ -375,10 +378,11 @@ For advanced distributed aggregations with `max_workers > 1`, use `store_state()
 
 **ScalarFunction:**
 
-| Method | When to Override | Default |
-|--------|------------------|---------|
+| Method/Attribute | When to Override | Default |
+|------------------|------------------|---------|
+| `Meta.output_type` | Always required (pa.DataType or AnyArrow) | Required |
 | `bind()` | Process input schema, compute dynamic output type | No-op |
-| `output_type` | Define output column type | Required |
+| `output_type` | Override if Meta.output_type is AnyArrow | Uses Meta.output_type |
 | `compute(batch)` | Transform batch to single array | Required |
 | `setup()` | Acquire resources | No-op |
 | `teardown()` | Release resources | No-op |
@@ -409,7 +413,7 @@ How will your function be used in SQL?
 
 1. SELECT my_func(col1, col2) FROM table
    → SCALAR FUNCTION: Returns one value per input row
-   → Use ScalarFunction, override output_type and compute()
+   → Use ScalarFunction, define Meta.output_type and compute()
    → Example: upper(), abs(), concat()
 
 2. SELECT * FROM my_func(args)

tests/scalar/test_function.py

@@ -10,7 +10,7 @@
 
 from tests.conftest import make_scalar_invocation
 from vgi import schema
-from vgi.arguments import AnyArrow, Arg, Arguments
+from vgi.arguments import Arg, Arguments
 from vgi.invocation import Invocation, InvocationType
 from vgi.log import Level, Message
 from vgi.scalar_function import (
@@ -139,11 +139,10 @@ def test_basic_compute(self) -> None:
         """Test basic compute() method."""
 
         class DoubleColumn(ScalarFunction):
-            column = Arg[str](0)
+            class Meta:
+                output_type = pa.int64()
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
-                return pa.int64()
+            column = Arg[str](0)
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 import pyarrow.compute as pc
@@ -177,9 +176,8 @@ def test_log_method(self) -> None:
         """Test self.log() method."""
 
         class LoggingFunc(ScalarFunction):
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
-                return pa.int64()
+            class Meta:
+                output_type = pa.int64()
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 import pyarrow.compute as pc
@@ -211,9 +209,8 @@ def test_row_count_validation(self) -> None:
         """Test that row count mismatch raises error."""
 
         class WrongRowCount(ScalarFunction):
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
-                return pa.int64()
+            class Meta:
+                output_type = pa.int64()
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 # Return wrong number of rows
@@ -238,9 +235,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):
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
-                return pa.int64()
+            class Meta:
+                output_type = pa.int64()
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 # Return MORE rows than input (expanding rows is not allowed)
@@ -267,9 +263,8 @@ def test_empty_batch(self) -> None:
         """Test handling of empty batches."""
 
         class DoubleFunc(ScalarFunction):
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
-                return pa.int64()
+            class Meta:
+                output_type = pa.int64()
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 import pyarrow.compute as pc

tests/test_type_bounds.py

@@ -38,11 +38,10 @@ def test_type_bound_passes_for_valid_type(self) -> None:
         """Type bound validation should pass when predicate returns True."""
 
         class TestFunc(ScalarFunction):
-            col = Arg[AnyArrow](0, type_bound=pa.types.is_integer)
+            class Meta:
+                output_type = pa.int64()
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.int64()
+            col = Arg[AnyArrow](0, type_bound=pa.types.is_integer)
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 return batch.column(self.col.value)
@@ -58,11 +57,10 @@ def test_type_bound_fails_for_invalid_type(self) -> None:
         """Type bound validation should raise when predicate returns False."""
 
         class TestFunc(ScalarFunction):
-            col = Arg[AnyArrow](0, type_bound=pa.types.is_integer)
+            class Meta:
+                output_type = pa.int64()
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.int64()
+            col = Arg[AnyArrow](0, type_bound=pa.types.is_integer)
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 return batch.column(self.col.value)
@@ -78,14 +76,13 @@ def test_multiple_type_bounds_or_logic_passes(self) -> None:
         """When multiple predicates are given, any match should pass (OR logic)."""
 
         class TestFunc(ScalarFunction):
+            class Meta:
+                output_type = pa.float64()
+
             col = Arg[AnyArrow](
                 0, type_bound=[pa.types.is_integer, pa.types.is_floating]
             )
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.float64()
-
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 return batch.column(self.col.value)
 
@@ -100,14 +97,13 @@ def test_multiple_type_bounds_or_logic_fails(self) -> None:
         """When multiple predicates fail, validation should raise."""
 
         class TestFunc(ScalarFunction):
+            class Meta:
+                output_type = pa.float64()
+
             col = Arg[AnyArrow](
                 0, type_bound=[pa.types.is_integer, pa.types.is_floating]
             )
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.float64()
-
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 return batch.column(self.col.value)
 
@@ -123,11 +119,10 @@ def test_type_bound_on_non_anyarrow_warns(self) -> None:
         with pytest.warns(UserWarning, match="only meaningful for Arg\\[AnyArrow\\]"):
 
             class TestFunc(ScalarFunction):
-                col = Arg[str](0, type_bound=pa.types.is_integer)
+                class Meta:
+                    output_type = pa.string()
 
-                @classmethod
-                def catalog_output_type(cls) -> pa.DataType:
-                    return pa.string()
+                col = Arg[str](0, type_bound=pa.types.is_integer)
 
                 def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                     return batch.column(self.col)
@@ -174,11 +169,10 @@ def test_error_message_includes_context(self) -> None:
         """Error messages should include argument name and predicate names."""
 
         class TestFunc(ScalarFunction):
-            my_column = Arg[AnyArrow](0, type_bound=pa.types.is_integer)
+            class Meta:
+                output_type = pa.int64()
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.int64()
+            my_column = Arg[AnyArrow](0, type_bound=pa.types.is_integer)
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 return batch.column(self.my_column.value)
@@ -205,11 +199,10 @@ def is_large_int(dtype: pa.DataType) -> bool:
             return dtype in (pa.int64(), pa.uint64())
 
         class TestFunc(ScalarFunction):
-            col = Arg[AnyArrow](0, type_bound=is_large_int)
+            class Meta:
+                output_type = pa.int64()
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.int64()
+            col = Arg[AnyArrow](0, type_bound=is_large_int)
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 return batch.column(self.col.value)
@@ -230,11 +223,10 @@ def test_type_bound_with_lambda(self) -> None:
         """Lambda predicates should work and show in error messages."""
 
         class TestFunc(ScalarFunction):
-            col = Arg[AnyArrow](0, type_bound=lambda t: pa.types.is_integer(t))
+            class Meta:
+                output_type = pa.int64()
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.int64()
+            col = Arg[AnyArrow](0, type_bound=lambda t: pa.types.is_integer(t))
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 return batch.column(self.col.value)
@@ -261,11 +253,10 @@ def test_varargs_type_bound_passes_for_all_valid_types(self) -> None:
         """Type bound validation passes when all varargs elements are valid."""
 
         class TestFunc(ScalarFunction):
-            columns = Arg[AnyArrow](0, varargs=True, type_bound=pa.types.is_integer)
+            class Meta:
+                output_type = pa.int64()
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.int64()
+            columns = Arg[AnyArrow](0, varargs=True, type_bound=pa.types.is_integer)
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 # Sum all integer columns (varargs returns tuple at runtime)
@@ -288,11 +279,10 @@ def test_varargs_type_bound_fails_when_any_element_invalid(self) -> None:
         """Type bound validation should fail if any varargs element has invalid type."""
 
         class TestFunc(ScalarFunction):
-            columns = Arg[AnyArrow](0, varargs=True, type_bound=pa.types.is_integer)
+            class Meta:
+                output_type = pa.int64()
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.int64()
+            columns = Arg[AnyArrow](0, varargs=True, type_bound=pa.types.is_integer)
 
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 result = batch.column(self.columns[0])  # type: ignore[index]
@@ -313,14 +303,13 @@ def test_varargs_type_bound_with_multiple_predicates(self) -> None:
         """Varargs with multiple type bounds should use OR logic per element."""
 
         class TestFunc(ScalarFunction):
+            class Meta:
+                output_type = pa.float64()
+
             columns = Arg[AnyArrow](
                 0, varargs=True, type_bound=[pa.types.is_integer, pa.types.is_floating]
             )
 
-            @classmethod
-            def catalog_output_type(cls) -> pa.DataType:
-                return pa.float64()
-
             def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
                 return batch.column(self.columns[0])  # type: ignore[index]
 

vgi/examples/scalar.py

@@ -45,6 +45,7 @@ class Meta:
 
         name = "double_column"
         description = "Doubles values in a numeric column"
+        output_type = AnyArrow  # Output type depends on input column type
         examples = [
             FunctionExample(
                 sql="SELECT double_column(price) FROM products",
@@ -59,11 +60,6 @@ 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."""
@@ -134,6 +130,7 @@ class Meta:
 
         name = "add_columns"
         description = "Adds two numeric columns"
+        output_type = AnyArrow  # Output type depends on input column types
         examples = [
             FunctionExample(
                 sql="SELECT add_columns(price, tax) FROM orders",
@@ -165,11 +162,6 @@ def bind(self) -> None:
         ).type
         self._output_type = _promote_for_addition(common_type)
 
-    @classmethod
-    def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
-        """Output type depends on input column types."""
-        return AnyArrow
-
     @property
     def output_type(self) -> pa.DataType:
         """Return the computed output type based on input column types."""
@@ -195,6 +187,7 @@ class Meta:
 
         name = "upper_case"
         description = "Converts string column to uppercase"
+        output_type = pa.string()  # Static output type
         examples = [
             FunctionExample(
                 sql="SELECT upper_case(name) FROM users",
@@ -208,12 +201,7 @@ class Meta:
 
     column = Arg[str](0, doc="Column name to uppercase")
 
-    @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()
+    # Note: No need to override output_type - default uses Meta.output_type
 
     def compute(self, batch: pa.RecordBatch) -> pa.Array[Any]:
         """Convert the column values to uppercase."""
@@ -238,6 +226,7 @@ class Meta:
 
         name = "sum_columns"
         description = "Sum values from multiple numeric columns"
+        output_type = AnyArrow  # Output type depends on input column types
         examples = [
             FunctionExample(
                 sql="SELECT sum_columns(price, tax, shipping) FROM orders",
@@ -262,11 +251,6 @@ def bind(self) -> None:
         first_type = self.input_schema.field(first_col).type
         self._output_type = _promote_for_addition(first_type)
 
-    @classmethod
-    def catalog_output_type(cls) -> pa.DataType | type[AnyArrow]:
-        """Output type depends on input column types."""
-        return AnyArrow
-
     @property
     def output_type(self) -> pa.DataType:
         """Return the computed output type based on first column."""

vgi/metadata.py

@@ -667,6 +667,7 @@ def _normalize_examples(
         "distinct_dependent",
         # Scalar function specific
         "return_type",
+        "output_type",  # pa.DataType | type[AnyArrow] for scalar functions
     }
 )