mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-21 11:12:27 +00:00
Merge branch 'merge-insert-source-stats' into merge-insert-source-stats-bindings
This commit is contained in:
Will Jones
+1
-1
@@ -1,7 +1,7 @@
|
||||
[package]
|
||||
name = "lancedb-nodejs"
|
||||
edition.workspace = true
|
||||
version = "0.41.0-beta.7"
|
||||
version = "0.41.0-beta.9"
|
||||
publish = false
|
||||
license.workspace = true
|
||||
description.workspace = true
|
||||
|
||||
@@ -21,7 +21,41 @@ import { BlobFile, blob, coerceBlobValue, isBlobField } from "../lancedb/blob";
|
||||
import { EmbeddingFunction, LanceSchema } from "../lancedb/embedding";
|
||||
import { getRegistry } from "../lancedb/embedding/registry";
|
||||
|
||||
function legacyBlob(name: string): Field {
|
||||
const field = blob(name);
|
||||
return new Field(
|
||||
name,
|
||||
new Struct((field.type as Struct).children.slice(0, 2)),
|
||||
field.nullable,
|
||||
field.metadata,
|
||||
);
|
||||
}
|
||||
|
||||
describe("blob()", () => {
|
||||
it("declares the four-field range layout", () => {
|
||||
const field = blob("image", {
|
||||
nullable: false,
|
||||
inlineSizeThreshold: 12,
|
||||
});
|
||||
expect(field.nullable).toBe(false);
|
||||
expect(isBlobField(field)).toBe(true);
|
||||
expect(
|
||||
field.metadata.get("lance-encoding:blob-inline-size-threshold"),
|
||||
).toBe("12");
|
||||
expect(
|
||||
(field.type as Struct).children.map((f) => [
|
||||
f.name,
|
||||
f.type.toString(),
|
||||
f.nullable,
|
||||
]),
|
||||
).toEqual([
|
||||
["data", "LargeBinary", true],
|
||||
["uri", "Utf8", true],
|
||||
["position", "Uint64", true],
|
||||
["size", "Uint64", true],
|
||||
]);
|
||||
});
|
||||
|
||||
it("marks the field as lance.blob.v2", () => {
|
||||
const field = blob("image", { nullable: false });
|
||||
expect(field.nullable).toBe(false);
|
||||
@@ -156,7 +190,7 @@ describe("coerceBlobValue", () => {
|
||||
[
|
||||
"unknown struct field",
|
||||
{ data: Buffer.from("y"), mime: "image/png" },
|
||||
/Blob struct values only support 'data' and 'uri', got 'mime'/,
|
||||
/Blob struct values only support .*got 'mime'/,
|
||||
],
|
||||
[
|
||||
"misspelled struct field",
|
||||
@@ -174,6 +208,158 @@ describe("coerceBlobValue", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("external blob ranges", () => {
|
||||
const uri = "s3://bucket/key";
|
||||
|
||||
it.each([0, 4, 0n, 4n, 9007199254740993n])(
|
||||
"preserves position %s",
|
||||
(position) => {
|
||||
expect(coerceBlobValue({ uri, position, size: 6 })).toEqual({
|
||||
data: null,
|
||||
uri,
|
||||
position: BigInt(position),
|
||||
size: 6n,
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it.each([
|
||||
[{ uri, position: 0 }, /both position and size/],
|
||||
[{ uri, size: 6 }, /both position and size/],
|
||||
[{ uri, position: null, size: 6 }, /both position and size/],
|
||||
[{ uri, position: 0, size: 0 }, /size must be positive/],
|
||||
[{ uri, position: -1, size: 6 }, /position must be non-negative/],
|
||||
[{ uri, position: 0, size: -1n }, /size must be positive/],
|
||||
[{ uri, position: 0.5, size: 6 }, /safe integer or bigint/],
|
||||
[
|
||||
{ uri, position: Number.MAX_SAFE_INTEGER + 1, size: 6 },
|
||||
/safe integer or bigint/,
|
||||
],
|
||||
[{ uri, position: 0, size: Number.NaN }, /safe integer or bigint/],
|
||||
[{ uri, position: "4", size: 6 }, /safe integer or bigint/],
|
||||
[{ uri, position: false, size: 6 }, /safe integer or bigint/],
|
||||
[{ uri, position: 1n << 64n, size: 6 }, /fit in uint64/],
|
||||
[{ uri, position: 0xffffffffffffffffn, size: 1 }, /overflows uint64/],
|
||||
[{ data: Buffer.from("x"), position: 0, size: 6 }, /require a uri/],
|
||||
])("rejects malformed input %p", (value, message) => {
|
||||
expect(() => coerceBlobValue(value)).toThrow(message);
|
||||
});
|
||||
|
||||
it("preserves ranges while normalizing a URL held by getters", () => {
|
||||
class Range {
|
||||
get uri() {
|
||||
return new URL(uri);
|
||||
}
|
||||
get position() {
|
||||
return 4;
|
||||
}
|
||||
get size() {
|
||||
return 6;
|
||||
}
|
||||
}
|
||||
expect(coerceBlobValue(new Range())).toEqual({
|
||||
data: null,
|
||||
uri,
|
||||
position: 4n,
|
||||
size: 6n,
|
||||
});
|
||||
});
|
||||
|
||||
it("writes ranges and mixed whole-file, inline, and null values to Arrow", () => {
|
||||
const schema = new Schema([blob("image")]);
|
||||
const rows = [
|
||||
null,
|
||||
Buffer.from("x"),
|
||||
uri,
|
||||
{ uri, position: 4, size: 6 },
|
||||
].map((image) => ({ image }));
|
||||
const column = makeArrowTable(rows, { schema }).getChild("image")!;
|
||||
expect(column.get(0)).toBeNull();
|
||||
expect(Buffer.from(column.getChild("data")!.get(1))).toEqual(
|
||||
Buffer.from("x"),
|
||||
);
|
||||
expect(column.getChild("uri")!.get(2)).toBe(uri);
|
||||
expect(column.getChild("position")!.toArray()).toEqual(
|
||||
new BigUint64Array([0n, 0n, 0n, 4n]),
|
||||
);
|
||||
expect(column.getChild("position")!.get(2)).toBeNull();
|
||||
expect(column.getChild("size")!.get(3)).toBe(6n);
|
||||
});
|
||||
|
||||
it.each([
|
||||
Buffer.from("inline"),
|
||||
uri,
|
||||
{ uri, position: null, size: null },
|
||||
null,
|
||||
])("retains the full declared range layout when seeded with %p", (image) => {
|
||||
const schema = new Schema([blob("image")]);
|
||||
const table = makeArrowTable([{ image }], { schema });
|
||||
expect(table.schema).toEqual(schema);
|
||||
expect(table.getChild("image")!.getChild("position")!.get(0)).toBeNull();
|
||||
});
|
||||
|
||||
it.each([0, 4])(
|
||||
"rejects position %s on a minimal schema before dropping keys",
|
||||
(position) => {
|
||||
expect(() =>
|
||||
makeArrowTable(
|
||||
[
|
||||
{ image: Buffer.from("inline") },
|
||||
{ image: { uri, position, size: 6 } },
|
||||
],
|
||||
{ schema: new Schema([legacyBlob("image")]) },
|
||||
),
|
||||
).toThrow(
|
||||
/field image at row 1: Blob storage schema cannot store byte ranges/,
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it("accepts omitted or null range fields on the minimal schema", () => {
|
||||
const column = makeArrowTable(
|
||||
[{ image: { uri, position: null, size: null } }],
|
||||
{
|
||||
schema: new Schema([legacyBlob("image")]),
|
||||
},
|
||||
).getChild("image")!;
|
||||
expect(column.getChild("uri")!.get(0)).toBe(uri);
|
||||
});
|
||||
|
||||
it("validates ranges nested in structs and lists against their storage fields", () => {
|
||||
const value = { uri, position: 4, size: 6 };
|
||||
for (const field of [legacyBlob("image"), blob("image")]) {
|
||||
const schema = new Schema([
|
||||
new Field("info", new Struct([field]), true),
|
||||
new Field("images", new List(field), true),
|
||||
]);
|
||||
if ((field.type as Struct).children.length === 2) {
|
||||
expect(() =>
|
||||
makeArrowTable([{ info: { image: value } }], { schema }),
|
||||
).toThrow(/field info.image.*storage schema/);
|
||||
expect(() => makeArrowTable([{ images: [value] }], { schema })).toThrow(
|
||||
/field images\[0\].*storage schema/,
|
||||
);
|
||||
} else {
|
||||
const table = makeArrowTable(
|
||||
[{ info: { image: value }, images: [value] }],
|
||||
{ schema },
|
||||
);
|
||||
const info = table.getChild("info")!.get(0) as {
|
||||
image: { position: bigint; size: bigint };
|
||||
};
|
||||
expect(info.image.position).toBe(4n);
|
||||
const images = Array.from(
|
||||
table.getChild("images")!.get(0) as Iterable<{
|
||||
position: bigint;
|
||||
size: bigint;
|
||||
}>,
|
||||
);
|
||||
expect(images[0].size).toBe(6n);
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("resolveBlobInputs", () => {
|
||||
it("reads Blob and File values in blob columns only", async () => {
|
||||
const schema = new Schema([
|
||||
|
||||
@@ -12,6 +12,7 @@ import {
|
||||
OAuthConfig,
|
||||
OAuthFlowType,
|
||||
TlsConfig,
|
||||
blob,
|
||||
connect,
|
||||
} from "../lancedb";
|
||||
import {
|
||||
@@ -19,7 +20,10 @@ import {
|
||||
Field,
|
||||
Int64,
|
||||
RecordBatch,
|
||||
RecordBatchReader,
|
||||
Schema,
|
||||
Struct,
|
||||
makeArrowTable,
|
||||
tableToIPC,
|
||||
} from "../lancedb/arrow";
|
||||
import {
|
||||
@@ -94,6 +98,137 @@ async function withMockDatabase(
|
||||
}
|
||||
|
||||
describe("remote connection", () => {
|
||||
it.each([false, true])(
|
||||
"normalizes remote blob writes (range target: %s)",
|
||||
async (wide) => {
|
||||
const field = (full: boolean) => {
|
||||
const f = blob("image");
|
||||
return full
|
||||
? f
|
||||
: new Field(
|
||||
f.name,
|
||||
new Struct((f.type as Struct).children.slice(0, 2)),
|
||||
f.nullable,
|
||||
f.metadata,
|
||||
);
|
||||
};
|
||||
const schema = (full: boolean) =>
|
||||
new Schema([new Field("id", new Int64(), true), field(full)]);
|
||||
const writes: { path: string; schema: Schema }[] = [];
|
||||
let storedWide = wide;
|
||||
const describe = () => ({
|
||||
version: writes.length + 1,
|
||||
schema: {
|
||||
fields: [
|
||||
{ name: "id", type: { type: "int64" }, nullable: true },
|
||||
{
|
||||
name: "image",
|
||||
nullable: true,
|
||||
metadata: { "ARROW:extension:name": "lance.blob.v2" },
|
||||
type: {
|
||||
type: "struct",
|
||||
fields: [
|
||||
{
|
||||
name: "data",
|
||||
nullable: true,
|
||||
type: { type: "large_binary" },
|
||||
},
|
||||
{ name: "uri", nullable: true, type: { type: "string" } },
|
||||
...(storedWide
|
||||
? ["position", "size"].map((name) => ({
|
||||
name,
|
||||
nullable: true,
|
||||
type: { type: "uint64" },
|
||||
}))
|
||||
: []),
|
||||
],
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
await withMockDatabase(
|
||||
(req, res) => {
|
||||
if (req.url?.includes("/describe/")) {
|
||||
res
|
||||
.writeHead(200, { "content-type": "application/json" })
|
||||
.end(JSON.stringify(describe()));
|
||||
return;
|
||||
}
|
||||
const chunks: Buffer[] = [];
|
||||
// Failed streaming input aborts before end; only complete bodies commit.
|
||||
req.on("error", () => {
|
||||
/* Aborted inputs must not commit. */
|
||||
});
|
||||
req.on("data", (chunk) => chunks.push(chunk));
|
||||
req.on("end", () => {
|
||||
const reader = RecordBatchReader.from(Buffer.concat(chunks)).open();
|
||||
// Only read the schema: Rust separately checks compressed wire values.
|
||||
writes.push({ path: req.url ?? "", schema: reader.schema });
|
||||
storedWide =
|
||||
(reader.schema.fields[1].type as Struct).children.length === 4;
|
||||
res.writeHead(200, { "content-type": "application/json" }).end(
|
||||
JSON.stringify({
|
||||
version: writes.length + 1,
|
||||
// biome-ignore lint/style/useNamingConvention: server wire format
|
||||
num_inserted_rows: 1,
|
||||
}),
|
||||
);
|
||||
});
|
||||
},
|
||||
async (db) => {
|
||||
const table = await db.openTable("test");
|
||||
const input = makeArrowTable(
|
||||
[{ id: 1n, image: Buffer.from("payload") }],
|
||||
{ schema: schema(!wide) },
|
||||
);
|
||||
await table.add(input);
|
||||
await table
|
||||
.mergeInsert("id")
|
||||
.whenMatchedUpdateAll()
|
||||
.whenNotMatchedInsertAll()
|
||||
.execute(input);
|
||||
expect(writes).toHaveLength(2);
|
||||
for (const write of writes)
|
||||
expect(write.schema).toEqual(schema(wide));
|
||||
expect(writes[0].path).toContain("/insert/");
|
||||
expect(writes[1].path).toContain("/merge_insert/");
|
||||
if (!wide) {
|
||||
const invalid = makeArrowTable(
|
||||
[
|
||||
{
|
||||
id: 1n,
|
||||
image: {
|
||||
uri: "s3://registered/payload",
|
||||
position: 4,
|
||||
size: 6,
|
||||
},
|
||||
},
|
||||
],
|
||||
{ schema: schema(true) },
|
||||
);
|
||||
await expect(table.add(invalid)).rejects.toThrow(
|
||||
/cannot store byte ranges/,
|
||||
);
|
||||
await expect(
|
||||
table
|
||||
.mergeInsert("id")
|
||||
.whenMatchedUpdateAll()
|
||||
.whenNotMatchedInsertAll()
|
||||
.execute(invalid),
|
||||
).rejects.toThrow(/cannot store byte ranges/);
|
||||
expect(writes).toHaveLength(2);
|
||||
}
|
||||
await table.add(input, { mode: "overwrite" });
|
||||
expect(writes).toHaveLength(3);
|
||||
expect(writes[2].path).toContain("mode=overwrite");
|
||||
expect(writes[2].schema).toEqual(schema(!wide));
|
||||
expect(await table.schema()).toEqual(schema(!wide));
|
||||
},
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it("reports unsupported operations with backend-neutral errors", async () => {
|
||||
await withMockDatabase(
|
||||
(req, res) => {
|
||||
|
||||
@@ -2593,6 +2593,16 @@ describe("when dealing with versioning", () => {
|
||||
});
|
||||
});
|
||||
|
||||
function legacyBlob(name: string): Field {
|
||||
const field = blob(name);
|
||||
return new Field(
|
||||
name,
|
||||
new Struct((field.type as Struct).children.slice(0, 2)),
|
||||
field.nullable,
|
||||
field.metadata,
|
||||
);
|
||||
}
|
||||
|
||||
describe("when dealing with blob columns", () => {
|
||||
let tmpDir: tmp.DirResult;
|
||||
beforeEach(() => {
|
||||
@@ -2602,6 +2612,89 @@ describe("when dealing with blob columns", () => {
|
||||
tmpDir.removeCallback();
|
||||
});
|
||||
|
||||
it.each([false, true])(
|
||||
"normalizes Arrow blob layouts for append and merge (range target: %s)",
|
||||
async (wide) => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = (full: boolean) =>
|
||||
new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
full ? blob("image") : legacyBlob("image"),
|
||||
]);
|
||||
const table = await db.createTable(
|
||||
"layouts",
|
||||
[{ id: 1n, image: Buffer.from("old") }],
|
||||
{ schema: schema(wide) },
|
||||
);
|
||||
await table.add(
|
||||
makeArrowTable([{ id: 2n, image: Buffer.from("appended") }], {
|
||||
schema: schema(!wide),
|
||||
}),
|
||||
);
|
||||
await table
|
||||
.mergeInsert("id")
|
||||
.whenMatchedUpdateAll()
|
||||
.whenNotMatchedInsertAll()
|
||||
.execute(
|
||||
makeArrowTable(
|
||||
[
|
||||
{ id: 1n, image: Buffer.from("updated") },
|
||||
{ id: 3n, image: Buffer.from("inserted") },
|
||||
],
|
||||
{ schema: schema(!wide) },
|
||||
),
|
||||
);
|
||||
expect(await table.schema()).toEqual(schema(wide));
|
||||
const rows = await table.query().withRowId().toArray();
|
||||
rows.sort((a, b) => Number(a.id - b.id));
|
||||
expect(
|
||||
await table.fetchBlobs(
|
||||
"image",
|
||||
rows.map((row) => row._rowid as bigint),
|
||||
),
|
||||
).toEqual([
|
||||
Buffer.from("updated"),
|
||||
Buffer.from("appended"),
|
||||
Buffer.from("inserted"),
|
||||
]);
|
||||
if (!wide) {
|
||||
const bad = makeArrowTable(
|
||||
[
|
||||
{
|
||||
id: 1n,
|
||||
image: { uri: "s3://registered/file", position: 0, size: 6 },
|
||||
},
|
||||
],
|
||||
{ schema: schema(true) },
|
||||
);
|
||||
const version = await table.version();
|
||||
await expect(table.add(bad)).rejects.toThrow(
|
||||
/cannot store byte ranges/,
|
||||
);
|
||||
await expect(
|
||||
table
|
||||
.mergeInsert("id")
|
||||
.whenMatchedUpdateAll()
|
||||
.whenNotMatchedInsertAll()
|
||||
.execute(bad),
|
||||
).rejects.toThrow(/cannot store byte ranges/);
|
||||
expect(await table.version()).toBe(version);
|
||||
expect(await table.countRows()).toBe(3);
|
||||
}
|
||||
const replacement = makeArrowTable(
|
||||
[{ id: 4n, image: Buffer.from("replacement") }],
|
||||
{ schema: schema(!wide) },
|
||||
);
|
||||
await table.add(replacement, { mode: "overwrite" });
|
||||
expect(await table.schema()).toEqual(schema(!wide));
|
||||
expect(await table.countRows()).toBe(1);
|
||||
const [row] = await table.query().withRowId().toArray();
|
||||
expect(await table.fetchBlobs("image", [row._rowid as bigint])).toEqual([
|
||||
Buffer.from("replacement"),
|
||||
]);
|
||||
},
|
||||
);
|
||||
|
||||
it("rejects unsupported blob values in createTable and add", async () => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = new Schema([
|
||||
@@ -2678,6 +2771,100 @@ describe("when dealing with blob columns", () => {
|
||||
expect(bytes[0]).toEqual(payload);
|
||||
});
|
||||
|
||||
it.each([0, 4])(
|
||||
"fetches the exact external range at position %s after reopening",
|
||||
async (position) => {
|
||||
const { db, uri, payload } = await openExternalBlobTable();
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
blob("payload"),
|
||||
]);
|
||||
const rows: { id: bigint; payload: BlobInput }[] = [
|
||||
{ id: 1n, payload: Buffer.from("inline") },
|
||||
{ id: 2n, payload: { uri: new URL(uri), position, size: 6 } },
|
||||
{ id: 3n, payload: uri },
|
||||
{ id: 4n, payload: null },
|
||||
];
|
||||
const table = await db.createTable("ranges", rows.slice(0, 1), {
|
||||
schema,
|
||||
});
|
||||
await table.add(rows.slice(1), { allowExternalBlobOutsideBases: true });
|
||||
table.close();
|
||||
const reopened = await db.openTable("ranges");
|
||||
const results = await reopened.query().withRowId().toArray();
|
||||
results.sort((a, b) => Number(a.id - b.id));
|
||||
expect(results[1].payload.position).toBe(BigInt(position));
|
||||
expect(results[1].payload.size).toBe(6n);
|
||||
expect(
|
||||
await reopened.fetchBlobs(
|
||||
"payload",
|
||||
results.map((r) => r._rowid as bigint),
|
||||
),
|
||||
).toEqual([
|
||||
Buffer.from("inline"),
|
||||
payload.subarray(position, position + 6),
|
||||
payload,
|
||||
null,
|
||||
]);
|
||||
},
|
||||
);
|
||||
|
||||
it.each([0, 4])(
|
||||
"rejects ranges at position %s on reopened minimal tables without writing",
|
||||
async (position) => {
|
||||
const { db, uri, payload } = await openExternalBlobTable();
|
||||
const table = await db.createEmptyTable(
|
||||
"legacy_blobs",
|
||||
new Schema([new Field("id", new Int64(), true), legacyBlob("payload")]),
|
||||
);
|
||||
table.close();
|
||||
const reopened = await db.openTable("legacy_blobs");
|
||||
await expect(
|
||||
reopened.add(
|
||||
[
|
||||
{ id: 1n, payload: Buffer.from("inline") },
|
||||
{ id: 2n, payload: { uri, position, size: 6 } },
|
||||
],
|
||||
{ allowExternalBlobOutsideBases: true },
|
||||
),
|
||||
).rejects.toThrow(/storage schema cannot store byte ranges/);
|
||||
expect(await reopened.countRows()).toBe(0);
|
||||
await reopened.add(
|
||||
[
|
||||
{ id: 3n, payload: uri },
|
||||
{ id: 4n, payload: Buffer.from("inline") },
|
||||
],
|
||||
{ allowExternalBlobOutsideBases: true },
|
||||
);
|
||||
const rows = await reopened.query().withRowId().toArray();
|
||||
rows.sort((a, b) => Number(a.id - b.id));
|
||||
expect(
|
||||
await reopened.fetchBlobs(
|
||||
"payload",
|
||||
rows.map((r) => r._rowid as bigint),
|
||||
),
|
||||
).toEqual([payload, Buffer.from("inline")]);
|
||||
},
|
||||
);
|
||||
|
||||
it.each([{ position: 0 }, { size: 6 }, { position: 0, size: 0 }])(
|
||||
"rejects partial or empty ranges without writing: %p",
|
||||
async (range) => {
|
||||
const { db, uri } = await openExternalBlobTable();
|
||||
const table = await db.createEmptyTable(
|
||||
"invalid_ranges",
|
||||
new Schema([blob("payload")]),
|
||||
);
|
||||
await expect(
|
||||
table.add(
|
||||
[{ payload: Buffer.from("inline") }, { payload: { uri, ...range } }],
|
||||
{ allowExternalBlobOutsideBases: true },
|
||||
),
|
||||
).rejects.toThrow(/both position and size|size must be positive/);
|
||||
expect(await table.countRows()).toBe(0);
|
||||
},
|
||||
);
|
||||
|
||||
it("applies the external blob opt-in only to the current add", async () => {
|
||||
const { table, uri } = await openExternalBlobTable();
|
||||
|
||||
|
||||
+27
-7
@@ -456,14 +456,29 @@ 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
|
||||
// storage 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) => {
|
||||
data = mapBlobInputs(data, schema, (value, field, row, storageField) => {
|
||||
if (value === undefined) {
|
||||
return value;
|
||||
}
|
||||
try {
|
||||
return coerceBlobValue(value);
|
||||
const normalized = coerceBlobValue(value, storageField);
|
||||
if (
|
||||
normalized === null ||
|
||||
!(storageField.type instanceof Struct) ||
|
||||
storageField.type.children.length === 2
|
||||
) {
|
||||
return normalized;
|
||||
}
|
||||
// Keep even unused range children during schema inference. Otherwise
|
||||
// creating a range-capable table with inline data would lose them.
|
||||
return Object.fromEntries(
|
||||
storageField.type.children.map((child) => [
|
||||
child.name,
|
||||
normalized[child.name as keyof typeof normalized] ?? null,
|
||||
]),
|
||||
);
|
||||
} catch (e) {
|
||||
throw new Error(
|
||||
`Invalid value for blob field ${field} at row ${row}: ${(e as Error).message}`,
|
||||
@@ -533,7 +548,12 @@ function containsBlobField(field: Field): boolean {
|
||||
);
|
||||
}
|
||||
|
||||
type BlobInputVisitor = (value: unknown, field: string, row: number) => unknown;
|
||||
type BlobInputVisitor = (
|
||||
value: unknown,
|
||||
field: string,
|
||||
row: number,
|
||||
storageField: Field,
|
||||
) => unknown;
|
||||
|
||||
/**
|
||||
* Calls `visit` on every value that lands in a blob column of `schema`,
|
||||
@@ -592,7 +612,7 @@ function mapBlobField(
|
||||
visit: BlobInputVisitor,
|
||||
): unknown {
|
||||
if (isBlobField(field)) {
|
||||
return visit(value, label, row);
|
||||
return visit(value, label, row, field);
|
||||
}
|
||||
if (field.type instanceof Struct && isObject(value)) {
|
||||
return mapBlobFields(value, field.type.children, `${label}.`, row, visit);
|
||||
@@ -697,11 +717,11 @@ function transposeData(
|
||||
const valuesPath = [...path, field.name];
|
||||
if (isBlobField(field) && field.type instanceof Struct) {
|
||||
const blobRows = data.map((datum) =>
|
||||
coerceBlobValue(valueAtPath(datum, valuesPath)),
|
||||
coerceBlobValue(valueAtPath(datum, valuesPath), field),
|
||||
);
|
||||
const childVectors = field.type.children.map((child) => {
|
||||
const values = blobRows.map((row) =>
|
||||
row == null ? null : (row[child.name as "data" | "uri"] ?? null),
|
||||
row == null ? null : (row[child.name as keyof typeof row] ?? null),
|
||||
);
|
||||
return makeVector(values, child.type, undefined, child.nullable);
|
||||
});
|
||||
|
||||
+91
-8
@@ -1,7 +1,7 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import { Field, LargeBinary, Struct, Utf8 } from "apache-arrow";
|
||||
import { Field, LargeBinary, Struct, Uint64, Utf8 } from "apache-arrow";
|
||||
import { BlobFile as NativeBlobFile } from "./native";
|
||||
|
||||
const BLOB_V2_EXTENSION_NAME = "lance.blob.v2";
|
||||
@@ -28,6 +28,9 @@ export type BlobUri = string | URL;
|
||||
*
|
||||
* 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.
|
||||
* External byte ranges require both
|
||||
* `position` (non-negative) and `size` (positive). Use safe integer numbers or
|
||||
* `bigint` values for these unsigned 64-bit byte counts.
|
||||
*
|
||||
* @example
|
||||
* ```ts
|
||||
@@ -45,8 +48,14 @@ export type BlobUri = string | URL;
|
||||
export type BlobInput =
|
||||
| BlobData
|
||||
| BlobUri
|
||||
| { data: BlobData; uri?: null }
|
||||
| { data?: null; uri: BlobUri }
|
||||
| { data: BlobData; uri?: null; position?: null; size?: null }
|
||||
| { data?: null; uri: BlobUri; position?: null; size?: null }
|
||||
| {
|
||||
data?: null;
|
||||
uri: BlobUri;
|
||||
position: number | bigint;
|
||||
size: number | bigint;
|
||||
}
|
||||
| null
|
||||
| undefined;
|
||||
|
||||
@@ -54,6 +63,8 @@ export type BlobInput =
|
||||
export type BlobValue = {
|
||||
data: Buffer | Uint8Array | null;
|
||||
uri: string | null;
|
||||
position?: bigint;
|
||||
size?: bigint;
|
||||
};
|
||||
|
||||
export type BlobOptions = {
|
||||
@@ -78,6 +89,15 @@ export type BlobOptions = {
|
||||
|
||||
/**
|
||||
* Declares a `lance.blob.v2` column.
|
||||
* Uses nullable `data`, `uri`, `position`, and `size` fields for inline bytes,
|
||||
* whole external files, and external byte ranges. Set both `position` and
|
||||
* positive `size` for a range, or omit both for a whole file.
|
||||
*
|
||||
* Existing two-field columns remain writable. Appends and merge inserts fill
|
||||
* missing range fields with nulls for four-field targets and drop all-null
|
||||
* range fields for two-field targets. Non-null ranges require a four-field
|
||||
* target. Overwrite with Arrow input replaces the table's schema; row objects
|
||||
* are converted using the current table schema.
|
||||
*
|
||||
* Query results are descriptors, not payload bytes. Use {@link Table.fetchBlobs}
|
||||
* or {@link Table.fetchBlobFiles} to read bytes.
|
||||
@@ -142,6 +162,8 @@ export function blob(name: string, options: BlobOptions = {}): Field {
|
||||
new Struct([
|
||||
new Field("data", new LargeBinary(), true),
|
||||
new Field("uri", new Utf8(), true),
|
||||
new Field("position", new Uint64(), true),
|
||||
new Field("size", new Uint64(), true),
|
||||
]),
|
||||
options.nullable ?? true,
|
||||
metadata,
|
||||
@@ -317,6 +339,8 @@ export function normalizeBlobInput(value: unknown): unknown {
|
||||
...value,
|
||||
data: value.data,
|
||||
uri: value.uri,
|
||||
position: value.position,
|
||||
size: value.size,
|
||||
};
|
||||
let changed = false;
|
||||
if (value.data instanceof ArrayBuffer) {
|
||||
@@ -355,10 +379,19 @@ export function withBlobBytes(value: unknown, bytes: Uint8Array): unknown {
|
||||
return bytes;
|
||||
}
|
||||
const record = value as Record<string, unknown>;
|
||||
return { ...record, uri: record.uri, data: bytes };
|
||||
return {
|
||||
...record,
|
||||
uri: record.uri,
|
||||
position: record.position,
|
||||
size: record.size,
|
||||
data: bytes,
|
||||
};
|
||||
}
|
||||
|
||||
export function coerceBlobValue(input: unknown): BlobValue | null {
|
||||
export function coerceBlobValue(
|
||||
input: unknown,
|
||||
field?: Field,
|
||||
): BlobValue | null {
|
||||
const value = normalizeBlobInput(input);
|
||||
if (value == null) {
|
||||
return null;
|
||||
@@ -385,11 +418,11 @@ export function coerceBlobValue(input: unknown): BlobValue | null {
|
||||
);
|
||||
}
|
||||
const unknownKeys = Object.keys(record).filter(
|
||||
(key) => key !== "data" && key !== "uri",
|
||||
(key) => !["data", "uri", "position", "size"].includes(key),
|
||||
);
|
||||
if (unknownKeys.length > 0) {
|
||||
throw new Error(
|
||||
`Blob struct values only support 'data' and 'uri', got ${unknownKeys
|
||||
`Blob struct values only support 'data', 'uri', 'position', and 'size', got ${unknownKeys
|
||||
.map((key) => `'${key}'`)
|
||||
.join(", ")}`,
|
||||
);
|
||||
@@ -416,13 +449,63 @@ export function coerceBlobValue(input: unknown): BlobValue | null {
|
||||
"Blob struct values must set exactly one of 'data' or 'uri'",
|
||||
);
|
||||
}
|
||||
const hasPosition = record.position != null;
|
||||
const hasSize = record.size != null;
|
||||
if (hasPosition !== hasSize) {
|
||||
throw new Error(
|
||||
"External blob must set both position and size, or neither",
|
||||
);
|
||||
}
|
||||
if (hasPosition) {
|
||||
if (uriValue == null) {
|
||||
throw new Error("External blob byte ranges require a uri");
|
||||
}
|
||||
const position = blobRangeInteger(record.position, "position", 0n);
|
||||
const size = blobRangeInteger(record.size, "size", 1n);
|
||||
if (position + size > 0xffffffffffffffffn) {
|
||||
throw new Error("External blob range overflows uint64");
|
||||
}
|
||||
if (
|
||||
field !== undefined &&
|
||||
(!(field.type instanceof Struct) ||
|
||||
["position", "size"].some(
|
||||
(name) =>
|
||||
!field.type.children.some((child: Field) => child.name === name),
|
||||
))
|
||||
) {
|
||||
throw new Error(
|
||||
"Blob storage schema cannot store byte ranges. Use a four-field blob() schema to store position and size",
|
||||
);
|
||||
}
|
||||
return { data: null, uri: uriValue, position, size };
|
||||
}
|
||||
return { data: bytes, uri: uriValue };
|
||||
}
|
||||
throw new Error(
|
||||
"Blob column values must be Buffer, Uint8Array, ArrayBuffer, Blob, a URI string or URL, null, or { data?, uri? }",
|
||||
"Blob column values must be Buffer, Uint8Array, ArrayBuffer, Blob, a URI string or URL, null, or { data?, uri?, position?, size? }",
|
||||
);
|
||||
}
|
||||
|
||||
function blobRangeInteger(
|
||||
value: unknown,
|
||||
name: string,
|
||||
minimum: bigint,
|
||||
): bigint {
|
||||
if (
|
||||
typeof value !== "bigint" &&
|
||||
(typeof value !== "number" || !Number.isSafeInteger(value))
|
||||
) {
|
||||
throw new Error(`Blob ${name} must be a safe integer or bigint`);
|
||||
}
|
||||
const integer = BigInt(value);
|
||||
if (integer < minimum || integer > 0xffffffffffffffffn) {
|
||||
throw new Error(
|
||||
`Blob ${name} must be ${minimum === 0n ? "non-negative" : "positive"} and fit in uint64`,
|
||||
);
|
||||
}
|
||||
return integer;
|
||||
}
|
||||
|
||||
function isBlobLike(value: unknown): value is Blob {
|
||||
return typeof Blob !== "undefined" && value instanceof Blob;
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-darwin-arm64",
|
||||
"version": "0.41.0-beta.7",
|
||||
"version": "0.41.0-beta.9",
|
||||
"os": ["darwin"],
|
||||
"cpu": ["arm64"],
|
||||
"main": "lancedb.darwin-arm64.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-arm64-gnu",
|
||||
"version": "0.41.0-beta.7",
|
||||
"version": "0.41.0-beta.9",
|
||||
"os": ["linux"],
|
||||
"cpu": ["arm64"],
|
||||
"main": "lancedb.linux-arm64-gnu.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-arm64-musl",
|
||||
"version": "0.41.0-beta.7",
|
||||
"version": "0.41.0-beta.9",
|
||||
"os": ["linux"],
|
||||
"cpu": ["arm64"],
|
||||
"main": "lancedb.linux-arm64-musl.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-x64-gnu",
|
||||
"version": "0.41.0-beta.7",
|
||||
"version": "0.41.0-beta.9",
|
||||
"os": ["linux"],
|
||||
"cpu": ["x64"],
|
||||
"main": "lancedb.linux-x64-gnu.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-x64-musl",
|
||||
"version": "0.41.0-beta.7",
|
||||
"version": "0.41.0-beta.9",
|
||||
"os": ["linux"],
|
||||
"cpu": ["x64"],
|
||||
"main": "lancedb.linux-x64-musl.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-win32-arm64-msvc",
|
||||
"version": "0.41.0-beta.7",
|
||||
"version": "0.41.0-beta.9",
|
||||
"os": [
|
||||
"win32"
|
||||
],
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-win32-x64-msvc",
|
||||
"version": "0.41.0-beta.7",
|
||||
"version": "0.41.0-beta.9",
|
||||
"os": ["win32"],
|
||||
"cpu": ["x64"],
|
||||
"main": "lancedb.win32-x64-msvc.node",
|
||||
|
||||
+1
-1
@@ -11,7 +11,7 @@
|
||||
"ann"
|
||||
],
|
||||
"private": false,
|
||||
"version": "0.41.0-beta.7",
|
||||
"version": "0.41.0-beta.9",
|
||||
"main": "dist/index.js",
|
||||
"exports": {
|
||||
".": "./dist/index.js",
|
||||
|
||||
Reference in New Issue
Block a user