Add Polars functions to example worker and fix scalar finalization

- Add all Polars scalar functions to ExampleWorker for vgi-client testing
- Fix scalar function worker to send FINISHED status after finalize
  (was hanging because client expected a response)
- Add position parameter to ConstParam for PolarsScalarFunction support
- Update PolarsMultiplyFunction to use proper ConstParam declaration
- Extract ConstParam from _polars_params in metadata extraction

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

Author: rustyconover

vgi/arguments.py

@@ -1588,6 +1588,8 @@ class ConstParam:
         doc: Documentation string describing this parameter.
         arrow_type: Optional explicit Arrow type. If not provided, type is
             inferred from the Annotated first argument.
+        position: Position in the argument list (required for PolarsScalarFunction,
+            optional for ScalarFunction where position is inferred from signature).
 
     Example:
         class FormatNumber(ScalarFunction):
@@ -1600,10 +1602,20 @@ def compute(
                 fmt = f"%.{precision}f"
                 return pa.array([fmt % v for v in value.to_pylist()])
 
+    Example for PolarsScalarFunction:
+        class Multiply(PolarsScalarFunction):
+            value: Annotated[pl.Float64, Param(position=0, doc="Value to multiply")]
+            factor: Annotated[float, ConstParam("Factor", position=0)]
+
+            def compute_polars(self) -> pl.Expr:
+                return pl.col("value") * self.factor
+
     """
 
     doc: str = ""
     arrow_type: pa.DataType | type | None = None
+    # Position in the argument list (for PolarsScalarFunction class-level attributes)
+    position: int | None = None
 
 
 @dataclass(frozen=True, slots=True)

vgi/examples/scalar_polars.py

@@ -62,7 +62,7 @@ def compute_polars(self) -> pl.Expr:
 import polars as pl
 import pyarrow.types as pat
 
-from vgi.arguments import Param
+from vgi.arguments import ConstParam, Param
 from vgi.metadata import FunctionExample
 from vgi.scalar_function_polars import AnyPolars, PolarsScalarFunction
 
@@ -253,17 +253,15 @@ class PolarsMultiplyFunction(PolarsScalarFunction):
     """Multiply a numeric column by a constant factor using Polars.
 
     Demonstrates combining a column parameter with a constant (scalar) argument.
-    The constant factor is passed via SQL and accessed through
-    ``self.invocation.arguments.positional``.
+    Use ``ConstParam`` to declare constant arguments that appear in function
+    signatures and metadata.
 
     This pattern is useful when you need a user-specified value that doesn't
     come from a table column (e.g., scaling factor, threshold, limit).
 
     Attributes:
-        value: Numeric column to multiply (position 0).
-
-    Properties:
-        factor: The constant multiplication factor from arguments.
+        value: Numeric column to multiply (position 0 in input batch).
+        factor: Constant multiplication factor (position 0 in function arguments).
 
     SQL Usage:
         ``SELECT polars_multiply(price, 1.1) FROM products``
@@ -272,9 +270,15 @@ class PolarsMultiplyFunction(PolarsScalarFunction):
         >>> input: value=[10.0, 20.0, 30.0], factor=2.0
         >>> output: [20.0, 40.0, 60.0]
 
+    Note:
+        The ``factor`` is a constant argument passed from SQL, not a table column.
+        It's declared with ``ConstParam(position=0)`` to indicate it's the first
+        positional argument in the function call.
+
     """
 
     value: Annotated[pl.Float64, Param(position=0, doc="Value to multiply")]
+    factor: Annotated[float, ConstParam("Multiplication factor", position=0)]
 
     class Meta:
         """Function metadata."""
@@ -290,13 +294,13 @@ class Meta:
         ]
 
     @property
-    def factor(self) -> float:
+    def _factor(self) -> float:
         """Get the multiplication factor from arguments."""
         return self.invocation.arguments.positional[0].as_py()  # type: ignore
 
     def compute_polars(self) -> pl.Expr:
         """Multiply the values by the constant factor."""
-        return pl.col("value") * self.factor
+        return pl.col("value") * self._factor
 
 
 # =============================================================================

vgi/examples/worker.py

@@ -29,6 +29,15 @@
     SumValuesFunction,
     UpperCaseFunction,
 )
+from vgi.examples.scalar_polars import (
+    PolarsAddValuesFunction,
+    PolarsDoubleFunction,
+    PolarsMultiplyFunction,
+    PolarsNormalizeFunction,
+    PolarsStringLengthFunction,
+    PolarsSumValuesFunction,
+    PolarsUpperCaseFunction,
+)
 from vgi.examples.table import (
     ConstantColumnsFunction,
     DoubleSequenceFunction,
@@ -103,6 +112,14 @@ class Settings:
         RandomIntFunction,
         SumValuesFunction,
         UpperCaseFunction,
+        # PolarsScalarFunction - Polars-based scalar functions
+        PolarsAddValuesFunction,
+        PolarsDoubleFunction,
+        PolarsMultiplyFunction,
+        PolarsNormalizeFunction,
+        PolarsStringLengthFunction,
+        PolarsSumValuesFunction,
+        PolarsUpperCaseFunction,
     ]
 
 

vgi/metadata.py

@@ -480,30 +480,30 @@ def extract_parameters(
     parameters: list[ParameterInfo] = []
     seen_names: set[str] = set()
 
-    # Check for _polars_params (PolarsScalarFunction subclasses)
-    # These store PolarsParamInfo objects with Polars types
+    # Note: _polars_params (PolarsScalarFunction subclasses) with is_const=False
+    # are NOT extracted as function parameters. They're column position bindings
+    # that describe which columns from the input batch map to expression variables.
+    # However, ConstParam entries (is_const=True) ARE function arguments and must
+    # be extracted for the function signature.
     polars_params = getattr(cls, "_polars_params", {})
     for name, param_info in polars_params.items():
+        if not param_info.is_const:
+            continue  # Skip column bindings, only extract ConstParam
         seen_names.add(name)
-        # Convert Polars type to string representation
-        type_name: str | None = None
-        if param_info.polars_type is not None:
-            type_name = str(param_info.polars_type)
-        else:
-            type_name = "any"  # Dynamic type
-
+        # ConstParam is always required (no default support currently)
+        type_name = "double"  # TODO: Infer from annotation type
         parameters.append(
             ParameterInfo(
                 name=name,
                 position=param_info.position,
                 type_name=type_name,
                 description=param_info.doc,
-                required=True,  # Polars params are always required
+                required=True,
                 default=None,
-                constraints={},
+                constraints=None,
                 is_table_input=False,
-                is_varargs=param_info.varargs,
-                is_const=param_info.is_const,
+                is_varargs=False,
+                is_const=True,
             )
         )
 

vgi/worker.py

@@ -745,6 +745,18 @@ def _process_scalar_batches(
                     if protocol_input.is_finalize:
                         fn_log.debug("finalize_received")
                         generator.close()
+                        # Send FINISHED status response (client expects this)
+                        empty_output = pa.RecordBatch.from_pydict(
+                            {
+                                instance.output_schema.field(0).name: pa.array(
+                                    [], type=instance.output_schema.field(0).type
+                                )
+                            },
+                            schema=instance.output_schema,
+                        )
+                        finished_md = pa.KeyValueMetadata({b"vgi.status": b"FINISHED"})
+                        writer.write_batch(empty_output, custom_metadata=finished_md)
+                        fn_log.debug("finished_status_sent")
                         # Read end-of-stream marker before returning to invocation loop
                         try:
                             data_reader.read_next_batch_with_custom_metadata()