diff --git a/python/python/lancedb/functions.py b/python/python/lancedb/functions.py index 19d4171fb..92be658a8 100644 --- a/python/python/lancedb/functions.py +++ b/python/python/lancedb/functions.py @@ -1529,6 +1529,17 @@ def _infer_signature( return FunctionSignature(inputs=tuple(inputs), output=output) +def _is_vectorized_callable(function: Callable[..., Any], *, method: bool) -> bool: + parameters = _callable_parameters(function) + if method: + parameters = parameters[1:] + annotations = get_type_hints(function, include_extras=True) + annotated = tuple(annotations.get(parameter.name) for parameter in parameters) + return (len(annotated) == 1 and annotated[0] is pa.RecordBatch) or ( + bool(annotated) and all(annotation is pa.Array for annotation in annotated) + ) + + # Both spellings strip from packaged source; `udf` is the deprecated alias. _DECORATOR_NAMES = frozenset({"function", "udf"}) @@ -2302,7 +2313,9 @@ class FunctionDefinition: once per row. A callable whose parameters are all annotated as ``pyarrow.Array`` receives whole Arrow arrays and returns a ``pyarrow.Array`` instead. This authoring choice is independent of the - internal artifact adapter in the registration request. + internal artifact adapter in the registration request. GPU Functions should + use the Arrow form when the model supports batches: a scalar GPU callable is + invoked once per row and leaves the model's own batching unavailable. """ def __init__( @@ -2424,6 +2437,18 @@ class FunctionDefinition: signature = signature._copy(update={"initialization": initialization}) else: signature = _infer_signature(function, input_schema, output_schema) + if gpu_marker is not None: + call = _class_call(function) if is_class else function + if not _is_vectorized_callable(call, method=is_class): + warnings.warn( + f"GPU Function {function_name!r} uses a scalar callable, so each " + "row invokes it separately and its model batch is not inferred. " + "For GPU throughput, annotate every input as pyarrow.Array or " + "take one pyarrow.RecordBatch, provide input_schema and " + "output_schema, and batch inside that callable.", + UserWarning, + stacklevel=3, + ) code_files, code_modules = _code_files(code) entry_source = _SourcePackager( code_modules, frozenset(_requirement_name(package) for package in packages) diff --git a/python/python/tests/test_first_class_function_slice2.py b/python/python/tests/test_first_class_function_slice2.py index 97c0a5f3c..a3a9e3e20 100644 --- a/python/python/tests/test_first_class_function_slice2.py +++ b/python/python/tests/test_first_class_function_slice2.py @@ -18,6 +18,7 @@ import subprocess import sys import typing import urllib.parse +import warnings from typing import Optional import pyarrow as pa @@ -658,9 +659,11 @@ def test_function_image_requires_digest_reference(image): def test_function_gpu_marker_uses_gpu_runtime(): - @function(pip=["cupy-cuda12x"], gpu=True, gpu_class="a10g") - def double_on_gpu(value: int) -> int: - return value * 2 + with pytest.warns(UserWarning, match="uses a scalar callable"): + + @function(pip=["cupy-cuda12x"], gpu=True, gpu_class="a10g") + def double_on_gpu(value: int) -> int: + return value * 2 request = json.loads(double_on_gpu.registration_request.to_canonical_json()) assert request["runtime"]["kind"] == "python_v2" @@ -710,6 +713,60 @@ def test_function_gpu_marker_uses_gpu_runtime(): ) +class _GpuScalarFunction: + def __call__(self, value: str) -> int: + return len(value) + + +class _GpuArrayFunction: + def __call__(self, value: pa.Array) -> pa.Array: + return value + + +def test_gpu_scalar_warning_names_the_vectorized_alternative(): + with pytest.warns(UserWarning, match="model batch is not inferred"): + + @function(gpu=True) + def scalar_embedding(value: str) -> list[int]: + return [len(value)] + + input_schema = pa.schema([pa.field("value", pa.string(), nullable=False)]) + with pytest.warns(UserWarning, match="uses a scalar callable"): + class_scalar = function(_GpuScalarFunction, gpu=True) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + + @function( + gpu=True, input_schema=input_schema, output_schema=pa.list_(pa.int64()) + ) + def array_embedding(value: pa.Array) -> pa.Array: + return value + + @function( + gpu=True, input_schema=input_schema, output_schema=pa.list_(pa.int64()) + ) + def record_batch_embedding(value: pa.RecordBatch) -> pa.Array: + return value.column(0) + + @function + def cpu_scalar(value: str) -> int: + return len(value) + + class_embedding = function( + _GpuArrayFunction, + gpu=True, + input_schema=input_schema, + output_schema=pa.list_(pa.int64()), + ) + + assert not caught + assert array_embedding.registration_request.batch_size is None + assert record_batch_embedding.registration_request.batch_size is None + assert cpu_scalar.registration_request.batch_size is None + assert class_scalar.registration_request.batch_size is None + assert class_embedding.registration_request.batch_size is None + + def test_unknown_runtime_discards_payload_before_known_field_validation(): for payload in [ {