Compare commits

..
Author SHA1 Message Date
lancedb automation a3f314960f chore: update lance dependency to v12.0.0-beta.18 2026-09-12 04:50:09 +00:00
Drew 6702e3fec1 feat(node): add blob v2 fetch and field helpers (#4155)
this PR blob v2 field helpers and reads to the Node SDK.

`blob()` marks a field as blob v2 and lets you set the storage
thresholds. Inputs can be bytes, a URI, or a data/uri struct.

Queries return descriptors. `fetchBlobs()` reads the bytes by row ID,
and `fetchBlobFiles()` gives you lazy handles for full or range reads.
`blobColumns()` lists the blob fields, including nested ones.

Fetch uses the table’s current checkout. It preserves order, duplicates,
and nulls. Holding row IDs across compaction still requires stable row
IDs.

```javascript
const db = await connect("./data");
const video = await readFile("clip.mp4");
const table = await db.createTable(
  "videos",
  [{ id: 1n, video }],
  {
    schema: new Schema([
      new Field("id", new Int64()),
      blob("video"),
    ]),
  },
);
const rows = await table.query().select(["id"]).withRowId().toArray();
const rowIds = rows.map((row) => row._rowid as bigint);
const bytes = await table.fetchBlobs("video", rowIds);
const [handle] = await table.fetchBlobFiles("video", rowIds);
const header = await handle!.readRange(0n, 65536n);

```

### Testing

- cover input validation, thresholds, nested fields, fetch ordering,
nulls, and range reads.
2026-09-11 14:46:27 -07:00
LanceDB Robot e0bd4b5fa1 chore: update lance dependency to v12.0.0-beta.17 (#4162)
Update the Rust workspace Lance dependencies and Java lance-core to
[v12.0.0-beta.17](https://github.com/lance-format/lance/releases/tag/v12.0.0-beta.17).
Align object_store to 0.14.1 for compatibility with Lance and refresh
the Cargo lockfile, including the required reqsign updates. Validation
passed: `cargo clippy --quiet --workspace --tests --all-features -- -D
warnings` and `cargo fmt --all --quiet`.
2026-09-11 09:59:19 -07:00
28 changed files with 1635 additions and 2324 deletions
Generated
+250 -240
View File
File diff suppressed because it is too large Load Diff
+15 -15
View File
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
rust-version = "1.91.0"
[workspace.dependencies]
lance = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
lance = { "version" = "=12.0.0-beta.18", default-features = false, "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=12.0.0-beta.18", default-features = false, "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=12.0.0-beta.18", default-features = false, "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=12.0.0-beta.18", "tag" = "v12.0.0-beta.18", "git" = "https://github.com/lance-format/lance.git" }
lancedb = { path = "rust/lancedb", default-features = false }
ahash = "0.8"
# Note that this one does not include pyarrow
@@ -60,7 +60,7 @@ log = "0.4"
metrics = "0.24"
metrics-util = "0.19"
moka = { version = "0.12", features = ["future"] }
object_store = "0.13.2"
object_store = "0.14.1"
pin-project = "1.0.7"
rand = "0.9"
snafu = "0.8"
+62
View File
@@ -0,0 +1,62 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / BlobFile
# Class: BlobFile
A lazy handle to blob bytes. Create one with [Table.fetchBlobFiles](Table.md#fetchblobfiles).
## Methods
### read()
```ts
read(): Promise<Buffer>
```
Reads from the cursor to the end and advances the cursor.
A second call returns an empty buffer. [BlobFile.readRange](BlobFile.md#readrange) does
not move the cursor.
#### Returns
`Promise`&lt;`Buffer`&gt;
***
### readRange()
```ts
readRange(start, end): Promise<Buffer>
```
Reads the half-open byte range `[start, end)`.
Fails when `end` is past the blob size. Does not move the cursor.
#### Parameters
* **start**: `bigint`
* **end**: `bigint`
#### Returns
`Promise`&lt;`Buffer`&gt;
***
### size()
```ts
size(): bigint
```
Returns the blob size in bytes.
#### Returns
`bigint`
+70 -8
View File
@@ -74,10 +74,10 @@ now: the column is committed with no values, and rows get them from
[Table#refreshColumn](Table.md#refreshcolumn). Declaring one therefore costs the same on a
large table as on an empty one.
A refresh also recomputes the rows whose inputs changed since they were
computed, so a mutated input is reflected by the next refresh. While a
declaration reads a column, that column cannot be renamed, retyped or
dropped.
A refresh does not revisit rows it has already filled, so mutating an
input leaves the value computed at fill time; recomputing means dropping
the column and declaring it again. While a declaration reads a column,
that column cannot be renamed, retyped or dropped.
On LanceDB Cloud and Enterprise the expression is planned by the
server, and the refresh runs as a server job -- see
@@ -137,6 +137,20 @@ containing the new version number of the table after altering the columns.
***
### blobColumns()
```ts
abstract blobColumns(): Promise<string[]>
```
Blob v2 columns, including nested dotted paths.
#### Returns
`Promise`&lt;`string`[]&gt;
***
### branches()
```ts
@@ -499,6 +513,54 @@ Drop an index from the table.
***
### fetchBlobFiles()
```ts
abstract fetchBlobFiles(column, rowIds): Promise<(null | BlobFile)[]>
```
Opens lazy blob handles for `column` at the given row IDs using the
table's current checkout.
Preserves input order, duplicates, and nulls. Use this for large payloads.
See [Table.fetchBlobs](Table.md#fetchblobs) for row-ID validity across versions.
#### Parameters
* **column**: `string`
* **rowIds**: readonly (`number` \| `bigint`)[]
#### Returns
`Promise`&lt;(`null` \| [`BlobFile`](BlobFile.md))[]&gt;
***
### fetchBlobs()
```ts
abstract fetchBlobs(column, rowIds): Promise<(null | Buffer)[]>
```
Bytes for `column` at row IDs from [Query.withRowId](Query.md#withrowid).
Reads the table's current checkout. IDs from another version can fail after
compaction unless stable row ids are enabled. Results keep input order and
duplicates. Null blobs are `null`. Empty blobs are empty buffers.
#### Parameters
* **column**: `string`
* **rowIds**: readonly (`number` \| `bigint`)[]
#### Returns
`Promise`&lt;(`null` \| `Buffer`)[]&gt;
***
### flushLsm()
```ts
@@ -854,10 +916,10 @@ abstract refreshColumn(column): Promise<RefreshColumnResult>
Fill the rows of a computed column that hold no value yet.
Rows appended since the last refresh are filled by the next one, and
rows whose inputs changed since they were computed are recomputed;
everything else is left as it is. Local tables only: a remote refresh
runs as a server job, through [Table#refreshColumnAsync](Table.md#refreshcolumnasync).
Rows appended since the last refresh are filled by the next one; rows
already filled are left as they are, so the call is idempotent and does
not observe a mutated input. Local tables only: a remote refresh runs
as a server job, through [Table#refreshColumnAsync](Table.md#refreshcolumnasync).
#### Parameters
+55
View File
@@ -0,0 +1,55 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / blob
# Function: blob()
```ts
function blob(name, options): Field
```
Declares a `lance.blob.v2` column.
Query results are descriptors, not payload bytes. Use [Table.fetchBlobs](../classes/Table.md#fetchblobs)
or [Table.fetchBlobFiles](../classes/Table.md#fetchblobfiles) to read bytes.
## Parameters
* **name**: `string`
* **options**: [`BlobOptions`](../type-aliases/BlobOptions.md) = `{}`
## Returns
`Field`
## Example
```ts
import { readFile } from "node:fs/promises";
import { Field, Int64, Schema } from "apache-arrow";
import { blob, connect } from "@lancedb/lancedb";
const db = await connect("./data");
const video = await readFile("clip.mp4");
const table = await db.createTable(
"videos",
[{ id: 1n, video }],
{
schema: new Schema([
new Field("id", new Int64()),
blob("video"),
]),
},
);
const rows = await table.query().select(["id"]).withRowId().toArray();
const rowIds = rows.map((row) => row._rowid as bigint);
const bytes = await table.fetchBlobs("video", rowIds);
const [handle] = await table.fetchBlobFiles("video", rowIds);
const size = handle!.size();
const header = await handle!.readRange(0n, size < 65536n ? size : 65536n);
```
+22
View File
@@ -0,0 +1,22 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / isBlobField
# Function: isBlobField()
```ts
function isBlobField(field): boolean
```
Checks for the `lance.blob.v2` extension marker. Does not validate the
field's storage type.
## Parameters
* **field**: `Field`&lt;`any`&gt;
## Returns
`boolean`
+4
View File
@@ -19,6 +19,7 @@
## Classes
- [AutoQuery](classes/AutoQuery.md)
- [BlobFile](classes/BlobFile.md)
- [BooleanQuery](classes/BooleanQuery.md)
- [BoostQuery](classes/BoostQuery.md)
- [BranchContents](classes/BranchContents.md)
@@ -143,6 +144,7 @@
- [AnalyzePlanDistributedMetrics](type-aliases/AnalyzePlanDistributedMetrics.md)
- [BaseTokenizer](type-aliases/BaseTokenizer.md)
- [BlobOptions](type-aliases/BlobOptions.md)
- [Data](type-aliases/Data.md)
- [DataLike](type-aliases/DataLike.md)
- [FieldLike](type-aliases/FieldLike.md)
@@ -158,9 +160,11 @@
## Functions
- [RecordBatchIterator](functions/RecordBatchIterator.md)
- [blob](functions/blob.md)
- [connect](functions/connect.md)
- [connectNamespace](functions/connectNamespace.md)
- [instrumentLanceDbMetrics](functions/instrumentLanceDbMetrics.md)
- [isBlobField](functions/isBlobField.md)
- [makeArrowTable](functions/makeArrowTable.md)
- [packBits](functions/packBits.md)
- [permutationBuilder](functions/permutationBuilder.md)
+48
View File
@@ -0,0 +1,48 @@
[**@lancedb/lancedb**](../README.md) • **Docs**
***
[@lancedb/lancedb](../globals.md) / BlobOptions
# Type Alias: BlobOptions
```ts
type BlobOptions: object;
```
## Type declaration
### dedicatedSizeThreshold?
```ts
optional dedicatedSizeThreshold: number;
```
Max payload bytes stored in a packed sidecar before a dedicated file. Must
be a positive safe integer.
### inlineSizeThreshold?
```ts
optional inlineSizeThreshold: number;
```
Max payload bytes kept inline in the data file. Zero is allowed. Must be a
safe integer.
### nullable?
```ts
optional nullable: boolean;
```
Defaults to true.
### packFileSizeThreshold?
```ts
optional packFileSizeThreshold: number;
```
Max bytes in one packed sidecar before starting another. Must be a positive
safe integer.
+1 -1
View File
@@ -28,7 +28,7 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<arrow.version>15.0.0</arrow.version>
<lance-core.version>12.0.0-beta.16</lance-core.version>
<lance-core.version>12.0.0-beta.18</lance-core.version>
<spotless.skip>false</spotless.skip>
<spotless.version>2.30.0</spotless.version>
<spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
+185
View File
@@ -0,0 +1,185 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import { Field, Int64, List, Schema, Struct, Utf8 } from "apache-arrow";
import { makeArrowTable } from "../lancedb/arrow";
import { BlobFile, blob, coerceBlobValue, isBlobField } from "../lancedb/blob";
describe("blob()", () => {
it("marks the field as lance.blob.v2", () => {
const field = blob("image", { nullable: false });
expect(field.nullable).toBe(false);
expect(isBlobField(field)).toBe(true);
expect(field.metadata.get("ARROW:extension:name")).toBe("lance.blob.v2");
});
it("writes encoding thresholds as field metadata", () => {
const field = blob("video", {
inlineSizeThreshold: 1024,
dedicatedSizeThreshold: 2 * 1024 * 1024,
packFileSizeThreshold: 64 * 1024 * 1024,
});
expect(
field.metadata.get("lance-encoding:blob-inline-size-threshold"),
).toBe("1024");
expect(
field.metadata.get("lance-encoding:blob-dedicated-size-threshold"),
).toBe(String(2 * 1024 * 1024));
expect(
field.metadata.get("lance-encoding:blob-pack-file-size-threshold"),
).toBe(String(64 * 1024 * 1024));
});
it("rejects invalid thresholds", () => {
expect(() => blob("image", { inlineSizeThreshold: -1 })).toThrow(
/inlineSizeThreshold must be non-negative/,
);
expect(() => blob("image", { dedicatedSizeThreshold: 0 })).toThrow(
/dedicatedSizeThreshold must be positive/,
);
expect(() => blob("image", { packFileSizeThreshold: 1.5 })).toThrow(
/packFileSizeThreshold must be a safe integer/,
);
expect(() =>
blob("image", { dedicatedSizeThreshold: Number.MAX_SAFE_INTEGER + 1 }),
).toThrow(/dedicatedSizeThreshold must be a safe integer/);
});
});
describe("coerceBlobValue", () => {
it.each([
["Buffer", Buffer.from("x"), { data: Buffer.from("x"), uri: null }],
[
"Uint8Array",
new Uint8Array([120]),
{ data: new Uint8Array([120]), uri: null },
],
["URI string", "s3://bucket/key", { data: null, uri: "s3://bucket/key" }],
[
"data struct",
{ data: Buffer.from("y") },
{ data: Buffer.from("y"), uri: null },
],
[
"uri struct",
{ uri: "s3://bucket/key" },
{ data: null, uri: "s3://bucket/key" },
],
["null", null, null],
])("accepts %s", (_name, input, expected) => {
expect(coerceBlobValue(input)).toEqual(expected);
});
it.each([
["empty URI", "", /uri cannot be empty/],
["object without data or uri", { position: 0 }, /data' or 'uri/],
[
"Int16Array",
new Int16Array([1]),
/Blob data must be Buffer or Uint8Array/,
],
[
"both data and uri",
{ data: Buffer.from("y"), uri: "s3://bucket/key" },
/exactly one of 'data' or 'uri'/,
],
[
"neither data nor uri",
{ data: null, uri: null },
/exactly one of 'data' or 'uri'/,
],
])("rejects %s", (_name, input, message) => {
expect(() => coerceBlobValue(input)).toThrow(message);
});
});
describe("BlobFile", () => {
it("rejects constructing BlobFile without a native handle", () => {
expect(() => new (BlobFile as unknown as { new (): BlobFile })()).toThrow(
/fetchBlobFiles/,
);
});
});
describe("makeArrowTable blob columns", () => {
it("coerces Buffer input onto a blob field", () => {
const schema = new Schema([
new Field("id", new Int64(), true),
blob("image"),
]);
const table = makeArrowTable([{ id: 1n, image: Buffer.from("hello") }], {
schema,
});
expect(isBlobField(table.schema.fields[1])).toBe(true);
const image = table.getChild("image")!;
expect(image.nullCount).toBe(0);
expect(image.getChild("uri")!.get(0)).toBeNull();
expect(image.getChild("data")!.nullCount).toBe(0);
expect(Buffer.from(image.getChild("data")!.get(0)!).toString()).toBe(
"hello",
);
});
it("coerces Buffer elements inside a list and keeps null slots", () => {
const schema = new Schema([
new Field("id", new Int64(), true),
new Field("images", new List(blob("image")), true),
]);
const table = makeArrowTable(
[
{ id: 1n, images: [Buffer.from("a"), Buffer.from("bb")] },
{ id: 2n, images: null },
{ id: 3n, images: [Buffer.from("c"), null] },
{ id: 4n, images: [] },
],
{ schema },
);
const images = table.getChild("images")!;
expect(images.nullCount).toBe(1);
const rows = images.toArray();
expect(rows[1]).toBeNull();
expect(Array.from(rows[3] as Iterable<unknown>)).toHaveLength(0);
const first = Array.from(rows[0] as Iterable<{ data: Uint8Array | null }>);
expect(Buffer.from(first[0].data!).toString()).toBe("a");
expect(Buffer.from(first[1].data!).toString()).toBe("bb");
const third = Array.from(
rows[2] as Iterable<{ data: Uint8Array | null } | null>,
);
expect(Buffer.from(third[0]!.data!).toString()).toBe("c");
expect(third[1]).toBeNull();
});
it("coerces Buffer fields inside list structs", () => {
const schema = new Schema([
new Field("id", new Int64(), true),
new Field(
"items",
new List(
new Field(
"item",
new Struct([new Field("name", new Utf8(), true), blob("image")]),
true,
),
),
true,
),
]);
const table = makeArrowTable(
[
{
id: 1n,
items: [{ name: "one", image: Buffer.from("alpha") }],
},
],
{ schema },
);
const items = Array.from(
table.getChild("items")!.toArray()[0] as Iterable<{
name: string;
image: { data: Uint8Array | null };
}>,
);
expect(items[0].name).toBe("one");
expect(Buffer.from(items[0].image.data!).toString()).toBe("alpha");
});
});
+271
View File
@@ -18,6 +18,7 @@ import {
Query,
Table,
VectorQuery,
blob,
connect,
tokenize,
} from "../lancedb";
@@ -2401,6 +2402,276 @@ describe("when dealing with versioning", () => {
});
});
describe("when dealing with blob columns", () => {
let tmpDir: tmp.DirResult;
beforeEach(() => {
tmpDir = tmp.dirSync({ unsafeCleanup: true });
});
afterEach(() => {
tmpDir.removeCallback();
});
it("discovers blob columns", async () => {
const { table } = await openBlobTable();
expect(await table.blobColumns()).toEqual(["image"]);
});
it("preserves order, duplicates, and nulls", async () => {
const { table, rowIds } = await openBlobTable();
const [alphaId, betaId, nullId] = rowIds;
const bytes = await table.fetchBlobs("image", [
betaId,
alphaId,
betaId,
nullId,
]);
expect(bytes.map((b) => (b == null ? null : b.toString()))).toEqual([
"beta",
"alpha",
"beta",
null,
]);
const files = await table.fetchBlobFiles("image", [
betaId,
nullId,
alphaId,
]);
expect(files.map((f) => f == null)).toEqual([false, true, false]);
});
it("reads full blob contents", async () => {
const { table, rowIds, alpha, beta } = await openBlobTable();
const bytes = await table.fetchBlobs("image", rowIds);
expect(bytes[0]!.equals(alpha)).toBe(true);
expect(bytes[1]!.equals(beta)).toBe(true);
const files = await table.fetchBlobFiles("image", rowIds);
expect(files[0]!.size()).toBe(BigInt(alpha.length));
expect(Buffer.from(await files[0]!.read()).toString()).toBe("alpha");
expect(Buffer.from(await files[1]!.read()).toString()).toBe("beta");
});
it("reads a half-open range", async () => {
const { table, rowIds } = await openBlobTable();
const files = await table.fetchBlobFiles("image", rowIds);
expect(Buffer.from(await files[0]!.readRange(0n, 2n)).toString()).toBe(
"al",
);
});
it("readRange does not move the cursor", async () => {
const { table, rowIds, alpha } = await openBlobTable();
const [handle] = await table.fetchBlobFiles("image", rowIds);
expect((await handle!.readRange(1n, 3n)).toString()).toBe("lp");
expect(await handle!.read()).toEqual(alpha);
expect(await handle!.read()).toEqual(Buffer.alloc(0));
});
it("fails when readRange end is past the blob size", async () => {
const { table, rowIds, alpha } = await openBlobTable();
const files = await table.fetchBlobFiles("image", rowIds);
await expect(
files[0]!.readRange(0n, BigInt(alpha.length + 1)),
).rejects.toThrow(/exceeds blob size/);
});
it("rejects fetchBlobs on a non-blob column", async () => {
const { table, rowIds } = await openBlobTable();
await expect(table.fetchBlobs("id", rowIds)).rejects.toThrow(/blob/i);
});
it("discovers and fetches nested blob columns", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
new Field("info", new Struct([blob("image")]), true),
]);
const payload = Buffer.from("nested");
const table = await db.createTable(
"nested_blobs",
[{ id: 1n, info: { image: payload } }],
{ schema },
);
expect(await table.blobColumns()).toEqual(["info.image"]);
const rows = await table.query().withRowId().toArray();
const bytes = await table.fetchBlobs("info.image", [
rows[0]._rowid as bigint,
]);
expect(bytes[0]!.equals(payload)).toBe(true);
});
it("creates and adds list blob columns", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
new Field("images", new List(blob("image")), true),
]);
const alpha = Buffer.from("alpha");
const beta = Buffer.from("beta");
const gamma = Buffer.from("gamma");
const table = await db.createTable(
"list_blobs",
[{ id: 1n, images: [alpha, beta] }],
{ schema },
);
await table.add([
{ id: 2n, images: null },
{ id: 3n, images: [gamma, null] },
{ id: 4n, images: [] },
]);
expect(await table.blobColumns()).toEqual(["images.image"]);
const rows = await table.query().toArray();
const byId = new Map(rows.map((row) => [Number(row.id), row]));
expect(descriptorSizes(byId.get(1)!.images)).toEqual([
alpha.length,
beta.length,
]);
expect(byId.get(2)!.images).toBeNull();
expect(descriptorSizes(byId.get(3)!.images)).toEqual([gamma.length, null]);
expect(Array.from(byId.get(4)!.images as Iterable<unknown>)).toHaveLength(
0,
);
});
it("creates and adds list struct blob columns", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
new Field(
"items",
new List(
new Field(
"item",
new Struct([new Field("name", new Utf8(), true), blob("image")]),
true,
),
),
true,
),
]);
const alpha = Buffer.from("nested-alpha");
const beta = Buffer.from("nested-beta");
const table = await db.createTable(
"list_struct_blobs",
[{ id: 1n, items: [{ name: "one", image: alpha }] }],
{ schema },
);
await table.add([
{
id: 2n,
items: [
{ name: "two", image: beta },
{ name: "three", image: null },
],
},
]);
const rows = await table.query().toArray();
const byId = new Map(rows.map((row) => [Number(row.id), row]));
expect(
descriptorSizes(
Array.from(byId.get(1)!.items as Iterable<{ image: unknown }>).map(
(item) => item.image,
),
),
).toEqual([alpha.length]);
expect(
descriptorSizes(
Array.from(byId.get(2)!.items as Iterable<{ image: unknown }>).map(
(item) => item.image,
),
),
).toEqual([beta.length, null]);
});
it("rejects blob fields inside a fixed-size list", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
new Field("frames", new FixedSizeList(2, blob("frame")), true),
]);
await expect(
db.createTable(
"fsl_blobs",
[{ id: 1n, frames: [Buffer.from("a"), Buffer.from("b")] }],
{ schema },
),
).rejects.toThrow(
"Blob fields inside FixedSizeList are not supported. Use List instead.",
);
});
it("rejects blob fields inside a nested fixed-size list", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
new Field(
"clip",
new Struct([
new Field("frames", new FixedSizeList(2, blob("frame")), true),
]),
true,
),
]);
await expect(
db.createTable(
"nested_fsl_blobs",
[
{
id: 1n,
clip: { frames: [Buffer.from("a"), Buffer.from("b")] },
},
],
{ schema },
),
).rejects.toThrow(
"Blob fields inside FixedSizeList are not supported. Use List instead.",
);
});
it("rejects an Arrow table with blob fields inside a fixed-size list", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
new Field("frames", new FixedSizeList(2, blob("frame")), true),
]);
await expect(
db.createTable("fsl_blobs_ipc", new ArrowTable(schema)),
).rejects.toThrow(
"Blob fields inside FixedSizeList are not supported. Use List instead.",
);
});
function descriptorSizes(values: unknown): (number | null)[] {
return Array.from(
values as Iterable<{ size?: bigint | number } | null>,
).map((value) => (value == null ? null : Number(value.size)));
}
async function openBlobTable() {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
blob("image"),
]);
const alpha = Buffer.from("alpha");
const beta = Buffer.from("beta");
const table = await db.createTable(
"blobs",
[
{ id: 1n, image: alpha },
{ id: 2n, image: beta },
{ id: 3n, image: null },
],
{ schema },
);
const rows = await table.query().withRowId().toArray();
const rowIdById = new Map(
rows.map((r) => [Number(r.id), r._rowid as bigint]),
);
const rowIds = [1, 2, 3].map((id) => rowIdById.get(id)!);
return { table, rowIds, alpha, beta };
}
});
describe("when dealing with tags", () => {
let tmpDir: tmp.DirResult;
beforeEach(() => {
+102 -1
View File
@@ -40,6 +40,7 @@ import {
} from "apache-arrow";
import { Buffers } from "apache-arrow/data";
import { typedArrayToArrowType } from "./arrow_type";
import { coerceBlobValue, isBlobField } from "./blob";
import { type EmbeddingFunction } from "./embedding/embedding_function";
import {
EmbeddingFunctionConfig,
@@ -430,12 +431,14 @@ export function makeArrowTable(
throw new Error("A schema must be provided if data is empty");
} else {
schema = new Schema(schema.fields, schemaMetadata);
validateBlobSchema(schema);
return new ArrowTable(schema);
}
}
let inferredSchema = inferSchema(data, schema, opt);
inferredSchema = new Schema(inferredSchema.fields, schemaMetadata);
validateBlobSchema(inferredSchema);
const finalColumns: Record<string, Vector> = {};
for (const field of inferredSchema.fields) {
@@ -445,6 +448,35 @@ export function makeArrowTable(
return new ArrowTable(inferredSchema, finalColumns);
}
function validateBlobSchema(schema: Schema): void {
for (const field of schema.fields) {
validateBlobField(field);
}
}
function validateBlobField(field: Field): void {
if (
isFixedSizeList(field.type) &&
containsBlobField(field.type.children[0])
) {
throw new Error(
"Blob fields inside FixedSizeList are not supported. Use List instead.",
);
}
for (const child of field.type.children ?? []) {
validateBlobField(child);
}
}
function containsBlobField(field: Field): boolean {
if (isBlobField(field)) {
return true;
}
return (field.type.children ?? []).some((child: Field) =>
containsBlobField(child),
);
}
function isObject(value: unknown): value is Record<string, unknown> {
return (
typeof value === "object" &&
@@ -480,6 +512,32 @@ function transposeData(
path: string[] = [],
): Vector {
const valuesPath = [...path, field.name];
if (isBlobField(field) && field.type instanceof Struct) {
const blobRows = data.map((datum) =>
coerceBlobValue(valueAtPath(datum, valuesPath)),
);
const childVectors = field.type.children.map((child) => {
const values = blobRows.map((row) =>
row == null ? null : (row[child.name as "data" | "uri"] ?? null),
);
return makeVector(values, child.type, undefined, child.nullable);
});
const nullCount = blobRows.filter((row) => row === null).length;
const structData = makeData({
type: field.type,
length: blobRows.length,
nullCount,
nullBitmap:
nullCount > 0
? arrowUtil.packBools(blobRows.map((row) => row !== null))
: undefined,
children: childVectors.map((v) => v.data[0]),
});
return arrowMakeVector(structData);
}
if (isList(field.type) && containsBlobField(field.type.children[0])) {
return transposeListData(data, field, valuesPath);
}
const values = data.map((datum) => valueAtPath(datum, valuesPath));
if (field.type instanceof Struct) {
const childFields = field.type.children;
@@ -495,7 +553,7 @@ function transposeData(
nullCount > 0
? arrowUtil.packBools(values.map((value) => value !== null))
: undefined,
children: childVectors as unknown as ArrowData<DataType>[],
children: childVectors.map((v) => v.data[0]),
});
return arrowMakeVector(structData);
} else {
@@ -503,6 +561,48 @@ function transposeData(
}
}
function transposeListData(
data: Record<string, unknown>[],
field: Field,
valuesPath: string[],
): Vector {
const listType = field.type as List;
const childField = listType.children[0];
const lists = data.map((datum) => valueAtPath(datum, valuesPath));
const flattened: Record<string, unknown>[] = [];
const validity: boolean[] = [];
const offsets: number[] = [0];
for (const list of lists) {
if (list == null) {
validity.push(false);
offsets.push(flattened.length);
continue;
}
if (!Array.isArray(list)) {
throw new Error(`expected an array for list field '${field.name}'`);
}
validity.push(true);
for (const element of list) {
flattened.push({ [childField.name]: element });
}
offsets.push(flattened.length);
}
const childVector = transposeData(flattened, childField, []);
const nullCount = validity.filter((valid) => !valid).length;
return arrowMakeVector(
makeData({
type: listType,
length: lists.length,
nullCount,
nullBitmap: nullCount > 0 ? arrowUtil.packBools(validity) : undefined,
valueOffsets: Int32Array.from(offsets),
child: childVector.data[0],
}),
);
}
/**
* Create an empty Arrow table with the provided schema
*/
@@ -952,6 +1052,7 @@ export async function fromTableToBuffer(
schema = sanitizeSchema(schema);
}
const tableWithEmbeddings = await applyEmbeddings(table, embeddings, schema);
validateBlobSchema(tableWithEmbeddings.schema);
const writer = RecordBatchFileWriter.writeAll(tableWithEmbeddings);
return Buffer.from(await writer.toUint8Array());
}
+236
View File
@@ -0,0 +1,236 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import { Field, LargeBinary, Struct, Utf8 } from "apache-arrow";
import { BlobFile as NativeBlobFile } from "./native";
const BLOB_V2_EXTENSION_NAME = "lance.blob.v2";
const INLINE_SIZE_THRESHOLD_KEY = "lance-encoding:blob-inline-size-threshold";
const DEDICATED_SIZE_THRESHOLD_KEY =
"lance-encoding:blob-dedicated-size-threshold";
const PACK_FILE_SIZE_THRESHOLD_KEY =
"lance-encoding:blob-pack-file-size-threshold";
export type BlobInput = {
data: Buffer | Uint8Array | null;
uri: string | null;
};
export type BlobOptions = {
/** Defaults to true. */
nullable?: boolean;
/**
* Max payload bytes kept inline in the data file. Zero is allowed. Must be a
* safe integer.
*/
inlineSizeThreshold?: number;
/**
* Max payload bytes stored in a packed sidecar before a dedicated file. Must
* be a positive safe integer.
*/
dedicatedSizeThreshold?: number;
/**
* Max bytes in one packed sidecar before starting another. Must be a positive
* safe integer.
*/
packFileSizeThreshold?: number;
};
/**
* Declares a `lance.blob.v2` column.
*
* Query results are descriptors, not payload bytes. Use {@link Table.fetchBlobs}
* or {@link Table.fetchBlobFiles} to read bytes.
*
* @example
* ```ts
* import { readFile } from "node:fs/promises";
* import { Field, Int64, Schema } from "apache-arrow";
* import { blob, connect } from "@lancedb/lancedb";
*
* const db = await connect("./data");
* const video = await readFile("clip.mp4");
* const table = await db.createTable(
* "videos",
* [{ id: 1n, video }],
* {
* schema: new Schema([
* new Field("id", new Int64()),
* blob("video"),
* ]),
* },
* );
*
* const rows = await table.query().select(["id"]).withRowId().toArray();
* const rowIds = rows.map((row) => row._rowid as bigint);
* const bytes = await table.fetchBlobs("video", rowIds);
*
* const [handle] = await table.fetchBlobFiles("video", rowIds);
* const size = handle!.size();
* const header = await handle!.readRange(0n, size < 65536n ? size : 65536n);
* ```
*/
export function blob(name: string, options: BlobOptions = {}): Field {
const metadata = new Map<string, string>([
["ARROW:extension:name", BLOB_V2_EXTENSION_NAME],
]);
setThreshold(
metadata,
INLINE_SIZE_THRESHOLD_KEY,
"inlineSizeThreshold",
options.inlineSizeThreshold,
0,
);
setThreshold(
metadata,
DEDICATED_SIZE_THRESHOLD_KEY,
"dedicatedSizeThreshold",
options.dedicatedSizeThreshold,
1,
);
setThreshold(
metadata,
PACK_FILE_SIZE_THRESHOLD_KEY,
"packFileSizeThreshold",
options.packFileSizeThreshold,
1,
);
return new Field(
name,
new Struct([
new Field("data", new LargeBinary(), true),
new Field("uri", new Utf8(), true),
]),
options.nullable ?? true,
metadata,
);
}
/**
* Checks for the `lance.blob.v2` extension marker. Does not validate the
* field's storage type.
*/
export function isBlobField(field: Field): boolean {
return field.metadata?.get("ARROW:extension:name") === BLOB_V2_EXTENSION_NAME;
}
/**
* A lazy handle to blob bytes. Create one with {@link Table.fetchBlobFiles}.
*
* @hideconstructor
*/
export class BlobFile {
private readonly inner: NativeBlobFile;
private constructor(inner: NativeBlobFile) {
if (!(inner instanceof NativeBlobFile)) {
throw new Error("BlobFile handles come from Table.fetchBlobFiles");
}
this.inner = inner;
}
/** @ignore */
static fromNative(inner: NativeBlobFile): BlobFile {
return new BlobFile(inner);
}
/** Returns the blob size in bytes. */
size(): bigint {
return this.inner.size();
}
/**
* Reads from the cursor to the end and advances the cursor.
*
* A second call returns an empty buffer. {@link BlobFile.readRange} does
* not move the cursor.
*/
read(): Promise<Buffer> {
return this.inner.read();
}
/**
* Reads the half-open byte range `[start, end)`.
*
* Fails when `end` is past the blob size. Does not move the cursor.
*/
readRange(start: bigint, end: bigint): Promise<Buffer> {
return this.inner.readRange(start, end);
}
}
export function coerceBlobValue(value: unknown): BlobInput | null {
if (value == null) {
return null;
}
if (isBlobBytes(value)) {
return { data: value, uri: null };
}
if (ArrayBuffer.isView(value)) {
throw new Error("Blob data must be Buffer or Uint8Array");
}
if (typeof value === "string") {
if (value === "") {
throw new Error("Blob uri cannot be empty");
}
return { data: null, uri: value };
}
if (typeof value === "object") {
const record = value as Record<string, unknown>;
if (!("data" in record) && !("uri" in record)) {
throw new Error(
"Blob struct values must include a 'data' or 'uri' field",
);
}
const uri = record.uri;
if (uri === "") {
throw new Error("Blob uri cannot be empty");
}
if (uri != null && typeof uri !== "string") {
throw new Error(`Blob uri must be a string or null, got ${typeof uri}`);
}
const data = record.data;
if (data != null && !isBlobBytes(data)) {
throw new Error("Blob data must be Buffer, Uint8Array, or null");
}
const bytes = (data as Buffer | Uint8Array | null | undefined) ?? null;
const uriValue = uri ?? null;
if ((bytes == null) === (uriValue == null)) {
throw new Error(
"Blob struct values must set exactly one of 'data' or 'uri'",
);
}
return { data: bytes, uri: uriValue };
}
throw new Error(
"Blob column values must be Buffer, Uint8Array, a URI string, null, or { data?, uri? }",
);
}
function isBlobBytes(value: unknown): value is Buffer | Uint8Array {
return Buffer.isBuffer(value) || value instanceof Uint8Array;
}
function setThreshold(
metadata: Map<string, string>,
key: string,
optionName: string,
value: number | undefined,
minimum: number,
): void {
if (value === undefined) {
return;
}
if (!Number.isSafeInteger(value)) {
throw new Error(`${optionName} must be a safe integer`);
}
if (value < minimum) {
throw new Error(
minimum <= 0
? `${optionName} must be non-negative`
: `${optionName} must be positive`,
);
}
metadata.set(key, String(value));
}
+3
View File
@@ -77,6 +77,9 @@ export {
VectorColumnOptions,
} from "./arrow";
export { blob, isBlobField, BlobFile } from "./blob";
export type { BlobOptions } from "./blob";
export {
Connection,
CreateTableOptions,
+83 -24
View File
@@ -17,6 +17,7 @@ import {
tableFromIPC,
} from "./arrow";
import { BlobFile } from "./blob";
import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry";
import { IndexOptions } from "./indices";
import { Job } from "./job";
@@ -510,6 +511,35 @@ export abstract class Table {
*/
abstract takeRowIds(rowIds: readonly (bigint | number)[]): TakeQuery;
/**
* Blob v2 columns, including nested dotted paths.
*/
abstract blobColumns(): Promise<string[]>;
/**
* Bytes for `column` at row IDs from {@link Query.withRowId}.
*
* Reads the table's current checkout. IDs from another version can fail after
* compaction unless stable row ids are enabled. Results keep input order and
* duplicates. Null blobs are `null`. Empty blobs are empty buffers.
*/
abstract fetchBlobs(
column: string,
rowIds: readonly (bigint | number)[],
): Promise<(Buffer | null)[]>;
/**
* Opens lazy blob handles for `column` at the given row IDs using the
* table's current checkout.
*
* Preserves input order, duplicates, and nulls. Use this for large payloads.
* See {@link Table.fetchBlobs} for row-ID validity across versions.
*/
abstract fetchBlobFiles(
column: string,
rowIds: readonly (bigint | number)[],
): Promise<(BlobFile | null)[]>;
/**
* Create a search query to find the nearest neighbors
* of the given query
@@ -542,10 +572,10 @@ export abstract class Table {
* {@link Table#refreshColumn}. Declaring one therefore costs the same on a
* large table as on an empty one.
*
* A refresh also recomputes the rows whose inputs changed since they were
* computed, so a mutated input is reflected by the next refresh. While a
* declaration reads a column, that column cannot be renamed, retyped or
* dropped.
* A refresh does not revisit rows it has already filled, so mutating an
* input leaves the value computed at fill time; recomputing means dropping
* the column and declaring it again. While a declaration reads a column,
* that column cannot be renamed, retyped or dropped.
*
* On LanceDB Cloud and Enterprise the expression is planned by the
* server, and the refresh runs as a server job -- see
@@ -576,10 +606,10 @@ export abstract class Table {
/**
* Fill the rows of a computed column that hold no value yet.
*
* Rows appended since the last refresh are filled by the next one, and
* rows whose inputs changed since they were computed are recomputed;
* everything else is left as it is. Local tables only: a remote refresh
* runs as a server job, through {@link Table#refreshColumnAsync}.
* Rows appended since the last refresh are filled by the next one; rows
* already filled are left as they are, so the call is idempotent and does
* not observe a mutated input. Local tables only: a remote refresh runs
* as a server job, through {@link Table#refreshColumnAsync}.
* @param {string} column The name of the computed column to fill.
* @returns {Promise<RefreshColumnResult>} A promise that resolves to the
* number of rows filled and the new version number of the table.
@@ -1160,23 +1190,34 @@ export class LocalTable extends Table {
}
takeRowIds(rowIds: readonly (bigint | number)[]): TakeQuery {
const ids = rowIds.map((id) => {
if (typeof id === "bigint") {
return id;
}
if (!Number.isInteger(id)) {
throw new Error("Row id must be an integer (or bigint)");
}
if (id < 0) {
throw new Error("Row id cannot be negative");
}
if (!Number.isSafeInteger(id)) {
throw new Error("Row id is too large for number; use bigint instead");
}
return BigInt(id);
});
return new TakeQuery(this.inner.takeRowIds(rowIdsToBigInts(rowIds)));
}
return new TakeQuery(this.inner.takeRowIds(ids));
blobColumns(): Promise<string[]> {
return this.inner.blobColumns();
}
async fetchBlobs(
column: string,
rowIds: readonly (bigint | number)[],
): Promise<(Buffer | null)[]> {
const values = await this.inner.fetchBlobs(column, rowIdsToBigInts(rowIds));
// N-API Option maps missing values to undefined. Collapse those to null.
return values.map((value) => value ?? null);
}
async fetchBlobFiles(
column: string,
rowIds: readonly (bigint | number)[],
): Promise<(BlobFile | null)[]> {
const files = await this.inner.fetchBlobFiles(
column,
rowIdsToBigInts(rowIds),
);
// N-API Option maps missing values to undefined. Collapse those to null.
return files.map((file) =>
file == null ? null : BlobFile.fromNative(file),
);
}
query(): Query {
@@ -1733,3 +1774,21 @@ export class Branches {
)) as unknown as CherryPickResult;
}
}
function rowIdsToBigInts(rowIds: readonly (bigint | number)[]): bigint[] {
return rowIds.map((id) => {
if (typeof id === "bigint") {
return id;
}
if (!Number.isInteger(id)) {
throw new Error("Row id must be an integer (or bigint)");
}
if (id < 0) {
throw new Error("Row id cannot be negative");
}
if (!Number.isSafeInteger(id)) {
throw new Error("Row id is too large for number; use bigint instead");
}
return BigInt(id);
});
}
+95
View File
@@ -0,0 +1,95 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
use std::ops::Range;
use std::sync::Arc;
use arrow_array::{Array, LargeBinaryArray};
use lancedb::blob::BlobFile as LanceBlobFile;
use napi::bindgen_prelude::*;
use napi_derive::napi;
use crate::error::convert_error;
#[napi]
pub struct BlobFile {
inner: Arc<LanceBlobFile>,
}
impl BlobFile {
pub(crate) fn new(inner: LanceBlobFile) -> Self {
Self {
inner: Arc::new(inner),
}
}
}
#[napi]
impl BlobFile {
#[napi]
pub fn size(&self) -> BigInt {
BigInt::from(self.inner.size())
}
#[napi]
pub async fn read(&self) -> napi::Result<Buffer> {
let bytes = self.inner.read().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<Buffer> {
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()))
}
}
fn bigint_range(start: BigInt, end: BigInt) -> napi::Result<Range<u64>> {
let start = parse_u64(start, "start")?;
let end = parse_u64(end, "end")?;
if start > end {
return Err(napi::Error::from_reason(format!(
"invalid blob range: start ({start}) > end ({end})"
)));
}
Ok(start..end)
}
fn parse_u64(value: BigInt, name: &str) -> napi::Result<u64> {
let (negative, value, lossless) = value.get_u64();
if negative {
return Err(napi::Error::from_reason(format!(
"{name} cannot be negative"
)));
}
if !lossless {
return Err(napi::Error::from_reason(format!(
"{name} is too large to fit in u64"
)));
}
Ok(value)
}
pub fn parse_row_ids(row_ids: Vec<BigInt>) -> napi::Result<Vec<u64>> {
row_ids
.into_iter()
.map(|id| parse_u64(id, "row id"))
.collect()
}
pub fn copy_blob_buffers(array: LargeBinaryArray) -> Vec<Option<Buffer>> {
(0..array.len())
.map(|i| {
if array.is_null(i) {
None
} else {
Some(Buffer::from(array.value(i).to_vec()))
}
})
.collect()
}
+1
View File
@@ -10,6 +10,7 @@ use std::collections::HashMap;
use env_logger::Env;
use napi_derive::*;
mod blob;
mod connection;
mod error;
mod header;
+39
View File
@@ -15,6 +15,7 @@ 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::error::NapiErrorExt;
use crate::index::Index;
use crate::merge::NativeMergeInsertBuilder;
@@ -329,6 +330,44 @@ impl Table {
))
}
#[napi(catch_unwind)]
pub async fn blob_columns(&self) -> napi::Result<Vec<String>> {
self.inner_ref()?.blob_columns().await.default_error()
}
#[napi(catch_unwind)]
pub async fn fetch_blobs(
&self,
column: String,
row_ids: Vec<BigInt>,
) -> napi::Result<Vec<Option<Buffer>>> {
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))
}
#[napi(catch_unwind)]
pub async fn fetch_blob_files(
&self,
column: String,
row_ids: Vec<BigInt>,
) -> napi::Result<Vec<Option<BlobFile>>> {
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())
}
#[napi(catch_unwind)]
pub fn vector_search(&self, vector: Float32Array) -> napi::Result<VectorQuery> {
self.query()?.nearest_to(vector)
+16 -17
View File
@@ -2188,10 +2188,10 @@ class Table(ABC):
Declaring one therefore costs the same on a large table as on an
empty one.
A refresh also recomputes the rows whose inputs changed since they
were computed, so a mutated input is reflected by the next refresh.
While a declaration reads a column, that column cannot be renamed,
retyped or dropped.
A refresh does not revisit rows it has already filled, so mutating
an input leaves the value computed at fill time; recomputing means
dropping the column and declaring it again. While a declaration
reads a column, that column cannot be renamed, retyped or dropped.
On LanceDB Cloud and Enterprise the expression is planned by the
server, and the refresh runs as a server job -- see
@@ -2211,7 +2211,7 @@ class Table(ABC):
>>> table.add_columns(computed={"doubled": "x * 2"})
AddColumnsResult(version=2)
>>> table.refresh_column("doubled")
RefreshColumnResult(rows_filled=2, version=4)
RefreshColumnResult(rows_filled=2, version=3)
>>> table.to_arrow().sort_by("x").to_pandas()
x doubled
0 1 2
@@ -2225,8 +2225,8 @@ class Table(ABC):
Declared with ``add_columns(computed=...)``, a column starts empty and
gets its values here. Rows appended since the last refresh are filled
by the next one, and rows whose inputs changed since they were computed
are recomputed; everything else is left as it is.
by the next one; rows already filled are left as they are, so the call
is idempotent and does not observe a mutated input.
Local tables only: a remote refresh runs as a server job, through
[`refresh_column_async`][lancedb.table.Table.refresh_column_async].
@@ -4318,14 +4318,13 @@ class LanceTable(Table):
return LOOP.run(self._table.add_columns(transforms, computed=computed))
def refresh_column(self, column: str) -> "RefreshColumnResult":
"""Fill a computed column's unfilled rows and recompute those whose
inputs changed. See
"""Fill a computed column's unfilled rows. See
[`AsyncTable.refresh_column`][lancedb.AsyncTable.refresh_column]."""
return LOOP.run(self._table.refresh_column(column))
def refresh_column_async(self, column: str) -> Job[RefreshColumnJobResult]:
"""Fill a computed column's unfilled rows and recompute those whose
inputs changed, returning a handle to the refresh job. See
"""Fill a computed column's unfilled rows, returning a handle to the
refresh job. See
[`Table.refresh_column_async`][lancedb.table.Table.refresh_column_async].
"""
return Job(LOOP.run(self._table.refresh_column_async(column)))
@@ -6313,10 +6312,10 @@ class AsyncTable:
them from
[`refresh_column`][lancedb.table.AsyncTable.refresh_column].
A refresh also recomputes the rows whose inputs changed since they
were computed, so a mutated input is reflected by the next refresh.
While a declaration reads a column, that column cannot be renamed,
retyped or dropped.
A refresh does not revisit rows it has already filled, so mutating
an input leaves the value computed at fill time. While a
declaration reads a column, that column cannot be renamed, retyped
or dropped.
On LanceDB Cloud and Enterprise the expression is planned by
the server. Cannot be combined with ``transforms``.
@@ -6378,8 +6377,8 @@ class AsyncTable:
Declared with ``add_columns(computed=...)``, a column starts empty and
gets its values here. Rows appended since the last refresh are filled
by the next one, and rows whose inputs changed since they were computed
are recomputed; everything else is left as it is.
by the next one; rows already filled are left as they are, so the call
is idempotent and does not observe a mutated input.
Local tables only: a remote refresh runs as a server job, through
[`refresh_column_async`][lancedb.table.Table.refresh_column_async].
+3 -4
View File
@@ -4183,14 +4183,13 @@ def test_refresh_column_async_returns_job(tmp_path):
assert result.rows_failed == 0
assert result.rows_remaining == 0
assert result.source_version == 2
# The fill lands at 3; the stamp recording its inputs is published at 4.
assert result.published_version == 4
assert result.published_version == 3
assert job.status() == "finished"
assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4]
no_op = table.refresh_column_async("doubled").wait()
assert no_op.rows_assigned == 0
assert no_op.source_version == 4
assert no_op.source_version == 3
assert no_op.published_version is None
# Bad input raises at the call, not through the job.
@@ -4209,6 +4208,6 @@ async def test_refresh_column_async_job_async_table(tmp_path):
assert isinstance(result, lancedb.RefreshColumnResult)
assert result.rows_assigned == 1
assert result.source_version == 2
assert result.published_version == 4
assert result.published_version == 3
assert await job.status() == "finished"
assert (await table.to_arrow())["tripled"].to_pylist() == [9]
+1 -2
View File
@@ -95,14 +95,13 @@ candle-transformers = { version = "0.9.1", optional = true }
candle-nn = { version = "0.9.1", optional = true }
tokenizers = { version = "0.19.1", optional = true }
semver = { workspace = true }
roaring = "0.11.4"
sha2 = "0.10"
[dev-dependencies]
anyhow = "1"
lance-testing = { workspace = true }
tempfile = { workspace = true }
random_word = { version = "0.4.3", features = ["en"] }
roaring = "0.11.4"
tokio = { workspace = true, features = ["io-util", "macros", "net", "test-util"] }
uuid = { workspace = true }
walkdir = "2"
+5 -59
View File
@@ -1124,9 +1124,8 @@ struct RowScope {
/// Whether every commit on the view after `recorded` is a fill of its
/// computed columns: a column rewrite or data replacement touching only
/// those fields and neither adding nor removing rows, or the freshness
/// stamp a fill leaves on them. A version whose transaction cannot be read
/// is not proven, so it counts as drift.
/// those fields and neither adding nor removing rows. A version whose
/// transaction cannot be read is not proven, so it counts as drift.
async fn only_computed_rewrites_since(view_ds: &Dataset, recorded: u64) -> Result<bool> {
// A fill may write any field under a computed column, so the whole
// subtree counts, not only the root.
@@ -1177,19 +1176,6 @@ async fn only_computed_rewrites_since(view_ds: &Dataset, recorded: u64) -> Resul
.all(|field| computed_fields.contains(&(*field as u32)))
})
}
// The stamp `refresh_column` writes after its fill (see
// `table::freshness`): field metadata on computed columns, no data.
Operation::UpdateConfig {
config_updates: None,
table_metadata_updates: None,
schema_metadata_updates: None,
field_metadata_updates,
} => {
!field_metadata_updates.is_empty()
&& field_metadata_updates
.keys()
.all(|field| computed_fields.contains(&(*field as u32)))
}
_ => false,
};
if !fill {
@@ -3373,46 +3359,6 @@ mod tests {
);
}
/// Field metadata on `field` only, the commit shape of the freshness
/// stamp `refresh_column` leaves after its fill.
async fn commit_field_metadata(view: &MaterializedView, field: &str, key: &str) {
let native = view.table().as_native().unwrap();
native.dataset.reload().await.unwrap();
let mut dataset = native.dataset.get().await.unwrap().as_ref().clone();
dataset
.update_field_metadata()
.update(field, [(key.to_string(), "{}".to_string())])
.unwrap()
.await
.unwrap();
}
/// The stamp is metadata on the computed column and rewrites nothing
/// refresh certifies, so it is not drift; the same commit shape on a
/// projected column is, like any other write to it.
#[tokio::test]
async fn test_a_freshness_stamp_is_not_drift() {
let conn = connect("memory://").execute().await.unwrap();
let view = refreshed_computed_view(&conn).await;
commit_field_metadata(
&view,
"emb",
crate::table::computed_columns::SOURCE_SIGNATURE_META_KEY,
)
.await;
assert_eq!(
view.refresh().execute().await.unwrap().mode,
RefreshMode::NoOp
);
commit_field_metadata(&view, "id", "probe").await;
assert_eq!(
view.refresh().execute().await.unwrap().mode,
RefreshMode::Rebuild
);
}
/// The fill job's commit rewrites only computed columns. It is the one
/// commit on a view that is not drift: the next refresh carries on from
/// its watermark instead of rebuilding, which would null what the fill
@@ -3578,9 +3524,9 @@ mod tests {
}
/// A SQL declaration is filled by `refresh_column` on the view, which
/// commits a data replacement and then its freshness stamp; the next
/// refresh continues from its watermark and keeps what the fill wrote,
/// and only rows the view added since come back unfilled.
/// commits a data replacement; the next refresh continues from its
/// watermark and keeps what the fill wrote, and only rows the view added
/// since come back unfilled.
#[tokio::test]
async fn test_a_sql_fill_is_not_drift() {
use crate::materialized_view::tests::{people, sql_field};
+6 -9
View File
@@ -75,7 +75,6 @@ mod create_index;
pub mod datafusion;
pub(crate) mod dataset;
pub mod delete;
pub mod freshness;
pub mod lsm_stats;
pub mod merge;
pub mod optimize;
@@ -779,8 +778,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
message: "Function columns are supported only on LanceDB Cloud and Enterprise".into(),
})
}
/// Fill a computed column's unfilled rows and recompute those whose
/// inputs changed.
/// Fill a computed column's unfilled rows.
///
/// The default returns `NotSupported`; Lance-backed tables override it.
async fn refresh_column(&self, _column: &str) -> Result<RefreshColumnResult> {
@@ -788,8 +786,8 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
message: "computed columns are supported only on local tables".into(),
})
}
/// Fill a computed column's unfilled rows and recompute those whose
/// inputs changed, returning a [`Job`] tracking the operation.
/// Fill a computed column's unfilled rows, returning a [`Job`] tracking
/// the operation.
async fn refresh_column_async(
&self,
_column: &str,
@@ -1751,10 +1749,9 @@ impl Table {
/// Declared with
/// [`AddColumnsBuilder::computed`](add_columns::AddColumnsBuilder::computed),
/// a column starts empty and gets its values here. Fragments appended
/// since the last refresh are filled by the next one, and fragments whose
/// inputs changed since they were computed are recomputed (see
/// [`freshness`](crate::table::freshness)); everything else is left as
/// it is.
/// since the last refresh are filled by the next one; fragments already
/// filled are left as they are, so the call is idempotent and does not
/// observe a mutated input.
///
/// Local tables only: a remote refresh runs as a server job, through
/// [`Table::refresh_column_async`].
+4 -5
View File
@@ -60,11 +60,10 @@ impl AddColumnsBuilder {
/// every fragment that has none -- including fragments appended since the
/// last refresh.
///
/// A refresh also recomputes the rows of a fragment whose inputs changed
/// since it was computed (see [`freshness`](super::freshness)), so a
/// mutated input is reflected by the next refresh. An input cannot be
/// renamed, retyped or dropped while a declaration reads it, since the
/// expression names it.
/// Refresh does not revisit a fragment it has filled, so mutating an input
/// leaves the value computed at fill time; recomputing means dropping the
/// column and declaring it again. An input cannot be renamed, retyped or
/// dropped while a declaration reads it, since the expression names it.
///
/// On LanceDB Cloud and Enterprise the expression is planned by the
/// server, and the refresh runs as a server job -- see
+2 -23
View File
@@ -71,22 +71,6 @@ pub const FUNCTION_BINDINGS_META_KEY: &str = "lancedb::function_bindings";
/// Version of the schema-level Function binding envelope.
pub const FUNCTION_BINDINGS_VERSION: u32 = 1;
/// Field metadata key holding `{fragment id -> input signature}` as JSON,
/// recorded by the refresh that last computed each fragment. Outside the
/// declaration namespace on purpose: a declaration is immutable through
/// metadata edits, this is rewritten by every refresh. Seeded empty at
/// declaration, so a column is tracked from birth; a column without it was
/// declared before signatures existed.
pub const SOURCE_SIGNATURE_META_KEY: &str = "computed_refresh.source_signature";
/// Field metadata key holding the definition digest a column was last
/// computed under. A change to it makes every row stale.
pub const DEFINITION_VERSION_META_KEY: &str = "computed_refresh.definition_version";
/// Field metadata key holding the table version the signature map describes:
/// where a refresh starts following compactions to carry freshness forward.
pub const RECORDED_AT_VERSION_META_KEY: &str = "computed_refresh.recorded_at_version";
/// Value of [`KIND_META_KEY`] for a column defined by a SQL expression.
pub const SQL_KIND: &str = "sql";
@@ -155,7 +139,6 @@ fn computed_column_metadata(expression: &str, inputs: &[String]) -> HashMap<Stri
INPUTS_META_KEY.to_string(),
serde_json::to_string(inputs).unwrap_or_else(|_| "[]".to_string()),
),
(SOURCE_SIGNATURE_META_KEY.to_string(), "{}".to_string()),
])
}
@@ -180,7 +163,6 @@ pub fn function_computed_column_metadata(
INPUTS_META_KEY.to_string(),
serde_json::to_string(inputs).unwrap_or_else(|_| "[]".to_string()),
),
(SOURCE_SIGNATURE_META_KEY.to_string(), "{}".to_string()),
])
}
@@ -1313,8 +1295,7 @@ pub(crate) fn ensure_not_an_input(schema: &SchemaRef, paths: &[&str]) -> Result<
}
/// Reject a write that supplies values for a computed column directly:
/// only refresh materializes one, and only refresh decides what it
/// recomputes.
/// only refresh materializes one, and refresh never revisits a filled row.
pub(crate) fn ensure_not_written<'a>(
schema: &ArrowSchema,
written: impl IntoIterator<Item = &'a str>,
@@ -1489,9 +1470,7 @@ fn ensure_no_foreign_declaration(field: &ArrowField) -> Result<()> {
/// kind, the expression, the inputs -- would bypass that validation or move
/// a binding out from under a refresh. Drop the column and declare it again.
pub(crate) fn is_declaration_key(key: &str) -> bool {
key == COMPUTED_COLUMN_META_KEY
|| key.starts_with("computed_column.")
|| key.starts_with("computed_refresh.")
key == COMPUTED_COLUMN_META_KEY || key.starts_with("computed_column.")
}
/// Reject retyping a computed column itself.
File diff suppressed because it is too large Load Diff
+2 -10
View File
@@ -134,17 +134,9 @@ pub(crate) async fn cleanup_old_versions(
) -> Result<RemovalStats> {
table.dataset.ensure_mutable()?;
let dataset = table.dataset.get().await?;
let stats = dataset
Ok(dataset
.cleanup_old_versions(older_than, delete_unverified, error_if_tagged_old_versions)
.await?;
// Computed-column signature sidecars live outside lance's directories;
// drop the ones the surviving versions no longer reference.
let removed =
super::freshness::prune_sidecars(&dataset, delete_unverified.unwrap_or(false)).await?;
if removed > 0 {
log::debug!("removed {removed} unreferenced computed-column signature sidecars");
}
Ok(stats)
.await?)
}
/// Compact files in the dataset.
+54 -370
View File
@@ -3,9 +3,9 @@
//! Filling computed columns.
//!
//! A row without a value gets one; a row that has one keeps it unless its
//! fragment's inputs moved since it was computed, which `freshness` decides
//! from the manifest and stamps after every fill.
//! A row without a value gets one; a row that has one keeps it. Refresh is
//! therefore idempotent and does not observe input mutation -- once a row is
//! filled, changing what the expression reads leaves the stored result alone.
//!
//! A column's computed inputs are filled first -- the dependency graph is
//! walked once, each reachable column filled once in dependency order, each
@@ -31,7 +31,6 @@
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use arrow_array::{
Array, ArrayRef, BooleanArray, LargeBinaryArray, RecordBatch, RecordBatchOptions, StructArray,
@@ -49,7 +48,6 @@ use lance_core::datatypes::{BlobHandling, Schema as LanceSchema};
use serde::{Deserialize, Serialize};
use super::computed_columns::{BoundExpression, ComputedColumnKind, computed_column_from_field};
use super::freshness::{self, SignatureMap, StalenessPlan};
use super::{BaseTable, NativeTable};
use crate::job::Job;
use crate::{Error, Result};
@@ -112,81 +110,28 @@ async fn execute_refresh_column_with_source(
};
let output_is_blob = field.is_blob_v2();
// Which fragments the null filter cannot speak for: their inputs moved
// since they were computed, or the definition did. Decided once, from
// the manifest the values are read from.
let inputs = freshness::fields_for_paths(dataset.schema(), &bound.inputs)?;
let definition = freshness::definition_version(&expression);
let staleness = freshness::staleness_against(&dataset, column, &definition, &inputs).await?;
let mut rows_filled = 0u64;
let mut replacements = Vec::new();
// Fragments this refresh computed in full, signed at the version read.
let mut computed = SignatureMap::new();
for fragment in dataset.get_fragments() {
let fragment_id = u32::try_from(fragment.id()).map_err(|_| Error::Runtime {
message: format!("fragment id {} does not fit a signature map", fragment.id()),
})?;
// A recompute rewrites every live row, so it is staged without the
// probe and counted as it fills; a null fill probes first, since a
// fragment with nothing to gain is not worth a write.
let recompute = staleness.is_dirty(fragment_id);
let whole = recompute || {
let (gained, unfilled) =
count_fragment_gains(&dataset, &fragment, &bound, column).await?;
if gained == 0 {
continue;
}
rows_filled += gained;
unfilled == u64::try_from(fragment.count_rows(None).await?).unwrap_or(u64::MAX)
};
if whole {
computed.insert(
fragment_id,
freshness::fragment_input_signature(fragment.metadata(), &inputs)?,
);
let gained = count_fragment_gains(&dataset, &fragment, &bound, column).await?;
if gained == 0 {
continue;
}
let gained = Arc::new(AtomicU64::new(0));
let values = fill_stream(
&dataset,
&fragment,
bound.clone(),
column,
output_is_blob,
recompute,
gained.clone(),
)
.await?;
rows_filled += gained;
let values =
fill_stream(&dataset, &fragment, bound.clone(), column, output_is_blob).await?;
replacements.push(fragment.write_columns(values, &column_schema).await?);
if recompute {
rows_filled += gained.load(Ordering::Relaxed);
}
}
let source_version = dataset.version().version;
if replacements.is_empty() {
// Nothing to fill; the stamp may still have something to record -- a
// column not yet enrolled, or fragments a compaction carried.
let mut latest = (*dataset).clone();
let stamped = record(
&mut latest,
(&dataset, &staleness),
column,
&definition,
&inputs,
computed,
)
.await;
if stamped.is_some() {
table.dataset.update(latest);
}
return Ok(RefreshExecution {
result: RefreshColumnResult {
rows_filled: 0,
version: stamped.unwrap_or(source_version),
version: source_version,
},
source_version,
published_version: stamped,
published_version: None,
});
}
@@ -204,17 +149,7 @@ async fn execute_refresh_column_with_source(
)
.await?;
let mut new_dataset = new_dataset;
let version = record(
&mut new_dataset,
(&dataset, &staleness),
column,
&definition,
&inputs,
computed,
)
.await
.unwrap_or(new_dataset.version().version);
let version = new_dataset.version().version;
table.dataset.update(new_dataset);
Ok(RefreshExecution {
result: RefreshColumnResult {
@@ -226,31 +161,6 @@ async fn execute_refresh_column_with_source(
})
}
/// Stamp the input state the refresh computed from (see
/// [`freshness::record_freshness`]); the version the stamp landed at, which
/// is the last one the refresh wrote. Never fails the refresh: the values
/// are committed, and a missing stamp only costs a recompute next time.
async fn record(
latest: &mut Dataset,
pinned: (&Dataset, &StalenessPlan),
column: &str,
definition: &str,
inputs: &freshness::InputFields,
computed: SignatureMap,
) -> Option<u64> {
match freshness::record_freshness(latest, Some(pinned), column, definition, inputs, computed)
.await
{
Ok(record) => record.version,
Err(error) => {
log::warn!(
"could not record the input state computed column '{column}' was refreshed from ({error}); its fragments will recompute on the next refresh"
);
None
}
}
}
/// Refuse while a computed input still has rows a refresh of it would fill:
/// read now, its placeholder null would be evaluated as a value and kept.
async fn ensure_inputs_filled(
@@ -278,9 +188,7 @@ async fn ensure_inputs_filled(
let input_bound = super::computed_columns::bind(schema.clone(), input, expression)?;
let mut unfilled = 0u64;
for fragment in dataset.get_fragments() {
unfilled += count_fragment_gains(dataset, &fragment, &input_bound, input)
.await?
.0;
unfilled += count_fragment_gains(dataset, &fragment, &input_bound, input).await?;
}
if unfilled > 0 {
return Err(Error::InvalidInput {
@@ -528,35 +436,32 @@ fn blob_array_from_binary(
/// Scans only the unfilled live rows -- deleted rows never reach the
/// expression here, the filter having already excluded them -- and counts the
/// non-null results. Exact, so it is both the staging decision and the
/// fragment's contribution to `rows_filled`. Returns the gains and the rows
/// scanned.
/// fragment's contribution to `rows_filled`.
async fn count_fragment_gains(
dataset: &Dataset,
fragment: &FileFragment,
bound: &BoundExpression,
column: &str,
) -> Result<(u64, u64)> {
) -> Result<u64> {
let mut scanner = dataset.scan();
scanner
.with_fragments(vec![fragment.metadata().clone()])
.with_row_id()
.project(&bound.roots)?
.filter(&format!("{} IS NULL", quote_identifier(column)))?;
.filter(&format!("{} IS NULL", quote_identifier(column)))?
.project(&bound.roots)?;
configure_blob_inputs(&mut scanner, dataset.schema(), bound, None)?;
let mut gained = 0u64;
let mut considered = 0u64;
let mut batches = scanner.try_into_stream().await?;
while let Some(batch) = batches.try_next().await? {
let evaluated = evaluate(bound, &evaluation_batch(&batch, bound, None)?)?;
gained += (batch.num_rows() - evaluated.null_count()) as u64;
considered += batch.num_rows() as u64;
}
Ok((gained, considered))
Ok(gained)
}
/// Stream one fragment's column in physical order, filling the unfilled live
/// rows -- every live row, for a recompute -- and keeping every other value.
/// rows and keeping every other value.
///
/// Deleted rows are carried through so the values line up positionally with
/// the fragment's data files; they are never read back, but the column file
@@ -567,8 +472,6 @@ async fn fill_stream(
bound: Arc<BoundExpression>,
column: &str,
output_is_blob: bool,
recompute: bool,
gained: Arc<AtomicU64>,
) -> Result<impl Stream<Item = lance_core::Result<RecordBatch>> + Send + use<>> {
let mut projection: Vec<String> = bound.roots.clone();
projection.push(column.to_string());
@@ -618,20 +521,14 @@ async fn fill_stream(
.column_by_name(ROW_ID)
.ok_or_else(|| missing(ROW_ID))?;
// Only an unfilled live row gains a value, or every live row under a
// recompute; a deleted row has a null row id and keeps its (null) slot.
// Only an unfilled live row gains a value; a deleted row has a null
// row id and keeps its (null) slot.
let unfilled = arrow::compute::is_null(existing.as_ref())?;
let live = arrow::compute::is_not_null(row_ids.as_ref())?;
let fill = if recompute {
live
} else {
let unfilled = arrow::compute::is_null(existing.as_ref())?;
arrow::compute::and(&unfilled, &live)?
};
let fill = arrow::compute::and(&unfilled, &live)?;
let keep = arrow::compute::not(&fill)?;
let computed = evaluate(&bound, &evaluation_batch(&batch, &bound, Some(&keep))?)?;
let values = arrow::compute::and(&fill, &arrow::compute::is_not_null(&computed)?)?;
gained.fetch_add(values.true_count() as u64, Ordering::Relaxed);
let merged = arrow_select::zip::zip(&fill, &computed, existing)?;
let merged = if output_is_blob {
blob_array_from_binary(&merged, projected.field(0))?
@@ -840,8 +737,7 @@ mod tests {
.await
.unwrap();
assert_eq!(no_op.rows_assigned, 0);
// The fill, then the stamp recording what it computed from.
assert_eq!(no_op.source_version, 4);
assert_eq!(no_op.source_version, 3);
assert_eq!(no_op.published_version, None);
}
@@ -881,8 +777,8 @@ mod tests {
}
/// A row is filled only by gaining a value, so an expression yielding null
/// settles at once instead of re-selecting the same rows forever: the
/// second refresh finds the fragment signed and moves nothing.
/// settles at once instead of re-selecting the same rows forever. Nothing
/// is staged, so the version does not move either.
#[tokio::test]
async fn test_refresh_converges_on_a_null_result() {
let table = table_with("refresh_null_result", vec![1, 2, 3]).await;
@@ -896,186 +792,28 @@ mod tests {
let first = table.refresh_column("maybe").await.unwrap();
assert_eq!(first.rows_filled, 0);
assert!(first.version > declared);
assert_eq!(first.version, declared);
assert_eq!(read(&table, "maybe").await, vec![None, None, None]);
let again = table.refresh_column("maybe").await.unwrap();
assert_eq!(again.rows_filled, 0);
assert_eq!(again.version, first.version);
assert_eq!(again.version, declared);
}
/// A filled row whose input moved is recomputed: the update rewrites
/// the row into a fragment the stamp never signed, and only that one.
/// The contract's boundary: a filled fragment is not revisited, so
/// mutating an input leaves the value computed at fill time.
#[tokio::test]
async fn test_refresh_recomputes_a_row_whose_input_moved() {
let table = table_with("refresh_mutation", vec![1, 2]).await;
declare_doubled(&table).await.unwrap();
append(&table, vec![5]).await;
table.refresh_column("doubled").await.unwrap();
assert_eq!(
read(&table, "doubled").await,
vec![Some(2), Some(4), Some(10)]
);
table
.update()
.column("x", "7")
.only_if("x = 5")
.execute()
.await
.unwrap();
let again = table.refresh_column("doubled").await.unwrap();
assert_eq!(again.rows_filled, 1);
assert_eq!(
read(&table, "doubled").await,
vec![Some(2), Some(4), Some(14)]
);
let settled = table.refresh_column("doubled").await.unwrap();
assert_eq!(settled.rows_filled, 0);
assert_eq!(settled.version, again.version);
}
/// Each stamp is a sidecar under `_computed/`; pruning old versions
/// removes the sidecars only they referenced, and keeps the current one.
#[tokio::test]
async fn test_pruning_drops_the_sidecars_of_pruned_versions() {
let dir = tempfile::tempdir().unwrap();
let conn = connect(dir.path().to_str().unwrap())
.execute()
.await
.unwrap();
let batch = record_batch!(("x", Int32, [1, 2])).unwrap();
let table = conn
.create_table("sidecars", batch)
.execute()
.await
.unwrap();
async fn test_refresh_does_not_observe_input_mutation() {
let table = table_with("refresh_mutation", vec![1]).await;
declare_doubled(&table).await.unwrap();
table.refresh_column("doubled").await.unwrap();
append(&table, vec![5]).await;
table.refresh_column("doubled").await.unwrap();
let sidecars = || {
std::fs::read_dir(dir.path().join("sidecars.lance").join("_computed"))
.unwrap()
.count()
};
assert_eq!(sidecars(), 2);
assert_eq!(read(&table, "doubled").await, vec![Some(2)]);
table
.optimize(crate::table::OptimizeAction::Prune {
older_than: Some(chrono::Duration::zero()),
delete_unverified: Some(true),
error_if_tagged_old_versions: None,
})
.await
.unwrap();
assert_eq!(sidecars(), 1);
assert_eq!(
table.refresh_column("doubled").await.unwrap().rows_filled,
0
);
}
/// A deleted row is never computed and the rows that stay keep their
/// values: a delete recomputes nothing and stamps nothing.
#[tokio::test]
async fn test_a_delete_recomputes_nothing() {
let table = table_with("refresh_delete", vec![1, 2, 3]).await;
declare_doubled(&table).await.unwrap();
let filled = table.refresh_column("doubled").await.unwrap();
table.delete("x = 2").await.unwrap();
let deleted = table.version().await.unwrap();
table.update().column("x", "3").execute().await.unwrap();
let again = table.refresh_column("doubled").await.unwrap();
assert_eq!(again.rows_filled, 0);
assert_eq!(again.version, deleted);
assert!(deleted > filled.version);
assert_eq!(read(&table, "doubled").await, vec![Some(2), Some(6)]);
}
/// Compaction copies inputs unchanged, so a fragment it builds from
/// signed ones is fresh: the refresh recomputes nothing and only records
/// the new fragment.
#[tokio::test]
async fn test_a_compaction_of_signed_fragments_recomputes_nothing() {
let table = table_with("refresh_compact_signed", vec![1, 2]).await;
declare_doubled(&table).await.unwrap();
append(&table, vec![5]).await;
table.refresh_column("doubled").await.unwrap();
table
.optimize(crate::table::OptimizeAction::Compact {
options: crate::table::CompactionOptions::default(),
remap_options: None,
})
.await
.unwrap();
let compacted = table.version().await.unwrap();
let carried = table.refresh_column("doubled").await.unwrap();
assert_eq!(carried.rows_filled, 0);
assert_eq!(carried.version, compacted + 1);
let settled = table.refresh_column("doubled").await.unwrap();
assert_eq!(settled.version, carried.version);
assert_eq!(
read(&table, "doubled").await,
vec![Some(2), Some(4), Some(10)]
);
}
/// A column declared before signatures existed has no map. Its first
/// refresh keeps the null-fill contract and enrolls what it read from;
/// from then on a moved input is recomputed like any other.
#[tokio::test]
async fn test_an_unsigned_column_is_enrolled_by_its_first_refresh() {
let table = table_with("refresh_legacy", vec![1, 2]).await;
declare_doubled(&table).await.unwrap();
table.refresh_column("doubled").await.unwrap();
table
.update()
.column("x", "3")
.only_if("x = 1")
.execute()
.await
.unwrap();
let native = table.as_native().unwrap();
let mut dataset = (*native.dataset.get().await.unwrap()).clone();
let declaration = dataset
.schema()
.field("doubled")
.unwrap()
.metadata
.iter()
.filter(|(key, _)| !key.starts_with("computed_refresh."))
.map(|(key, value)| (key.clone(), value.clone()))
.collect::<Vec<_>>();
dataset
.update_field_metadata()
.replace("doubled", declaration)
.unwrap()
.await
.unwrap();
table.checkout_latest().await.unwrap();
// Null-fill only: the moved row keeps the value it was filled with.
let enrolled = table.refresh_column("doubled").await.unwrap();
assert_eq!(enrolled.rows_filled, 0);
assert_eq!(read(&table, "doubled").await, vec![Some(2), Some(4)]);
table
.update()
.column("x", "5")
.only_if("x = 3")
.execute()
.await
.unwrap();
let again = table.refresh_column("doubled").await.unwrap();
assert_eq!(again.rows_filled, 1);
assert_eq!(read(&table, "doubled").await, vec![Some(4), Some(10)]);
assert_eq!(read(&table, "doubled").await, vec![Some(2)]);
}
/// A row rewrite before the first refresh materializes the declared
@@ -1093,11 +831,11 @@ mod tests {
assert_eq!(read(&table, "doubled").await, vec![Some(6)]);
}
/// A fragment compacted out of one the stamp never signed cannot vouch
/// for any of its rows: every live row is recomputed, the moved one
/// included.
/// The contract holds row by row, not fragment by fragment: revisiting a
/// fragment to fill one row must not recompute a filled row sitting beside
/// it, even where the input behind it has since changed.
#[tokio::test]
async fn test_a_compaction_of_an_unsigned_fragment_recomputes_it() {
async fn test_refresh_does_not_recompute_a_filled_row_beside_an_unfilled_one() {
let table = table_with("refresh_mixed", vec![1, 2]).await;
declare_doubled(&table).await.unwrap();
table.refresh_column("doubled").await.unwrap();
@@ -1119,70 +857,18 @@ mod tests {
.unwrap();
let result = table.refresh_column("doubled").await.unwrap();
assert_eq!(result.rows_filled, 3);
assert_eq!(
read(&table, "doubled").await,
vec![Some(4), Some(10), Some(200)]
);
}
/// The gate's reproducer: a raw lance append may carry a value for the
/// computed column. Compaction cannot certify it, so the product is
/// recomputed and the supplied value replaced.
#[tokio::test]
async fn test_raw_append_values_are_not_trusted_after_compaction() {
use arrow_array::RecordBatchIterator;
use lance::Dataset;
use lance::dataset::{WriteMode, WriteParams};
let dir = tempfile::tempdir().unwrap();
let conn = connect(dir.path().to_str().unwrap())
.execute()
.await
.unwrap();
let batch = record_batch!(("x", Int32, [1, 2])).unwrap();
let table = conn
.create_table("raw_append", batch)
.execute()
.await
.unwrap();
declare_doubled(&table).await.unwrap();
table.refresh_column("doubled").await.unwrap();
let batch = record_batch!(("x", Int32, [5]), ("doubled", Int32, [Some(999_i32)])).unwrap();
let schema = batch.schema();
let uri = table.uri().await.unwrap();
Dataset::write(
RecordBatchIterator::new(vec![Ok(batch)], schema),
&uri,
Some(WriteParams {
mode: WriteMode::Append,
..Default::default()
}),
)
.await
.unwrap();
table.checkout_latest().await.unwrap();
table
.optimize(crate::table::OptimizeAction::Compact {
options: crate::table::CompactionOptions::default(),
remap_options: None,
})
.await
.unwrap();
let result = table.refresh_column("doubled").await.unwrap();
assert_eq!(result.rows_filled, 3);
assert_eq!(result.rows_filled, 1);
// 2 is the mutated row keeping the value it was filled with, not 200.
assert_eq!(
read(&table, "doubled").await,
vec![Some(2), Some(4), Some(10)]
);
}
/// An appended fragment holds no values, so compacting it into a signed
/// one leaves the product fresh: only the appended rows are filled.
/// Filling a fragment must not disturb the values it already holds, which
/// is what makes a compaction-mixed fragment safe to revisit.
#[tokio::test]
async fn test_a_compaction_with_an_appended_fragment_fills_only_its_rows() {
async fn test_refresh_preserves_already_filled_rows() {
let table = table_with("refresh_preserves", vec![1, 2]).await;
declare_doubled(&table).await.unwrap();
table.refresh_column("doubled").await.unwrap();
@@ -1309,8 +995,7 @@ mod tests {
assert_eq!(result.rows_failed, 0);
assert_eq!(result.rows_remaining, 0);
assert_eq!(result.source_version, 2);
// The fill lands at 3; the stamp recording its inputs is published at 4.
assert_eq!(result.published_version, Some(4));
assert_eq!(result.published_version, Some(3));
assert_eq!(job.status().await.unwrap(), "finished");
assert_eq!(
read(&table, "doubled").await,
@@ -1374,32 +1059,31 @@ mod tests {
assert_eq!(read(&table, "quotient").await, vec![Some(10)]);
}
/// A filled row whose input moved is re-evaluated, and a row whose
/// input did not move is not: the untouched fragment is never read, so
/// its poison input is never reached.
/// The gate's reproducer: an already-filled row's value must not be
/// re-evaluated either -- its input may have mutated into one the
/// expression chokes on.
#[tokio::test]
async fn test_only_a_moved_rows_value_is_re_evaluated() {
let table = table_with("refresh_filled_poison", vec![1, 0]).await;
async fn test_a_filled_rows_value_is_never_evaluated() {
let table = table_with("refresh_filled_poison", vec![1, 2]).await;
table
.add_columns()
.computed("quotient", "10 / coalesce(nullif(x, 0), 1)")
.computed("quotient", "10 / x")
.execute()
.await
.unwrap();
table.refresh_column("quotient").await.unwrap();
assert_eq!(read(&table, "quotient").await, vec![Some(10), Some(10)]);
append(&table, vec![5]).await;
table
.update()
.column("x", "2")
.column("x", "0")
.only_if("x = 1")
.execute()
.await
.unwrap();
append(&table, vec![5]).await;
let result = table.refresh_column("quotient").await.unwrap();
assert_eq!(result.rows_filled, 2);
assert_eq!(result.rows_filled, 1);
assert_eq!(
read(&table, "quotient").await,
vec![Some(2), Some(5), Some(10)]