mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-13 23:32:27 +00:00
feat(node): accept ArrayBuffer, Blob/File, and URL as blob column inputs (#4313)
Closes #4270. ## Summary Blob v2 columns now accept more of the containers TypeScript code commonly has, both as the column value and inside `{ data }` / `{ uri }`: | Input | Handling | |---|---| | `ArrayBuffer` | `new Uint8Array(value)`, no copy | | `URL` | stored as `value.href` | | `Blob` / `File` | `await value.arrayBuffer()` in the async write paths: `createTable`, `add`, `mergeInsert` | `makeArrowTable` is synchronous, so it can't read a `Blob`. Given one, it throws an error that names the field and row and points to the async APIs or `new Uint8Array(await value.arrayBuffer())`. It does not write a null blob. ### Types `BlobInput` is now the accepted input type. It is widened to the types above and exported from the package entry point with `BlobData` and `BlobUri`, so callers can annotate rows and `tsc` catches bad inputs. The struct forms are a union that requires exactly one of `data` / `uri`. The old normalized `{ data, uri }` shape is renamed to `BlobValue` (internal). ### Implementation - `blob.ts`: `normalizeBlobInput` (sync: `ArrayBuffer`/`URL`), `resolveBlobInput` (async: `Blob`/`File`). `coerceBlobValue` normalizes first. - `arrow.ts`: `mapBlobInputs` walks only the blob positions in the schema, including blobs nested in structs and lists. It copies records and arrays only where a value changed, so caller data is never mutated. `resolveBlobInputs` runs before `makeArrowTable` in `convertToTable` (used by `add` / `mergeInsert`) and in `createTable`. Each distinct `Blob` is read once. - `makeArrowTable` now coerces every blob value to the full `{ data, uri }` struct **before** schema inference. This also fixes two bugs that existed before this change: - Rows mixing plain bytes with `{ data }` failed inference. - A `{ uri }`-only struct produced a blob struct missing its `data` child. Blob validation errors now include the field path and row. - Also fixed along the way: - `add` / `mergeInsert` on a table with an embedding function rejected most blob values, because `ensureNestedFieldsExist` filled blob columns in like ordinary structs. It now skips them. - With a schema, a struct field that is `null` in one row and a record in a later row failed inference. Primitives and arrays given for a struct field are now rejected instead of being written as an all-null struct. - Blob structs with keys other than `data` / `uri` are rejected, which catches typos. Class instances with `data` / `uri` are handled like plain objects. ## Testing - Unit tests in `__test__/blob.test.ts`: coercion of the new types, rejection and error messages for `Blob` in the sync path, nested lists and structs, `resolveBlobInputs` (only blob columns touched, caller data untouched, shared `Blob` read once). - Table tests in `__test__/table.test.ts`: - `createTable`, `add`, and `mergeInsert` round-trip `ArrayBuffer` / `Blob` / `File` bytes through `fetchBlobs`. - `URL` and `{ uri: URL }` are added to the external-blob round-trip test from #4317 (`allowExternalBlobOutsideBases`). - Full `jest` suite passes locally after the rebase (938 passed). `pnpm lint` is clean and `pnpm run docs` has been regenerated. Streaming inputs (`ReadableStream`, Node `Readable`, async iterables) remain out of scope, as the issue says. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_012U4KdHofT5edv3LcXHXbLv --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -157,9 +157,12 @@
|
|||||||
|
|
||||||
- [AnalyzePlanDistributedMetrics](type-aliases/AnalyzePlanDistributedMetrics.md)
|
- [AnalyzePlanDistributedMetrics](type-aliases/AnalyzePlanDistributedMetrics.md)
|
||||||
- [BaseTokenizer](type-aliases/BaseTokenizer.md)
|
- [BaseTokenizer](type-aliases/BaseTokenizer.md)
|
||||||
|
- [BlobData](type-aliases/BlobData.md)
|
||||||
|
- [BlobInput](type-aliases/BlobInput.md)
|
||||||
- [BlobOptions](type-aliases/BlobOptions.md)
|
- [BlobOptions](type-aliases/BlobOptions.md)
|
||||||
- [BlobRange](type-aliases/BlobRange.md)
|
- [BlobRange](type-aliases/BlobRange.md)
|
||||||
- [BlobReadOptions](type-aliases/BlobReadOptions.md)
|
- [BlobReadOptions](type-aliases/BlobReadOptions.md)
|
||||||
|
- [BlobUri](type-aliases/BlobUri.md)
|
||||||
- [Data](type-aliases/Data.md)
|
- [Data](type-aliases/Data.md)
|
||||||
- [DataLike](type-aliases/DataLike.md)
|
- [DataLike](type-aliases/DataLike.md)
|
||||||
- [DataTypeLike](type-aliases/DataTypeLike.md)
|
- [DataTypeLike](type-aliases/DataTypeLike.md)
|
||||||
|
|||||||
@@ -0,0 +1,16 @@
|
|||||||
|
[**@lancedb/lancedb**](../README.md) • **Docs**
|
||||||
|
|
||||||
|
***
|
||||||
|
|
||||||
|
[@lancedb/lancedb](../globals.md) / BlobData
|
||||||
|
|
||||||
|
# Type Alias: BlobData
|
||||||
|
|
||||||
|
```ts
|
||||||
|
type BlobData: Buffer | Uint8Array | ArrayBuffer | Blob;
|
||||||
|
```
|
||||||
|
|
||||||
|
Bytes accepted for a blob value. `ArrayBuffer` is wrapped without copying.
|
||||||
|
`Blob` and `File` are read with `arrayBuffer()`, which only the async write
|
||||||
|
paths ([Connection.createTable](../classes/Connection.md#createtable), [Table.add](../classes/Table.md#add),
|
||||||
|
[Table.mergeInsert](../classes/Table.md#mergeinsert)) can do.
|
||||||
@@ -0,0 +1,36 @@
|
|||||||
|
[**@lancedb/lancedb**](../README.md) • **Docs**
|
||||||
|
|
||||||
|
***
|
||||||
|
|
||||||
|
[@lancedb/lancedb](../globals.md) / BlobInput
|
||||||
|
|
||||||
|
# Type Alias: BlobInput
|
||||||
|
|
||||||
|
```ts
|
||||||
|
type BlobInput:
|
||||||
|
| BlobData
|
||||||
|
| BlobUri
|
||||||
|
| object
|
||||||
|
| object
|
||||||
|
| null
|
||||||
|
| undefined;
|
||||||
|
```
|
||||||
|
|
||||||
|
A value accepted for a `lance.blob.v2` column (see [blob](../functions/blob.md)).
|
||||||
|
|
||||||
|
Either inline bytes, a URI pointing at external bytes, or a struct that sets
|
||||||
|
exactly one of `data` or `uri`. `null` and `undefined` write a null blob.
|
||||||
|
|
||||||
|
## Example
|
||||||
|
|
||||||
|
```ts
|
||||||
|
import { pathToFileURL } from "node:url";
|
||||||
|
import type { BlobInput } from "@lancedb/lancedb";
|
||||||
|
|
||||||
|
const rows: { id: bigint; image: BlobInput }[] = [
|
||||||
|
{ id: 1n, image: await (await fetch(url)).arrayBuffer() },
|
||||||
|
{ id: 2n, image: pathToFileURL("/data/cat.png") },
|
||||||
|
{ id: 3n, image: { data: new Blob(["hello"]) } },
|
||||||
|
];
|
||||||
|
await table.add(rows);
|
||||||
|
```
|
||||||
@@ -0,0 +1,13 @@
|
|||||||
|
[**@lancedb/lancedb**](../README.md) • **Docs**
|
||||||
|
|
||||||
|
***
|
||||||
|
|
||||||
|
[@lancedb/lancedb](../globals.md) / BlobUri
|
||||||
|
|
||||||
|
# Type Alias: BlobUri
|
||||||
|
|
||||||
|
```ts
|
||||||
|
type BlobUri: string | URL;
|
||||||
|
```
|
||||||
|
|
||||||
|
A URI accepted for a blob value. A `URL` is stored as its `href`.
|
||||||
@@ -790,6 +790,43 @@ describe.each(arrowVersions)("Arrow %#", (arrow: ApacheArrow) => {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("will allow null values around struct values with a schema", function () {
|
||||||
|
const schema = new Schema([
|
||||||
|
new Field("value", new Struct([new Field("nested", new Int32())])),
|
||||||
|
]);
|
||||||
|
for (const { records, nullIndex } of [
|
||||||
|
{
|
||||||
|
records: [{ value: null }, { value: { nested: 2 } }],
|
||||||
|
nullIndex: 0,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
records: [{ value: { nested: 1 } }, { value: null }],
|
||||||
|
nullIndex: 1,
|
||||||
|
},
|
||||||
|
]) {
|
||||||
|
const table = makeArrowTable(records, { schema });
|
||||||
|
const values = table.getChild("value");
|
||||||
|
|
||||||
|
expect(values?.nullCount).toBe(1);
|
||||||
|
expect(values?.get(nullIndex)).toBeNull();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it("will reject non-object values for struct fields in a schema", function () {
|
||||||
|
const schema = new Schema([
|
||||||
|
new Field("value", new Struct([new Field("nested", new Int32())])),
|
||||||
|
]);
|
||||||
|
for (const records of [
|
||||||
|
[{ value: 5 }],
|
||||||
|
[{ value: 5 }, { value: { nested: 1 } }],
|
||||||
|
[{ value: { nested: 1 } }, { value: "oops" }],
|
||||||
|
]) {
|
||||||
|
expect(() => makeArrowTable(records, { schema })).toThrow(
|
||||||
|
/Expected a struct value for field value at row \d, got/,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
it("will allow a schema to be provided", async function () {
|
it("will allow a schema to be provided", async function () {
|
||||||
await checkTableCreation(
|
await checkTableCreation(
|
||||||
async (records, _, schema) =>
|
async (records, _, schema) =>
|
||||||
|
|||||||
+316
-13
@@ -1,9 +1,25 @@
|
|||||||
// SPDX-License-Identifier: Apache-2.0
|
// SPDX-License-Identifier: Apache-2.0
|
||||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||||
|
|
||||||
import { Field, Int64, List, Schema, Struct, Utf8 } from "apache-arrow";
|
import {
|
||||||
import { makeArrowTable } from "../lancedb/arrow";
|
Field,
|
||||||
|
Float,
|
||||||
|
Float32,
|
||||||
|
Int32,
|
||||||
|
Int64,
|
||||||
|
List,
|
||||||
|
Schema,
|
||||||
|
Struct,
|
||||||
|
Utf8,
|
||||||
|
} from "apache-arrow";
|
||||||
|
import {
|
||||||
|
convertToTable,
|
||||||
|
makeArrowTable,
|
||||||
|
resolveBlobInputs,
|
||||||
|
} from "../lancedb/arrow";
|
||||||
import { BlobFile, blob, coerceBlobValue, isBlobField } from "../lancedb/blob";
|
import { BlobFile, blob, coerceBlobValue, isBlobField } from "../lancedb/blob";
|
||||||
|
import { EmbeddingFunction, LanceSchema } from "../lancedb/embedding";
|
||||||
|
import { getRegistry } from "../lancedb/embedding/registry";
|
||||||
|
|
||||||
describe("blob()", () => {
|
describe("blob()", () => {
|
||||||
it("marks the field as lance.blob.v2", () => {
|
it("marks the field as lance.blob.v2", () => {
|
||||||
@@ -46,6 +62,13 @@ describe("blob()", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
class BlobStruct {
|
||||||
|
constructor(
|
||||||
|
public data?: unknown,
|
||||||
|
public uri?: unknown,
|
||||||
|
) {}
|
||||||
|
}
|
||||||
|
|
||||||
describe("coerceBlobValue", () => {
|
describe("coerceBlobValue", () => {
|
||||||
it.each([
|
it.each([
|
||||||
["Buffer", Buffer.from("x"), { data: Buffer.from("x"), uri: null }],
|
["Buffer", Buffer.from("x"), { data: Buffer.from("x"), uri: null }],
|
||||||
@@ -65,6 +88,41 @@ describe("coerceBlobValue", () => {
|
|||||||
{ uri: "s3://bucket/key" },
|
{ uri: "s3://bucket/key" },
|
||||||
{ data: null, uri: "s3://bucket/key" },
|
{ data: null, uri: "s3://bucket/key" },
|
||||||
],
|
],
|
||||||
|
[
|
||||||
|
"ArrayBuffer",
|
||||||
|
new Uint8Array([120]).buffer,
|
||||||
|
{ data: new Uint8Array([120]), uri: null },
|
||||||
|
],
|
||||||
|
[
|
||||||
|
"URL",
|
||||||
|
new URL("https://example.com/blob"),
|
||||||
|
{ data: null, uri: "https://example.com/blob" },
|
||||||
|
],
|
||||||
|
[
|
||||||
|
"data struct with ArrayBuffer",
|
||||||
|
{ data: new Uint8Array([121]).buffer },
|
||||||
|
{ data: new Uint8Array([121]), uri: null },
|
||||||
|
],
|
||||||
|
[
|
||||||
|
"uri struct with URL",
|
||||||
|
{ uri: new URL("file:///tmp/a.png") },
|
||||||
|
{ data: null, uri: "file:///tmp/a.png" },
|
||||||
|
],
|
||||||
|
[
|
||||||
|
"class instance with Uint8Array data",
|
||||||
|
new BlobStruct(new Uint8Array([122])),
|
||||||
|
{ data: new Uint8Array([122]), uri: null },
|
||||||
|
],
|
||||||
|
[
|
||||||
|
"class instance with ArrayBuffer data",
|
||||||
|
new BlobStruct(new Uint8Array([122]).buffer),
|
||||||
|
{ data: new Uint8Array([122]), uri: null },
|
||||||
|
],
|
||||||
|
[
|
||||||
|
"class instance with URL uri",
|
||||||
|
new BlobStruct(undefined, new URL("s3://bucket/key")),
|
||||||
|
{ data: null, uri: "s3://bucket/key" },
|
||||||
|
],
|
||||||
["null", null, null],
|
["null", null, null],
|
||||||
])("accepts %s", (_name, input, expected) => {
|
])("accepts %s", (_name, input, expected) => {
|
||||||
expect(coerceBlobValue(input)).toEqual(expected);
|
expect(coerceBlobValue(input)).toEqual(expected);
|
||||||
@@ -76,7 +134,7 @@ describe("coerceBlobValue", () => {
|
|||||||
[
|
[
|
||||||
"Int16Array",
|
"Int16Array",
|
||||||
new Int16Array([1]),
|
new Int16Array([1]),
|
||||||
/Blob data must be Buffer or Uint8Array/,
|
/Blob data must be Buffer, Uint8Array, ArrayBuffer, or Blob/,
|
||||||
],
|
],
|
||||||
[
|
[
|
||||||
"both data and uri",
|
"both data and uri",
|
||||||
@@ -88,11 +146,160 @@ describe("coerceBlobValue", () => {
|
|||||||
{ data: null, uri: null },
|
{ data: null, uri: null },
|
||||||
/exactly one of 'data' or 'uri'/,
|
/exactly one of 'data' or 'uri'/,
|
||||||
],
|
],
|
||||||
|
["Blob", new Blob(["x"]), /must be read asynchronously/],
|
||||||
|
["data struct with Blob", { data: new Blob(["x"]) }, /asynchronously/],
|
||||||
|
[
|
||||||
|
"class instance with Blob data",
|
||||||
|
new BlobStruct(new Blob(["x"])),
|
||||||
|
/asynchronously/,
|
||||||
|
],
|
||||||
|
[
|
||||||
|
"unknown struct field",
|
||||||
|
{ data: Buffer.from("y"), mime: "image/png" },
|
||||||
|
/Blob struct values only support 'data' and 'uri', got 'mime'/,
|
||||||
|
],
|
||||||
|
[
|
||||||
|
"misspelled struct field",
|
||||||
|
{ uri: "s3://bucket/key", dat: Buffer.from("y") },
|
||||||
|
/got 'dat'/,
|
||||||
|
],
|
||||||
|
[
|
||||||
|
"SharedArrayBuffer",
|
||||||
|
new SharedArrayBuffer(2),
|
||||||
|
/Blob column values must be Buffer, Uint8Array, ArrayBuffer, Blob/,
|
||||||
|
],
|
||||||
|
["Date", new Date(0), /Blob column values must be/],
|
||||||
])("rejects %s", (_name, input, message) => {
|
])("rejects %s", (_name, input, message) => {
|
||||||
expect(() => coerceBlobValue(input)).toThrow(message);
|
expect(() => coerceBlobValue(input)).toThrow(message);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("resolveBlobInputs", () => {
|
||||||
|
it("reads Blob and File values in blob columns only", async () => {
|
||||||
|
const schema = new Schema([
|
||||||
|
new Field("name", new Utf8(), true),
|
||||||
|
blob("image"),
|
||||||
|
new Field("images", new List(blob("image")), true),
|
||||||
|
]);
|
||||||
|
const shared = new Blob(["same"]);
|
||||||
|
const notABlobColumn = new Blob(["other"]);
|
||||||
|
const data = [
|
||||||
|
{
|
||||||
|
name: notABlobColumn,
|
||||||
|
image: new File(["file"], "a.txt"),
|
||||||
|
images: [shared, { data: shared }, null],
|
||||||
|
},
|
||||||
|
{ name: "x", image: Buffer.from("buf"), images: null },
|
||||||
|
];
|
||||||
|
const resolved = await resolveBlobInputs(data, schema);
|
||||||
|
|
||||||
|
expect(resolved[0].name).toBe(notABlobColumn);
|
||||||
|
expect(Buffer.from(resolved[0].image as Uint8Array).toString()).toBe(
|
||||||
|
"file",
|
||||||
|
);
|
||||||
|
const images = resolved[0].images as [Uint8Array, { data: Uint8Array }];
|
||||||
|
expect(Buffer.from(images[0]).toString()).toBe("same");
|
||||||
|
expect(Buffer.from(images[1].data).toString()).toBe("same");
|
||||||
|
// Rows without Blob values and the caller's records are left alone.
|
||||||
|
expect(resolved[1]).toBe(data[1]);
|
||||||
|
expect(data[0].images?.[0]).toBe(shared);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("reads each distinct Blob once, bare or inside data structs", async () => {
|
||||||
|
class CountingBlob extends Blob {
|
||||||
|
calls = 0;
|
||||||
|
async arrayBuffer(): Promise<ArrayBuffer> {
|
||||||
|
this.calls++;
|
||||||
|
return super.arrayBuffer();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
const shared = new CountingBlob(["same"]);
|
||||||
|
const schema = new Schema([blob("image")]);
|
||||||
|
const resolved = await resolveBlobInputs(
|
||||||
|
[
|
||||||
|
{ image: { data: shared } },
|
||||||
|
{ image: { data: shared } },
|
||||||
|
{ image: shared },
|
||||||
|
],
|
||||||
|
schema,
|
||||||
|
);
|
||||||
|
expect(shared.calls).toBe(1);
|
||||||
|
const first = (resolved[0].image as { data: Uint8Array }).data;
|
||||||
|
expect(Buffer.from(first).toString()).toBe("same");
|
||||||
|
expect((resolved[1].image as { data: Uint8Array }).data).toBe(first);
|
||||||
|
expect(resolved[2].image).toBe(first);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("reads a Blob held by a class instance", async () => {
|
||||||
|
const schema = new Schema([blob("image")]);
|
||||||
|
const [row] = await resolveBlobInputs(
|
||||||
|
[{ image: new BlobStruct(new Blob(["hi"])) }],
|
||||||
|
schema,
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
Buffer.from((row.image as { data: Uint8Array }).data).toString(),
|
||||||
|
).toBe("hi");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("returns the input when there is nothing to read", async () => {
|
||||||
|
const data = [{ image: Buffer.from("x") }];
|
||||||
|
const schema = new Schema([blob("image")]);
|
||||||
|
await expect(resolveBlobInputs(data, schema)).resolves.toBe(data);
|
||||||
|
await expect(resolveBlobInputs(data)).resolves.toBe(data);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe("convertToTable blob columns with embedding functions", () => {
|
||||||
|
// Embedding metadata routes rows through ensureNestedFieldsExist, which must
|
||||||
|
// leave blob values for the blob coercion instead of treating them as structs.
|
||||||
|
class LengthEmbedding extends EmbeddingFunction<string> {
|
||||||
|
ndims() {
|
||||||
|
return 2;
|
||||||
|
}
|
||||||
|
embeddingDataType(): Float {
|
||||||
|
return new Float32();
|
||||||
|
}
|
||||||
|
async computeSourceEmbeddings(data: string[]) {
|
||||||
|
return data.map((text) => [text.length, 1]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
getRegistry().register("blob-test-length")(LengthEmbedding);
|
||||||
|
const func = getRegistry().get<LengthEmbedding>("blob-test-length")!.create();
|
||||||
|
const embeddingSchema = LanceSchema({
|
||||||
|
text: func.sourceField(new Utf8()),
|
||||||
|
vector: func.vectorField(),
|
||||||
|
});
|
||||||
|
const schema = new Schema(
|
||||||
|
[...embeddingSchema.fields, blob("image")],
|
||||||
|
embeddingSchema.metadata,
|
||||||
|
);
|
||||||
|
|
||||||
|
it.each([
|
||||||
|
["Uint8Array", new Uint8Array([104, 105]), "hi", null],
|
||||||
|
["ArrayBuffer", new Uint8Array([104, 105]).buffer, "hi", null],
|
||||||
|
["Blob", new Blob(["hi"]), "hi", null],
|
||||||
|
["URI string", "s3://b/k", null, "s3://b/k"],
|
||||||
|
["URL", new URL("s3://b/k"), null, "s3://b/k"],
|
||||||
|
["data struct", { data: new Uint8Array([104, 105]) }, "hi", null],
|
||||||
|
])("accepts %s input", async (_name, image, data, uri) => {
|
||||||
|
const table = await convertToTable([{ text: "a", image }], undefined, {
|
||||||
|
schema,
|
||||||
|
});
|
||||||
|
const column = table.getChild("image")!;
|
||||||
|
expect(column.nullCount).toBe(0);
|
||||||
|
const bytes = column.getChild("data")!.get(0);
|
||||||
|
expect(bytes == null ? null : Buffer.from(bytes).toString()).toBe(data);
|
||||||
|
expect(column.getChild("uri")!.get(0)).toBe(uri);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("writes a missing blob as null", async () => {
|
||||||
|
const table = await convertToTable([{ text: "a" }], undefined, { schema });
|
||||||
|
const column = table.getChild("image")!;
|
||||||
|
expect(column.nullCount).toBe(1);
|
||||||
|
expect(column.get(0)).toBeNull();
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("BlobFile", () => {
|
describe("BlobFile", () => {
|
||||||
it("rejects constructing BlobFile without a native handle", () => {
|
it("rejects constructing BlobFile without a native handle", () => {
|
||||||
expect(() => new (BlobFile as unknown as { new (): BlobFile })()).toThrow(
|
expect(() => new (BlobFile as unknown as { new (): BlobFile })()).toThrow(
|
||||||
@@ -103,26 +310,122 @@ describe("BlobFile", () => {
|
|||||||
|
|
||||||
describe("makeArrowTable blob columns", () => {
|
describe("makeArrowTable blob columns", () => {
|
||||||
it.each([
|
it.each([
|
||||||
["ArrayBuffer", new Uint8Array([104]).buffer],
|
|
||||||
["Blob", new Blob(["hello"])],
|
|
||||||
["File", new File(["hello"], "hello.txt")],
|
|
||||||
["URL", new URL("https://example.com/blob")],
|
|
||||||
["ReadableStream", new ReadableStream()],
|
["ReadableStream", new ReadableStream()],
|
||||||
["data struct with ArrayBuffer", { data: new Uint8Array([104]).buffer }],
|
["Int16Array", new Int16Array([1])],
|
||||||
["uri struct with URL", { uri: new URL("https://example.com/blob") }],
|
|
||||||
])("rejects %s input rather than writing a null blob", (_name, image) => {
|
])("rejects %s input rather than writing a null blob", (_name, image) => {
|
||||||
const schema = new Schema([blob("image")]);
|
const schema = new Schema([blob("image")]);
|
||||||
expect(() => makeArrowTable([{ image }], { schema })).toThrow(/image/);
|
expect(() => makeArrowTable([{ image }], { schema })).toThrow(/image/);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("rejects Blob instances with enumerable fields as one invalid value", () => {
|
it.each([
|
||||||
|
["Blob", new Blob(["hello"])],
|
||||||
|
["File", new File(["hello"], "hello.txt")],
|
||||||
|
["data struct with Blob", { data: new Blob(["hello"]) }],
|
||||||
|
[
|
||||||
|
"Blob with enumerable fields",
|
||||||
|
Object.assign(new Blob(["hello"]), { position: 0 }),
|
||||||
|
],
|
||||||
|
])("points %s input at the async APIs", (_name, image) => {
|
||||||
const schema = new Schema([blob("image")]);
|
const schema = new Schema([blob("image")]);
|
||||||
const image = Object.assign(new Blob(["hello"]), { position: 0 });
|
expect(() => makeArrowTable([{ id: 1, image }], { schema })).toThrow(
|
||||||
expect(() => makeArrowTable([{ image }], { schema })).toThrow(
|
/blob field image at row 0: Blob and File values must be read asynchronously/,
|
||||||
"Unsupported object value for field image at row 0.",
|
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it.each([
|
||||||
|
["ArrayBuffer", new Uint8Array([104, 105]).buffer, "hi", null],
|
||||||
|
[
|
||||||
|
"data struct with ArrayBuffer",
|
||||||
|
{ data: new Uint8Array([104, 105]).buffer },
|
||||||
|
"hi",
|
||||||
|
null,
|
||||||
|
],
|
||||||
|
["URL", new URL("https://example.com/a"), null, "https://example.com/a"],
|
||||||
|
[
|
||||||
|
"uri struct with URL",
|
||||||
|
{ uri: new URL("file:///tmp/a.png") },
|
||||||
|
null,
|
||||||
|
"file:///tmp/a.png",
|
||||||
|
],
|
||||||
|
])("accepts %s input", (_name, image, data, uri) => {
|
||||||
|
const schema = new Schema([blob("image")]);
|
||||||
|
const input = { image };
|
||||||
|
const table = makeArrowTable([input], { schema });
|
||||||
|
const column = table.getChild("image")!;
|
||||||
|
expect(column.nullCount).toBe(0);
|
||||||
|
const bytes = column.getChild("data")!.get(0);
|
||||||
|
expect(bytes == null ? null : Buffer.from(bytes).toString()).toBe(data);
|
||||||
|
expect(column.getChild("uri")!.get(0)).toBe(uri);
|
||||||
|
// The caller's record is left untouched.
|
||||||
|
expect(input.image).toBe(image);
|
||||||
|
});
|
||||||
|
|
||||||
|
it.each([
|
||||||
|
["Uint8Array", new Uint8Array([104, 105]), "hi", null],
|
||||||
|
["URI string", "s3://b/k", null, "s3://b/k"],
|
||||||
|
["data struct", { data: new Uint8Array([104, 105]) }, "hi", null],
|
||||||
|
])("accepts %s input after a null row", (_name, image, data, uri) => {
|
||||||
|
const schema = new Schema([blob("image")]);
|
||||||
|
const table = makeArrowTable([{ image: null }, { image }], { schema });
|
||||||
|
const column = table.getChild("image")!;
|
||||||
|
expect(column.nullCount).toBe(1);
|
||||||
|
expect(column.get(0)).toBeNull();
|
||||||
|
const bytes = column.getChild("data")!.get(1);
|
||||||
|
expect(bytes == null ? null : Buffer.from(bytes).toString()).toBe(data);
|
||||||
|
expect(column.getChild("uri")!.get(1)).toBe(uri);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("accepts a null struct before one holding a blob", () => {
|
||||||
|
const schema = new Schema([
|
||||||
|
new Field("info", new Struct([blob("image")]), true),
|
||||||
|
]);
|
||||||
|
const table = makeArrowTable(
|
||||||
|
[{ info: null }, { info: { image: new Uint8Array([97]) } }],
|
||||||
|
{ schema },
|
||||||
|
);
|
||||||
|
const info = table.getChild("info")!;
|
||||||
|
expect(info.get(0)).toBeNull();
|
||||||
|
const image = (info.get(1) as { image: { data: Uint8Array } }).image;
|
||||||
|
expect(Buffer.from(image.data).toString()).toBe("a");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("skips a missing blob column named like an Object.prototype member", () => {
|
||||||
|
const schema = new Schema([
|
||||||
|
new Field("id", new Int32(), true),
|
||||||
|
blob("constructor"),
|
||||||
|
]);
|
||||||
|
const table = makeArrowTable([{ id: 1 }], { schema });
|
||||||
|
expect(table.schema.fields.map((f) => f.name)).toEqual(["id"]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("accepts ArrayBuffer and URL inside lists and structs", () => {
|
||||||
|
const schema = new Schema([
|
||||||
|
new Field("images", new List(blob("image")), true),
|
||||||
|
new Field("info", new Struct([blob("image")]), true),
|
||||||
|
]);
|
||||||
|
const table = makeArrowTable(
|
||||||
|
[
|
||||||
|
{
|
||||||
|
images: [new Uint8Array([97]).buffer, new URL("s3://b/k")],
|
||||||
|
info: { image: new Uint8Array([98]).buffer },
|
||||||
|
},
|
||||||
|
],
|
||||||
|
{ schema },
|
||||||
|
);
|
||||||
|
const images = Array.from(
|
||||||
|
table.getChild("images")!.get(0) as Iterable<{
|
||||||
|
data: Uint8Array | null;
|
||||||
|
uri: string | null;
|
||||||
|
}>,
|
||||||
|
);
|
||||||
|
expect(Buffer.from(images[0].data!).toString()).toBe("a");
|
||||||
|
expect(images[1].uri).toBe("s3://b/k");
|
||||||
|
const info = table.getChild("info")!.get(0) as {
|
||||||
|
image: { data: Uint8Array };
|
||||||
|
};
|
||||||
|
expect(Buffer.from(info.image.data).toString()).toBe("b");
|
||||||
|
});
|
||||||
|
|
||||||
it("coerces Buffer input onto a blob field", () => {
|
it("coerces Buffer input onto a blob field", () => {
|
||||||
const schema = new Schema([
|
const schema = new Schema([
|
||||||
new Field("id", new Int64(), true),
|
new Field("id", new Int64(), true),
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import * as tmp from "tmp";
|
|||||||
|
|
||||||
import {
|
import {
|
||||||
AutoQuery,
|
AutoQuery,
|
||||||
|
type BlobInput,
|
||||||
Connection,
|
Connection,
|
||||||
MatchQuery,
|
MatchQuery,
|
||||||
PhraseQuery,
|
PhraseQuery,
|
||||||
@@ -2583,7 +2584,7 @@ describe("when dealing with blob columns", () => {
|
|||||||
new Field("id", new Int64(), true),
|
new Field("id", new Int64(), true),
|
||||||
blob("image"),
|
blob("image"),
|
||||||
]);
|
]);
|
||||||
const row = { id: 1n, image: new Uint8Array([104]).buffer };
|
const row = { id: 1n, image: new ReadableStream() };
|
||||||
|
|
||||||
await expect(db.createTable("invalid", [row], { schema })).rejects.toThrow(
|
await expect(db.createTable("invalid", [row], { schema })).rejects.toThrow(
|
||||||
/field image at row 0/,
|
/field image at row 0/,
|
||||||
@@ -2613,6 +2614,8 @@ describe("when dealing with blob columns", () => {
|
|||||||
it.each([
|
it.each([
|
||||||
["URI string", (uri: string) => uri],
|
["URI string", (uri: string) => uri],
|
||||||
["URI struct", (uri: string) => ({ uri })],
|
["URI struct", (uri: string) => ({ uri })],
|
||||||
|
["URL", (uri: string) => new URL(uri)],
|
||||||
|
["URL struct", (uri: string) => ({ uri: new URL(uri) })],
|
||||||
])(
|
])(
|
||||||
"round-trips an external blob from a %s after reopening",
|
"round-trips an external blob from a %s after reopening",
|
||||||
async (_label, blobValue) => {
|
async (_label, blobValue) => {
|
||||||
@@ -2706,6 +2709,63 @@ describe("when dealing with blob columns", () => {
|
|||||||
expect(bytes[0]).toEqual(payload);
|
expect(bytes[0]).toEqual(payload);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("accepts ArrayBuffer, Blob, and File in createTable and add", async () => {
|
||||||
|
const db = await connect(tmpDir.name);
|
||||||
|
const schema = new Schema([
|
||||||
|
new Field("id", new Int64(), true),
|
||||||
|
blob("image"),
|
||||||
|
]);
|
||||||
|
const rows: { id: bigint; image: BlobInput }[] = [
|
||||||
|
{ id: 1n, image: new TextEncoder().encode("array-buffer").buffer },
|
||||||
|
{ id: 2n, image: new Blob(["blob"]) },
|
||||||
|
{ id: 3n, image: { data: new File(["file"], "f.txt") } },
|
||||||
|
];
|
||||||
|
const table = await db.createTable("widened", rows.slice(0, 2), {
|
||||||
|
schema,
|
||||||
|
});
|
||||||
|
await table.add(rows.slice(2));
|
||||||
|
|
||||||
|
const results = await table.query().select(["id"]).withRowId().toArray();
|
||||||
|
results.sort((a, b) => Number(a.id - b.id));
|
||||||
|
const bytes = await table.fetchBlobs(
|
||||||
|
"image",
|
||||||
|
results.map((row) => row._rowid as bigint),
|
||||||
|
);
|
||||||
|
expect(bytes.map((b) => b?.toString())).toEqual([
|
||||||
|
"array-buffer",
|
||||||
|
"blob",
|
||||||
|
"file",
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("accepts Blob values in mergeInsert", async () => {
|
||||||
|
const db = await connect(tmpDir.name);
|
||||||
|
const schema = new Schema([
|
||||||
|
new Field("id", new Int64(), true),
|
||||||
|
blob("image"),
|
||||||
|
]);
|
||||||
|
const table = await db.createTable(
|
||||||
|
"merge_blobs",
|
||||||
|
[{ id: 1n, image: Buffer.from("old") }],
|
||||||
|
{ schema },
|
||||||
|
);
|
||||||
|
await table
|
||||||
|
.mergeInsert("id")
|
||||||
|
.whenMatchedUpdateAll()
|
||||||
|
.whenNotMatchedInsertAll()
|
||||||
|
.execute([
|
||||||
|
{ id: 1n, image: new Blob(["new"]) },
|
||||||
|
{ id: 2n, image: new Uint8Array([104, 105]).buffer },
|
||||||
|
]);
|
||||||
|
const results = await table.query().select(["id"]).withRowId().toArray();
|
||||||
|
results.sort((a, b) => Number(a.id - b.id));
|
||||||
|
const bytes = await table.fetchBlobs(
|
||||||
|
"image",
|
||||||
|
results.map((row) => row._rowid as bigint),
|
||||||
|
);
|
||||||
|
expect(bytes.map((b) => b?.toString())).toEqual(["new", "hi"]);
|
||||||
|
});
|
||||||
|
|
||||||
it("discovers blob columns", async () => {
|
it("discovers blob columns", async () => {
|
||||||
const { table } = await openBlobTable();
|
const { table } = await openBlobTable();
|
||||||
expect(await table.blobColumns()).toEqual(["image"]);
|
expect(await table.blobColumns()).toEqual(["image"]);
|
||||||
|
|||||||
+173
-11
@@ -40,7 +40,12 @@ import {
|
|||||||
} from "apache-arrow";
|
} from "apache-arrow";
|
||||||
import { Buffers } from "apache-arrow/data";
|
import { Buffers } from "apache-arrow/data";
|
||||||
import { typedArrayToArrowType } from "./arrow_type";
|
import { typedArrayToArrowType } from "./arrow_type";
|
||||||
import { coerceBlobValue, isBlobField } from "./blob";
|
import {
|
||||||
|
blobToRead,
|
||||||
|
coerceBlobValue,
|
||||||
|
isBlobField,
|
||||||
|
withBlobBytes,
|
||||||
|
} from "./blob";
|
||||||
import { type EmbeddingFunction } from "./embedding/embedding_function";
|
import { type EmbeddingFunction } from "./embedding/embedding_function";
|
||||||
import {
|
import {
|
||||||
EmbeddingFunctionConfig,
|
EmbeddingFunctionConfig,
|
||||||
@@ -449,6 +454,24 @@ export function makeArrowTable(
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (schema !== undefined) {
|
||||||
|
// Validate blob values up front and give every one the full
|
||||||
|
// `{ data, uri }` shape, so inference sees the same struct whether a row
|
||||||
|
// passed bytes, a URI, or a partial struct.
|
||||||
|
data = mapBlobInputs(data, schema, (value, field, row) => {
|
||||||
|
if (value === undefined) {
|
||||||
|
return value;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
return coerceBlobValue(value);
|
||||||
|
} catch (e) {
|
||||||
|
throw new Error(
|
||||||
|
`Invalid value for blob field ${field} at row ${row}: ${(e as Error).message}`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
let schemaMetadata = schema?.metadata || new Map<string, string>();
|
let schemaMetadata = schema?.metadata || new Map<string, string>();
|
||||||
if (metadata !== undefined) {
|
if (metadata !== undefined) {
|
||||||
schemaMetadata = new Map([...schemaMetadata, ...metadata]);
|
schemaMetadata = new Map([...schemaMetadata, ...metadata]);
|
||||||
@@ -510,6 +533,133 @@ function containsBlobField(field: Field): boolean {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type BlobInputVisitor = (value: unknown, field: string, row: number) => unknown;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Calls `visit` on every value that lands in a blob column of `schema`,
|
||||||
|
* including blobs nested in structs and lists, and replaces it with the
|
||||||
|
* result. Records and arrays are copied only where a value changed, so the
|
||||||
|
* caller's data is never mutated.
|
||||||
|
*/
|
||||||
|
function mapBlobInputs(
|
||||||
|
data: Array<Record<string, unknown>>,
|
||||||
|
schema: Schema,
|
||||||
|
visit: BlobInputVisitor,
|
||||||
|
): Array<Record<string, unknown>> {
|
||||||
|
if (!schema.fields.some(containsBlobField)) {
|
||||||
|
return data;
|
||||||
|
}
|
||||||
|
return data.map((record, row) =>
|
||||||
|
isObject(record)
|
||||||
|
? mapBlobFields(record, schema.fields, "", row, visit)
|
||||||
|
: record,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
function mapBlobFields(
|
||||||
|
record: Record<string, unknown>,
|
||||||
|
fields: Field[],
|
||||||
|
prefix: string,
|
||||||
|
row: number,
|
||||||
|
visit: BlobInputVisitor,
|
||||||
|
): Record<string, unknown> {
|
||||||
|
let out: Record<string, unknown> | undefined;
|
||||||
|
for (const field of fields) {
|
||||||
|
if (!containsBlobField(field) || !Object.hasOwn(record, field.name)) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
const value = record[field.name];
|
||||||
|
const mapped = mapBlobField(
|
||||||
|
value,
|
||||||
|
field,
|
||||||
|
`${prefix}${field.name}`,
|
||||||
|
row,
|
||||||
|
visit,
|
||||||
|
);
|
||||||
|
if (mapped !== value) {
|
||||||
|
out ??= { ...record };
|
||||||
|
out[field.name] = mapped;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out ?? record;
|
||||||
|
}
|
||||||
|
|
||||||
|
function mapBlobField(
|
||||||
|
value: unknown,
|
||||||
|
field: Field,
|
||||||
|
label: string,
|
||||||
|
row: number,
|
||||||
|
visit: BlobInputVisitor,
|
||||||
|
): unknown {
|
||||||
|
if (isBlobField(field)) {
|
||||||
|
return visit(value, label, row);
|
||||||
|
}
|
||||||
|
if (field.type instanceof Struct && isObject(value)) {
|
||||||
|
return mapBlobFields(value, field.type.children, `${label}.`, row, visit);
|
||||||
|
}
|
||||||
|
if (isList(field.type) && Array.isArray(value)) {
|
||||||
|
const child = field.type.children[0];
|
||||||
|
let out: unknown[] | undefined;
|
||||||
|
for (const [index, element] of value.entries()) {
|
||||||
|
const mapped = mapBlobField(
|
||||||
|
element,
|
||||||
|
child,
|
||||||
|
`${label}[${index}]`,
|
||||||
|
row,
|
||||||
|
visit,
|
||||||
|
);
|
||||||
|
if (mapped !== element) {
|
||||||
|
out ??= [...value];
|
||||||
|
out[index] = mapped;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out ?? value;
|
||||||
|
}
|
||||||
|
return value;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Reads `Blob` / `File` values in the blob columns of `schema` into bytes so
|
||||||
|
* the synchronous conversion in {@link makeArrowTable} can accept them.
|
||||||
|
* Returns `data` itself when there is nothing to read.
|
||||||
|
*/
|
||||||
|
export async function resolveBlobInputs(
|
||||||
|
data: Array<Record<string, unknown>>,
|
||||||
|
schema?: SchemaLike,
|
||||||
|
): Promise<Array<Record<string, unknown>>> {
|
||||||
|
if (schema === undefined || schema === null) {
|
||||||
|
return data;
|
||||||
|
}
|
||||||
|
const sanitized = sanitizeSchema(schema);
|
||||||
|
// Keyed by the Blob itself, so rows sharing one Blob (bare or as
|
||||||
|
// `{ data }`) read it once and share the bytes.
|
||||||
|
const reads = new Map<Blob, Promise<Uint8Array>>();
|
||||||
|
mapBlobInputs(data, sanitized, (value) => {
|
||||||
|
const source = blobToRead(value);
|
||||||
|
if (source !== undefined && !reads.has(source)) {
|
||||||
|
reads.set(
|
||||||
|
source,
|
||||||
|
source.arrayBuffer().then((buffer) => new Uint8Array(buffer)),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
return value;
|
||||||
|
});
|
||||||
|
if (reads.size === 0) {
|
||||||
|
return data;
|
||||||
|
}
|
||||||
|
const bytes = new Map(
|
||||||
|
await Promise.all(
|
||||||
|
[...reads].map(async ([source, read]) => [source, await read] as const),
|
||||||
|
),
|
||||||
|
);
|
||||||
|
return mapBlobInputs(data, sanitized, (value) => {
|
||||||
|
const source = blobToRead(value);
|
||||||
|
return source === undefined
|
||||||
|
? value
|
||||||
|
: withBlobBytes(value, bytes.get(source) as Uint8Array);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
function isObject(value: unknown): value is Record<string, unknown> {
|
function isObject(value: unknown): value is Record<string, unknown> {
|
||||||
return (
|
return (
|
||||||
typeof value === "object" &&
|
typeof value === "object" &&
|
||||||
@@ -1013,6 +1163,11 @@ export async function convertToTable(
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
processedData = await resolveBlobInputs(
|
||||||
|
processedData,
|
||||||
|
makeTableOptions?.schema,
|
||||||
|
);
|
||||||
|
|
||||||
const table = makeArrowTable(processedData, makeTableOptions);
|
const table = makeArrowTable(processedData, makeTableOptions);
|
||||||
return await applyEmbeddings(table, embeddings, makeTableOptions?.schema);
|
return await applyEmbeddings(table, embeddings, makeTableOptions?.schema);
|
||||||
}
|
}
|
||||||
@@ -1369,7 +1524,7 @@ export function ensureNestedFieldsExist(
|
|||||||
for (const field of schema.fields) {
|
for (const field of schema.fields) {
|
||||||
if (field.name in row) {
|
if (field.name in row) {
|
||||||
if (
|
if (
|
||||||
field.type.constructor.name === "Struct" &&
|
isPlainStructField(field) &&
|
||||||
row[field.name] !== null &&
|
row[field.name] !== null &&
|
||||||
row[field.name] !== undefined
|
row[field.name] !== undefined
|
||||||
) {
|
) {
|
||||||
@@ -1386,10 +1541,9 @@ export function ensureNestedFieldsExist(
|
|||||||
} else {
|
} else {
|
||||||
// Keep a missing struct valid while filling each of its children with
|
// Keep a missing struct valid while filling each of its children with
|
||||||
// null. This is distinct from an explicitly null struct value.
|
// null. This is distinct from an explicitly null struct value.
|
||||||
completeRow[field.name] =
|
completeRow[field.name] = isPlainStructField(field)
|
||||||
field.type.constructor.name === "Struct"
|
? ensureStructFieldsExist({}, field.type as Struct)
|
||||||
? ensureStructFieldsExist({}, field.type as Struct)
|
: null;
|
||||||
: null;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1397,6 +1551,15 @@ export function ensureNestedFieldsExist(
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Blob fields are Arrow structs, but their values are bytes, URIs, or
|
||||||
|
* `{ data } | { uri }` inputs that the blob coercion in `makeArrowTable`
|
||||||
|
* handles, so they must not be filled in like ordinary structs.
|
||||||
|
*/
|
||||||
|
function isPlainStructField(field: Field): boolean {
|
||||||
|
return field.type.constructor.name === "Struct" && !isBlobField(field);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Recursively ensures that all fields in a struct type exist in the data,
|
* Recursively ensures that all fields in a struct type exist in the data,
|
||||||
* filling missing fields with null values.
|
* filling missing fields with null values.
|
||||||
@@ -1410,7 +1573,7 @@ function ensureStructFieldsExist(
|
|||||||
for (const childField of structType.children) {
|
for (const childField of structType.children) {
|
||||||
if (childField.name in data) {
|
if (childField.name in data) {
|
||||||
if (
|
if (
|
||||||
childField.type.constructor.name === "Struct" &&
|
isPlainStructField(childField) &&
|
||||||
data[childField.name] !== null &&
|
data[childField.name] !== null &&
|
||||||
data[childField.name] !== undefined
|
data[childField.name] !== undefined
|
||||||
) {
|
) {
|
||||||
@@ -1426,10 +1589,9 @@ function ensureStructFieldsExist(
|
|||||||
} else {
|
} else {
|
||||||
// Keep a missing struct valid while filling each of its children with
|
// Keep a missing struct valid while filling each of its children with
|
||||||
// null. This is distinct from an explicitly null struct value.
|
// null. This is distinct from an explicitly null struct value.
|
||||||
completeStruct[childField.name] =
|
completeStruct[childField.name] = isPlainStructField(childField)
|
||||||
childField.type.constructor.name === "Struct"
|
? ensureStructFieldsExist({}, childField.type as Struct)
|
||||||
? ensureStructFieldsExist({}, childField.type as Struct)
|
: null;
|
||||||
: null;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+165
-8
@@ -12,7 +12,46 @@ const DEDICATED_SIZE_THRESHOLD_KEY =
|
|||||||
const PACK_FILE_SIZE_THRESHOLD_KEY =
|
const PACK_FILE_SIZE_THRESHOLD_KEY =
|
||||||
"lance-encoding:blob-pack-file-size-threshold";
|
"lance-encoding:blob-pack-file-size-threshold";
|
||||||
|
|
||||||
export type BlobInput = {
|
/**
|
||||||
|
* Bytes accepted for a blob value. `ArrayBuffer` is wrapped without copying.
|
||||||
|
* `Blob` and `File` are read with `arrayBuffer()`, which only the async write
|
||||||
|
* paths ({@link Connection.createTable}, {@link Table.add},
|
||||||
|
* {@link Table.mergeInsert}) can do.
|
||||||
|
*/
|
||||||
|
export type BlobData = Buffer | Uint8Array | ArrayBuffer | Blob;
|
||||||
|
|
||||||
|
/** A URI accepted for a blob value. A `URL` is stored as its `href`. */
|
||||||
|
export type BlobUri = string | URL;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A value accepted for a `lance.blob.v2` column (see {@link blob}).
|
||||||
|
*
|
||||||
|
* Either inline bytes, a URI pointing at external bytes, or a struct that sets
|
||||||
|
* exactly one of `data` or `uri`. `null` and `undefined` write a null blob.
|
||||||
|
*
|
||||||
|
* @example
|
||||||
|
* ```ts
|
||||||
|
* import { pathToFileURL } from "node:url";
|
||||||
|
* import type { BlobInput } from "@lancedb/lancedb";
|
||||||
|
*
|
||||||
|
* const rows: { id: bigint; image: BlobInput }[] = [
|
||||||
|
* { id: 1n, image: await (await fetch(url)).arrayBuffer() },
|
||||||
|
* { id: 2n, image: pathToFileURL("/data/cat.png") },
|
||||||
|
* { id: 3n, image: { data: new Blob(["hello"]) } },
|
||||||
|
* ];
|
||||||
|
* await table.add(rows);
|
||||||
|
* ```
|
||||||
|
*/
|
||||||
|
export type BlobInput =
|
||||||
|
| BlobData
|
||||||
|
| BlobUri
|
||||||
|
| { data: BlobData; uri?: null }
|
||||||
|
| { data?: null; uri: BlobUri }
|
||||||
|
| null
|
||||||
|
| undefined;
|
||||||
|
|
||||||
|
/** A blob value normalized to the columns of the blob struct. */
|
||||||
|
export type BlobValue = {
|
||||||
data: Buffer | Uint8Array | null;
|
data: Buffer | Uint8Array | null;
|
||||||
uri: string | null;
|
uri: string | null;
|
||||||
};
|
};
|
||||||
@@ -248,7 +287,77 @@ export type BlobRange = {
|
|||||||
end: bigint;
|
end: bigint;
|
||||||
};
|
};
|
||||||
|
|
||||||
export function coerceBlobValue(value: unknown): BlobInput | null {
|
/**
|
||||||
|
* Rewrites the synchronous-only widenings of {@link BlobInput} (`ArrayBuffer`,
|
||||||
|
* `URL`) into `Uint8Array` and URI strings, keeping the value's shape. Other
|
||||||
|
* values are returned unchanged and validated later by {@link coerceBlobValue}.
|
||||||
|
*
|
||||||
|
* Throws on `Blob` / `File`, which must be read into bytes first.
|
||||||
|
*/
|
||||||
|
export function normalizeBlobInput(value: unknown): unknown {
|
||||||
|
if (blobToRead(value) !== undefined) {
|
||||||
|
throw new Error(
|
||||||
|
"Blob and File values must be read asynchronously. Pass them to " +
|
||||||
|
"createTable, add, or mergeInsert, or convert them with " +
|
||||||
|
"`new Uint8Array(await value.arrayBuffer())`",
|
||||||
|
);
|
||||||
|
}
|
||||||
|
if (value instanceof ArrayBuffer) {
|
||||||
|
return new Uint8Array(value);
|
||||||
|
}
|
||||||
|
if (value instanceof URL) {
|
||||||
|
return value.href;
|
||||||
|
}
|
||||||
|
if (isBlobStruct(value)) {
|
||||||
|
// Read `data` and `uri` explicitly: spreading a class instance would drop
|
||||||
|
// fields defined as getters on its prototype.
|
||||||
|
const out: Record<string, unknown> = {
|
||||||
|
...value,
|
||||||
|
data: value.data,
|
||||||
|
uri: value.uri,
|
||||||
|
};
|
||||||
|
let changed = false;
|
||||||
|
if (value.data instanceof ArrayBuffer) {
|
||||||
|
out.data = new Uint8Array(value.data);
|
||||||
|
changed = true;
|
||||||
|
}
|
||||||
|
if (value.uri instanceof URL) {
|
||||||
|
out.uri = value.uri.href;
|
||||||
|
changed = true;
|
||||||
|
}
|
||||||
|
return changed ? out : value;
|
||||||
|
}
|
||||||
|
return value;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the `Blob` / `File` a blob value needs read, either the value itself
|
||||||
|
* or its `data` field, or `undefined` when there is nothing to read.
|
||||||
|
*/
|
||||||
|
export function blobToRead(value: unknown): Blob | undefined {
|
||||||
|
if (isBlobLike(value)) {
|
||||||
|
return value;
|
||||||
|
}
|
||||||
|
if (isBlobStruct(value) && isBlobLike(value.data)) {
|
||||||
|
return value.data;
|
||||||
|
}
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Replaces the `Blob` that {@link blobToRead} found in `value` with `bytes`,
|
||||||
|
* keeping the value's shape.
|
||||||
|
*/
|
||||||
|
export function withBlobBytes(value: unknown, bytes: Uint8Array): unknown {
|
||||||
|
if (isBlobLike(value)) {
|
||||||
|
return bytes;
|
||||||
|
}
|
||||||
|
const record = value as Record<string, unknown>;
|
||||||
|
return { ...record, uri: record.uri, data: bytes };
|
||||||
|
}
|
||||||
|
|
||||||
|
export function coerceBlobValue(input: unknown): BlobValue | null {
|
||||||
|
const value = normalizeBlobInput(input);
|
||||||
if (value == null) {
|
if (value == null) {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
@@ -256,7 +365,9 @@ export function coerceBlobValue(value: unknown): BlobInput | null {
|
|||||||
return { data: value, uri: null };
|
return { data: value, uri: null };
|
||||||
}
|
}
|
||||||
if (ArrayBuffer.isView(value)) {
|
if (ArrayBuffer.isView(value)) {
|
||||||
throw new Error("Blob data must be Buffer or Uint8Array");
|
throw new Error(
|
||||||
|
"Blob data must be Buffer, Uint8Array, ArrayBuffer, or Blob",
|
||||||
|
);
|
||||||
}
|
}
|
||||||
if (typeof value === "string") {
|
if (typeof value === "string") {
|
||||||
if (value === "") {
|
if (value === "") {
|
||||||
@@ -264,23 +375,37 @@ export function coerceBlobValue(value: unknown): BlobInput | null {
|
|||||||
}
|
}
|
||||||
return { data: null, uri: value };
|
return { data: null, uri: value };
|
||||||
}
|
}
|
||||||
if (typeof value === "object") {
|
if (isBlobStruct(value)) {
|
||||||
const record = value as Record<string, unknown>;
|
const record = value;
|
||||||
if (!("data" in record) && !("uri" in record)) {
|
if (!("data" in record) && !("uri" in record)) {
|
||||||
throw new Error(
|
throw new Error(
|
||||||
"Blob struct values must include a 'data' or 'uri' field",
|
"Blob struct values must include a 'data' or 'uri' field",
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
const unknownKeys = Object.keys(record).filter(
|
||||||
|
(key) => key !== "data" && key !== "uri",
|
||||||
|
);
|
||||||
|
if (unknownKeys.length > 0) {
|
||||||
|
throw new Error(
|
||||||
|
`Blob struct values only support 'data' and 'uri', got ${unknownKeys
|
||||||
|
.map((key) => `'${key}'`)
|
||||||
|
.join(", ")}`,
|
||||||
|
);
|
||||||
|
}
|
||||||
const uri = record.uri;
|
const uri = record.uri;
|
||||||
if (uri === "") {
|
if (uri === "") {
|
||||||
throw new Error("Blob uri cannot be empty");
|
throw new Error("Blob uri cannot be empty");
|
||||||
}
|
}
|
||||||
if (uri != null && typeof uri !== "string") {
|
if (uri != null && typeof uri !== "string") {
|
||||||
throw new Error(`Blob uri must be a string or null, got ${typeof uri}`);
|
throw new Error(
|
||||||
|
`Blob uri must be a string, URL, or null, got ${typeof uri}`,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
const data = record.data;
|
const data = record.data;
|
||||||
if (data != null && !isBlobBytes(data)) {
|
if (data != null && !isBlobBytes(data)) {
|
||||||
throw new Error("Blob data must be Buffer, Uint8Array, or null");
|
throw new Error(
|
||||||
|
"Blob data must be Buffer, Uint8Array, ArrayBuffer, Blob, or null",
|
||||||
|
);
|
||||||
}
|
}
|
||||||
const bytes = (data as Buffer | Uint8Array | null | undefined) ?? null;
|
const bytes = (data as Buffer | Uint8Array | null | undefined) ?? null;
|
||||||
const uriValue = uri ?? null;
|
const uriValue = uri ?? null;
|
||||||
@@ -292,7 +417,39 @@ export function coerceBlobValue(value: unknown): BlobInput | null {
|
|||||||
return { data: bytes, uri: uriValue };
|
return { data: bytes, uri: uriValue };
|
||||||
}
|
}
|
||||||
throw new Error(
|
throw new Error(
|
||||||
"Blob column values must be Buffer, Uint8Array, a URI string, null, or { data?, uri? }",
|
"Blob column values must be Buffer, Uint8Array, ArrayBuffer, Blob, a URI string or URL, null, or { data?, uri? }",
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
function isBlobLike(value: unknown): value is Blob {
|
||||||
|
return typeof Blob !== "undefined" && value instanceof Blob;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Whether `value` is a `{ data?, uri? }` struct: any plain object, or any
|
||||||
|
* other object with a `data` or `uri` field (such as a class instance), apart
|
||||||
|
* from the binary, `Blob`, and `URL` values that are blob inputs themselves.
|
||||||
|
*/
|
||||||
|
function isBlobStruct(value: unknown): value is Record<string, unknown> {
|
||||||
|
if (
|
||||||
|
typeof value !== "object" ||
|
||||||
|
value === null ||
|
||||||
|
Array.isArray(value) ||
|
||||||
|
ArrayBuffer.isView(value) ||
|
||||||
|
value instanceof ArrayBuffer ||
|
||||||
|
(typeof SharedArrayBuffer !== "undefined" &&
|
||||||
|
value instanceof SharedArrayBuffer) ||
|
||||||
|
isBlobLike(value) ||
|
||||||
|
value instanceof URL
|
||||||
|
) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
const prototype = Object.getPrototypeOf(value);
|
||||||
|
return (
|
||||||
|
prototype === Object.prototype ||
|
||||||
|
prototype === null ||
|
||||||
|
"data" in value ||
|
||||||
|
"uri" in value
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import {
|
|||||||
fromTableToStreamBuffer,
|
fromTableToStreamBuffer,
|
||||||
isArrowTable,
|
isArrowTable,
|
||||||
makeArrowTable,
|
makeArrowTable,
|
||||||
|
resolveBlobInputs,
|
||||||
} from "./arrow";
|
} from "./arrow";
|
||||||
import {
|
import {
|
||||||
Table as ArrowTable,
|
Table as ArrowTable,
|
||||||
@@ -1133,7 +1134,11 @@ async function parseTableData(
|
|||||||
if (isArrowTable(data)) {
|
if (isArrowTable(data)) {
|
||||||
table = sanitizeTable(data);
|
table = sanitizeTable(data);
|
||||||
} else {
|
} else {
|
||||||
table = makeArrowTable(data as Record<string, unknown>[], options);
|
const records = await resolveBlobInputs(
|
||||||
|
data as Record<string, unknown>[],
|
||||||
|
options?.schema,
|
||||||
|
);
|
||||||
|
table = makeArrowTable(records, options);
|
||||||
}
|
}
|
||||||
if (streaming) {
|
if (streaming) {
|
||||||
const buf = await fromTableToStreamBuffer(
|
const buf = await fromTableToStreamBuffer(
|
||||||
|
|||||||
@@ -84,7 +84,14 @@ export {
|
|||||||
} from "./arrow";
|
} from "./arrow";
|
||||||
|
|
||||||
export { blob, isBlobField, BlobFile } from "./blob";
|
export { blob, isBlobField, BlobFile } from "./blob";
|
||||||
export type { BlobOptions, BlobRange, BlobReadOptions } from "./blob";
|
export type {
|
||||||
|
BlobData,
|
||||||
|
BlobInput,
|
||||||
|
BlobOptions,
|
||||||
|
BlobRange,
|
||||||
|
BlobReadOptions,
|
||||||
|
BlobUri,
|
||||||
|
} from "./blob";
|
||||||
|
|
||||||
export {
|
export {
|
||||||
Connection,
|
Connection,
|
||||||
|
|||||||
@@ -69,12 +69,18 @@ class SchemaInferrer {
|
|||||||
this.addField(path, value, row);
|
this.addField(path, value, row);
|
||||||
} else if (this.providedSchema === undefined) {
|
} else if (this.providedSchema === undefined) {
|
||||||
this.updateInferredField(path, value, row, current);
|
this.updateInferredField(path, value, row, current);
|
||||||
|
} else if (current instanceof DeferredTypeEvidence) {
|
||||||
|
if (value != null) {
|
||||||
|
this.addSchemaField(this.providedSchema, path, value, row);
|
||||||
|
}
|
||||||
|
} else if (current instanceof FieldTree || DataType.isStruct(current)) {
|
||||||
|
assertStructValue(path, value, row);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private addField(path: string[], value: unknown, row: number): void {
|
private addField(path: string[], value: unknown, row: number): void {
|
||||||
if (this.providedSchema !== undefined) {
|
if (this.providedSchema !== undefined) {
|
||||||
this.addSchemaField(this.providedSchema, path, row);
|
this.addSchemaField(this.providedSchema, path, value, row);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -95,15 +101,31 @@ class SchemaInferrer {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private addSchemaField(schema: Schema, path: string[], row: number): void {
|
private addSchemaField(
|
||||||
|
schema: Schema,
|
||||||
|
path: string[],
|
||||||
|
value: unknown,
|
||||||
|
row: number,
|
||||||
|
): void {
|
||||||
const field = fieldAtPath(schema, path);
|
const field = fieldAtPath(schema, path);
|
||||||
if (field === undefined) {
|
if (field === undefined) {
|
||||||
throw new Error(
|
throw new Error(
|
||||||
`Found field not in schema: ${path.join(".")} at row ${row}`,
|
`Found field not in schema: ${path.join(".")} at row ${row}`,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
if (DataType.isStruct(field.type)) {
|
||||||
|
assertStructValue(path, value, row);
|
||||||
|
}
|
||||||
|
|
||||||
const conflict = this.fields.set(path, field.type);
|
// Nulls are kept as deferred evidence so that a struct field seen first as
|
||||||
|
// null can be replaced by a record in a later row, just as a record seen
|
||||||
|
// first already ignores later nulls.
|
||||||
|
const conflict = this.fields.set(
|
||||||
|
path,
|
||||||
|
DeferredTypeEvidence.from(value, row) ?? field.type,
|
||||||
|
(existing) =>
|
||||||
|
existing instanceof DeferredTypeEvidence && existing.isOnlyNulls(),
|
||||||
|
);
|
||||||
if (conflict !== undefined) {
|
if (conflict !== undefined) {
|
||||||
throw branchConflictError(conflict, row, "Struct");
|
throw branchConflictError(conflict, row, "Struct");
|
||||||
}
|
}
|
||||||
@@ -483,6 +505,26 @@ function assertSupportedValue(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Rejects primitives, arrays, and typed arrays given for a struct field, which
|
||||||
|
* would otherwise be written as a struct with every child null.
|
||||||
|
*/
|
||||||
|
function assertStructValue(path: string[], value: unknown, row: number): void {
|
||||||
|
if (
|
||||||
|
value == null ||
|
||||||
|
(typeof value === "object" &&
|
||||||
|
!Array.isArray(value) &&
|
||||||
|
!ArrayBuffer.isView(value))
|
||||||
|
) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
const kind =
|
||||||
|
typeof value === "object" ? value.constructor.name : typeof value;
|
||||||
|
throw new Error(
|
||||||
|
`Expected a struct value for field ${path.join(".")} at row ${row}, got ${kind}.`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
function fieldAtPath(schema: Schema, path: string[]): Field | undefined {
|
function fieldAtPath(schema: Schema, path: string[]): Field | undefined {
|
||||||
let fields = schema.fields;
|
let fields = schema.fields;
|
||||||
let field: Field | undefined;
|
let field: Field | undefined;
|
||||||
|
|||||||
Reference in New Issue
Block a user