Compare commits

..

3 Commits

Author SHA1 Message Date
Lance Release 2761e6108e Bump version: 0.38.0-beta.11 → 0.38.0-beta.12 2026-08-28 00:51:15 +00:00
Wyatt Alt 6c8aa22704 feat: let a computed-column batch read its own earlier declarations (#4072)
`add_columns().computed()` accepted several columns in one call but
bound each against the table's schema as it stood before the call, so
`a` and `b = a + 1` had to be two commits. A server staging declarations
behind other schema work has no atomic way to do that, and a caller
reading the builder's plural signature reasonably expects the batch to
be one.

Each accepted column now joins the schema the next one resolves against,
so the batch is planned and committed as one. Order is the dependency
order; reading ahead is still an unknown column. `validate_declarations`
exposes the schema-level checks -- the Function-binding guard and the
planning -- without a commit, for callers that must reject before
earlier work in the same request lands; LSM state is table state and
stays a commit-time check.

Refresh order matters for a dependent column: `b = coalesce(a, 0)`
refreshed before `a` would bake zeros from `a`'s placeholder null, and
the fill-once contract keeps them. Refresh now refuses, naming the
input, while a computed input still has rows a refresh of it would fill
-- the same probe refresh already uses to detect a no-op. Otherwise it
is one snapshot and one commit, as before; a concurrent append is not in
the commit and waits for the next refresh. Refreshing dependencies on
the caller's behalf was considered and rejected: it is not how
materialized views or our own backfill scheduler behave, and it needs
multi-commit fencing that an explicit per-row fill marker would make
unnecessary.
2026-08-27 17:47:27 -07:00
Will Jones 84f46df876 ci(nodejs): fix nightly OOM on the aarch64 publish legs (#4077)
The nightly `NPM Publish` run has failed every night since at least Aug
23, always on the same two legs: `aarch64-unknown-linux-gnu` and
`aarch64-unknown-linux-musl`. The other five targets pass. rustc is
OOM-killed during the fat-LTO codegen of the cdylib — `signal: 9` with
no diagnostic, about 27 minutes in — and on the musl leg that takes the
whole runner down with `The runner has received a shutdown signal`.

Both legs now pass:

| leg | before | peak memory | wall time |
| --- | --- | --- | --- |
| `aarch64-unknown-linux-gnu` | OOM-killed at ~27 min | 31391 → 22851
MiB | 38m43s → 22m04s |
| `aarch64-unknown-linux-musl` | runner killed at ~28 min | >32 GiB →
16516 MiB | ~40 min → 20m50s |

**ThinLTO** is most of that. Fat LTO is single-threaded, and its peak is
consumed inside rustc's LLVM before any linker process is spawned —
which is why it is the whole fix on musl, and why lld alone left the gnu
leg still peaking at 31391 MiB against the runner's 32 GiB. Both legs
now use the `lto: thin` / `codegen_units: 16` settings that darwin and
both Windows legs already use, at a cost of a few percent runtime
performance.

**lld** covers the rest, on the gnu leg. arm64 Linux otherwise links
through GNU `ld` where x86_64 already defaults to `rust-lld`, which is
why only the arm64 legs hit this at all; on a comparable arm64 build
(`lancedb/sophon#7313`) it cut the largest single linker process from
7.0 to 4.0 GiB and wall time by 35%. The flags live in a small wrapper
script used as the linker rather than in `-C link-arg`, because the
per-target rustflags variable does not reach every unit that links:
dependency crates linking a dylib (`crc-fast`, `lance-arrow`) were
invoked as bare `clang`, which targets the x86_64 host and fails with
`Relocations in generic ELF (EM: 183)`.

Separately, and affecting five legs rather than two: the three ThinLTO
targets exported `CARGO_PROFILE_RELEASE_LTO` and
`CARGO_PROFILE_RELEASE_CODEGEN_UNITS` from `pre_build`, which runs
inside the build step — after the cache step. `Swatinem/rust-cache`
computes its key when the action runs, before any step, so step-local
values are invisible to it. The result is a loop that never converges:
the key never changes, so restores are exact hits, an exact hit makes
the post-run save a no-op, and cargo invalidates the restored artifacts
anyway because the flags differ. Those legs have been rebuilding cold on
every run. Both values move to job-level `env:` ahead of the cache step,
driven by new `lto:`/`codegen_units:` matrix fields, and are forwarded
into the containers with `-e` since `docker run` inherits nothing.

Every leg's cache key shifts once as a result, so expect one cold
rebuild.

A `Report peak memory` step is added so whether these legs fit is a
number rather than an inference from whether the runner survived. It
produced the figures above.

## Not included

Moving these legs to native arm64 runners. It would retire the zig cross
path, the `AT_HWCAP2` workaround and the `TARGET_CC` override, and arm64
runners are billed roughly 37% below x64 at equal core count — but the
`lts-debian-aarch64` image exists to link against the manylinux2014
sysroot's glibc 2.17, and building natively on ubuntu-24.04 would raise
the minimum glibc for every published aarch64 binary. That is a
user-facing decision, not a CI cleanup.

Dropping these legs to smaller runners, which is where the real cost
saving is — larger runners are billed even on public repos. On these
numbers it is not available yet: musl at 16516 MiB is about 130 MiB over
what a 16 GB standard runner has. Worth revisiting as a follow-up.

## Testing

Cargo's rustflags precedence was checked locally rather than taken from
the docs, since getting it wrong would silently change the published
binaries. With a throwaway crate carrying both a `target.'cfg(all())'`
and a per-target rustflags table: setting `RUSTFLAGS` discards both, and
setting it to the empty string discards them too. That rules out routing
the linker flag through a job-level `RUSTFLAGS`, because `env:` keys
cannot be conditionally omitted and every other leg would then silently
lose the `target-cpu`/`target-feature` settings in `.cargo/config.toml`
— `+avx2` on x86_64 and `-crt-static` on aarch64-musl.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-27 17:35:06 -07:00
32 changed files with 405 additions and 981 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
[tool.bumpversion]
current_version = "0.38.0-beta.11"
current_version = "0.38.0-beta.12"
parse = """(?x)
(?P<major>0|[1-9]\\d*)\\.
(?P<minor>0|[1-9]\\d*)\\.
+74 -28
View File
@@ -40,40 +40,31 @@ jobs:
- target: aarch64-apple-darwin
host: macos-latest
features: fp16kernels
# Fat LTO was ~111 of this job's ~113 minutes.
lto: thin
codegen_units: 16
pre_build: |-
brew install protobuf
# Fat LTO (the workspace default in .cargo/config.toml) is
# single-threaded and is the peak-memory step of the build. On
# this runner it accounted for ~111 of the job's ~113 minutes,
# making it the critical path of the entire publish pipeline.
# ThinLTO parallelizes it across the runner's cores, for a few
# percent of runtime performance.
export CARGO_PROFILE_RELEASE_LTO=thin
export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
- target: x86_64-pc-windows-msvc
host: windows-2025
features: ","
# The lower peak also keeps this on the standard 4-core runner.
lto: thin
codegen_units: 16
pre_build: |-
choco install --no-progress protoc ninja nasm
tail -n 1000 /c/ProgramData/chocolatey/logs/chocolatey.log
# There is an issue where choco doesn't add nasm to the path
export PATH="$PATH:/c/Program Files/NASM"
nasm -v
# See the ThinLTO note on aarch64-apple-darwin above. Keeping
# peak memory down is also what lets this run on the standard
# 4-core runner: the 8-core larger runner was only needed to
# stop fat LTO from OOMing rustc-LLVM.
export CARGO_PROFILE_RELEASE_LTO=thin
export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
- target: aarch64-pc-windows-msvc
host: windows-2025
features: ","
lto: thin
codegen_units: 16
pre_build: |-
choco install --no-progress protoc
rustup target add aarch64-pc-windows-msvc
# See the ThinLTO note on aarch64-apple-darwin above.
export CARGO_PROFILE_RELEASE_LTO=thin
export CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16
- target: x86_64-unknown-linux-gnu
host: ubuntu-latest
features: fp16kernels
@@ -103,6 +94,14 @@ jobs:
# https://github.com/napi-rs/napi-rs/blob/main/debian-aarch64.Dockerfile
docker: ghcr.io/napi-rs/napi-rs/nodejs-rust:lts-debian-aarch64
features: "fp16kernels"
# Fat LTO OOM-killed rustc every nightly; even with lld it peaked
# at 31391 MiB of the runner's 32 GiB.
lto: thin
codegen_units: 16
# arm64 Linux links through GNU `ld` where x86_64 defaults to
# `rust-lld`, which is why only arm64 OOM'd. lld cut the largest
# linker process 7.0 -> 4.0 GiB (lancedb/sophon#7313).
linker: /tmp/aarch64-lld-clang
pre_build: |-
set -e &&
apt-get update &&
@@ -112,9 +111,30 @@ jobs:
# AT_HWCAP2 (added in Linux 3.17). Define it for aws-lc-sys.
export CFLAGS="$CFLAGS -DAT_HWCAP2=26" &&
rustup target add aarch64-unknown-linux-gnu
# Not `&&`-chained: in dash, errexit does not fire for a
# non-final command in an `&&` list, so failures were ignored.
#
# A wrapper rather than `-C link-arg` because the per-target
# rustflags variable does not reach every unit that links, while
# the linker variable does. `clang` because GCC silently ignores
# `-fuse-ld=lld` unless built with lld support. Two echoes
# because printf's newline escape gets rewritten to `;` between
# here and the container.
echo '#!/bin/sh' > /tmp/aarch64-lld-clang
echo 'exec clang --target=aarch64-unknown-linux-gnu --sysroot=/usr/aarch64-unknown-linux-gnu/aarch64-unknown-linux-gnu/sysroot --gcc-toolchain=/usr/aarch64-unknown-linux-gnu -fuse-ld=lld "$@"' >> /tmp/aarch64-lld-clang
chmod 0755 /tmp/aarch64-lld-clang
# Fail now, not at the cdylib link ~30 minutes later. Linking at
# all also proves lld resolved; clang errors out when it cannot.
echo 'int main(void){return 0;}' > /tmp/probe.c
/tmp/aarch64-lld-clang /tmp/probe.c -o /tmp/probe
readelf -h /tmp/probe | grep AArch64
- target: aarch64-unknown-linux-musl
host: ubuntu-2404-8x-x64
features: ","
# Fat LTO took the whole runner down. lld cannot help: it died
# inside rustc's LLVM, before any linker was spawned.
lto: thin
codegen_units: 16
pre_build: |-
set -e &&
sudo apt-get update &&
@@ -123,6 +143,19 @@ jobs:
export EXTRA_ARGS="-x"
name: build - ${{ matrix.settings.target }}
runs-on: ${{ matrix.settings.host }}
# On the job, not exported from `pre_build`: `Swatinem/rust-cache` hashes
# `CARGO_*` into its cache key before any step runs, so a step-local export
# leaves the key unchanged while cargo still rebuilds cold. The ThinLTO
# legs had been doing that every run.
#
# Not `RUSTFLAGS`: setting it, even to "", discards every config-file
# rustflag, silently dropping .cargo/config.toml's `target-cpu` and
# `target-feature` from the published binaries.
env:
CARGO_PROFILE_RELEASE_LTO: ${{ matrix.settings.lto || 'fat' }}
CARGO_PROFILE_RELEASE_CODEGEN_UNITS: ${{ matrix.settings.codegen_units || '1' }}
# Empty elsewhere: a per-target variable is only read for that triple.
CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER: ${{ matrix.settings.linker }}
defaults:
run:
working-directory: nodejs
@@ -169,19 +202,15 @@ jobs:
# creating ref). The nightly cadence also keeps entries inside
# GitHub's 7-day eviction window, which a tag-only trigger would not.
save-if: ${{ github.ref == 'refs/heads/main' }}
# Docker builds can use rust-cache too. `target/` already lives on the
# host because the whole workspace is bind-mounted into the container, and
# rust-cache's prune and save run host-side, so they can manage it -- which
# is what keeps the entry to dependency artifacts rather than a multi-GB
# copy of everything.
# Docker builds can use rust-cache too: the workspace is bind-mounted, so
# `target/` lives on the host and rust-cache's prune keeps the entry
# small.
#
# Two differences from the native builds. The container's CARGO_HOME is
# bind-mounted from `.cargo-cache` rather than the host's ~/.cargo, so that
# has to be cached explicitly. And the key is derived from the *host* rustc
# version, which is not the compiler that produced these artifacts; that is
# safe because cargo fingerprints the real compiler and rebuilds on a
# mismatch, it just means a base-image toolchain bump costs one cold build
# instead of invalidating the key.
# bind-mounted from `.cargo-cache` rather than ~/.cargo, so that is cached
# explicitly. And the key uses the *host* rustc version, not the compiler
# that built these artifacts -- safe, since cargo fingerprints the real
# one; a base-image bump just costs one cold build.
- name: Cache cargo (docker builds)
uses: Swatinem/rust-cache@v2
if: ${{ matrix.settings.docker }}
@@ -210,9 +239,14 @@ jobs:
# cache step above saves. Previously the registry mounts pointed at
# `.cargo/...`, a path nothing cached, so the container re-downloaded
# the whole crate registry on every run.
#
# `docker run` inherits nothing; `-e NAME` carries the job's `env:` in.
options: "--user 0:0 -v ${{ github.workspace }}/.cargo-cache/git/db:/usr/local/cargo/git/db \
-v ${{ github.workspace }}/.cargo-cache/registry/cache:/usr/local/cargo/registry/cache \
-v ${{ github.workspace }}/.cargo-cache/registry/index:/usr/local/cargo/registry/index \
-e CARGO_PROFILE_RELEASE_LTO \
-e CARGO_PROFILE_RELEASE_CODEGEN_UNITS \
-e CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER \
-v ${{ github.workspace }}:/build -w /build/nodejs"
run: |
set -e
@@ -256,6 +290,18 @@ jobs:
if: always()
run: df -h
shell: bash
- name: Report peak memory
if: always() && runner.os == 'Linux'
shell: bash
run: |
peak=$(find /sys/fs/cgroup -name memory.peak -readable \
-exec cat {} + 2>/dev/null | sort -n | tail -1)
if [ -n "$peak" ]; then
echo "peak memory: $((peak / 1024 / 1024)) MiB"
else
echo "peak memory: unavailable (no readable cgroup v2 memory.peak)"
fi
free -g || true
- name: Upload artifact
uses: actions/upload-artifact@v7
with:
+1 -1
View File
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
<dependency>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-core</artifactId>
<version>0.38.0-beta.11</version>
<version>0.38.0-beta.12</version>
</dependency>
```
+1 -1
View File
@@ -8,7 +8,7 @@
<parent>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.38.0-beta.11</version>
<version>0.38.0-beta.12</version>
<relativePath>../pom.xml</relativePath>
</parent>
+1 -1
View File
@@ -6,7 +6,7 @@
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.38.0-beta.11</version>
<version>0.38.0-beta.12</version>
<packaging>pom</packaging>
<name>${project.artifactId}</name>
<description>LanceDB Java SDK Parent POM</description>
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
name = "lancedb-nodejs"
edition.workspace = true
version = "0.38.0-beta.11"
version = "0.38.0-beta.12"
publish = false
license.workspace = true
description.workspace = true
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-darwin-arm64",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.12",
"os": ["darwin"],
"cpu": ["arm64"],
"main": "lancedb.darwin-arm64.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-gnu",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.12",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-musl",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.12",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-gnu",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.12",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-musl",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.12",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-arm64-msvc",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.12",
"os": [
"win32"
],
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-x64-msvc",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.12",
"os": ["win32"],
"cpu": ["x64"],
"main": "lancedb.win32-x64-msvc.node",
+1 -1
View File
@@ -11,7 +11,7 @@
"ann"
],
"private": false,
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.12",
"main": "dist/index.js",
"exports": {
".": "./dist/index.js",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.38.0-beta.11"
version = "0.38.0-beta.12"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
+7 -38
View File
@@ -16,7 +16,6 @@ from typing import (
Iterable,
List,
Literal,
Mapping,
Optional,
Union,
)
@@ -688,35 +687,17 @@ class DBConnection(EnforceOverrides):
"""
raise NotImplementedError("serialize is not supported for this connection type")
def create_function(
self,
definition: UdfDefinition,
*,
secrets: Optional[Mapping[str, str]] = None,
) -> FunctionVersion:
def create_function(self, definition: UdfDefinition) -> FunctionVersion:
"""Register a scalar Python UDF and wait for its immutable version.
``secrets`` must contain exactly the names declared by
``@udf(secrets=[...])``. Values are sent in the create request and
stored server-side in the private execution artifact; returned
Function and Job metadata contain only the declared names.
This is the blocking counterpart of :meth:`create_function_async`.
Local connections raise ``NotImplementedError``.
"""
return self.create_function_async(definition, secrets=secrets).wait()
return self.create_function_async(definition).wait()
def create_function_async(
self,
definition: UdfDefinition,
*,
secrets: Optional[Mapping[str, str]] = None,
) -> Job[FunctionVersion]:
def create_function_async(self, definition: UdfDefinition) -> Job[FunctionVersion]:
"""Register a scalar Python UDF through the remote Function catalog.
``secrets`` must contain exactly the names declared by
``@udf(secrets=[...])``. Values are sent in the create request and
stored server-side in the private execution artifact; returned
Function and Job metadata contain only the declared names.
Submission returns a typed job. The immutable Function version becomes
available only when :meth:`Job.wait` succeeds. Local connections raise
``NotImplementedError``.
@@ -1424,13 +1405,8 @@ class LanceDBConnection(DBConnection):
return Job(self._conn.job(job_id))
@override
def create_function_async(
self,
definition: UdfDefinition,
*,
secrets: Optional[Mapping[str, str]] = None,
) -> Job[FunctionVersion]:
job = LOOP.run(self._conn.create_function_async(definition, secrets=secrets))
def create_function_async(self, definition: UdfDefinition) -> Job[FunctionVersion]:
job = LOOP.run(self._conn.create_function_async(definition))
return Job(job)
@override
@@ -2249,24 +2225,17 @@ class AsyncConnection(object):
return AsyncJob(self._inner.job(job_id))
async def create_function_async(
self,
definition: UdfDefinition,
*,
secrets: Optional[Mapping[str, str]] = None,
self, definition: UdfDefinition
) -> AsyncJob[FunctionVersion]:
"""Register a scalar Python UDF through the remote Function catalog.
``secrets`` must contain exactly the names declared by
``@udf(secrets=[...])``. Values are sent in the create request and
stored server-side in the private execution artifact; returned
Function and Job metadata contain only the declared names.
The returned typed job resolves to the immutable Function version.
Local connections raise ``NotImplementedError``.
"""
if not isinstance(definition, UdfDefinition):
raise TypeError("create_function_async requires a @udf definition")
inner = await self._inner.create_function_async(
definition._submission_json(secrets)
definition.registration_request.to_canonical_json()
)
return _typed_job(inner, FunctionVersion.from_json)
+6 -106
View File
@@ -4,7 +4,7 @@
"""Canonical Function values exchanged with LanceDB Enterprise services.
These immutable models contain client/wire state only. Catalog persistence,
environment bake, secret resolution, and execution are owned by Sophon.
environment bake, and execution are owned by Sophon.
``RefreshColumnResult`` is also the backend-neutral result of a local
expression-backed refresh job.
"""
@@ -229,7 +229,7 @@ class PythonEnvironmentSpec(_RemoteValue):
class PythonRuntimeSpec(_RemoteValue):
"""Remote runtime definition with non-secret environment values.
"""Remote runtime definition with environment values.
V1 supports ``kind="python"``. Newer runtime kinds remain readable, while
their unknown payload fields are intentionally not retained by the client.
@@ -268,7 +268,6 @@ class FunctionVersion(_RemoteValue):
runtime: PythonRuntimeSpec
runtime_digest: str
environment_digest: str
required_secrets: tuple[str, ...] = ()
created_at: str
def __call__(self, **inputs: Any) -> FunctionApplication:
@@ -330,17 +329,12 @@ class FunctionVersion(_RemoteValue):
class FunctionRegistrationRequest(_RemoteValue):
"""Stable remote registration envelope produced by :func:`udf`.
Only secret names are represented. Secret values are supplied separately
when the definition is submitted and are not part of this durable value.
"""
"""Stable remote registration envelope produced by :func:`udf`."""
name: str
artifact: FunctionArtifactRequest
signature: FunctionSignature
runtime: PythonRuntimeSpec
required_secrets: tuple[str, ...] = ()
class FunctionVersionRef(_OpenRemoteValue):
@@ -485,27 +479,6 @@ class RefreshColumnResult(_RemoteValue):
_FUNCTION_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*$")
_SECRET_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
# Keep this byte limit aligned with Sophon's MAX_FUNCTION_SECRET_VALUE_BYTES.
_MAX_FUNCTION_SECRET_VALUE_BYTES = 64 * 1024
_MAX_FUNCTION_SECRET_VALUES_BYTES = 512 * 1024
def _validate_secret_value(name: str, value: Any) -> str:
"""Validate one secret value before building the create request."""
if not isinstance(value, str):
raise TypeError(f"Function secret {name!r} value must be a string")
if not value:
raise ValueError(f"Function secret {name!r} value must be non-empty")
if "\0" in value:
raise ValueError(f"Function secret {name!r} value must not contain NUL")
value_bytes = len(value.encode("utf-8"))
if value_bytes > _MAX_FUNCTION_SECRET_VALUE_BYTES:
raise ValueError(
f"Function secret {name!r} value exceeds the "
f"{_MAX_FUNCTION_SECRET_VALUE_BYTES}-byte limit"
)
return value
_GRAMMAR_PRIMITIVES = (
@@ -936,7 +909,6 @@ class UdfDefinition:
output_schema: Optional[pa.DataType | pa.Field | pa.Schema],
pip: tuple[str, ...],
env: Mapping[str, str],
secrets: tuple[str, ...],
python_version: Optional[str],
conda: tuple[str, ...] = (),
conda_channels: tuple[str, ...] = (),
@@ -963,17 +935,6 @@ class UdfDefinition:
for key, value in environment.items()
):
raise TypeError("Function env keys and values must be strings")
required_secrets = tuple(sorted(set(secrets)))
invalid_secrets = [
secret for secret in required_secrets if not _SECRET_NAME.fullmatch(secret)
]
if invalid_secrets:
raise ValueError(f"invalid Function secret names: {invalid_secrets!r}")
overlap = set(environment) & set(required_secrets)
if overlap:
raise ValueError(
f"Function env and secret names must be disjoint: {sorted(overlap)!r}"
)
signature = _infer_signature(function, input_schema, output_schema)
source = _package_source(function)
digest = f"sha256:{hashlib.sha256(source).hexdigest()}"
@@ -1002,65 +963,14 @@ class UdfDefinition:
),
signature=signature,
runtime=runtime,
required_secrets=required_secrets,
)
functools.update_wrapper(self, function)
@property
def registration_request(self) -> FunctionRegistrationRequest:
"""The immutable, value-free client model for a Function submission."""
"""The immutable request sent by ``create_function_async``."""
return self._request
def _submission_json(self, secrets: Optional[Mapping[str, str]]) -> str:
"""Build one registration submission without retaining values on self."""
if secrets is None:
secret_values: Mapping[str, str] = {}
elif not isinstance(secrets, Mapping):
raise TypeError("Function secrets must be a mapping of names to strings")
else:
secret_values = secrets
if any(not isinstance(name, str) for name in secret_values):
raise TypeError("Function secret names must be strings")
expected = set(self._request.required_secrets)
provided = set(secret_values)
if provided != expected:
missing = sorted(expected - provided)
unexpected = sorted(provided - expected)
details = []
if missing:
details.append(f"missing: {missing!r}")
if unexpected:
details.append(f"unexpected: {unexpected!r}")
raise ValueError(
"Function secret values must exactly match the declared secrets ("
+ "; ".join(details)
+ ")"
)
canonical_values = {}
total_bytes = 0
for name in sorted(secret_values):
value = _validate_secret_value(name, secret_values[name])
total_bytes += len(value.encode("utf-8"))
if total_bytes > _MAX_FUNCTION_SECRET_VALUES_BYTES:
raise ValueError(
"Function secret values exceed the "
f"{_MAX_FUNCTION_SECRET_VALUES_BYTES}-byte request limit"
)
canonical_values[name] = value
submission = self._request._known_dict()
if canonical_values:
submission["secret_values"] = canonical_values
return json.dumps(
submission,
ensure_ascii=False,
allow_nan=False,
sort_keys=True,
separators=(",", ":"),
)
def __call__(self, *args, **kwargs):
return self._function(*args, **kwargs)
@@ -1078,7 +988,6 @@ def udf(
output_schema: Optional[pa.DataType | pa.Field | pa.Schema] = None,
pip: tuple[str, ...] | list[str] = (),
env: Optional[Mapping[str, str]] = None,
secrets: tuple[str, ...] | list[str] = (),
python_version: Optional[str] = None,
conda: tuple[str, ...] | list[str] = (),
conda_channels: tuple[str, ...] | list[str] = (),
@@ -1093,7 +1002,6 @@ def udf(
output_schema: Optional[pa.DataType | pa.Field | pa.Schema] = None,
pip: tuple[str, ...] | list[str] = (),
env: Optional[Mapping[str, str]] = None,
secrets: tuple[str, ...] | list[str] = (),
python_version: Optional[str] = None,
conda: tuple[str, ...] | list[str] = (),
conda_channels: tuple[str, ...] | list[str] = (),
@@ -1124,10 +1032,7 @@ def udf(
conda_channels : sequence of str, optional
Conda channels in priority order; requires ``conda``.
env : mapping of str to str, optional
Non-secret environment variables. Use ``secrets`` for credentials.
secrets : sequence of str, optional
Names of secrets required by the callable. Supply their values separately
to ``create_function`` or ``create_function_async``.
Environment variables included in the Function definition.
python_version : str, optional
Remote Python major/minor version. Defaults to the client version.
@@ -1149,15 +1054,11 @@ def udf(
Examples
--------
>>> from lancedb import udf
>>> @udf(pip=["numpy==2.2.0"], secrets=["MODEL_TOKEN"])
>>> @udf(pip=["numpy==2.2.0"])
... def score(value: float) -> float:
... return value * 2
>>> score(1.5)
3.0
>>> db.create_function( # doctest: +SKIP
... score, secrets={"MODEL_TOKEN": "user-secret-value"}
... )
"""
def decorate(target: Callable[..., Any]) -> UdfDefinition:
@@ -1168,7 +1069,6 @@ def udf(
output_schema=output_schema,
pip=tuple(pip),
env={} if env is None else env,
secrets=tuple(secrets),
python_version=python_version,
conda=tuple(conda),
conda_channels=tuple(conda_channels),
+3 -10
View File
@@ -7,7 +7,7 @@ import json
import logging
from concurrent.futures import ThreadPoolExecutor
import sys
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Mapping, Optional, Union
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Union
from urllib.parse import urlparse
import warnings
@@ -742,15 +742,8 @@ class RemoteDBConnection(DBConnection):
return Job(self._conn.job(job_id))
@override
def create_function_async(
self,
definition: UdfDefinition,
*,
secrets: Optional[Mapping[str, str]] = None,
) -> Job[FunctionVersion]:
return Job(
LOOP.run(self._conn.create_function_async(definition, secrets=secrets))
)
def create_function_async(self, definition: UdfDefinition) -> Job[FunctionVersion]:
return Job(LOOP.run(self._conn.create_function_async(definition)))
@override
def get_function(self, name: str, *, version: str) -> FunctionVersion:
@@ -37,21 +37,6 @@ def job_result(name: str) -> dict:
return json.loads(fixture(name))["result"]
def assert_no_secret_values(value):
if isinstance(value, dict):
for key, child in value.items():
assert key not in {
"secret_value",
"secret_values",
"resolved_secret",
"resolved_secrets",
}
assert_no_secret_values(child)
elif isinstance(value, list):
for child in value:
assert_no_secret_values(child)
def test_public_function_values_are_in_api_reference():
docs = Path(__file__).parents[3] / "docs" / "src" / "python" / "python.md"
rendered = docs.read_text()
@@ -109,7 +94,6 @@ def test_function_version_identity_is_immutable_and_exact():
version = FunctionVersion.from_json(json.dumps(value))
assert version.name == "embed"
assert version.version == "fv_01K3EXACT"
assert version.required_secrets == ("HF_TOKEN",)
with pytest.raises((TypeError, ValueError)):
version.version = "fv_changed"
@@ -292,15 +276,6 @@ def test_refresh_result_rejects_non_u64_values(field):
RefreshColumnResult.from_json(json.dumps(value))
def test_canonical_client_values_contain_secret_names_only():
version = FunctionVersion.from_json(
json.dumps(job_result("remote_function_job.json"))
)
canonical = json.loads(version.to_canonical_json())
assert canonical["required_secrets"] == ["HF_TOKEN"]
assert_no_secret_values(canonical)
class _FunctionDeclarationInner:
def __init__(self):
self.calls = []
@@ -19,13 +19,7 @@ import pyarrow as pa
import pytest
import lancedb
from lancedb.functions import (
_MAX_FUNCTION_SECRET_VALUE_BYTES,
_MAX_FUNCTION_SECRET_VALUES_BYTES,
FunctionRegistrationRequest,
UdfDefinition,
udf,
)
from lancedb.functions import UdfDefinition, udf
THRESHOLD = 20
_CACHE = None
@@ -45,28 +39,12 @@ FIXTURES = (
@udf(
pip=["numpy>=2"],
env={"MODE": "test"},
secrets=["API_TOKEN"],
python_version="3.12",
)
def normalize_score(value: float) -> float:
return value / 100.0
def _assert_no_secret_values(value):
if isinstance(value, dict):
for key, child in value.items():
assert key not in {
"secret_value",
"secret_values",
"resolved_secret",
"resolved_secrets",
}
_assert_no_secret_values(child)
elif isinstance(value, list):
for child in value:
_assert_no_secret_values(child)
def test_scalar_udf_matches_shared_registration_golden_and_remains_callable():
assert isinstance(normalize_score, UdfDefinition)
assert normalize_score(25.0) == 0.25
@@ -81,8 +59,6 @@ def test_scalar_udf_matches_shared_registration_golden_and_remains_callable():
"kind": "scalar_to_arrow_batch",
"version": 1,
}
assert request["required_secrets"] == ["API_TOKEN"]
_assert_no_secret_values(request)
def _run_packaged(definition, *args):
@@ -396,7 +372,6 @@ def test_udf_recursion_versus_a_rebound_module_name(tmp_path):
output_schema=None,
pip=(),
env={},
secrets=(),
python_version=None,
)
with pytest.raises(ValueError, match="binds that name to another value"):
@@ -551,54 +526,13 @@ def test_annotation_and_explicit_schema_validation_fail_closed():
return value
def test_secret_names_are_canonical_and_disjoint_from_environment():
@udf(secrets=["Z_TOKEN", "A_TOKEN", "Z_TOKEN"])
def canonical_secrets(value: int) -> int:
return value
assert canonical_secrets.registration_request.required_secrets == (
"A_TOKEN",
"Z_TOKEN",
)
with pytest.raises(ValueError, match="must be disjoint"):
@udf(env={"TOKEN": "plaintext"}, secrets=["TOKEN"])
def overlapping(value: int) -> int:
return value
def test_declared_secret_api_still_requires_explicit_create_values():
@udf(secrets=["API_TOKEN"])
def declared_secret(value: int) -> int:
return value
with pytest.raises(ValueError, match="missing"):
declared_secret._submission_json(None)
submission = json.loads(
declared_secret._submission_json({"API_TOKEN": "explicit-secret"})
)
assert submission["required_secrets"] == ["API_TOKEN"]
assert submission["secret_values"] == {"API_TOKEN": "explicit-secret"}
def test_no_secrets_preserve_canonical_registration_shape():
@udf
def no_secrets(value: int) -> int:
return value
canonical = json.loads(no_secrets.registration_request.to_canonical_json())
assert "required_secrets" not in canonical
assert json.loads(no_secrets._submission_json(None)) == canonical
def test_local_function_catalog_operations_are_not_supported(tmp_path):
db = lancedb.connect(tmp_path)
message = "Function catalog operations are not supported by this database"
with pytest.raises(NotImplementedError, match=message):
db.create_function(normalize_score, secrets={"API_TOKEN": "value"})
db.create_function(normalize_score)
with pytest.raises(NotImplementedError, match=message):
db.create_function_async(normalize_score, secrets={"API_TOKEN": "value"})
db.create_function_async(normalize_score)
with pytest.raises(NotImplementedError, match=message):
db.get_function("normalize_score", version="fv_exact")
@@ -628,7 +562,6 @@ def _mock_remote_function_catalog():
"runtime": body["runtime"],
"runtime_digest": "sha256:runtime",
"environment_digest": "sha256:environment",
"required_secrets": body.get("required_secrets", []),
"created_at": "2026-08-21T00:00:00Z",
}
response = {"job_id": "job-register"}
@@ -675,9 +608,7 @@ def test_remote_registration_job_and_exact_version_reopen_round_trip():
host_override=host,
client_config={"retry_config": {"retries": 0}},
)
registration = db.create_function_async(
normalize_score, secrets={"API_TOKEN": "secret-value"}
)
registration = db.create_function_async(normalize_score)
assert registration.id == "job-register"
created = registration.wait()
reopened = db.get_function("normalize_score", version=created.version)
@@ -686,18 +617,9 @@ def test_remote_registration_job_and_exact_version_reopen_round_trip():
assert reopened.name == "normalize_score"
assert reopened.version == "fv_exact"
create_request = state["requests"][0][1]
expected = json.loads(normalize_score.registration_request.to_canonical_json())
expected["secret_values"] = {"API_TOKEN": "secret-value"}
assert create_request == expected
durable_request = FunctionRegistrationRequest.from_json(json.dumps(create_request))
assert not hasattr(durable_request, "secret_values")
assert "secret_values" not in json.loads(durable_request.to_canonical_json())
assert "secret_values" not in json.loads(
assert create_request == json.loads(
normalize_score.registration_request.to_canonical_json()
)
assert "secret-value" not in repr(normalize_score)
assert "secret-value" not in repr(normalize_score.registration_request)
assert not hasattr(created, "secret_values")
def test_blocking_remote_registration_returns_function_version():
@@ -708,9 +630,7 @@ def test_blocking_remote_registration_returns_function_version():
host_override=host,
client_config={"retry_config": {"retries": 0}},
)
created = db.create_function(
normalize_score, secrets={"API_TOKEN": "blocking-secret"}
)
created = db.create_function(normalize_score)
assert created.name == "normalize_score"
assert created.version == "fv_exact"
@@ -718,121 +638,3 @@ def test_blocking_remote_registration_returns_function_version():
"/v1/functions/create",
"/v1/jobs/describe",
]
assert state["requests"][0][1]["secret_values"] == {"API_TOKEN": "blocking-secret"}
@pytest.mark.parametrize(
("secret_values", "error_type", "message"),
[
(None, ValueError, "missing"),
({}, ValueError, "missing"),
({"OTHER": "value"}, ValueError, "missing.*unexpected"),
({"API_TOKEN": ""}, ValueError, "non-empty"),
({"API_TOKEN": "bad\0value"}, ValueError, "NUL"),
({"API_TOKEN": 123}, TypeError, "must be a string"),
([("API_TOKEN", "value")], TypeError, "must be a mapping"),
],
)
def test_secret_values_are_validated_before_remote_request(
secret_values, error_type, message
):
with _mock_remote_function_catalog() as (host, state):
db = lancedb.connect(
"db://dev",
api_key="fake",
host_override=host,
client_config={"retry_config": {"retries": 0}},
)
with pytest.raises(error_type, match=message):
db.create_function_async(normalize_score, secrets=secret_values)
assert state["requests"] == []
@pytest.mark.parametrize(
"value",
[
"x" * _MAX_FUNCTION_SECRET_VALUE_BYTES,
"é" * (_MAX_FUNCTION_SECRET_VALUE_BYTES // len("é".encode("utf-8"))),
],
ids=["ascii", "multibyte"],
)
def test_secret_value_accepts_exact_utf8_byte_limit(value):
submission = json.loads(normalize_score._submission_json({"API_TOKEN": value}))
assert submission["secret_values"]["API_TOKEN"] == value
assert len(value.encode("utf-8")) == _MAX_FUNCTION_SECRET_VALUE_BYTES
@pytest.mark.parametrize(
"value",
[
"x" * (_MAX_FUNCTION_SECRET_VALUE_BYTES + 1),
"é" * (_MAX_FUNCTION_SECRET_VALUE_BYTES // len("é".encode("utf-8")) + 1),
],
ids=["ascii", "multibyte"],
)
def test_secret_value_rejects_over_utf8_byte_limit_before_json_construction(
monkeypatch, value
):
def fail_if_json_construction_starts(self):
pytest.fail("oversized secret reached JSON construction")
monkeypatch.setattr(
FunctionRegistrationRequest, "_known_dict", fail_if_json_construction_starts
)
with pytest.raises(ValueError, match=r"exceeds the 65536-byte limit"):
normalize_score._submission_json({"API_TOKEN": value})
def test_secret_values_accept_exact_aggregate_utf8_byte_limit(monkeypatch):
names = tuple(f"SECRET_{index}" for index in range(8))
value = "é" * (_MAX_FUNCTION_SECRET_VALUE_BYTES // len("é".encode("utf-8")))
values = {name: value for name in names}
monkeypatch.setattr(
normalize_score,
"_request",
normalize_score._request._copy(update={"required_secrets": names}),
)
submission = json.loads(normalize_score._submission_json(values))
assert submission["secret_values"] == values
assert sum(len(item.encode("utf-8")) for item in values.values()) == (
_MAX_FUNCTION_SECRET_VALUES_BYTES
)
def test_secret_values_reject_aggregate_over_limit_before_construction(monkeypatch):
names = tuple(f"SECRET_{index}" for index in range(9))
values = {name: "x" * _MAX_FUNCTION_SECRET_VALUE_BYTES for name in names}
monkeypatch.setattr(
normalize_score,
"_request",
normalize_score._request._copy(update={"required_secrets": names}),
)
def fail_if_json_construction_starts(self):
pytest.fail("oversized aggregate reached JSON construction")
monkeypatch.setattr(
FunctionRegistrationRequest, "_known_dict", fail_if_json_construction_starts
)
with pytest.raises(ValueError, match=r"exceed.*524288-byte request limit"):
normalize_score._submission_json(values)
@pytest.mark.asyncio
async def test_async_remote_registration_submits_secret_values_only_once():
with _mock_remote_function_catalog() as (host, state):
db = await lancedb.connect_async(
"db://dev",
api_key="fake",
host_override=host,
client_config={"retry_config": {"retries": 0}},
)
registration = await db.create_function_async(
normalize_score, secrets={"API_TOKEN": "async-secret"}
)
created = await registration.wait()
assert state["requests"][0][1]["secret_values"] == {"API_TOKEN": "async-secret"}
assert not hasattr(created, "secret_values")
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb"
version = "0.38.0-beta.11"
version = "0.38.0-beta.12"
edition.workspace = true
description = "LanceDB: A serverless, low-latency vector database for AI applications"
license.workspace = true
+4 -309
View File
@@ -5,9 +5,9 @@
//! backend-neutral terminal result of a computed-column refresh.
//!
//! This module contains client/wire values only. Catalog persistence,
//! environment bake, secret resolution, and execution are owned by Sophon.
//! environment bake, and execution are owned by Sophon.
use std::collections::{BTreeMap, BTreeSet};
use std::collections::BTreeMap;
use serde::de::{self, DeserializeOwned};
use serde::{Deserialize, Deserializer, Serialize, Serializer};
@@ -15,16 +15,6 @@ use serde_json::Value;
use crate::{Error, Result};
// Keep these byte limits aligned with Sophon's Function submission validation.
pub(crate) const MAX_FUNCTION_SECRET_VALUE_BYTES: usize = 64 * 1024;
const MAX_FUNCTION_SECRET_VALUES_BYTES: usize = 512 * 1024;
fn is_portable_environment_name(name: &str) -> bool {
let mut bytes = name.bytes();
matches!(bytes.next(), Some(b'A'..=b'Z' | b'a'..=b'z' | b'_'))
&& bytes.all(|byte| matches!(byte, b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'_'))
}
fn invalid_json(error: impl std::fmt::Display) -> Error {
Error::InvalidInput {
message: format!("invalid remote Function JSON: {error}"),
@@ -208,11 +198,6 @@ pub struct PythonEnvironmentSpec {
}
/// Reproducible Python runtime definition understood by Sophon.
///
/// `env` contains non-secret values. Secret values are submission-only in the
/// client model and do not become part of this public runtime identity;
/// [`FunctionVersion::required_secrets`] contains names only. Sophon persists
/// submitted values separately in the private execution artifact.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum PythonRuntimeSpec {
@@ -254,7 +239,7 @@ impl PythonRuntimeSpec {
}
}
/// Non-secret environment variables, or `None` for an unknown kind.
/// Environment variables, or `None` for an unknown kind.
pub fn env(&self) -> Option<&BTreeMap<String, String>> {
match self {
Self::Python { env, .. } => Some(env),
@@ -339,8 +324,6 @@ pub struct FunctionVersion {
runtime: PythonRuntimeSpec,
runtime_digest: String,
environment_digest: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
required_secrets: Vec<String>,
created_at: String,
}
@@ -373,12 +356,6 @@ impl FunctionVersion {
&self.environment_digest
}
/// Required secret names. Resolved values exist only in Sophon's private
/// execution artifact and worker launch path.
pub fn required_secrets(&self) -> &[String] {
&self.required_secrets
}
pub fn created_at(&self) -> &str {
&self.created_at
}
@@ -420,115 +397,12 @@ pub struct FunctionArtifactRequest {
}
/// Stable request envelope for remote immutable Function registration.
///
/// Secret values are submission-only in the client model. Sophon persists them
/// in the database-scoped private execution artifact; returned
/// [`FunctionVersion`] and Job metadata contain only
/// [`Self::required_secrets`] names. Debug formatting always redacts values.
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FunctionRegistrationRequest {
pub name: String,
pub artifact: FunctionArtifactRequest,
pub signature: FunctionSignature,
pub runtime: PythonRuntimeSpec,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub required_secrets: Vec<String>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub secret_values: BTreeMap<String, String>,
}
impl FunctionRegistrationRequest {
pub(crate) fn validate_secret_values(&self) -> Result<()> {
let mut required = BTreeSet::new();
for name in &self.required_secrets {
if !is_portable_environment_name(name) {
return Err(Error::InvalidInput {
message: format!(
"Function secret name {name:?} must be a portable environment variable name"
),
});
}
if !required.insert(name) {
return Err(Error::InvalidInput {
message: format!("Function required_secrets contains duplicate name {name:?}"),
});
}
}
if let PythonRuntimeSpec::Python { env, .. } = &self.runtime
&& let Some(name) = required.iter().find(|name| env.contains_key(**name))
{
return Err(Error::InvalidInput {
message: format!(
"Function runtime env and secret names must be disjoint: {name:?}"
),
});
}
let provided = self.secret_values.keys().collect::<BTreeSet<_>>();
if required != provided {
return Err(Error::InvalidInput {
message: "Function secret_values keys must exactly match required_secrets"
.to_string(),
});
}
let mut total_bytes = 0usize;
for (name, value) in &self.secret_values {
if value.is_empty() {
return Err(Error::InvalidInput {
message: format!("Function secret {name:?} value must be non-empty"),
});
}
if value.contains('\0') {
return Err(Error::InvalidInput {
message: format!("Function secret {name:?} value must not contain NUL"),
});
}
if value.len() > MAX_FUNCTION_SECRET_VALUE_BYTES {
return Err(Error::InvalidInput {
message: format!(
"Function secret {name:?} value exceeds the \
{MAX_FUNCTION_SECRET_VALUE_BYTES}-byte limit"
),
});
}
total_bytes =
total_bytes
.checked_add(value.len())
.ok_or_else(|| Error::InvalidInput {
message: "Function secret values exceed the request byte limit".to_string(),
})?;
}
if total_bytes > MAX_FUNCTION_SECRET_VALUES_BYTES {
return Err(Error::InvalidInput {
message: format!(
"Function secret values exceed the \
{MAX_FUNCTION_SECRET_VALUES_BYTES}-byte request limit"
),
});
}
Ok(())
}
}
impl std::fmt::Debug for FunctionRegistrationRequest {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let secret_values = self
.secret_values
.keys()
.map(|name| (name, "[REDACTED]"))
.collect::<BTreeMap<_, _>>();
formatter
.debug_struct("FunctionRegistrationRequest")
.field("name", &self.name)
.field("artifact", &self.artifact)
.field("signature", &self.signature)
.field("runtime", &self.runtime)
.field("required_secrets", &self.required_secrets)
.field("secret_values", &secret_values)
.finish()
}
}
impl_json!(FunctionRegistrationRequest);
@@ -713,185 +587,6 @@ impl RefreshColumnResult {
impl_json!(RefreshColumnResult);
#[cfg(test)]
mod secret_value_tests {
use super::{
FunctionRegistrationRequest, MAX_FUNCTION_SECRET_VALUE_BYTES,
MAX_FUNCTION_SECRET_VALUES_BYTES, PythonRuntimeSpec,
};
use crate::Error;
fn request() -> FunctionRegistrationRequest {
FunctionRegistrationRequest::from_json(include_str!(
"../tests/fixtures/first_class_functions/v1/remote_function_registration_request.json"
))
.unwrap()
}
#[test]
fn validates_secret_name_and_value_invariants() {
let missing = request();
assert!(matches!(
missing.validate_secret_values(),
Err(Error::InvalidInput { message }) if message.contains("exactly match")
));
let mut empty = request();
empty
.secret_values
.insert("API_TOKEN".to_string(), String::new());
assert!(matches!(
empty.validate_secret_values(),
Err(Error::InvalidInput { message }) if message.contains("non-empty")
));
let mut nul = request();
nul.secret_values
.insert("API_TOKEN".to_string(), "before\0after".to_string());
assert!(matches!(
nul.validate_secret_values(),
Err(Error::InvalidInput { message }) if message.contains("NUL")
));
let mut unexpected = request();
unexpected
.secret_values
.insert("OTHER".to_string(), "value".to_string());
assert!(matches!(
unexpected.validate_secret_values(),
Err(Error::InvalidInput { message }) if message.contains("exactly match")
));
}
#[test]
fn rejects_invalid_duplicate_and_overlapping_secret_declarations() {
let mut invalid_name = request();
invalid_name.required_secrets = vec!["BAD=NAME".to_string()];
invalid_name
.secret_values
.insert("BAD=NAME".to_string(), "secret".to_string());
let mut duplicate = request();
duplicate.required_secrets = vec!["API_TOKEN".to_string(), "API_TOKEN".to_string()];
duplicate
.secret_values
.insert("API_TOKEN".to_string(), "secret".to_string());
let mut overlap = request();
overlap
.secret_values
.insert("API_TOKEN".to_string(), "secret".to_string());
if let PythonRuntimeSpec::Python { env, .. } = &mut overlap.runtime {
env.insert("API_TOKEN".to_string(), "public".to_string());
}
assert!(matches!(
invalid_name.validate_secret_values(),
Err(Error::InvalidInput { message }) if message.contains("portable environment variable")
));
assert!(matches!(
duplicate.validate_secret_values(),
Err(Error::InvalidInput { message }) if message.contains("duplicate")
));
assert!(matches!(
overlap.validate_secret_values(),
Err(Error::InvalidInput { message }) if message.contains("must be disjoint")
));
}
#[test]
fn enforces_portable_secret_name_boundaries() {
for name in ["A", "_", "A0_"] {
let mut request = request();
request.required_secrets = vec![name.to_string()];
request
.secret_values
.insert(name.to_string(), "secret".to_string());
request.validate_secret_values().unwrap();
}
for name in ["", "0TOKEN", "BAD-NAME", "TÖKEN"] {
let mut request = request();
request.required_secrets = vec![name.to_string()];
request
.secret_values
.insert(name.to_string(), "secret".to_string());
assert!(matches!(
request.validate_secret_values(),
Err(Error::InvalidInput { message })
if message.contains("portable environment variable")
));
}
}
#[test]
fn accepts_exact_secret_value_utf8_byte_limit() {
for value in [
"x".repeat(MAX_FUNCTION_SECRET_VALUE_BYTES),
"é".repeat(MAX_FUNCTION_SECRET_VALUE_BYTES / "é".len()),
] {
assert_eq!(value.len(), MAX_FUNCTION_SECRET_VALUE_BYTES);
let mut request = request();
request.secret_values.insert("API_TOKEN".to_string(), value);
request.validate_secret_values().unwrap();
}
}
#[test]
fn rejects_secret_value_over_utf8_byte_limit() {
for value in [
"x".repeat(MAX_FUNCTION_SECRET_VALUE_BYTES + 1),
"é".repeat(MAX_FUNCTION_SECRET_VALUE_BYTES / "é".len() + 1),
] {
assert!(value.len() > MAX_FUNCTION_SECRET_VALUE_BYTES);
let mut request = request();
request.secret_values.insert("API_TOKEN".to_string(), value);
assert!(matches!(
request.validate_secret_values(),
Err(Error::InvalidInput { message }) if message.contains("65536-byte limit")
));
}
}
#[test]
fn rejects_aggregate_secret_value_bytes_over_server_limit() {
let mut request = request();
request.required_secrets = (0..9).map(|index| format!("SECRET_{index}")).collect();
request.secret_values = request
.required_secrets
.iter()
.map(|name| (name.clone(), "x".repeat(MAX_FUNCTION_SECRET_VALUE_BYTES)))
.collect();
assert!(matches!(
request.validate_secret_values(),
Err(Error::InvalidInput { message })
if message.contains(&format!("{MAX_FUNCTION_SECRET_VALUES_BYTES}-byte request limit"))
));
}
#[test]
fn accepts_exact_aggregate_secret_value_byte_limit() {
let mut request = request();
request.required_secrets = (0..8).map(|index| format!("SECRET_{index}")).collect();
request.secret_values = request
.required_secrets
.iter()
.map(|name| (name.clone(), "x".repeat(MAX_FUNCTION_SECRET_VALUE_BYTES)))
.collect();
assert_eq!(
request
.secret_values
.values()
.map(String::len)
.sum::<usize>(),
MAX_FUNCTION_SECRET_VALUES_BYTES
);
request.validate_secret_values().unwrap();
}
}
#[cfg(test)]
mod conda_environment_tests {
use super::PythonEnvironmentSpec;
+15 -100
View File
@@ -7,7 +7,6 @@ use reqwest::{
Body, Request, RequestBuilder, Response,
header::{HeaderMap, HeaderValue},
};
use serde_json::Value;
use std::{collections::HashMap, future::Future, str::FromStr, sync::Arc, time::Duration};
use crate::error::{Error, Result};
@@ -15,60 +14,6 @@ use crate::remote::db::RemoteOptions;
use crate::remote::retry::{ResolvedRetryConfig, RetryCounter};
const REQUEST_ID_HEADER: HeaderName = HeaderName::from_static("x-request-id");
const REDACTED_JSON_VALUE: &str = "[REDACTED]";
const SUPPRESSED_JSON_BODY: &str = "[JSON BODY SUPPRESSED]";
fn is_sensitive_json_field(name: &str) -> bool {
name.to_ascii_lowercase().contains("secret")
}
fn redact_sensitive_json_fields(value: &mut Value) {
match value {
Value::Object(fields) => {
for (name, child) in fields {
if is_sensitive_json_field(name) {
*child = Value::String(REDACTED_JSON_VALUE.to_string());
} else {
redact_sensitive_json_fields(child);
}
}
}
Value::Array(values) => values.iter_mut().for_each(redact_sensitive_json_fields),
_ => {}
}
}
fn redacted_json_body(request: &Request) -> Option<String> {
let body = request.body()?.as_bytes()?;
let mut value = serde_json::from_slice(body).ok()?;
redact_sensitive_json_fields(&mut value);
serde_json::to_string(&value).ok()
}
fn request_log_message(request: &Request, request_id: &str) -> String {
let prefix = format!(
"Sending request_id={}: {} {}",
request_id,
request.method(),
request.url()
);
let content_type = request
.headers()
.get("content-type")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.split(';').next());
if content_type.is_some_and(|value| value.eq_ignore_ascii_case("application/json")) {
// Never format the raw Request here: its Debug representation is not a
// redaction boundary and may include the original body. If the JSON body
// cannot be structurally parsed, suppress it instead of logging raw bytes.
let body = redacted_json_body(request).unwrap_or_else(|| SUPPRESSED_JSON_BODY.to_string());
format!("{prefix} with body {body}")
} else {
// Method and URL are sufficient request context. Raw Request formatting
// may expose headers or a non-JSON body, so it is never a logging fallback.
prefix
}
}
/// Configuration for TLS/mTLS settings.
#[derive(Clone, Debug)]
@@ -894,9 +839,22 @@ impl<S: HttpSend> RestfulLanceDbClient<S> {
}
}
pub(crate) fn log_request(&self, request: &Request, request_id: &str) {
pub(crate) fn log_request(&self, request: &Request, request_id: &String) {
if log::log_enabled!(log::Level::Debug) {
debug!("{}", request_log_message(request, request_id));
let content_type = request
.headers()
.get("content-type")
.map(|v| v.to_str().unwrap());
if content_type == Some("application/json") {
let body = request.body().as_ref().unwrap().as_bytes().unwrap();
let body = String::from_utf8_lossy(body);
debug!(
"Sending request_id={}: {:?} with body {}",
request_id, request, body
);
} else {
debug!("Sending request_id={}: {:?}", request_id, request);
}
}
}
@@ -1119,49 +1077,6 @@ mod tests {
ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner())
}
#[test]
fn test_request_log_message_redacts_secrets_and_never_formats_raw_requests() {
const SECRET_SENTINEL: &str = "udf-secret-log-sentinel-7e4e";
const MALFORMED_SENTINEL: &str = "malformed-secret-log-sentinel-b652";
const NON_JSON_SENTINEL: &str = "non-json-secret-log-sentinel-7fd1";
let request = reqwest::Client::new()
.post("https://example.com/v1/functions/create")
.json(&serde_json::json!({
"name": "uses_secret",
"nested": {
"secret_values": {"OPENAI_API_KEY": SECRET_SENTINEL},
"safe": "visible-value"
}
}))
.build()
.unwrap();
let log_message = request_log_message(&request, "valid-json");
let malformed_request = reqwest::Client::new()
.post("https://example.com/v1/functions/create")
.header("content-type", "application/json; charset=utf-8")
.body(format!(r#"{{"secret_values":"{MALFORMED_SENTINEL}""#))
.build()
.unwrap();
let malformed_log_message = request_log_message(&malformed_request, "malformed-json");
let non_json_request = reqwest::Client::new()
.post("https://example.com/v1/functions/create")
.header("content-type", "text/plain")
.body(NON_JSON_SENTINEL)
.build()
.unwrap();
let non_json_log_message = request_log_message(&non_json_request, "non-json");
assert!(log_message.contains("visible-value"));
assert!(log_message.contains(REDACTED_JSON_VALUE));
assert!(!log_message.contains(SECRET_SENTINEL));
assert!(malformed_log_message.contains(SUPPRESSED_JSON_BODY));
assert!(!malformed_log_message.contains(MALFORMED_SENTINEL));
assert!(!non_json_log_message.contains(NON_JSON_SENTINEL));
}
#[test]
fn test_timeout_config_default() {
let config = TimeoutConfig::default();
+2 -33
View File
@@ -554,7 +554,6 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
&self,
request: FunctionRegistrationRequest,
) -> Result<Job<FunctionVersion>> {
request.validate_secret_values()?;
let req = self.client.post("/v1/functions/create").json(&request);
let (request_id, response) = self.client.send(req).await?;
let response = self.client.check_response(&request_id, response).await?;
@@ -2643,8 +2642,7 @@ mod tests {
);
const FUNCTION_JOB: &str =
include_str!("../../tests/fixtures/first_class_functions/v1/remote_function_job.json");
let mut expected: serde_json::Value = serde_json::from_str(REQUEST).unwrap();
expected["secret_values"] = serde_json::json!({"API_TOKEN": "secret-value"});
let expected: serde_json::Value = serde_json::from_str(REQUEST).unwrap();
let conn = Connection::new_with_handler(move |request| match request.url().path() {
"/v1/functions/create" => {
assert_eq!(request.method(), &reqwest::Method::POST);
@@ -2662,10 +2660,7 @@ mod tests {
.unwrap(),
path => panic!("unexpected path: {path}"),
});
let mut request = crate::function::FunctionRegistrationRequest::from_json(REQUEST).unwrap();
request
.secret_values
.insert("API_TOKEN".to_string(), "secret-value".to_string());
let request = crate::function::FunctionRegistrationRequest::from_json(REQUEST).unwrap();
let job = conn.create_function_async(request).await.unwrap();
assert_eq!(job.id(), Some("job-function-1"));
let version = job.wait().await.unwrap();
@@ -2673,32 +2668,6 @@ mod tests {
assert_eq!(version.version(), "fv_01K3EXACT");
}
#[tokio::test]
async fn test_create_function_async_validates_secrets_before_serialization_and_send() {
const REQUEST: &str = include_str!(
"../../tests/fixtures/first_class_functions/v1/remote_function_registration_request.json"
);
let sends = Arc::new(AtomicUsize::new(0));
let sends_ref = sends.clone();
let conn = Connection::new_with_handler(move |_| {
sends_ref.fetch_add(1, Ordering::SeqCst);
http::Response::builder().status(500).body("").unwrap()
});
let mut request = crate::function::FunctionRegistrationRequest::from_json(REQUEST).unwrap();
request.secret_values.insert(
"API_TOKEN".to_string(),
"x".repeat(crate::function::MAX_FUNCTION_SECRET_VALUE_BYTES + 1),
);
let error = conn.create_function_async(request).await.unwrap_err();
assert!(matches!(
error,
Error::InvalidInput { message } if message.contains("65536-byte limit")
));
assert_eq!(sends.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn test_get_function_requires_and_sends_exact_version() {
const VERSION: &str = include_str!(
+97 -8
View File
@@ -22,7 +22,7 @@
use std::collections::{BTreeSet, HashMap};
use std::sync::Arc;
use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema, SchemaRef};
use arrow_schema::{DataType, Field as ArrowField, Fields, Schema as ArrowSchema, SchemaRef};
use datafusion_common::tree_node::TreeNode;
use datafusion_physical_plan::PhysicalExpr;
use lance::dataset::NewColumnTransform;
@@ -1273,6 +1273,11 @@ pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result<
/// refresh time: that the expression parses, that every column it reads
/// exists, and that the target name is free. A declaration that survives this
/// is one a refresh can always act on.
///
/// Each accepted column joins the schema the next one resolves against, so a
/// batch may declare `a` and then `b = a + 1` in one commit. Refresh order
/// then matters, and refresh enforces it: `b` is refused while `a` still has
/// unfilled rows.
pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Vec<ArrowField>> {
if columns.is_empty() {
return Err(Error::InvalidInput {
@@ -1280,11 +1285,11 @@ pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Ve
});
}
let mut schema = schema;
let mut fields = Vec::with_capacity(columns.len());
let mut declared: Vec<&str> = Vec::with_capacity(columns.len());
for (name, expression) in columns {
if schema.field_with_name(name).is_ok() || declared.contains(&name.as_str()) {
if schema.field_with_name(name).is_ok() {
return Err(Error::ColumnAlreadyExists { name: name.clone() });
}
@@ -1292,16 +1297,50 @@ pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Ve
// Declared columns start entirely null, so nullability is a property
// of the declaration rather than of what the expression yields.
fields.push(
ArrowField::new(name, bound.data_type, true)
.with_metadata(computed_column_metadata(expression, &bound.inputs)),
);
declared.push(name);
let field = ArrowField::new(name, bound.data_type, true)
.with_metadata(computed_column_metadata(expression, &bound.inputs));
schema = Arc::new(ArrowSchema::new_with_metadata(
schema
.fields()
.iter()
.cloned()
.chain(std::iter::once(Arc::new(field.clone())))
.collect::<Fields>(),
schema.metadata().clone(),
));
fields.push(field);
}
Ok(fields)
}
/// Run the schema-level checks of
/// [`AddColumnsBuilder::computed`](super::AddColumnsBuilder::computed) against
/// `schema` without committing: the Function-binding guard and the planning of
/// every declaration. For callers that stage declarations behind other work
/// and need those rejections before any of it lands.
///
/// Only the schema is consulted. Declaring also refuses a table with an LSM
/// write spec or retained SSTables; that is table state, checked at commit.
///
/// ```
/// # use std::sync::Arc;
/// # use arrow_schema::{DataType, Field, Schema};
/// use lancedb::table::computed_columns::validate_declarations;
///
/// let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
/// let declarations = vec![
/// ("a".to_string(), "x + 1".to_string()),
/// ("b".to_string(), "a * 2".to_string()),
/// ];
/// assert!(validate_declarations(schema.clone(), &declarations).is_ok());
/// assert!(validate_declarations(schema, &[("c".into(), "random()".into())]).is_err());
/// ```
pub fn validate_declarations(schema: SchemaRef, columns: &[(String, String)]) -> Result<()> {
ensure_no_function_bindings_for_mutation(schema.as_ref(), "schema evolution")?;
plan(schema, columns).map(drop)
}
/// Build the transform that declares `columns` against `schema`.
///
/// An all-null column is how a binding with no values yet is carried into a
@@ -1340,6 +1379,22 @@ pub(super) async fn add_foreign_kind(table: &crate::Table, name: &str, kind: &st
#[cfg(test)]
mod tests {
/// The gate's reproducer: the validator applies the same schema-level
/// guard declaring does, so a staging caller is refused before it commits
/// anything else.
#[test]
fn test_validate_declarations_matches_schema_admission_barriers() {
let schema = Arc::new(ArrowSchema::new_with_metadata(
vec![ArrowField::new("x", DataType::Int32, true)],
HashMap::from([(
FUNCTION_BINDINGS_META_KEY.to_string(),
"not valid binding metadata".to_string(),
)]),
));
let declarations = vec![("a".to_string(), "x + 1".to_string())];
assert!(super::validate_declarations(schema, &declarations).is_err());
}
#[test]
fn output_arrow_type_grammar_matches_the_shared_golden() {
let golden: serde_json::Value = serde_json::from_str(include_str!(
@@ -1582,6 +1637,40 @@ mod tests {
assert!(declared(&table).await.is_empty());
}
/// A batch may build on itself: one commit, and the later entry's inputs
/// name the earlier one.
#[tokio::test]
async fn test_a_declaration_may_read_one_declared_before_it() {
let table = table_with_ints("chain").await;
let before = table.version().await.unwrap();
add_computed(
&table,
&[("a".into(), "x + 1".into()), ("b".into(), "a * 2".into())],
)
.await
.unwrap();
assert_eq!(table.version().await.unwrap(), before + 1);
let declared = declared(&table).await;
assert_eq!(declared[1].name, "b");
assert_eq!(declared[1].inputs, vec!["a".to_string()]);
// Order is the dependency order; reading ahead is still unknown.
let err = add_computed(
&table,
&[("c".into(), "d + 1".into()), ("d".into(), "x + 1".into())],
)
.await
.unwrap_err();
assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "c"));
assert!(
validate_declarations(
table.schema().await.unwrap(),
&[("e".into(), "random()".into())]
)
.is_err()
);
}
/// A column added by an ordinary transform is materialized, not bound, so
/// it carries no declaration to report.
#[tokio::test]
+173 -17
View File
@@ -7,6 +7,16 @@
//! therefore idempotent and does not observe input mutation -- once a row is
//! filled, changing what the expression reads leaves the stored result alone.
//!
//! A column's computed inputs are filled first -- the dependency graph is
//! walked once, each reachable column filled once in dependency order, each
//! fill its own commit. Every fill in the pass, the requested column's
//! included, covers only the fragments of the snapshot the pass started
//! from: a commit may rebase over a concurrent append, and the fragment that
//! admits carries placeholder nulls no earlier fill covered, so it waits for
//! a later refresh rather than being read as values. Two concurrent fills of
//! one input collide on its field in lance's conflict check, so a dependent
//! fill can only commit over inputs that were durable when it read them.
//!
//! Two passes per fragment. The first scans only the unfilled live rows and
//! evaluates the expression over them, which yields the exact fill count and
//! decides whether the fragment is staged at all -- a fragment where nothing
@@ -41,7 +51,8 @@ use crate::{Error, Result};
/// The result of refreshing a computed column.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct RefreshColumnResult {
/// Rows that had a value computed.
/// Rows that had a value computed, in the requested column only; inputs
/// filled on its behalf are not counted.
#[serde(default)]
pub rows_filled: u64,
/// The commit version associated with the operation.
@@ -52,6 +63,7 @@ pub struct RefreshColumnResult {
struct RefreshExecution {
result: RefreshColumnResult,
source_version: u64,
published_version: Option<u64>,
}
/// Internal implementation of the refresh logic.
@@ -74,7 +86,12 @@ async fn execute_refresh_column_with_source(
let expression = declared_expression(&dataset, column)?;
let schema = Arc::new(ArrowSchema::from(dataset.schema()));
let bound = Arc::new(super::computed_columns::bind(schema, column, &expression)?);
let bound = Arc::new(super::computed_columns::bind(
schema.clone(),
column,
&expression,
)?);
ensure_inputs_filled(&dataset, &schema, column, &bound).await?;
let field = dataset
.schema()
.field(column)
@@ -100,25 +117,25 @@ async fn execute_refresh_column_with_source(
replacements.push(fragment.write_columns(values, &column_schema).await?);
}
let source_version = dataset.version().version;
if replacements.is_empty() {
let source_version = dataset.version().version;
return Ok(RefreshExecution {
result: RefreshColumnResult {
rows_filled: 0,
version: source_version,
},
source_version,
published_version: None,
});
}
let read_version = dataset.version().version;
// The dataset's own session, so registrations and caches survive the
// commit being installed on the handle.
let session = dataset.session();
let new_dataset = Dataset::commit(
WriteDestination::Dataset(dataset.clone()),
Operation::DataReplacement { replacements },
Some(read_version),
Some(source_version),
None,
None,
session,
@@ -133,10 +150,52 @@ async fn execute_refresh_column_with_source(
rows_filled,
version,
},
source_version: read_version,
source_version,
published_version: Some(version),
})
}
/// Refuse while a computed input still has rows a refresh of it would fill:
/// read now, its placeholder null would be evaluated as a value and kept.
async fn ensure_inputs_filled(
dataset: &Dataset,
schema: &Arc<ArrowSchema>,
column: &str,
bound: &BoundExpression,
) -> Result<()> {
for input in &bound.roots {
let Some(declaration) = schema
.field_with_name(input)
.ok()
.and_then(computed_column_from_field)
else {
continue;
};
let ComputedColumnKind::Sql { expression } = &declaration.kind else {
return Err(Error::NotSupported {
message: format!(
"computed column '{column}' reads '{input}', whose fill state this \
refresh cannot check; refresh '{input}' first"
),
});
};
let input_bound = super::computed_columns::bind(schema.clone(), input, expression)?;
let mut unfilled = 0u64;
for fragment in dataset.get_fragments() {
unfilled += count_fragment_gains(dataset, &fragment, &input_bound, input).await?;
}
if unfilled > 0 {
return Err(Error::InvalidInput {
message: format!(
"computed column '{column}' reads '{input}', which has {unfilled} unfilled \
rows; refresh '{input}' first"
),
});
}
}
Ok(())
}
/// Run the refresh as a [`Job`] in this process.
pub(crate) async fn execute_refresh_column_async(
table: &NativeTable,
@@ -160,8 +219,7 @@ pub(crate) async fn execute_refresh_column_async(
rows_failed: 0,
rows_remaining: 0,
source_version: execution.source_version,
published_version: (execution.result.rows_filled > 0)
.then_some(execution.result.version),
published_version: execution.published_version,
})
})))
}
@@ -384,7 +442,8 @@ mod tests {
.version)
}
async fn read(table: &Table, column: &str) -> Vec<Option<i32>> {
async fn read(table: &Table, column: &str) -> Vec<Option<i64>> {
use arrow_array::{Array, Int64Array};
let batches = table
.query()
.select(Select::columns(&[column]))
@@ -394,15 +453,19 @@ mod tests {
.try_collect::<Vec<_>>()
.await
.unwrap();
let mut values: Vec<Option<i32>> = batches
let mut values: Vec<Option<i64>> = batches
.iter()
.flat_map(|batch| {
batch[column]
.as_any()
.downcast_ref::<Int32Array>()
.unwrap()
.iter()
.collect::<Vec<_>>()
let array = &batch[column];
match array.as_any().downcast_ref::<Int32Array>() {
Some(ints) => ints.iter().map(|v| v.map(i64::from)).collect::<Vec<_>>(),
None => array
.as_any()
.downcast_ref::<Int64Array>()
.unwrap()
.iter()
.collect::<Vec<_>>(),
}
})
.collect();
values.sort();
@@ -414,6 +477,98 @@ mod tests {
table.add(batch).execute().await.unwrap();
}
/// The gate's reproducer: `b = coalesce(a, 0)` refreshed before `a`
/// must not bake zeros from `a`'s placeholder null. It is refused, and
/// names the input, until `a` is filled -- after every append too.
#[tokio::test]
async fn test_dependent_refresh_refuses_an_unfilled_input() {
let table = table_with("dependent_refresh_order", vec![1, 2, 3]).await;
table
.add_columns()
.computed("a", "x + 1")
.computed("b", "coalesce(a, 0)")
.execute()
.await
.unwrap();
let err = table.refresh_column("b").await.unwrap_err();
assert!(
matches!(&err, Error::InvalidInput { message } if message.contains("refresh 'a' first")),
"{err}"
);
assert_eq!(read(&table, "b").await, vec![None, None, None]);
assert_eq!(table.refresh_column("a").await.unwrap().rows_filled, 3);
assert_eq!(table.refresh_column("b").await.unwrap().rows_filled, 3);
assert_eq!(read(&table, "b").await, vec![Some(2), Some(3), Some(4)]);
append(&table, vec![10]).await;
assert!(table.refresh_column("b").await.is_err());
table.refresh_column("a").await.unwrap();
assert_eq!(table.refresh_column("b").await.unwrap().rows_filled, 1);
assert_eq!(
table.count_rows(Some("b = 0".to_string())).await.unwrap(),
0
);
}
/// Names that need quoting, and a nested input, survive the trip through
/// declaration metadata and the dependency check: the recorded inputs
/// are matched by name, never re-parsed as SQL.
#[tokio::test]
async fn test_dependent_refresh_handles_awkward_column_names() {
use arrow_array::{Int32Array, StructArray};
use arrow_schema::{DataType, Field, Fields};
let conn = connect("memory://").execute().await.unwrap();
let age_fields = Fields::from(vec![Field::new("age", DataType::Int32, true)]);
let meta = StructArray::new(
age_fields.clone(),
vec![Arc::new(Int32Array::from(vec![10, 20])) as _],
None,
);
let schema = Arc::new(arrow_schema::Schema::new(vec![
Field::new("camelCase", DataType::Int32, true),
Field::new("with-hyphen", DataType::Int32, true),
Field::new("meta", DataType::Struct(age_fields), true),
]));
let batch = arrow_array::RecordBatch::try_new(
schema,
vec![
Arc::new(Int32Array::from(vec![1, 2])) as _,
Arc::new(Int32Array::from(vec![100, 200])) as _,
Arc::new(meta) as _,
],
)
.unwrap();
let table = conn
.create_table("awkward_names", batch)
.execute()
.await
.unwrap();
table
.add_columns()
.computed("y", "`camelCase` * 2")
.computed("z", "coalesce(y, 0) + `with-hyphen` + meta.age")
.execute()
.await
.unwrap();
let z = crate::table::computed_columns::computed_columns(
table.schema().await.unwrap().as_ref(),
)
.into_iter()
.find(|c| c.name == "z")
.unwrap();
assert_eq!(z.inputs, vec!["meta.age", "with-hyphen", "y"]);
let err = table.refresh_column("z").await.unwrap_err();
assert!(err.to_string().contains("refresh 'y' first"), "{err}");
assert_eq!(table.refresh_column("y").await.unwrap().rows_filled, 2);
assert_eq!(table.refresh_column("z").await.unwrap().rows_filled, 2);
assert_eq!(read(&table, "z").await, vec![Some(112), Some(224)]);
}
#[tokio::test]
async fn test_refresh_fills_a_declared_column() {
let table = table_with("refresh_fills", vec![1, 2, 3]).await;
@@ -651,7 +806,8 @@ mod tests {
let read_back = read(&table, "doubled").await;
assert_eq!(read_back.len(), 20_000);
let mut expected: Vec<Option<i32>> = values.iter().map(|v| Some(v * 2)).collect();
let mut expected: Vec<Option<i64>> =
values.iter().map(|v| Some(i64::from(v * 2))).collect();
expected.sort();
assert_eq!(read_back, expected);
}
@@ -20,25 +20,6 @@ fn job_result(name: &str) -> Value {
serde_json::from_str::<Value>(&fixture(name)).expect("remote Job fixture")["result"].clone()
}
fn assert_no_secret_values(value: &Value) {
match value {
Value::Object(values) => {
for (key, value) in values {
assert!(
!matches!(
key.as_str(),
"secret_value" | "secret_values" | "resolved_secret" | "resolved_secrets"
),
"client canonical value must not model resolved secret material"
);
assert_no_secret_values(value);
}
}
Value::Array(values) => values.iter().for_each(assert_no_secret_values),
_ => {}
}
}
#[test]
fn function_version_job_result_matches_shared_canonical_golden() {
let result = job_result("remote_function_job.json");
@@ -47,7 +28,6 @@ fn function_version_job_result_matches_shared_canonical_golden() {
assert_eq!(version.name(), "embed");
assert_eq!(version.version(), "fv_01K3EXACT");
assert_eq!(version.runtime_digest(), "sha256:runtime");
assert_eq!(version.required_secrets(), &["HF_TOKEN"]);
assert_eq!(
version.to_canonical_json().expect("canonical JSON"),
fixture("remote_function_version.canonical.json").trim()
@@ -162,21 +142,3 @@ fn floating_point_application_literals_are_rejected_consistently() {
.contains("floating-point Function literals")
);
}
#[test]
fn canonical_client_values_contain_secret_names_only() {
let result = job_result("remote_function_job.json");
let version = FunctionVersion::from_json(&result.to_string()).expect("FunctionVersion result");
let canonical: Value = serde_json::from_str(
&version
.to_canonical_json()
.expect("canonical FunctionVersion"),
)
.expect("canonical JSON");
assert_eq!(
canonical["required_secrets"],
serde_json::json!(["HF_TOKEN"])
);
assert_no_secret_values(&canonical);
}
@@ -6,7 +6,6 @@ use std::path::PathBuf;
use lancedb::Error;
use lancedb::function::FunctionRegistrationRequest;
use serde_json::Value;
fn fixture(name: &str) -> String {
let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
@@ -15,25 +14,6 @@ fn fixture(name: &str) -> String {
fs::read_to_string(path).expect("fixture must be readable")
}
fn assert_no_secret_values(value: &Value) {
match value {
Value::Object(values) => {
for (key, value) in values {
assert!(
!matches!(
key.as_str(),
"secret_value" | "secret_values" | "resolved_secret" | "resolved_secrets"
),
"registration requests must not model resolved secret material"
);
assert_no_secret_values(value);
}
}
Value::Array(values) => values.iter().for_each(assert_no_secret_values),
_ => {}
}
}
#[test]
fn registration_request_matches_shared_canonical_golden() {
let request = FunctionRegistrationRequest::from_json(&fixture(
@@ -42,33 +22,10 @@ fn registration_request_matches_shared_canonical_golden() {
.expect("registration request");
assert_eq!(request.name, "normalize_score");
assert_eq!(request.artifact.adapter.kind, "scalar_to_arrow_batch");
assert_eq!(request.required_secrets, ["API_TOKEN"]);
assert!(request.secret_values.is_empty());
assert_eq!(
request.to_canonical_json().expect("canonical request"),
fixture("remote_function_registration_request.canonical.json").trim()
);
let value: Value =
serde_json::from_str(&request.to_canonical_json().expect("canonical request"))
.expect("request JSON");
assert_no_secret_values(&value);
}
#[test]
fn registration_request_serializes_secret_values_but_redacts_debug_output() {
let mut value: Value =
serde_json::from_str(&fixture("remote_function_registration_request.json")).unwrap();
value["secret_values"] = serde_json::json!({"API_TOKEN": "secret-plaintext"});
let request = FunctionRegistrationRequest::from_json(&value.to_string()).unwrap();
assert_eq!(request.secret_values["API_TOKEN"], "secret-plaintext");
let canonical = request.to_canonical_json().unwrap();
assert!(canonical.contains("secret-plaintext"));
let debug = format!("{request:?}");
assert!(debug.contains("API_TOKEN"));
assert!(debug.contains("[REDACTED]"));
assert!(!debug.contains("secret-plaintext"));
}
#[tokio::test]
@@ -24,7 +24,6 @@
},
"runtime_digest": "sha256:runtime",
"environment_digest": "sha256:environment",
"required_secrets": ["HF_TOKEN"],
"created_at": "2026-08-21T00:00:00Z"
},
"future_job": {"trace_id": "trace-1"}
@@ -1 +1 @@
{"artifact":{"adapter":{"kind":"scalar_to_arrow_batch","version":1},"content":{"data":"ZnJvbSBfX2Z1dHVyZV9fIGltcG9ydCBhbm5vdGF0aW9ucwoKZGVmIG5vcm1hbGl6ZV9zY29yZSh2YWx1ZTogZmxvYXQpIC0+IGZsb2F0OgogICAgcmV0dXJuIHZhbHVlIC8gMTAwLjAK","encoding":"base64"},"digest":"sha256:760784bdcef57b802f389b97804cc0b618aae39e86044733451bcd0b13089a7f","entrypoint":"normalize_score","kind":"python_callable"},"name":"normalize_score","required_secrets":["API_TOKEN"],"runtime":{"env":{"MODE":"test"},"environment":{"kind":"pip","packages":["numpy>=2"]},"kind":"python","python_version":"3.12"},"signature":{"inputs":[{"arrow_type":"float64","name":"value","nullable":false}],"output":{"arrow_type":"float64","kind":"scalar","nullable":false}}}
{"artifact":{"adapter":{"kind":"scalar_to_arrow_batch","version":1},"content":{"data":"ZnJvbSBfX2Z1dHVyZV9fIGltcG9ydCBhbm5vdGF0aW9ucwoKZGVmIG5vcm1hbGl6ZV9zY29yZSh2YWx1ZTogZmxvYXQpIC0+IGZsb2F0OgogICAgcmV0dXJuIHZhbHVlIC8gMTAwLjAK","encoding":"base64"},"digest":"sha256:760784bdcef57b802f389b97804cc0b618aae39e86044733451bcd0b13089a7f","entrypoint":"normalize_score","kind":"python_callable"},"name":"normalize_score","runtime":{"env":{"MODE":"test"},"environment":{"kind":"pip","packages":["numpy>=2"]},"kind":"python","python_version":"3.12"},"signature":{"inputs":[{"arrow_type":"float64","name":"value","nullable":false}],"output":{"arrow_type":"float64","kind":"scalar","nullable":false}}}
@@ -39,8 +39,5 @@
"env": {
"MODE": "test"
}
},
"required_secrets": [
"API_TOKEN"
]
}
}
@@ -1 +1 @@
{"artifact":{"digest":"sha256:code","entrypoint":"embed","kind":"python_callable"},"created_at":"2026-08-21T00:00:00Z","environment_digest":"sha256:environment","name":"embed","required_secrets":["HF_TOKEN"],"runtime":{"env":{"TOKENIZERS_PARALLELISM":"false"},"environment":{"kind":"pip","packages":["sentence-transformers>=3"]},"kind":"python","python_version":"3.12"},"runtime_digest":"sha256:runtime","signature":{"inputs":[{"arrow_type":"utf8","name":"text","nullable":true}],"output":{"arrow_type":"list<float32>","kind":"scalar","nullable":false}},"version":"fv_01K3EXACT"}
{"artifact":{"digest":"sha256:code","entrypoint":"embed","kind":"python_callable"},"created_at":"2026-08-21T00:00:00Z","environment_digest":"sha256:environment","name":"embed","runtime":{"env":{"TOKENIZERS_PARALLELISM":"false"},"environment":{"kind":"pip","packages":["sentence-transformers>=3"]},"kind":"python","python_version":"3.12"},"runtime_digest":"sha256:runtime","signature":{"inputs":[{"arrow_type":"utf8","name":"text","nullable":true}],"output":{"arrow_type":"list<float32>","kind":"scalar","nullable":false}},"version":"fv_01K3EXACT"}