diff --git a/.bumpversion.toml b/.bumpversion.toml index e32f518c9..a6912bed5 100644 --- a/.bumpversion.toml +++ b/.bumpversion.toml @@ -1,5 +1,5 @@ [tool.bumpversion] -current_version = "0.40.0-beta.11" +current_version = "0.40.0-beta.12" parse = """(?x) (?P0|[1-9]\\d*)\\. (?P0|[1-9]\\d*)\\. diff --git a/Cargo.lock b/Cargo.lock index 2d626f9bf..bb986e05a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3588,8 +3588,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" [[package]] name = "fsst" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-array", "rand 0.9.5", @@ -5143,8 +5143,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a" [[package]] name = "lance" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arc-swap", "arrow", @@ -5217,7 +5217,7 @@ dependencies = [ [[package]] name = "lance-arrow" -version = "13.0.0-beta.18" +version = "13.1.0-beta.1" dependencies = [ "arrow-array", "arrow-buffer", @@ -5239,7 +5239,7 @@ dependencies = [ [[package]] name = "lance-arrow-scalar" version = "58.0.0" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-array", "arrow-buffer", @@ -5253,7 +5253,7 @@ dependencies = [ [[package]] name = "lance-arrow-stats" version = "58.0.0" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-array", "arrow-schema", @@ -5262,8 +5262,8 @@ dependencies = [ [[package]] name = "lance-bitpacking" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrayref", "crunchy", @@ -5273,8 +5273,8 @@ dependencies = [ [[package]] name = "lance-core" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-array", "arrow-buffer", @@ -5311,8 +5311,8 @@ dependencies = [ [[package]] name = "lance-datafusion" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow", "arrow-array", @@ -5343,8 +5343,8 @@ dependencies = [ [[package]] name = "lance-datagen" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow", "arrow-array", @@ -5361,8 +5361,8 @@ dependencies = [ [[package]] name = "lance-derive" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "proc-macro2", "quote", @@ -5371,8 +5371,8 @@ dependencies = [ [[package]] name = "lance-encoding" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-arith", "arrow-array", @@ -5405,8 +5405,8 @@ dependencies = [ [[package]] name = "lance-file" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-arith", "arrow-array", @@ -5438,8 +5438,8 @@ dependencies = [ [[package]] name = "lance-geo" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "datafusion", "geo-traits", @@ -5453,8 +5453,8 @@ dependencies = [ [[package]] name = "lance-index" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arc-swap", "arrow", @@ -5481,6 +5481,7 @@ dependencies = [ "geoarrow-array", "geoarrow-schema", "half", + "hashbrown 0.16.1", "hex", "itertools 0.14.0", "jieba-rs", @@ -5524,8 +5525,8 @@ dependencies = [ [[package]] name = "lance-index-core" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-array", "arrow-schema", @@ -5548,8 +5549,8 @@ dependencies = [ [[package]] name = "lance-io" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow", "arrow-array", @@ -5591,8 +5592,8 @@ dependencies = [ [[package]] name = "lance-linalg" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-array", "arrow-schema", @@ -5606,8 +5607,8 @@ dependencies = [ [[package]] name = "lance-namespace" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow", "async-trait", @@ -5621,8 +5622,8 @@ dependencies = [ [[package]] name = "lance-namespace-impls" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arc-swap", "arrow", @@ -5662,9 +5663,9 @@ dependencies = [ [[package]] name = "lance-namespace-reqwest-client" -version = "0.12.0" +version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8d23e54b1634d5bbb434f8dd33dc3c05f6e58d876a9a27b3b4aef58ddbe11af" +checksum = "828f291764feda98898452bea4ce3756b73e5d5c7c67b62edda27456b7cc72c1" dependencies = [ "reqwest 0.12.28", "serde", @@ -5676,8 +5677,8 @@ dependencies = [ [[package]] name = "lance-select" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-array", "arrow-buffer", @@ -5691,8 +5692,8 @@ dependencies = [ [[package]] name = "lance-table" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow", "arrow-array", @@ -5732,8 +5733,8 @@ dependencies = [ [[package]] name = "lance-testing" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "arrow-array", "arrow-schema", @@ -5746,8 +5747,8 @@ dependencies = [ [[package]] name = "lance-tokenizer" -version = "13.0.0-beta.18" -source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.18#b646eda6a891f481b6f6d6baef9cd17656aef65b" +version = "13.1.0-beta.1" +source = "git+https://github.com/lance-format/lance.git?tag=v13.1.0-beta.1#e5a3553b1699a7062ff7afe4ee268d596bbf2e0d" dependencies = [ "frostem", "icu_segmenter", @@ -5760,7 +5761,7 @@ dependencies = [ [[package]] name = "lancedb" -version = "0.40.0-beta.11" +version = "0.40.0-beta.12" dependencies = [ "ahash", "anyhow", @@ -5857,7 +5858,7 @@ dependencies = [ [[package]] name = "lancedb-nodejs" -version = "0.40.0-beta.11" +version = "0.40.0-beta.12" dependencies = [ "arrow-array", "arrow-buffer", @@ -5882,7 +5883,7 @@ dependencies = [ [[package]] name = "lancedb-python" -version = "0.40.0-beta.11" +version = "0.40.0-beta.12" dependencies = [ "arc-swap", "arrow", diff --git a/Cargo.toml b/Cargo.toml index 4aa1da696..1d072e654 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,20 +13,20 @@ categories = ["database-implementations"] rust-version = "1.91.0" [workspace.dependencies] -lance = { "version" = "=13.0.0-beta.18", default-features = false, "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-core = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-datagen = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-file = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-io = { "version" = "=13.0.0-beta.18", default-features = false, "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-index = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-linalg = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-namespace = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-namespace-impls = { "version" = "=13.0.0-beta.18", default-features = false, "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-table = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-testing = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-datafusion = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-encoding = { "version" = "=13.0.0-beta.18", "tag" = "v13.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" } -lance-arrow = { version = "=13.0.0-beta.18", path = "python/vendor/lance-arrow" } +lance = { "version" = "=13.1.0-beta.1", default-features = false, "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-core = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-datagen = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-file = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-io = { "version" = "=13.1.0-beta.1", default-features = false, "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-index = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-linalg = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-namespace = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-namespace-impls = { "version" = "=13.1.0-beta.1", default-features = false, "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-table = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-testing = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-datafusion = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-encoding = { "version" = "=13.1.0-beta.1", "tag" = "v13.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" } +lance-arrow = { version = "=13.1.0-beta.1", path = "python/vendor/lance-arrow" } lancedb = { path = "rust/lancedb", default-features = false } ahash = "0.8" # Note that this one does not include pyarrow diff --git a/docs/src/java/java.md b/docs/src/java/java.md index 5946e5862..0565156dc 100644 --- a/docs/src/java/java.md +++ b/docs/src/java/java.md @@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`: com.lancedb lancedb-core - 0.40.0-beta.11 + 0.40.0-beta.12 ``` diff --git a/docs/src/js/classes/BlobFile.md b/docs/src/js/classes/BlobFile.md index 84a596d1a..ee012be25 100644 --- a/docs/src/js/classes/BlobFile.md +++ b/docs/src/js/classes/BlobFile.md @@ -10,16 +10,52 @@ A lazy handle to blob bytes. Create one with [Table.fetchBlobFiles](Table.md#fet ## Methods +### close() + +```ts +close(): Promise +``` + +Releases the handle. Reads after `close()` fail. Calling it again does +nothing. + +#### Returns + +`Promise`<`void`> + +*** + +### isClosed() + +```ts +isClosed(): boolean +``` + +Returns true after [BlobFile.close](BlobFile.md#close). + +#### Returns + +`boolean` + +*** + ### read() ```ts -read(): Promise +read(maxBytes?, options?): Promise ``` -Reads from the cursor to the end and advances the cursor. +Reads from the cursor and advances the cursor. -A second call returns an empty buffer. [BlobFile.readRange](BlobFile.md#readrange) does -not move the cursor. +Reads to the end when `maxBytes` is omitted, or at most `maxBytes` bytes +otherwise. Returns an empty buffer at the end of the blob. +[BlobFile.readRange](BlobFile.md#readrange) does not move the cursor. + +#### Parameters + +* **maxBytes?**: `bigint` + +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) #### Returns @@ -30,7 +66,10 @@ not move the cursor. ### readRange() ```ts -readRange(start, end): Promise +readRange( + start, + end, + options?): Promise ``` Reads the half-open byte range `[start, end)`. @@ -43,12 +82,55 @@ Fails when `end` is past the blob size. Does not move the cursor. * **end**: `bigint` +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) + #### Returns `Promise`<`Buffer`> *** +### readRanges() + +```ts +readRanges(ranges, options?): Promise +``` + +Reads several half-open byte ranges. Returns one buffer per range, in +the order given. + +Fails when any `end` is past the blob size. Does not move the cursor. + +#### Parameters + +* **ranges**: [`BlobRange`](../type-aliases/BlobRange.md)[] + +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) + +#### Returns + +`Promise`<`Buffer`[]> + +*** + +### seek() + +```ts +seek(position): Promise +``` + +Moves the cursor to `position`, in bytes from the start of the blob. + +#### Parameters + +* **position**: `bigint` + +#### Returns + +`Promise`<`void`> + +*** + ### size() ```ts @@ -60,3 +142,17 @@ Returns the blob size in bytes. #### Returns `bigint` + +*** + +### tell() + +```ts +tell(): Promise +``` + +Returns the cursor position in bytes. + +#### Returns + +`Promise`<`bigint`> diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index 17e2d4f34..0d813ff1e 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -516,7 +516,10 @@ Drop an index from the table. ### fetchBlobFiles() ```ts -abstract fetchBlobFiles(column, rowIds): Promise<(null | BlobFile)[]> +abstract fetchBlobFiles( + column, + rowIds, + options?): Promise<(null | BlobFile)[]> ``` Opens lazy blob handles for `column` at the given row IDs using the @@ -531,6 +534,8 @@ See [Table.fetchBlobs](Table.md#fetchblobs) for row-ID validity across versions. * **rowIds**: readonly (`number` \| `bigint`)[] +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) + #### Returns `Promise`<(`null` \| [`BlobFile`](BlobFile.md))[]> @@ -540,7 +545,10 @@ See [Table.fetchBlobs](Table.md#fetchblobs) for row-ID validity across versions. ### fetchBlobs() ```ts -abstract fetchBlobs(column, rowIds): Promise<(null | Buffer)[]> +abstract fetchBlobs( + column, + rowIds, + options?): Promise<(null | Buffer)[]> ``` Bytes for `column` at row IDs from [Query.withRowId](Query.md#withrowid). @@ -559,6 +567,8 @@ memory; use [Table.fetchBlobFiles](Table.md#fetchblobfiles) for large values. * **rowIds**: readonly (`number` \| `bigint`)[] +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) + #### Returns `Promise`<(`null` \| `Buffer`)[]> diff --git a/docs/src/js/globals.md b/docs/src/js/globals.md index 2f49578b1..a82d9e334 100644 --- a/docs/src/js/globals.md +++ b/docs/src/js/globals.md @@ -158,6 +158,8 @@ - [AnalyzePlanDistributedMetrics](type-aliases/AnalyzePlanDistributedMetrics.md) - [BaseTokenizer](type-aliases/BaseTokenizer.md) - [BlobOptions](type-aliases/BlobOptions.md) +- [BlobRange](type-aliases/BlobRange.md) +- [BlobReadOptions](type-aliases/BlobReadOptions.md) - [Data](type-aliases/Data.md) - [DataLike](type-aliases/DataLike.md) - [FieldLike](type-aliases/FieldLike.md) diff --git a/docs/src/js/type-aliases/BlobRange.md b/docs/src/js/type-aliases/BlobRange.md new file mode 100644 index 000000000..85cd48242 --- /dev/null +++ b/docs/src/js/type-aliases/BlobRange.md @@ -0,0 +1,27 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / BlobRange + +# Type Alias: BlobRange + +```ts +type BlobRange: object; +``` + +A half-open byte range `[start, end)` for [BlobFile.readRanges](../classes/BlobFile.md#readranges). + +## Type declaration + +### end + +```ts +end: bigint; +``` + +### start + +```ts +start: bigint; +``` diff --git a/docs/src/js/type-aliases/BlobReadOptions.md b/docs/src/js/type-aliases/BlobReadOptions.md new file mode 100644 index 000000000..17264b0e4 --- /dev/null +++ b/docs/src/js/type-aliases/BlobReadOptions.md @@ -0,0 +1,24 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / BlobReadOptions + +# Type Alias: BlobReadOptions + +```ts +type BlobReadOptions: object; +``` + +Options for blob reads. + +## Type declaration + +### signal? + +```ts +optional signal: AbortSignal; +``` + +Cancels the read. The call rejects with `signal.reason`, and the native +read stops, including in-flight requests to a remote table. diff --git a/java/lancedb-core/pom.xml b/java/lancedb-core/pom.xml index b7cdb9e41..77635c74e 100644 --- a/java/lancedb-core/pom.xml +++ b/java/lancedb-core/pom.xml @@ -8,7 +8,7 @@ com.lancedb lancedb-parent - 0.40.0-beta.11 + 0.40.0-beta.12 ../pom.xml diff --git a/java/pom.xml b/java/pom.xml index 7b9009ccd..9d7249f92 100644 --- a/java/pom.xml +++ b/java/pom.xml @@ -6,7 +6,7 @@ com.lancedb lancedb-parent - 0.40.0-beta.11 + 0.40.0-beta.12 pom ${project.artifactId} LanceDB Java SDK Parent POM @@ -28,7 +28,7 @@ UTF-8 15.0.0 - 13.0.0-beta.18 + 13.1.0-beta.1 false 2.30.0 1.7 diff --git a/nodejs/Cargo.toml b/nodejs/Cargo.toml index 87d4973d2..051e16474 100644 --- a/nodejs/Cargo.toml +++ b/nodejs/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "lancedb-nodejs" edition.workspace = true -version = "0.40.0-beta.11" +version = "0.40.0-beta.12" publish = false license.workspace = true description.workspace = true diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts index ae257455d..904c282d4 100644 --- a/nodejs/__test__/table.test.ts +++ b/nodejs/__test__/table.test.ts @@ -2,6 +2,7 @@ // SPDX-FileCopyrightText: Copyright The LanceDB Authors import * as fs from "fs"; +import { getEventListeners } from "node:events"; import { pathToFileURL } from "node:url"; import * as path from "path"; import * as tmp from "tmp"; @@ -92,6 +93,38 @@ describe.each([arrow15, arrow16, arrow17, arrow18])( await expect(table.countRows()).resolves.toBe(3); }); + it("creates and adds Arrow tables with large string and binary columns", async () => { + const data = new arrow.Table({ + text: arrow.vectorFromArray(["alpha", "beta"], new arrow.LargeUtf8()), + bytes: arrow.vectorFromArray( + [Buffer.from("one"), Buffer.from("two")], + new arrow.LargeBinary(), + ), + }); + const conn = await connect(tmpDir.name); + const largeTable = await conn.createTable("large_columns", data); + await largeTable.add(data); + + const rows = await largeTable.query().toArray(); + expect(rows.map((row) => row.text)).toEqual([ + "alpha", + "beta", + "alpha", + "beta", + ]); + expect(rows.map((row) => Buffer.from(row.bytes).toString())).toEqual([ + "one", + "two", + "one", + "two", + ]); + const fields = (await largeTable.schema()).fields; + expect(fields.map((field) => field.type.typeId)).toEqual([ + arrow.Type.LargeUtf8, + arrow.Type.LargeBinary, + ]); + }); + it("should support a foreign Float64 vector schema end to end", async () => { const conn = await connect(tmpDir.name); const schema = new arrow.Schema([ @@ -2601,6 +2634,27 @@ describe("when dealing with blob columns", () => { expect(Buffer.from(await files[1]!.read()).toString()).toBe("beta"); }); + it("adds a LargeBinary Arrow column to a blob table", async () => { + const db = await connect(tmpDir.name); + const schema = new Schema([new Field("id", new Int64()), blob("image")]); + const table = await db.createTable( + "blobs", + [{ id: 1n, image: Buffer.from("alpha") }], + { schema }, + ); + const payload = Buffer.from("beta"); + const data = new arrow18.Table({ + id: arrow18.vectorFromArray([2n], new arrow18.Int64()), + image: arrow18.vectorFromArray([payload], new arrow18.LargeBinary()), + }); + await table.add(data); + + const rows = await table.query().withRowId().toArray(); + const rowId = rows.find((row) => row.id === 2n)!._rowid as bigint; + const [actual] = await table.fetchBlobs("image", [rowId]); + expect(actual).toEqual(payload); + }); + it("reads a half-open range", async () => { const { table, rowIds } = await openBlobTable(); const files = await table.fetchBlobFiles("image", rowIds); @@ -2625,6 +2679,148 @@ describe("when dealing with blob columns", () => { ).rejects.toThrow(/exceeds blob size/); }); + it("reads at most maxBytes and advances the cursor", async () => { + const { table, rowIds } = await openBlobTable(); + const [handle] = await table.fetchBlobFiles("image", rowIds); + expect((await handle!.read(2n)).toString()).toBe("al"); + expect(await handle!.tell()).toBe(2n); + expect((await handle!.read(2n)).toString()).toBe("ph"); + expect((await handle!.read(10n)).toString()).toBe("a"); + expect(await handle!.tell()).toBe(5n); + expect(await handle!.read(1n)).toEqual(Buffer.alloc(0)); + }); + + it("seeks the cursor", async () => { + const { table, rowIds, alpha } = await openBlobTable(); + const [handle] = await table.fetchBlobFiles("image", rowIds); + await handle!.seek(3n); + expect(await handle!.tell()).toBe(3n); + expect((await handle!.read()).toString()).toBe("ha"); + await handle!.seek(0n); + expect(await handle!.read()).toEqual(alpha); + }); + + it("rejects a negative seek position", async () => { + const { table, rowIds } = await openBlobTable(); + const [handle] = await table.fetchBlobFiles("image", rowIds); + await expect(handle!.seek(-1n)).rejects.toThrow(/cannot be negative/); + }); + + it("readRanges returns one buffer per range in order", async () => { + const { table, rowIds } = await openBlobTable(); + const [handle] = await table.fetchBlobFiles("image", rowIds); + const buffers = await handle!.readRanges([ + { start: 3n, end: 5n }, + { start: 0n, end: 2n }, + { start: 3n, end: 5n }, + ]); + expect(buffers.map((b) => b.toString())).toEqual(["ha", "al", "ha"]); + expect(await handle!.tell()).toBe(0n); + }); + + it("fails when a readRanges end is past the blob size", async () => { + const { table, rowIds, alpha } = await openBlobTable(); + const [handle] = await table.fetchBlobFiles("image", rowIds); + await expect( + handle!.readRanges([ + { start: 0n, end: 1n }, + { start: 0n, end: BigInt(alpha.length + 1) }, + ]), + ).rejects.toThrow(/exceeds blob size/); + }); + + it("fails reads after close and closes once", async () => { + const { table, rowIds } = await openBlobTable(); + const [handle] = await table.fetchBlobFiles("image", rowIds); + expect(handle!.isClosed()).toBe(false); + await handle!.close(); + expect(handle!.isClosed()).toBe(true); + await expect(handle!.close()).resolves.toBeUndefined(); + await expect(handle!.read()).rejects.toThrow(/already closed/); + await expect(handle!.read(1n)).rejects.toThrow(/already closed/); + await expect(handle!.readRange(0n, 1n)).rejects.toThrow(/already closed/); + await expect(handle!.readRanges([{ start: 0n, end: 1n }])).rejects.toThrow( + /already closed/, + ); + await expect(handle!.seek(0n)).rejects.toThrow(/already closed/); + await expect(handle!.tell()).rejects.toThrow(/already closed/); + expect(handle!.size()).toBe(5n); + }); + + it("rejects reads with an already aborted signal", async () => { + const { table, rowIds, alpha } = await openBlobTable(); + const [handle] = await table.fetchBlobFiles("image", rowIds); + const signal = AbortSignal.abort(); + const abortError = { name: "AbortError" }; + await expect(handle!.read(undefined, { signal })).rejects.toMatchObject( + abortError, + ); + await expect(handle!.read(2n, { signal })).rejects.toMatchObject( + abortError, + ); + await expect(handle!.readRange(0n, 1n, { signal })).rejects.toMatchObject( + abortError, + ); + await expect( + handle!.readRanges([{ start: 0n, end: 1n }], { signal }), + ).rejects.toMatchObject(abortError); + await expect( + table.fetchBlobs("image", rowIds, { signal }), + ).rejects.toMatchObject(abortError); + await expect( + table.fetchBlobFiles("image", rowIds, { signal }), + ).rejects.toMatchObject(abortError); + expect(await handle!.read()).toEqual(alpha); + }); + + it("rejects with the abort reason", async () => { + const { table, rowIds } = await openBlobTable(); + const [handle] = await table.fetchBlobFiles("image", rowIds); + const reason = new Error("caller gave up"); + await expect( + handle!.readRange(0n, 1n, { signal: AbortSignal.abort(reason) }), + ).rejects.toBe(reason); + }); + + it("reads with a signal and removes its abort listener", async () => { + const { table, rowIds, alpha, beta } = await openBlobTable(); + const controller = new AbortController(); + const { signal } = controller; + const [handle] = await table.fetchBlobFiles("image", rowIds, { signal }); + expect((await handle!.readRange(0n, 2n, { signal })).toString()).toBe("al"); + expect(await table.fetchBlobs("image", rowIds, { signal })).toEqual([ + alpha, + beta, + null, + ]); + expect(getEventListeners(signal, "abort")).toHaveLength(0); + controller.abort(); + expect(await handle!.read()).toEqual(alpha); + }); + + it("cancels an in-flight read", async () => { + const db = await connect(tmpDir.name); + const schema = new Schema([ + new Field("id", new Int64(), true), + blob("payload", { inlineSizeThreshold: 0, dedicatedSizeThreshold: 1 }), + ]); + const payload = Buffer.alloc(32 * 1024 * 1024, 7); + const table = await db.createTable("large_blobs", [{ id: 1n, payload }], { + schema, + }); + const [row] = await table.query().withRowId().toArray(); + const [handle] = await table.fetchBlobFiles("payload", [ + row._rowid as bigint, + ]); + const controller = new AbortController(); + const pending = handle!.readRange(0n, handle!.size(), { + signal: controller.signal, + }); + controller.abort(); + await expect(pending).rejects.toMatchObject({ name: "AbortError" }); + expect(await handle!.readRange(0n, 4n)).toEqual(payload.subarray(0, 4)); + }); + it("rejects fetchBlobs on a non-blob column", async () => { const { table, rowIds } = await openBlobTable(); await expect(table.fetchBlobs("id", rowIds)).rejects.toThrow(/blob/i); diff --git a/nodejs/lancedb/blob.ts b/nodejs/lancedb/blob.ts index 244faac09..e6c780ca9 100644 --- a/nodejs/lancedb/blob.ts +++ b/nodejs/lancedb/blob.ts @@ -141,13 +141,16 @@ export class BlobFile { } /** - * Reads from the cursor to the end and advances the cursor. + * Reads from the cursor and advances the cursor. * - * A second call returns an empty buffer. {@link BlobFile.readRange} does - * not move the cursor. + * Reads to the end when `maxBytes` is omitted, or at most `maxBytes` bytes + * otherwise. Returns an empty buffer at the end of the blob. + * {@link BlobFile.readRange} does not move the cursor. */ - read(): Promise { - return this.inner.read(); + read(maxBytes?: bigint, options?: BlobReadOptions): Promise { + return runWithSignal(options?.signal, (signal) => + this.inner.read(maxBytes, signal), + ); } /** @@ -155,11 +158,96 @@ export class BlobFile { * * Fails when `end` is past the blob size. Does not move the cursor. */ - readRange(start: bigint, end: bigint): Promise { - return this.inner.readRange(start, end); + readRange( + start: bigint, + end: bigint, + options?: BlobReadOptions, + ): Promise { + return runWithSignal(options?.signal, (signal) => + this.inner.readRange(start, end, signal), + ); + } + + /** + * Reads several half-open byte ranges. Returns one buffer per range, in + * the order given. + * + * Fails when any `end` is past the blob size. Does not move the cursor. + */ + readRanges( + ranges: BlobRange[], + options?: BlobReadOptions, + ): Promise { + return runWithSignal(options?.signal, (signal) => + this.inner.readRanges(ranges, signal), + ); + } + + /** Moves the cursor to `position`, in bytes from the start of the blob. */ + seek(position: bigint): Promise { + return this.inner.seek(position); + } + + /** Returns the cursor position in bytes. */ + tell(): Promise { + return this.inner.tell(); + } + + /** + * Releases the handle. Reads after `close()` fail. Calling it again does + * nothing. + */ + close(): Promise { + return this.inner.close(); + } + + /** Returns true after {@link BlobFile.close}. */ + isClosed(): boolean { + return this.inner.isClosed(); } } +/** Options for blob reads. */ +export type BlobReadOptions = { + /** + * Cancels the read. The call rejects with `signal.reason`, and the native + * read stops, including in-flight requests to a remote table. + */ + signal?: AbortSignal; +}; + +/** @ignore */ +export async function runWithSignal( + signal: AbortSignal | undefined, + run: (signal: AbortSignal | undefined) => Promise, +): Promise { + if (signal === undefined) { + return run(undefined); + } + signal.throwIfAborted(); + // napi-rs replaces `onabort` on the signal it receives, so pass a private + // signal and forward the caller's abort to it. + const controller = new AbortController(); + const onAbort = () => controller.abort(); + signal.addEventListener("abort", onAbort, { once: true }); + try { + return await run(controller.signal); + } catch (err) { + if (signal.aborted) { + throw signal.reason; + } + throw err; + } finally { + signal.removeEventListener("abort", onAbort); + } +} + +/** A half-open byte range `[start, end)` for {@link BlobFile.readRanges}. */ +export type BlobRange = { + start: bigint; + end: bigint; +}; + export function coerceBlobValue(value: unknown): BlobInput | null { if (value == null) { return null; diff --git a/nodejs/lancedb/index.ts b/nodejs/lancedb/index.ts index ed6889232..a1776e3f4 100644 --- a/nodejs/lancedb/index.ts +++ b/nodejs/lancedb/index.ts @@ -84,7 +84,7 @@ export { } from "./arrow"; export { blob, isBlobField, BlobFile } from "./blob"; -export type { BlobOptions } from "./blob"; +export type { BlobOptions, BlobRange, BlobReadOptions } from "./blob"; export { Connection, diff --git a/nodejs/lancedb/sanitize.ts b/nodejs/lancedb/sanitize.ts index 3e5d583c4..98a144451 100644 --- a/nodejs/lancedb/sanitize.ts +++ b/nodejs/lancedb/sanitize.ts @@ -43,6 +43,8 @@ import { Interval, IntervalDayTime, IntervalYearMonth, + LargeBinary, + LargeUtf8, List, Map_, Null, @@ -465,8 +467,12 @@ function sanitizeTypeById( return sanitizeFloat(typeLike); case Type.Binary: return new Binary(); + case Type.LargeBinary: + return new LargeBinary(); case Type.Utf8: return new Utf8(); + case Type.LargeUtf8: + return new LargeUtf8(); case Type.Bool: return new Bool(); case Type.Decimal: diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index 132e8bc20..d4c5db509 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -17,7 +17,7 @@ import { tableFromIPC, } from "./arrow"; -import { BlobFile } from "./blob"; +import { BlobFile, BlobReadOptions, runWithSignal } from "./blob"; import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry"; import { IndexOptions } from "./indices"; import { Job } from "./job"; @@ -545,6 +545,7 @@ export abstract class Table { abstract fetchBlobs( column: string, rowIds: readonly (bigint | number)[], + options?: BlobReadOptions, ): Promise<(Buffer | null)[]>; /** @@ -557,6 +558,7 @@ export abstract class Table { abstract fetchBlobFiles( column: string, rowIds: readonly (bigint | number)[], + options?: BlobReadOptions, ): Promise<(BlobFile | null)[]>; /** @@ -1248,8 +1250,11 @@ export class LocalTable extends Table { async fetchBlobs( column: string, rowIds: readonly (bigint | number)[], + options?: BlobReadOptions, ): Promise<(Buffer | null)[]> { - const values = await this.inner.fetchBlobs(column, rowIdsToBigInts(rowIds)); + const values = await runWithSignal(options?.signal, (signal) => + this.inner.fetchBlobs(column, rowIdsToBigInts(rowIds), signal), + ); // N-API Option maps missing values to undefined. Collapse those to null. return values.map((value) => value ?? null); } @@ -1257,10 +1262,10 @@ export class LocalTable extends Table { async fetchBlobFiles( column: string, rowIds: readonly (bigint | number)[], + options?: BlobReadOptions, ): Promise<(BlobFile | null)[]> { - const files = await this.inner.fetchBlobFiles( - column, - rowIdsToBigInts(rowIds), + const files = await runWithSignal(options?.signal, (signal) => + this.inner.fetchBlobFiles(column, rowIdsToBigInts(rowIds), signal), ); // N-API Option maps missing values to undefined. Collapse those to null. return files.map((file) => diff --git a/nodejs/npm/darwin-arm64/package.json b/nodejs/npm/darwin-arm64/package.json index 0e9d131b0..051bdb98f 100644 --- a/nodejs/npm/darwin-arm64/package.json +++ b/nodejs/npm/darwin-arm64/package.json @@ -1,6 +1,6 @@ { "name": "@lancedb/lancedb-darwin-arm64", - "version": "0.40.0-beta.11", + "version": "0.40.0-beta.12", "os": ["darwin"], "cpu": ["arm64"], "main": "lancedb.darwin-arm64.node", diff --git a/nodejs/npm/linux-arm64-gnu/package.json b/nodejs/npm/linux-arm64-gnu/package.json index 39002b700..755035ee2 100644 --- a/nodejs/npm/linux-arm64-gnu/package.json +++ b/nodejs/npm/linux-arm64-gnu/package.json @@ -1,6 +1,6 @@ { "name": "@lancedb/lancedb-linux-arm64-gnu", - "version": "0.40.0-beta.11", + "version": "0.40.0-beta.12", "os": ["linux"], "cpu": ["arm64"], "main": "lancedb.linux-arm64-gnu.node", diff --git a/nodejs/npm/linux-arm64-musl/package.json b/nodejs/npm/linux-arm64-musl/package.json index 232796091..d31737229 100644 --- a/nodejs/npm/linux-arm64-musl/package.json +++ b/nodejs/npm/linux-arm64-musl/package.json @@ -1,6 +1,6 @@ { "name": "@lancedb/lancedb-linux-arm64-musl", - "version": "0.40.0-beta.11", + "version": "0.40.0-beta.12", "os": ["linux"], "cpu": ["arm64"], "main": "lancedb.linux-arm64-musl.node", diff --git a/nodejs/npm/linux-x64-gnu/package.json b/nodejs/npm/linux-x64-gnu/package.json index cc9c003e0..618df080d 100644 --- a/nodejs/npm/linux-x64-gnu/package.json +++ b/nodejs/npm/linux-x64-gnu/package.json @@ -1,6 +1,6 @@ { "name": "@lancedb/lancedb-linux-x64-gnu", - "version": "0.40.0-beta.11", + "version": "0.40.0-beta.12", "os": ["linux"], "cpu": ["x64"], "main": "lancedb.linux-x64-gnu.node", diff --git a/nodejs/npm/linux-x64-musl/package.json b/nodejs/npm/linux-x64-musl/package.json index 2f0356da2..87628c0c3 100644 --- a/nodejs/npm/linux-x64-musl/package.json +++ b/nodejs/npm/linux-x64-musl/package.json @@ -1,6 +1,6 @@ { "name": "@lancedb/lancedb-linux-x64-musl", - "version": "0.40.0-beta.11", + "version": "0.40.0-beta.12", "os": ["linux"], "cpu": ["x64"], "main": "lancedb.linux-x64-musl.node", diff --git a/nodejs/npm/win32-arm64-msvc/package.json b/nodejs/npm/win32-arm64-msvc/package.json index 2f96d1019..65c7b500c 100644 --- a/nodejs/npm/win32-arm64-msvc/package.json +++ b/nodejs/npm/win32-arm64-msvc/package.json @@ -1,6 +1,6 @@ { "name": "@lancedb/lancedb-win32-arm64-msvc", - "version": "0.40.0-beta.11", + "version": "0.40.0-beta.12", "os": [ "win32" ], diff --git a/nodejs/npm/win32-x64-msvc/package.json b/nodejs/npm/win32-x64-msvc/package.json index c941d597f..ad6a81acc 100644 --- a/nodejs/npm/win32-x64-msvc/package.json +++ b/nodejs/npm/win32-x64-msvc/package.json @@ -1,6 +1,6 @@ { "name": "@lancedb/lancedb-win32-x64-msvc", - "version": "0.40.0-beta.11", + "version": "0.40.0-beta.12", "os": ["win32"], "cpu": ["x64"], "main": "lancedb.win32-x64-msvc.node", diff --git a/nodejs/package.json b/nodejs/package.json index bd37d97e2..29bbe4314 100644 --- a/nodejs/package.json +++ b/nodejs/package.json @@ -11,7 +11,7 @@ "ann" ], "private": false, - "version": "0.40.0-beta.11", + "version": "0.40.0-beta.12", "main": "dist/index.js", "exports": { ".": "./dist/index.js", diff --git a/nodejs/src/blob.rs b/nodejs/src/blob.rs index 230a11bae..59cf61af6 100644 --- a/nodejs/src/blob.rs +++ b/nodejs/src/blob.rs @@ -1,25 +1,45 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright The LanceDB Authors +use std::future::Future; use std::ops::Range; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use arrow_array::{Array, LargeBinaryArray}; +use futures::future::abortable; use lancedb::blob::BlobFile as LanceBlobFile; +use napi::Env; use napi::bindgen_prelude::*; use napi_derive::napi; use crate::error::convert_error; +#[napi(object)] +pub struct BlobRange { + pub start: BigInt, + pub end: BigInt, +} + #[napi] pub struct BlobFile { inner: Arc, + closed: AtomicBool, } impl BlobFile { pub(crate) fn new(inner: LanceBlobFile) -> Self { Self { inner: Arc::new(inner), + closed: AtomicBool::new(false), + } + } + + fn ensure_open(&self) -> napi::Result<()> { + if self.closed.load(Ordering::Acquire) { + Err(napi::Error::from_reason("blob file is already closed")) + } else { + Ok(()) } } } @@ -32,21 +52,124 @@ impl BlobFile { } #[napi] - pub async fn read(&self) -> napi::Result { - let bytes = self.inner.read().await.map_err(|err| convert_error(&err))?; - Ok(Buffer::from(bytes.as_ref())) + pub fn read<'env>( + &self, + env: &'env Env, + max_bytes: Option, + signal: Option, + ) -> napi::Result> { + self.ensure_open()?; + let max_bytes = max_bytes + .map(|max_bytes| { + usize::try_from(parse_u64(max_bytes, "maxBytes")?) + .map_err(|_| napi::Error::from_reason("maxBytes is too large")) + }) + .transpose()?; + let inner = self.inner.clone(); + spawn_abortable(env, signal, async move { + let bytes = match max_bytes { + None => inner.read().await, + Some(max_bytes) => inner.read_up_to(max_bytes).await, + } + .map_err(|err| convert_error(&err))?; + Ok(Buffer::from(bytes.as_ref())) + }) } #[napi] - pub async fn read_range(&self, start: BigInt, end: BigInt) -> napi::Result { + pub fn read_range<'env>( + &self, + env: &'env Env, + start: BigInt, + end: BigInt, + signal: Option, + ) -> napi::Result> { + self.ensure_open()?; let range = bigint_range(start, end)?; - let bytes = self - .inner - .read_range(range) - .await - .map_err(|err| convert_error(&err))?; - Ok(Buffer::from(bytes.as_ref())) + let inner = self.inner.clone(); + spawn_abortable(env, signal, async move { + let bytes = inner + .read_range(range) + .await + .map_err(|err| convert_error(&err))?; + Ok(Buffer::from(bytes.as_ref())) + }) } + + #[napi] + pub fn read_ranges<'env>( + &self, + env: &'env Env, + ranges: Vec, + signal: Option, + ) -> napi::Result>> { + self.ensure_open()?; + let ranges = ranges + .into_iter() + .map(|range| bigint_range(range.start, range.end)) + .collect::>>()?; + let inner = self.inner.clone(); + spawn_abortable(env, signal, async move { + let buffers = inner + .read_ranges(&ranges) + .await + .map_err(|err| convert_error(&err))?; + Ok(buffers + .iter() + .map(|bytes| Buffer::from(bytes.as_ref())) + .collect()) + }) + } + + #[napi] + pub async fn seek(&self, position: BigInt) -> napi::Result<()> { + self.ensure_open()?; + let position = parse_u64(position, "position")?; + self.inner + .seek(position) + .await + .map_err(|err| convert_error(&err)) + } + + #[napi] + pub async fn tell(&self) -> napi::Result { + self.ensure_open()?; + let position = self.inner.tell().await.map_err(|err| convert_error(&err))?; + Ok(BigInt::from(position)) + } + + #[napi] + pub async fn close(&self) -> napi::Result<()> { + if self.closed.swap(true, Ordering::AcqRel) { + return Ok(()); + } + self.inner.close().await.map_err(|err| convert_error(&err)) + } + + #[napi] + pub fn is_closed(&self) -> bool { + self.closed.load(Ordering::Acquire) + } +} + +/// Runs `fut` on the napi runtime. Aborting `signal` drops the future, which +/// cancels in-flight reads. +pub fn spawn_abortable<'env, T>( + env: &'env Env, + signal: Option, + fut: impl Future> + Send + 'static, +) -> napi::Result> +where + T: ToNapiValue + Send + 'static, +{ + let (fut, handle) = abortable(fut); + if let Some(signal) = signal { + signal.on_abort(move || handle.abort()); + } + env.spawn_future(async move { + fut.await + .unwrap_or_else(|_| Err(napi::Error::from_reason("blob read aborted"))) + }) } fn bigint_range(start: BigInt, end: BigInt) -> napi::Result> { diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index d381d98e5..f5c3b2987 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -11,11 +11,12 @@ use lancedb::table::{ FieldMetadataUpdate as LanceFieldMetadataUpdate, FtsToken as LanceDbFtsToken, NewColumnTransform, OptimizeAction, OptimizeOptions, Ref, Table as LanceDbTable, }; +use napi::Env; use napi::bindgen_prelude::*; use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode}; use napi_derive::napi; -use crate::blob::{BlobFile, copy_blob_buffers, parse_row_ids}; +use crate::blob::{BlobFile, copy_blob_buffers, parse_row_ids, spawn_abortable}; use crate::error::NapiErrorExt; use crate::index::Index; use crate::merge::NativeMergeInsertBuilder; @@ -340,36 +341,44 @@ impl Table { } #[napi(catch_unwind)] - pub async fn fetch_blobs( + pub fn fetch_blobs<'env>( &self, + env: &'env Env, column: String, row_ids: Vec, - ) -> napi::Result>> { + signal: Option, + ) -> napi::Result>>> { let row_ids = parse_row_ids(row_ids)?; - let array = self - .inner_ref()? - .fetch_blobs(column.as_str(), &row_ids) - .await - .default_error()?; - Ok(copy_blob_buffers(array)) + let table = self.inner_ref()?.clone(); + spawn_abortable(env, signal, async move { + let array = table + .fetch_blobs(column.as_str(), &row_ids) + .await + .default_error()?; + Ok(copy_blob_buffers(array)) + }) } #[napi(catch_unwind)] - pub async fn fetch_blob_files( + pub fn fetch_blob_files<'env>( &self, + env: &'env Env, column: String, row_ids: Vec, - ) -> napi::Result>> { + signal: Option, + ) -> napi::Result>>> { let row_ids = parse_row_ids(row_ids)?; - let files = self - .inner_ref()? - .fetch_blob_files(column.as_str(), &row_ids) - .await - .default_error()?; - Ok(files - .into_iter() - .map(|file| file.map(BlobFile::new)) - .collect()) + let table = self.inner_ref()?.clone(); + spawn_abortable(env, signal, async move { + let files = table + .fetch_blob_files(column.as_str(), &row_ids) + .await + .default_error()?; + Ok(files + .into_iter() + .map(|file| file.map(BlobFile::new)) + .collect()) + }) } #[napi(catch_unwind)] diff --git a/python/Cargo.toml b/python/Cargo.toml index 5add05d47..799a5df62 100644 --- a/python/Cargo.toml +++ b/python/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lancedb-python" -version = "0.40.0-beta.11" +version = "0.40.0-beta.12" publish = false edition.workspace = true description = "Python bindings for LanceDB" diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index cf63d7a9b..176e57ac3 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -149,12 +149,20 @@ class Connection(object): limit: Optional[int], ) -> list[str]: ... # Deprecated: Use list_tables instead async def open_job(self, job_id: str) -> Job: ... - async def create_function_async(self, request_json: str) -> Job: ... - async def get_function(self, name: str, version: str) -> str: ... - async def list_functions(self) -> List[str]: ... - async def drop_function(self, name: str, version: str) -> bool: ... + async def create_function_async( + self, request_json: str, namespace_path: Optional[List[str]] = None + ) -> Job: ... + async def get_function( + self, name: str, version: str, namespace_path: Optional[List[str]] = None + ) -> str: ... + async def list_functions( + self, namespace_path: Optional[List[str]] = None + ) -> List[str]: ... + async def drop_function( + self, name: str, version: str, namespace_path: Optional[List[str]] = None + ) -> bool: ... async def drop_function_async( - self, name: str, version: str + self, name: str, version: str, namespace_path: Optional[List[str]] = None ) -> Tuple[bool, Job]: ... async def create_secret( self, name: str, value: str, namespace_path: Optional[List[str]] = None diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 490987c3c..60e1780f0 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -784,6 +784,7 @@ class DBConnection(EnforceOverrides): definition: UdfDefinition, *, secrets: Optional[Sequence[EnvVarSecret]] = None, + namespace_path: Optional[List[str]] = None, ) -> FunctionVersion: """Build and register a scalar Python UDF, then return its version. @@ -800,6 +801,9 @@ class DBConnection(EnforceOverrides): Function needs, each naming a Secret and the environment variable its value arrives in. The Function's source is unchanged by this; it reads the variable the way it already did. + namespace_path : list of str, optional + The namespace holding the Function. None or an empty list is the + root namespace. Examples -------- @@ -815,13 +819,16 @@ class DBConnection(EnforceOverrides): ) ``` """ - return self.create_function_async(definition, secrets=secrets).wait() + return self.create_function_async( + definition, secrets=secrets, namespace_path=namespace_path + ).wait() def create_function_async( self, definition: UdfDefinition, *, secrets: Optional[Sequence[EnvVarSecret]] = None, + namespace_path: Optional[List[str]] = None, ) -> Job[FunctionVersion]: """Submit a scalar Python UDF for building and registration. @@ -833,17 +840,26 @@ class DBConnection(EnforceOverrides): "Function catalog operations are not supported for this connection type" ) - def get_function(self, name: str, *, version: str) -> FunctionVersion: + def get_function( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> FunctionVersion: """Open one exact immutable Function version from the remote catalog.""" raise NotImplementedError( "Function catalog operations are not supported for this connection type" ) - def list_functions(self) -> List[FunctionVersion]: - """List every published immutable Function version. + def list_functions( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[FunctionVersion]: + """List every published immutable Function version in a namespace. - Results are ordered by Function name then version. Local connections - raise ``NotImplementedError``. + Functions in child namespaces are not included. Results are ordered by + Function name then version. Local connections raise + ``NotImplementedError``. Examples -------- @@ -857,7 +873,13 @@ class DBConnection(EnforceOverrides): "Function catalog operations are not supported for this connection type" ) - def drop_function(self, name: str, *, version: str) -> bool: + def drop_function( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> bool: """Drop a Function name and the object it was bound to. The requested version must exist in the currently named object. Returns @@ -870,7 +892,13 @@ class DBConnection(EnforceOverrides): "Function catalog operations are not supported for this connection type" ) - def drop_function_async(self, name: str, *, version: str) -> Tuple[bool, Job]: + def drop_function_async( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> Tuple[bool, Job]: """Drop a Function name and return its cleanup job. The name is unbound before this returns; the object's content may still @@ -1808,25 +1836,62 @@ class LanceDBConnection(DBConnection): definition: UdfDefinition, *, secrets: Optional[Sequence[EnvVarSecret]] = None, + namespace_path: Optional[List[str]] = None, ) -> Job[FunctionVersion]: - job = LOOP.run(self._conn.create_function_async(definition, secrets=secrets)) + job = LOOP.run( + self._conn.create_function_async( + definition, secrets=secrets, namespace_path=namespace_path + ) + ) return Job(job) @override - def get_function(self, name: str, *, version: str) -> FunctionVersion: - return LOOP.run(self._conn.get_function(name, version=version)) + def get_function( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> FunctionVersion: + return LOOP.run( + self._conn.get_function( + name, version=version, namespace_path=namespace_path + ) + ) @override - def list_functions(self) -> List[FunctionVersion]: - return LOOP.run(self._conn.list_functions()) + def list_functions( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[FunctionVersion]: + return LOOP.run(self._conn.list_functions(namespace_path=namespace_path)) @override - def drop_function(self, name: str, *, version: str) -> bool: - return LOOP.run(self._conn.drop_function(name, version=version)) + def drop_function( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> bool: + return LOOP.run( + self._conn.drop_function( + name, version=version, namespace_path=namespace_path + ) + ) @override - def drop_function_async(self, name: str, *, version: str) -> Tuple[bool, Job]: - dropped, job = LOOP.run(self._conn.drop_function_async(name, version=version)) + def drop_function_async( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> Tuple[bool, Job]: + dropped, job = LOOP.run( + self._conn.drop_function_async( + name, version=version, namespace_path=namespace_path + ) + ) return dropped, Job(job) @override @@ -2787,6 +2852,7 @@ class AsyncConnection(object): definition: UdfDefinition, *, secrets: Optional[Sequence[EnvVarSecret]] = None, + namespace_path: Optional[List[str]] = None, ) -> AsyncJob[FunctionVersion]: """Submit a scalar Python UDF for building and registration. @@ -2794,36 +2860,60 @@ class AsyncConnection(object): artifact. Waiting on the job returns the immutable Function version. ``secrets`` is a sequence of [EnvVarSecret][lancedb.secrets.EnvVarSecret], each naming a Secret and - the environment variable its value arrives in. Local connections raise - ``NotImplementedError``. + the environment variable its value arrives in. ``namespace_path`` is + the namespace the Function is registered in; None or an empty list is + the root. Local connections raise ``NotImplementedError``. """ if not isinstance(definition, UdfDefinition): raise TypeError("create_function_async requires a @udf definition") request = definition.bind_secrets(secrets) - inner = await self._inner.create_function_async(request.to_canonical_json()) + inner = await self._inner.create_function_async( + request.to_canonical_json(), namespace_path + ) return _typed_job(inner, FunctionVersion.from_json) - async def get_function(self, name: str, *, version: str) -> FunctionVersion: + async def get_function( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> FunctionVersion: """Open one exact immutable Function version from the remote catalog.""" - return FunctionVersion.from_json(await self._inner.get_function(name, version)) + return FunctionVersion.from_json( + await self._inner.get_function(name, version, namespace_path) + ) - async def list_functions(self) -> List[FunctionVersion]: - """List every published immutable Function version. + async def list_functions( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[FunctionVersion]: + """List every published immutable Function version in a namespace. - Results are ordered by Function name then version. Local connections - raise ``NotImplementedError``. + Functions in child namespaces are not included. Results are ordered by + Function name then version. Local connections raise + ``NotImplementedError``. """ return [ FunctionVersion.from_json(value) - for value in await self._inner.list_functions() + for value in await self._inner.list_functions(namespace_path) ] - async def drop_function(self, name: str, *, version: str) -> bool: + async def drop_function( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> bool: """Drop a Function name and the object it was bound to.""" - return await self._inner.drop_function(name, version) + return await self._inner.drop_function(name, version, namespace_path) async def drop_function_async( - self, name: str, *, version: str + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, ) -> Tuple[bool, AsyncJob]: """Drop a Function name and return its cleanup job. @@ -2831,7 +2921,9 @@ class AsyncConnection(object): be being deleted. Await :meth:`AsyncJob.wait` to wait for that to finish. """ - dropped, job = await self._inner.drop_function_async(name, version) + dropped, job = await self._inner.drop_function_async( + name, version, namespace_path + ) return dropped, AsyncJob(job) async def create_secret( diff --git a/python/python/lancedb/embeddings/base.py b/python/python/lancedb/embeddings/base.py index 149b9c089..fbe43860d 100644 --- a/python/python/lancedb/embeddings/base.py +++ b/python/python/lancedb/embeddings/base.py @@ -182,7 +182,11 @@ class EmbeddingFunction(BaseModel, ABC): def VectorField(self, **kwargs): """ Creates a pydantic Field that can automatically annotate - the target vector column for this embedding function + the target vector column for this embedding function. + + The field can be annotated as ``list[float]`` for compatibility with + static type checkers. LanceDB will infer the fixed vector dimension from + this embedding function. """ return Field(json_schema_extra={"vector_column_for": self}, **kwargs) diff --git a/python/python/lancedb/functions.py b/python/python/lancedb/functions.py index 5f26b590e..c170c7375 100644 --- a/python/python/lancedb/functions.py +++ b/python/python/lancedb/functions.py @@ -356,9 +356,16 @@ _ObjectVersion = Annotated[ class FunctionVersion(_RemoteValue): - """A pinned object revision, independent of its executable image digest.""" + """A pinned object revision, independent of its executable image digest. + + ``name`` and ``namespace_path`` are the Function's parts, as a + [SecretReference][lancedb.functions.SecretReference] carries a Secret's; + they are what [get_function][lancedb.db.DBConnection.get_function] takes. + An empty ``namespace_path`` is the root namespace. + """ name: str + namespace_path: tuple[str, ...] = () object_id: str location: str version: _ObjectVersion @@ -444,6 +451,7 @@ class FunctionVersion(_RemoteValue): return FunctionApplication( function=FunctionVersionRef( name=self.name, + namespace_path=self.namespace_path, object_id=self.object_id, location=self.location, version=self.version, @@ -502,6 +510,7 @@ class FunctionRegistrationRequest(_RemoteValue): class FunctionVersionRef(_OpenRemoteValue): name: str + namespace_path: tuple[str, ...] = () object_id: str location: str version: _ObjectVersion diff --git a/python/python/lancedb/pydantic.py b/python/python/lancedb/pydantic.py index 528d865d7..313c514c3 100644 --- a/python/python/lancedb/pydantic.py +++ b/python/python/lancedb/pydantic.py @@ -88,6 +88,17 @@ def Vector( ... pa.field("url", pa.utf8(), False), ... pa.field("embeddings", pa.list_(pa.float32(), 768)) ... ]) + + Notes + ----- + ``Vector`` creates a type dynamically, so calls such as ``Vector(768)`` are + not valid static type annotations. For an embedding field, use the standard + ``list[float]`` annotation when running mypy; ``VectorField`` supplies the + fixed dimension to LanceDB:: + + class MyModel(LanceModel): + text: str = embeddings.SourceField() + vector: list[float] = embeddings.VectorField() """ # TODO: make a public parameterized type. @@ -313,6 +324,10 @@ def _unwrap_optional_annotation(annotation: Any) -> Any | None: def _pydantic_to_arrow_type(field: FieldInfo) -> pa.DataType: """Convert a Pydantic FieldInfo to Arrow DataType""" + embedding_vector_type = _embedding_vector_to_arrow_type(field) + if embedding_vector_type is not None: + return embedding_vector_type + unwrapped = _unwrap_optional_annotation(field.annotation) if unwrapped is not None: return _pydantic_type_to_arrow_type(unwrapped, field) @@ -326,8 +341,32 @@ def _pydantic_to_arrow_type(field: FieldInfo) -> pa.DataType: return _pydantic_type_to_arrow_type(field.annotation, field) +def _embedding_vector_to_arrow_type(field: FieldInfo) -> pa.DataType | None: + """Infer a fixed-size vector type from ``VectorField`` metadata.""" + if not _is_embedding_vector_annotation(field): + return None + + function = get_extras(field, "vector_column_for") + return pa.list_(pa.float32(), function.ndims()) + + +def _is_embedding_vector_annotation(field: FieldInfo) -> bool: + if get_extras(field, "vector_column_for") is None: + return False + + annotation = _unwrap_optional_annotation(field.annotation) + if annotation is None: + annotation = field.annotation + + origin = getattr(annotation, "__origin__", None) + args = getattr(annotation, "__args__", ()) + return origin is list and args == (float,) + + def is_nullable(field: FieldInfo) -> bool: """Check if a Pydantic FieldInfo is nullable.""" + if _is_embedding_vector_annotation(field): + return True if _unwrap_optional_annotation(field.annotation) is not None: return True if isinstance(field.annotation, (_GenericAlias, GenericAlias)): diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py index a1c71440c..3773ac547 100644 --- a/python/python/lancedb/remote/db.py +++ b/python/python/lancedb/remote/db.py @@ -862,25 +862,62 @@ class RemoteDBConnection(DBConnection): definition: UdfDefinition, *, secrets: Optional[Sequence[EnvVarSecret]] = None, + namespace_path: Optional[List[str]] = None, ) -> Job[FunctionVersion]: - job = LOOP.run(self._conn.create_function_async(definition, secrets=secrets)) + job = LOOP.run( + self._conn.create_function_async( + definition, secrets=secrets, namespace_path=namespace_path + ) + ) return Job(job) @override - def get_function(self, name: str, *, version: str) -> FunctionVersion: - return LOOP.run(self._conn.get_function(name, version=version)) + def get_function( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> FunctionVersion: + return LOOP.run( + self._conn.get_function( + name, version=version, namespace_path=namespace_path + ) + ) @override - def list_functions(self) -> List[FunctionVersion]: - return LOOP.run(self._conn.list_functions()) + def list_functions( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[FunctionVersion]: + return LOOP.run(self._conn.list_functions(namespace_path=namespace_path)) @override - def drop_function(self, name: str, *, version: str) -> bool: - return LOOP.run(self._conn.drop_function(name, version=version)) + def drop_function( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> bool: + return LOOP.run( + self._conn.drop_function( + name, version=version, namespace_path=namespace_path + ) + ) @override - def drop_function_async(self, name: str, *, version: str) -> Tuple[bool, Job]: - dropped, job = LOOP.run(self._conn.drop_function_async(name, version=version)) + def drop_function_async( + self, + name: str, + *, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> Tuple[bool, Job]: + dropped, job = LOOP.run( + self._conn.drop_function_async( + name, version=version, namespace_path=namespace_path + ) + ) return dropped, Job(job) @override diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py index e100597cf..72107364c 100644 --- a/python/python/tests/test_blob.py +++ b/python/python/tests/test_blob.py @@ -1506,7 +1506,7 @@ def test_merge_insert_mixed_python_blobs_validates_external_uri(tmp_path, as_pan {"id": 2, "image": blob_path.as_uri()}, ] - with pytest.raises(ValueError, match="allow_external_blob_outside_bases"): + with pytest.raises(ValueError, match="outside registered external bases"): ( table.merge_insert("id") .when_matched_update_all() diff --git a/python/python/tests/test_first_class_function_slice1.py b/python/python/tests/test_first_class_function_slice1.py index 1abd42e56..b08dadf36 100644 --- a/python/python/tests/test_first_class_function_slice1.py +++ b/python/python/tests/test_first_class_function_slice1.py @@ -162,6 +162,29 @@ def test_function_version_binds_named_columns_as_one_immutable_application(): ] == [("text", "column", "documents.body")] +def test_function_version_names_its_namespace_as_a_separate_field(): + """A Function is named by its parts, as a Secret is; a root one states no + namespace.""" + value = job_result("remote_function_job.json") + root = FunctionVersion.from_json(json.dumps(value)) + assert root.namespace_path == () + assert "namespace_path" not in json.loads(root.to_canonical_json()) + + nested = FunctionVersion(**{**value, "namespace_path": ["analytics", "features"]}) + assert nested.name == "embed" + assert nested.namespace_path == ("analytics", "features") + assert json.loads(nested.to_canonical_json())["namespace_path"] == [ + "analytics", + "features", + ] + application = nested(text=col("documents.body")) + assert application.function.name == "embed" + assert application.function.namespace_path == ("analytics", "features") + assert json.loads(application.to_canonical_json())["function"][ + "namespace_path" + ] == ["analytics", "features"] + + def test_function_version_binding_validates_names_and_direct_columns(): version = FunctionVersion.from_json( json.dumps(job_result("remote_function_job.json")) diff --git a/python/python/tests/test_first_class_function_slice2.py b/python/python/tests/test_first_class_function_slice2.py index 1cd8d4cf5..4610170f6 100644 --- a/python/python/tests/test_first_class_function_slice2.py +++ b/python/python/tests/test_first_class_function_slice2.py @@ -1539,8 +1539,10 @@ def _mock_remote_function_catalog(): else (None, None) ) if function_action[1] == "create": + *namespace_path, name = function_action[0].split("$") state["version"] = { - "name": "normalize_score", + "name": name, + **({"namespace_path": namespace_path} if namespace_path else {}), "version": FUNCTION_VERSION, "object_id": "fixture", "location": "memory:///fixture", @@ -1565,10 +1567,18 @@ def _mock_remote_function_catalog(): "job_state": "DONE", "result": state["version"], } - elif self.path == "/v1/function/normalize_score/describe": + # A namespaced Function's id is its namespace path and name joined + # by `$`, so the name is the last segment of it. + elif ( + function_action[1] == "describe" + and function_action[0].split("$")[-1] == "normalize_score" + ): assert body == {"version": FUNCTION_VERSION} response = state["version"] - elif self.path == "/v1/function/normalize_score/drop": + elif ( + function_action[1] == "drop" + and function_action[0].split("$")[-1] == "normalize_score" + ): assert body == {"version": FUNCTION_VERSION} response = {"dropped": True} elif secret_action[1] in ("create", "alter"): @@ -1602,7 +1612,8 @@ def _mock_remote_function_catalog(): assert query["page_token"] == "next" self._write_response(200, {"secrets": [{"name": "hf-prod"}]}) return - if url.path != "/v1/namespace/$/function/list": + parts = url.path.strip("/").split("/") + if parts[:2] != ["v1", "namespace"] or parts[3:] != ["function", "list"]: self._write_response(404, {"error": "not found"}) return assert query["include_definition"] == "true" @@ -1840,3 +1851,102 @@ async def test_async_remote_drop_function_sends_exact_version(): {"version": FUNCTION_VERSION}, ) ] + + +def test_remote_function_verbs_address_a_namespace_in_the_path(): + """A Function's namespace is part of the identifier each route addresses; + no body carries it, so a namespaced request differs only in its path.""" + namespace = ["analytics", "features"] + 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}}, + ) + created = db.create_function(normalize_score, namespace_path=namespace) + # The returned parts are what the other verbs take, so a created + # Function reopens from its own fields. + assert created.name == "normalize_score" + assert created.namespace_path == tuple(namespace) + reopened = db.get_function( + created.name, + version=created.version, + namespace_path=list(created.namespace_path), + ) + assert reopened == created + assert reopened(value=lancedb.col("score")).function.namespace_path == tuple( + namespace + ) + db.list_functions(namespace_path=namespace) + assert ( + db.drop_function( + "normalize_score", version=FUNCTION_VERSION, namespace_path=namespace + ) + is True + ) + dropped, _ = db.drop_function_async( + "normalize_score", version=FUNCTION_VERSION, namespace_path=namespace + ) + assert dropped is True + + assert [path for path, _ in state["requests"]] == [ + "/v1/function/analytics$features$normalize_score/create", + "/v1/jobs/describe", + "/v1/function/analytics$features$normalize_score/describe", + "/v1/namespace/analytics$features/function/list", + "/v1/namespace/analytics$features/function/list", + "/v1/function/analytics$features$normalize_score/drop", + "/v1/function/analytics$features$normalize_score/drop", + ] + + +@pytest.mark.asyncio +async def test_async_remote_function_verbs_address_a_namespace_in_the_path(): + namespace = ["analytics", "features"] + 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, namespace_path=namespace + ) + await registration.wait() + await db.get_function( + "normalize_score", version=FUNCTION_VERSION, namespace_path=namespace + ) + await db.list_functions(namespace_path=namespace) + assert ( + await db.drop_function( + "normalize_score", version=FUNCTION_VERSION, namespace_path=namespace + ) + is True + ) + + assert [path for path, _ in state["requests"]] == [ + "/v1/function/analytics$features$normalize_score/create", + "/v1/jobs/describe", + "/v1/function/analytics$features$normalize_score/describe", + "/v1/namespace/analytics$features/function/list", + "/v1/namespace/analytics$features/function/list", + "/v1/function/analytics$features$normalize_score/drop", + ] + + +def test_a_function_namespace_path_must_be_a_list_of_segments(): + """A bare string is refused rather than read as a sequence of one-character + segments, and nothing reaches the service.""" + 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(TypeError): + db.list_functions(namespace_path="analytics") + + assert state["requests"] == [] diff --git a/python/python/tests/test_pydantic.py b/python/python/tests/test_pydantic.py index ca03940f2..81674d735 100644 --- a/python/python/tests/test_pydantic.py +++ b/python/python/tests/test_pydantic.py @@ -9,6 +9,7 @@ from typing import List, Optional, Tuple import pyarrow as pa import pydantic import pytest +from lancedb.conftest import MockTextEmbeddingFunction from lancedb.pydantic import ( LanceModel, MultiVector, @@ -425,6 +426,25 @@ def test_bare_vector_raises_clear_error(): exec("class TestModel(LanceModel):\n vector: Vector", namespace) +def test_embedding_vector_list_annotation(): + embedding = MockTextEmbeddingFunction.create() + + class StaticTypingModel(LanceModel): + text: str = embedding.SourceField() + vector: list[float] = embedding.VectorField() + + schema = pydantic_to_schema(StaticTypingModel) + assert schema == pa.schema( + [ + pa.field("text", pa.utf8(), False), + pa.field("vector", pa.list_(pa.float32(), embedding.ndims()), True), + ] + ) + + model = StaticTypingModel(text="hello", vector=[0.0] * embedding.ndims()) + assert model.vector == [0.0] * embedding.ndims() + + def test_fixed_size_list_field(): class TestModel(pydantic.BaseModel): vec: Vector(16) diff --git a/python/src/connection.rs b/python/src/connection.rs index a0fc3b04b..744ab34b3 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -754,31 +754,37 @@ impl Connection { }) } + #[pyo3(signature = (request_json, namespace_path=None))] pub fn create_function_async( self_: PyRef<'_, Self>, request_json: String, + namespace_path: Option>, ) -> PyResult> { let inner = self_.get_inner()?.clone(); let request = lancedb::function::FunctionRegistrationRequest::from_json(&request_json) .infer_error()?; + let namespace_path = namespace_path.unwrap_or_default(); future_into_py(self_.py(), async move { inner - .create_function_async(request) + .create_function_async(request, &namespace_path) .await .infer_error() .map(crate::job::Job::new_typed) }) } + #[pyo3(signature = (name, version, namespace_path=None))] pub fn get_function( self_: PyRef<'_, Self>, name: String, version: String, + namespace_path: Option>, ) -> PyResult> { let inner = self_.get_inner()?.clone(); + let namespace_path = namespace_path.unwrap_or_default(); future_into_py(self_.py(), async move { inner - .get_function(name, version) + .get_function(name, version, &namespace_path) .await .infer_error()? .to_canonical_json() @@ -786,11 +792,16 @@ impl Connection { }) } - pub fn list_functions(self_: PyRef<'_, Self>) -> PyResult> { + #[pyo3(signature = (namespace_path=None))] + pub fn list_functions( + self_: PyRef<'_, Self>, + namespace_path: Option>, + ) -> PyResult> { let inner = self_.get_inner()?.clone(); + let namespace_path = namespace_path.unwrap_or_default(); future_into_py(self_.py(), async move { inner - .list_functions() + .list_functions(&namespace_path) .await .infer_error()? .into_iter() @@ -799,26 +810,35 @@ impl Connection { }) } + #[pyo3(signature = (name, version, namespace_path=None))] pub fn drop_function( self_: PyRef<'_, Self>, name: String, version: String, + namespace_path: Option>, ) -> PyResult> { let inner = self_.get_inner()?.clone(); + let namespace_path = namespace_path.unwrap_or_default(); future_into_py(self_.py(), async move { - inner.drop_function(name, version).await.infer_error() + inner + .drop_function(name, version, &namespace_path) + .await + .infer_error() }) } + #[pyo3(signature = (name, version, namespace_path=None))] pub fn drop_function_async( self_: PyRef<'_, Self>, name: String, version: String, + namespace_path: Option>, ) -> PyResult> { let inner = self_.get_inner()?.clone(); + let namespace_path = namespace_path.unwrap_or_default(); future_into_py(self_.py(), async move { inner - .drop_function_async(name, version) + .drop_function_async(name, version, &namespace_path) .await .infer_error() .map(|(dropped, job)| (dropped, crate::job::Job::new(job))) diff --git a/python/vendor/lance-arrow/Cargo.toml b/python/vendor/lance-arrow/Cargo.toml index 7f12336d8..83f9f0878 100644 --- a/python/vendor/lance-arrow/Cargo.toml +++ b/python/vendor/lance-arrow/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lance-arrow" -version = "13.0.0-beta.18" +version = "13.1.0-beta.1" edition = "2024" authors = ["Lance Devs "] license = "Apache-2.0" diff --git a/python/vendor/lance-arrow/README.lancedb.md b/python/vendor/lance-arrow/README.lancedb.md index 46b017c7e..d544b0a6d 100644 --- a/python/vendor/lance-arrow/README.lancedb.md +++ b/python/vendor/lance-arrow/README.lancedb.md @@ -1,5 +1,5 @@ -This directory contains `rust/lance-arrow` from Lance tag `v13.0.0-beta.18` -(commit `b646eda6a891f481b6f6d6baef9cd17656aef65b`), licensed under Apache-2.0. +This directory contains `rust/lance-arrow` from Lance tag `v13.1.0-beta.1` +(commit `e5a3553b1699a7062ff7afe4ee268d596bbf2e0d`), licensed under Apache-2.0. LanceDB carries a local fix in `src/lib.rs` for sliced child validity bitmaps until a Lance release includes the fix. The patch is selected in the root diff --git a/rust/lancedb/Cargo.toml b/rust/lancedb/Cargo.toml index fbc9ab5ee..cdb0609d3 100644 --- a/rust/lancedb/Cargo.toml +++ b/rust/lancedb/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lancedb" -version = "0.40.0-beta.11" +version = "0.40.0-beta.12" edition.workspace = true description = "LanceDB: A serverless, low-latency vector database for AI applications" license.workspace = true diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index 0e5bbf5da..985195c07 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -587,7 +587,8 @@ impl Connection { ) } - /// Build and register a Python callable as an immutable Function version. + /// Build and register a Python callable as an immutable Function version + /// in `namespace_path`, which is empty for the root namespace. /// /// The server-side job builds the OCI image, then registers the completed /// artifact. Waiting on the returned typed job yields the durable @@ -597,8 +598,11 @@ impl Connection { pub async fn create_function_async( &self, request: crate::function::FunctionRegistrationRequest, + namespace_path: &[String], ) -> Result> { - self.internal.create_function_async(request).await + self.internal + .create_function_async(request, namespace_path) + .await } /// Look up one exact immutable Function version in the remote catalog. @@ -610,13 +614,15 @@ impl Connection { &self, name: impl AsRef, version: impl AsRef, + namespace_path: &[String], ) -> Result { self.internal - .get_function(name.as_ref(), version.as_ref()) + .get_function(name.as_ref(), version.as_ref(), namespace_path) .await } - /// List every published immutable Function version in the remote catalog. + /// List every published immutable Function version in `namespace_path` of + /// the remote catalog. Functions in child namespaces are not included. /// /// Results are ordered by Function name then version. The client walks all /// server pages before returning. Local databases return @@ -628,14 +634,17 @@ impl Connection { /// # async fn list_functions( /// # connection: &lancedb::Connection, /// # ) -> Result<(), Box> { - /// for function in connection.list_functions().await? { + /// for function in connection.list_functions(&[]).await? { /// println!("{} {}", function.name(), function.version()); /// } /// # Ok(()) /// # } /// ``` - pub async fn list_functions(&self) -> Result> { - self.internal.list_functions().await + pub async fn list_functions( + &self, + namespace_path: &[String], + ) -> Result> { + self.internal.list_functions(namespace_path).await } /// Remove the current Function name binding, retaining the object history. @@ -647,9 +656,10 @@ impl Connection { &self, name: impl AsRef, version: impl AsRef, + namespace_path: &[String], ) -> Result { self.internal - .drop_function(name.as_ref(), version.as_ref()) + .drop_function(name.as_ref(), version.as_ref(), namespace_path) .await } @@ -663,9 +673,10 @@ impl Connection { &self, name: impl AsRef, version: impl AsRef, + namespace_path: &[String], ) -> Result<(bool, crate::job::Job)> { self.internal - .drop_function_async(name.as_ref(), version.as_ref()) + .drop_function_async(name.as_ref(), version.as_ref(), namespace_path) .await } diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index 224fd8770..2518edf13 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -333,10 +333,12 @@ pub trait Database: /// /// See [`CloneTableRequest`] for detailed documentation and examples. async fn clone_table(&self, request: CloneTableRequest) -> Result>; - /// Submit a Function creation job that builds an image and registers it. + /// Submit a Function creation job that builds an image and registers it + /// in `namespace_path`. async fn create_function_async( &self, _request: crate::function::FunctionRegistrationRequest, + _namespace_path: &[String], ) -> Result> { function_catalog_not_supported() } @@ -412,15 +414,25 @@ pub trait Database: &self, _name: &str, _version: &str, + _namespace_path: &[String], ) -> Result { function_catalog_not_supported() } - /// List every published immutable Function version in the remote catalog. - async fn list_functions(&self) -> Result> { + /// List every published immutable Function version in one namespace of the + /// remote catalog. + async fn list_functions( + &self, + _namespace_path: &[String], + ) -> Result> { function_catalog_not_supported() } /// Remove the current Function name binding, retaining the object history. - async fn drop_function(&self, _name: &str, _version: &str) -> Result { + async fn drop_function( + &self, + _name: &str, + _version: &str, + _namespace_path: &[String], + ) -> Result { function_catalog_not_supported() } /// Start dropping a Function and return a handle to the cleanup job. @@ -431,8 +443,9 @@ pub trait Database: &self, name: &str, version: &str, + namespace_path: &[String], ) -> Result<(bool, crate::job::Job)> { - let dropped = self.drop_function(name, version).await?; + let dropped = self.drop_function(name, version, namespace_path).await?; Ok((dropped, crate::job::Job::new_done())) } /// Create a named Secret in this database. Fails if the name is taken, so diff --git a/rust/lancedb/src/function.rs b/rust/lancedb/src/function.rs index 3b7209bd5..f05633281 100644 --- a/rust/lancedb/src/function.rs +++ b/rust/lancedb/src/function.rs @@ -85,22 +85,25 @@ fn has_unknown_keys(value: &Value, allowed: &[&str]) -> bool { .is_some_and(|object| object.keys().any(|key| !allowed.contains(&key.as_str()))) } +/// Every field a [`FunctionVersionRef`] document may carry. An application or +/// binding with any other key comes from a newer contract. +pub(crate) const FUNCTION_VERSION_REF_FIELDS: &[&str] = &[ + "name", + "namespace_path", + "object_id", + "location", + "version", + "manifest_digest", +]; + fn application_has_unknown_nested_fields(value: &Value) -> bool { let Some(application) = value.as_object() else { return false; }; - if application.get("function").is_some_and(|value| { - has_unknown_keys( - value, - &[ - "name", - "object_id", - "location", - "version", - "manifest_digest", - ], - ) - }) { + if application + .get("function") + .is_some_and(|value| has_unknown_keys(value, FUNCTION_VERSION_REF_FIELDS)) + { return true; } if application @@ -431,9 +434,17 @@ fn deserialize_object_version<'de, D: Deserializer<'de>>( } /// One immutable Function object revision and its executable artifact. +/// +/// The Function is named by its parts, `name` and `namespace_path`, the way a +/// [`SecretReference`](crate::secrets::SecretReference) names a Secret. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct FunctionVersion { name: String, + /// The namespace holding the Function. Empty is the root, and is absent + /// from the wire, as a [`SecretReference`](crate::secrets::SecretReference) + /// omits it. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + namespace_path: Vec, object_id: String, location: String, #[serde(deserialize_with = "deserialize_object_version")] @@ -463,6 +474,7 @@ impl FunctionVersion { pub fn reference(&self) -> FunctionVersionRef { FunctionVersionRef { name: self.name.clone(), + namespace_path: self.namespace_path.clone(), object_id: self.object_id.clone(), location: self.location.clone(), version: self.version.clone(), @@ -472,6 +484,9 @@ impl FunctionVersion { pub fn name(&self) -> &str { &self.name } + pub fn namespace_path(&self) -> &[String] { + &self.namespace_path + } pub fn version(&self) -> &str { &self.version } @@ -556,6 +571,10 @@ impl_json!(FunctionRegistrationRequest); #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct FunctionVersionRef { pub name: String, + /// The namespace holding the Function; empty is the root and is absent + /// from the wire. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub namespace_path: Vec, pub object_id: String, pub location: String, #[serde(deserialize_with = "deserialize_object_version")] @@ -954,6 +973,46 @@ mod tests { ); } + /// A Function is named by its parts, and a root one states no namespace. + #[test] + fn function_version_carries_its_namespace_as_a_separate_field() { + const VERSION: &str = include_str!( + "../tests/fixtures/first_class_functions/v1/remote_function_version.canonical.json" + ); + let root = FunctionVersion::from_json(VERSION).unwrap(); + assert!(root.namespace_path().is_empty()); + assert_eq!(root.to_canonical_json().unwrap(), VERSION.trim_end()); + + let mut document: Value = serde_json::from_str(VERSION).unwrap(); + document["namespace_path"] = serde_json::json!(["analytics", "features"]); + let nested = FunctionVersion::from_json(&document.to_string()).unwrap(); + assert_eq!(nested.name(), "embed"); + assert_eq!(nested.namespace_path(), ["analytics", "features"]); + let reference = nested.reference(); + assert_eq!(reference.name, "embed"); + assert_eq!(reference.namespace_path, ["analytics", "features"]); + let reference: Value = serde_json::to_value(&reference).unwrap(); + assert_eq!( + reference["namespace_path"], + serde_json::json!(["analytics", "features"]) + ); + let root_reference = serde_json::to_value(root.reference()).unwrap(); + assert!(root_reference.get("namespace_path").is_none()); + + // An application carrying the namespace is not from a newer contract. + let application = serde_json::json!({ + "function": reference, + "inputs": [], + "output": serde_json::to_value(nested.signature().output.clone()).unwrap(), + }); + let application = FunctionApplication::from_json(&application.to_string()).unwrap(); + assert!(!application.has_unknown_fields()); + assert_eq!( + application.function().namespace_path, + ["analytics", "features"] + ); + } + /// Arrays are ordered by the caller, so canonicalization must leave them /// alone -- sorting them would change what a signature means. #[test] diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 39bb5a528..27606cdfe 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -979,8 +979,9 @@ impl Database for RemoteDatabase { async fn create_function_async( &self, request: FunctionRegistrationRequest, + namespace_path: &[String], ) -> Result> { - let function_id = build_object_identifier("Function name", &request.name, &[])?; + let function_id = build_object_identifier("Function name", &request.name, namespace_path)?; let req = self .client .post(&format!("/v1/function/{function_id}/create")) @@ -1005,8 +1006,13 @@ impl Database for RemoteDatabase { )))) } - async fn get_function(&self, name: &str, version: &str) -> Result { - let function_id = build_object_identifier("Function name", name, &[])?; + async fn get_function( + &self, + name: &str, + version: &str, + namespace_path: &[String], + ) -> Result { + let function_id = build_object_identifier("Function name", name, namespace_path)?; let req = self .client .post(&format!("/v1/function/{function_id}/describe")) @@ -1018,8 +1024,8 @@ impl Database for RemoteDatabase { response.json().await.err_to_http(request_id) } - async fn list_functions(&self) -> Result> { - let namespace_id = build_namespace_identifier(&[])?; + async fn list_functions(&self, namespace_path: &[String]) -> Result> { + let namespace_id = build_namespace_identifier(namespace_path)?; let path = format!("/v1/namespace/{namespace_id}/function/list"); let mut functions = Vec::new(); let mut page_token: Option = None; @@ -1059,12 +1065,25 @@ impl Database for RemoteDatabase { Ok(functions) } - async fn drop_function(&self, name: &str, version: &str) -> Result { - Ok(self.drop_function_async(name, version).await?.0) + async fn drop_function( + &self, + name: &str, + version: &str, + namespace_path: &[String], + ) -> Result { + Ok(self + .drop_function_async(name, version, namespace_path) + .await? + .0) } - async fn drop_function_async(&self, name: &str, version: &str) -> Result<(bool, Job)> { - let function_id = build_object_identifier("Function name", name, &[])?; + async fn drop_function_async( + &self, + name: &str, + version: &str, + namespace_path: &[String], + ) -> Result<(bool, Job)> { + let function_id = build_object_identifier("Function name", name, namespace_path)?; let req = self .client .post(&format!("/v1/function/{function_id}/drop")) @@ -2004,7 +2023,7 @@ mod tests { .body(serde_json::json!({"dropped": true, "job_id": "j1-fn-drop"}).to_string()) .unwrap() }); - let (dropped, job) = db.drop_function_async("embed", "1").await.unwrap(); + let (dropped, job) = db.drop_function_async("embed", "1", &[]).await.unwrap(); assert!(dropped); assert_eq!(job.id(), Some("j1-fn-drop")); } @@ -2019,7 +2038,7 @@ mod tests { .body(serde_json::json!({"dropped": false}).to_string()) .unwrap() }); - let (dropped, job) = db.drop_function_async("embed", "1").await.unwrap(); + let (dropped, job) = db.drop_function_async("embed", "1", &[]).await.unwrap(); assert!(!dropped); assert_eq!(job.id(), None); assert_eq!(job.status().await.unwrap(), "finished"); @@ -2036,7 +2055,11 @@ mod tests { let db = super::RemoteDatabase::new_mock(move |_| { http::Response::builder().status(202).body(body).unwrap() }); - let error = db.drop_function_async("embed", "1").await.err().unwrap(); + let error = db + .drop_function_async("embed", "1", &[]) + .await + .err() + .unwrap(); assert!(error.to_string().contains("valid job_id")); } } @@ -3905,7 +3928,7 @@ mod tests { http::Response::builder().status(200).body("{}").unwrap() }); let error = conn - .drop_function(name, "fv_1") + .drop_function(name, "fv_1", &[]) .await .expect_err("a dot-only Function name must be refused"); assert!( @@ -4307,7 +4330,7 @@ mod tests { path => panic!("unexpected path: {path}"), }); let request = crate::function::FunctionRegistrationRequest::from_json(REQUEST).unwrap(); - let job = conn.create_function_async(request).await.unwrap(); + let job = conn.create_function_async(request, &[]).await.unwrap(); assert_eq!(job.id(), Some("job-function-1")); let version = job.wait().await.unwrap(); assert_eq!(version.name(), "embed"); @@ -4327,7 +4350,7 @@ mod tests { assert_eq!(body, serde_json::json!({"version": "1"})); http::Response::builder().status(200).body(VERSION).unwrap() }); - let version = conn.get_function("embed", "1").await.unwrap(); + let version = conn.get_function("embed", "1", &[]).await.unwrap(); assert_eq!(version.name(), "embed"); assert_eq!(version.version(), "1"); } @@ -4370,7 +4393,7 @@ mod tests { } } }); - let functions = conn.list_functions().await.unwrap(); + let functions = conn.list_functions(&[]).await.unwrap(); assert_eq!(functions.len(), 1); assert_eq!(functions[0].name(), "embed"); assert_eq!(functions[0].version(), "1"); @@ -4393,7 +4416,7 @@ mod tests { .unwrap() }); - let functions = conn.list_functions().await.unwrap(); + let functions = conn.list_functions(&[]).await.unwrap(); assert!(functions.is_empty()); assert_eq!(requests.load(Ordering::SeqCst), 1); } @@ -4434,7 +4457,7 @@ mod tests { .unwrap() }); - let error = conn.list_functions().await.unwrap_err(); + let error = conn.list_functions(&[]).await.unwrap_err(); assert!( matches!( &error, @@ -4461,7 +4484,82 @@ mod tests { .body(r#"{"dropped":false}"#) .unwrap() }); - assert!(!conn.drop_function("embed", "1").await.unwrap()); + assert!(!conn.drop_function("embed", "1", &[]).await.unwrap()); + } + + /// A Function's namespace is addressed in the path the way a Secret's is: + /// every route takes the joined identifier and no body carries the + /// namespace, so a namespaced request differs from a root one only in its + /// path. + #[tokio::test] + async fn test_a_function_namespace_path_is_addressed_in_the_path() { + const REQUEST: &str = include_str!( + "../../tests/fixtures/first_class_functions/v1/remote_function_registration_request.json" + ); + const VERSION: &str = include_str!( + "../../tests/fixtures/first_class_functions/v1/remote_function_version.canonical.json" + ); + let namespace = ["analytics".to_string(), "features".to_string()]; + let mut expected_create: serde_json::Value = serde_json::from_str(REQUEST).unwrap(); + expected_create.as_object_mut().unwrap().remove("name"); + let conn = Connection::new_with_handler(move |request| { + let body = request + .body() + .and_then(|body| body.as_bytes()) + .map(|bytes| serde_json::from_slice::(bytes).unwrap()); + match request.url().path() { + "/v1/function/analytics$features$normalize_score/create" => { + assert_eq!(body.unwrap(), expected_create); + http::Response::builder() + .status(202) + .body(r#"{"job_id":"job-function-1"}"#.to_string()) + .unwrap() + } + "/v1/function/analytics$features$embed/describe" => { + assert_eq!(body.unwrap(), serde_json::json!({"version": "1"})); + http::Response::builder() + .status(200) + .body(VERSION.to_string()) + .unwrap() + } + "/v1/function/analytics$features$embed/drop" => { + assert_eq!(body.unwrap(), serde_json::json!({"version": "1"})); + http::Response::builder() + .status(200) + .body(r#"{"dropped":true}"#.to_string()) + .unwrap() + } + // Listing is namespace-scoped, so the namespace is the whole + // identifier. + "/v1/namespace/analytics$features/function/list" => http::Response::builder() + .status(200) + .body(r#"{"functions":[]}"#.to_string()) + .unwrap(), + path => panic!("unexpected path: {path}"), + } + }); + let request = crate::function::FunctionRegistrationRequest::from_json(REQUEST).unwrap(); + let job = conn + .create_function_async(request, &namespace) + .await + .unwrap(); + assert_eq!(job.id(), Some("job-function-1")); + conn.get_function("embed", "1", &namespace).await.unwrap(); + assert!(conn.list_functions(&namespace).await.unwrap().is_empty()); + assert!(conn.drop_function("embed", "1", &namespace).await.unwrap()); + } + + /// Each namespace segment is checked before a request is built, so an + /// empty segment cannot collapse the identifier onto the parent namespace. + #[tokio::test] + async fn test_an_unaddressable_function_namespace_segment_is_refused() { + let conn = Connection::new_with_handler(|request| -> http::Response { + panic!("reached the transport: {}", request.url().path()) + }); + let namespace = ["analytics".to_string(), String::new()]; + assert!(conn.get_function("embed", "1", &namespace).await.is_err()); + assert!(conn.list_functions(&namespace).await.is_err()); + assert!(conn.drop_function("embed", "1", &namespace).await.is_err()); } #[tokio::test] diff --git a/rust/lancedb/src/table/computed_columns.rs b/rust/lancedb/src/table/computed_columns.rs index 98b4e1627..d92599294 100644 --- a/rust/lancedb/src/table/computed_columns.rs +++ b/rust/lancedb/src/table/computed_columns.rs @@ -704,13 +704,7 @@ fn ensure_known_binding_shape(value: &Value) -> Result<()> { object .get("function") .ok_or_else(|| invalid_function("Function binding is missing its exact version"))?, - &[ - "name", - "object_id", - "location", - "version", - "manifest_digest", - ], + crate::function::FUNCTION_VERSION_REF_FIELDS, "version reference", )?; for input in object @@ -3760,6 +3754,20 @@ mod tests { ensure_known_binding_shape(&raw_binding).unwrap(); } + /// A binding names its Function by parts, so one declared against a + /// namespaced Function is part of the known contract too. + #[test] + fn test_namespaced_bindings_are_a_known_shape() { + let mut raw_binding: Value = serde_json::from_str(include_str!( + "../../tests/fixtures/first_class_functions/v1/remote_initialized_function_binding.json" + )) + .unwrap(); + raw_binding["function"]["namespace_path"] = serde_json::json!(["analytics", "features"]); + ensure_known_binding_shape(&raw_binding).unwrap(); + let binding = FunctionBinding::from_json(&raw_binding.to_string()).unwrap(); + assert_eq!(binding.function().namespace_path, ["analytics", "features"]); + } + #[test] fn test_named_struct_can_be_kept_as_one_nullable_physical_column() { let application = named_struct_application("{}"); diff --git a/rust/lancedb/tests/first_class_function_slice2.rs b/rust/lancedb/tests/first_class_function_slice2.rs index 14ceb511d..380766e03 100644 --- a/rust/lancedb/tests/first_class_function_slice2.rs +++ b/rust/lancedb/tests/first_class_function_slice2.rs @@ -97,13 +97,16 @@ async fn local_function_catalog_operations_return_stable_not_supported() { )) .unwrap(); - let create_error = connection.create_function_async(request).await.unwrap_err(); + let create_error = connection + .create_function_async(request, &[]) + .await + .unwrap_err(); let lookup_error = connection - .get_function("normalize_score", "1") + .get_function("normalize_score", "1", &[]) .await .unwrap_err(); let drop_error = connection - .drop_function("normalize_score", "1") + .drop_function("normalize_score", "1", &[]) .await .unwrap_err(); for error in [create_error, lookup_error, drop_error] {