feat: support nullable named function outputs (#4123)

Supports fully nullable named Function outputs while preserving the
distinction between a valid all-null struct and a null/unassigned
result.

## Concrete example

This UDF contract is now valid:

```python
@udf(
    input_schema=pa.schema([
        pa.field("text", pa.string(), nullable=False),
    ]),
    output_schema=pa.schema([
        pa.field(
            "embedding",
            pa.list_(pa.float32(), list_size=1024),
            nullable=True,
        ),
        pa.field("embedding_failure_reason", pa.string(), nullable=True),
        pa.field("embedding_failure_code", pa.int32(), nullable=True),
    ]),
)
def embed(text):
    ...
```

A successful row can return:

```text
embedding = [0.12, ...]
embedding_failure_reason = NULL
embedding_failure_code = NULL
```

If remote inference still fails after retries, it can return:

```text
embedding = NULL
embedding_failure_reason = "HTTP 429: rate limited"
embedding_failure_code = 429
```

An all-null but valid result struct is also assigned; it is not mistaken
for unfinished work.

## Binding shapes

- Mapping the result to one output column stores the `StructArray`
directly, including its parent validity bitmap.
- Flattening the result into top-level columns stores the parent
validity in a reserved internal nullable Boolean assignment column that
is not part of the UDF result mapping.
- An outer null struct remains unassigned/skipped. A valid struct
remains assigned regardless of which child fields are null.
- Scalar Function outputs remain non-nullable.

The contract is preserved through Python registration, Rust application
planning, persisted `FunctionBinding` metadata, schema revalidation, and
Enterprise execution.

python/python/lancedb/__init__.py

@@ -26,6 +26,7 @@ from .sql import AsyncQuery as AsyncSqlQuery
 from .sql import Query as SqlQuery
 from .sql import QueryDescription
 from .functions import (
+    AssignmentMapping as AssignmentMapping,
     FunctionArtifactRequest as FunctionArtifactRequest,
     FunctionApplication as FunctionApplication,
     FunctionBinding as FunctionBinding,

python/python/lancedb/functions.py

@@ -470,11 +470,7 @@ class InputBinding(_RemoteValue):
 
 
 class OutputMapping(_RemoteValue):
-    """One stable result-field mapping.
-
-    Assignment state is outside the Slice 1 client contract. During the NULL
-    transition Lance exposes no public cell-flag identifier to persist here.
-    """
+    """One stable result-field mapping."""
 
     result_field: str
     output_name: str
@@ -484,6 +480,13 @@ class OutputMapping(_RemoteValue):
     nullable: bool
 
 
+class AssignmentMapping(_RemoteValue):
+    """Internal physical column preserving flattened struct validity."""
+
+    output_name: str
+    output_field_id: _Int32
+
+
 class FunctionBinding(_RemoteValue):
     """Immutable Function binding persisted by the Enterprise table service."""
 
@@ -491,6 +494,7 @@ class FunctionBinding(_RemoteValue):
     function: FunctionVersionRef
     inputs: tuple[InputBinding, ...]
     outputs: tuple[OutputMapping, ...]
+    assignment: Optional[AssignmentMapping] = None
     input_schema: Optional[Mapping[str, Any]] = None
     output_schema: Optional[Mapping[str, Any]] = None
 
@@ -911,8 +915,6 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
 
     if not fields:
         raise ValueError("named-struct Function output must contain at least one field")
-    if any(field.nullable for field in fields):
-        raise ValueError("Function output fields must be non-nullable")
     for field in fields:
         _validate_exact_arrow_field(field)
     names = [field.name for field in fields]
@@ -924,7 +926,7 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
             FunctionResultField(
                 name=field.name,
                 arrow_type=_canonical_arrow_field(field),
-                nullable=False,
+                nullable=field.nullable,
             )
             for field in fields
         ),
@@ -1307,8 +1309,9 @@ def udf(
 
     Input and output signatures are inferred from supported annotations. For
     Arrow types annotations cannot express precisely, pass ``input_schema``
-    and ``output_schema`` together. Nullable outputs are rejected because V1
-    uses physical NULL to represent unassigned computed-column rows.
+    and ``output_schema`` together. Scalar outputs must be non-nullable. Every
+    named-struct field may be nullable; Enterprise preserves the struct's
+    validity when the result is expanded into sibling columns.
 
     Parameters
     ----------
@@ -1320,8 +1323,8 @@ def udf(
         Explicit input fields in the exact order of the callable parameters.
         Must be provided together with ``output_schema``.
     output_schema : pyarrow.DataType, pyarrow.Field, or pyarrow.Schema, optional
-        Explicit scalar or named-struct output. Must be non-nullable and be
-        provided together with ``input_schema``.
+        Explicit scalar or named-struct output. Scalar outputs must be
+        non-nullable. Must be provided together with ``input_schema``.
     pip : sequence of str, optional
         Pip requirements for the remote environment.
     conda : sequence of str, optional
@@ -1387,6 +1390,7 @@ def udf(
 
 
 __all__ = [
+    "AssignmentMapping",
     "ApplicationInput",
     "FunctionApplication",
     "FunctionArtifact",

python/python/tests/test_first_class_function_slice2.py

@@ -850,6 +850,43 @@ def test_named_struct_function_can_include_a_blob_result_field():
     ]
 
 
+def test_named_struct_function_preserves_nullable_result_fields():
+    @udf(
+        input_schema=pa.schema([pa.field("value", pa.int64(), nullable=False)]),
+        output_schema=pa.schema(
+            [
+                pa.field("result", pa.int64(), nullable=True),
+                pa.field("failure_code", pa.int32(), nullable=False),
+            ]
+        ),
+    )
+    def nullable_result(value):
+        return {"result": value, "failure_code": 0}
+
+    output = nullable_result.registration_request.signature.output
+    assert [(field.name, field.nullable) for field in output.fields] == [
+        ("result", True),
+        ("failure_code", False),
+    ]
+
+    @udf(
+        input_schema=pa.schema([pa.field("value", pa.int64(), nullable=False)]),
+        output_schema=pa.schema(
+            [
+                pa.field("result", pa.int64(), nullable=True),
+                pa.field("failure_code", pa.int32(), nullable=True),
+            ]
+        ),
+    )
+    def all_nullable(value):
+        return {"result": value, "failure_code": None}
+
+    assert all(
+        field.nullable
+        for field in all_nullable.registration_request.signature.output.fields
+    )
+
+
 def test_metadata_marked_blob_field_uses_the_semantic_type():
     extension = lancedb.blob("image", nullable=False).type
     storage = (