chore: merge main to pick up Tokio advisory fix

This commit is contained in:
Xuanwo
2026-10-23 21:48:31 +02:00
135 changed files with 14115 additions and 1011 deletions
+1
View File
@@ -15,6 +15,7 @@ renovate.json
src
lancedb
examples
scripts
nodejs-artifacts
Cargo.toml
biome.json
+332 -56
View File
@@ -5,15 +5,15 @@ import { mkdirSync, readdirSync } from "fs";
import { join } from "path";
import { Field, Float64, Schema } from "apache-arrow";
import * as tmp from "tmp";
import {
Connection,
ListTablesResponse,
Table,
connect,
connectNamespace,
} from "../lancedb";
import { Connection, Table, connect, connectNamespace } from "../lancedb";
import { LocalTable } from "../lancedb/table";
async function collectNames(items: AsyncIterable<string>): Promise<string[]> {
const names: string[] = [];
for await (const name of items) names.push(name);
return names;
}
describe("when connecting", () => {
let tmpDir: tmp.DirResult;
beforeEach(() => {
@@ -36,6 +36,87 @@ describe("when connecting", () => {
});
});
describe.each([
[
"URI overload",
(uri: string, options: { storageOptions?: Record<string, string> }) =>
connect(uri, options),
],
[
"object overload",
async (
uri: string,
options: { storageOptions?: Record<string, string> },
) => {
const input = { uri, ...options };
if (Object.isFrozen(options)) {
Object.freeze(input);
}
const db = await connect(input);
expect(input).toEqual({ uri, ...options });
expect(input.storageOptions).toBe(options.storageOptions);
return db;
},
],
[
"namespace",
(uri: string, options: { storageOptions?: Record<string, string> }) =>
connectNamespace("dir", { root: uri }, options),
],
])("%s connection options", (_name, open) => {
let tmpDir: tmp.DirResult;
beforeEach(() => {
tmpDir = tmp.dirSync({ unsafeCleanup: true });
});
afterEach(() => tmpDir.removeCallback());
it("should accept frozen options without storage options", async () => {
const options = Object.freeze({});
const db = await open(tmpDir.name, options);
try {
await db.createTable("frozen", [{ id: 1 }]);
await expect(db.tableNames()).resolves.toContain("frozen");
expect(options).toEqual({});
} finally {
await db.close();
}
});
it("should accept frozen options with storage options", async () => {
const storageOptions = Object.freeze({
newTableDataStorageVersion: "stable",
});
const options = Object.freeze({ storageOptions });
const db = await open(tmpDir.name, options);
try {
const table = await db.createTable("frozen", [{ id: 1 }]);
await expect(table.countRows()).resolves.toBe(1);
expect(options.storageOptions).toBe(storageOptions);
expect(storageOptions).toEqual({ newTableDataStorageVersion: "stable" });
} finally {
await db.close();
}
});
it("should preserve reused options and their storage options", async () => {
const storageOptions = { newTableDataStorageVersion: "stable" };
const options = { storageOptions };
for (let i = 0; i < 2; i++) {
const db = await open(tmpDir.name, options);
try {
const table = await db.createTable(`reused_${i}`, [{ id: i }]);
await expect(table.countRows()).resolves.toBe(1);
expect(options.storageOptions).toBe(storageOptions);
expect(storageOptions).toEqual({
newTableDataStorageVersion: "stable",
});
} finally {
await db.close();
}
}
});
});
describe("given a connection", () => {
let tmpDir: tmp.DirResult;
let db: Connection;
@@ -54,7 +135,7 @@ describe("given a connection", () => {
await db.close();
expect(db.isOpen()).toBe(false);
await expect(db.tableNames()).rejects.toThrow("Connection is closed");
await expect(db.listTables()).rejects.toThrow("Connection is closed");
expect(() => db.listTables()).toThrow("Connection is closed");
await expect(db.renameTable("a", "b")).rejects.toThrow(
"Connection is closed",
);
@@ -117,6 +198,80 @@ describe("given a connection", () => {
await expect(tbl.countRows()).resolves.toBe(1);
});
describe.each([undefined, []])("root namespace %p", (namespacePath) => {
it("should preserve createTable options after the namespace", async () => {
await db.createTable("test", [{ id: 1 }, { id: 2 }]);
const table = await db.createTable("test", [{ id: 3 }], namespacePath, {
mode: "overwrite",
});
const rows = await table.query().toArray();
expect(rows.map((row) => row.id)).toEqual([3]);
});
it("should preserve createEmptyTable overwrite after the namespace", async () => {
await db.createTable("test", [{ id: 1 }]);
const schema = new Schema([new Field("id", new Float64(), true)]);
const table = await db.createEmptyTable("test", schema, namespacePath, {
mode: "overwrite",
});
await expect(table.countRows()).resolves.toBe(0);
});
it("should preserve createEmptyTable existOk after the namespace", async () => {
await db.createTable("test", [{ id: 1 }]);
const schema = new Schema([new Field("id", new Float64(), true)]);
const table = await db.createEmptyTable("test", schema, namespacePath, {
existOk: true,
});
await expect(table.countRows()).resolves.toBe(1);
});
it("should preserve tableNames options after the namespace", async () => {
for (const name of ["a", "b", "c"]) {
await db.createTable(name, [{ id: 1 }]);
}
await expect(
db.tableNames(namespacePath, { limit: 1, startAfter: "a" }),
).resolves.toEqual(["b"]);
});
it("should preserve listTables options after the namespace", async () => {
for (const name of ["a", "b", "c"]) {
await db.createTable(name, [{ id: 1 }]);
}
const first = db.listTables(namespacePath, { pageLimit: 1 });
expect(await first.next()).toEqual({ done: false, value: "a" });
expect(first.pageToken()).toBeDefined();
const second = db.listTables(namespacePath, {
pageLimit: 1,
pageToken: first.pageToken(),
});
expect(await second.next()).toEqual({ done: false, value: "b" });
});
});
it("should keep the existing table when mode is exist_ok", async () => {
const created = await db.createTable("test", [{ id: 1 }, { id: 2 }], {
mode: "exist_ok",
});
await expect(created.countRows()).resolves.toBe(2);
const reopened = await db.createTable("test", [{ id: 9 }], {
mode: "exist_ok",
});
await expect(reopened.countRows()).resolves.toBe(2);
const rows = (await reopened.query().toArray()) as { id: number }[];
expect(rows.map((row) => row.id).sort()).toEqual([1, 2]);
});
it("should respect limit and page token when listing tables", async () => {
const db = await connect(tmpDir.name);
@@ -144,19 +299,18 @@ describe("given a connection", () => {
await db.createTable("a", [{ id: 1 }]);
await db.createTable("c", [{ id: 1 }]);
const all = await db.listTables();
expect(all.tables).toEqual(["a", "b", "c"]);
expect(all.pageToken).toBeUndefined();
expect(await collectNames(db.listTables())).toEqual(["a", "b", "c"]);
const first = await db.listTables({ limit: 1 });
expect(first.tables).toEqual(["a"]);
expect(first.pageToken).toBeDefined();
const first = db.listTables([], { pageLimit: 1 });
expect(first.numPageResults()).toBe(0);
expect(await first.next()).toEqual({ done: false, value: "a" });
expect(first.pageToken()).toBeDefined();
const second = await db.listTables({
limit: 1,
pageToken: first.pageToken,
const second = db.listTables([], {
pageLimit: 1,
pageToken: first.pageToken(),
});
expect(second.tables).toEqual(["b"]);
expect(await second.next()).toEqual({ done: false, value: "b" });
});
it("should visit every table exactly once when walking pages", async () => {
@@ -167,18 +321,71 @@ describe("given a connection", () => {
await db.createTable(name, [{ id: 1 }]);
}
const seen: string[] = [];
let pageToken: string | undefined = undefined;
do {
const page: ListTablesResponse = await db.listTables({
limit: 2,
pageToken,
});
seen.push(...page.tables);
pageToken = page.pageToken;
} while (pageToken);
expect(await collectNames(db.listTables([], { pageLimit: 2 }))).toEqual(
created,
);
});
expect(seen).toEqual(created);
it.each([2 ** 32, 2 ** 32 + 1, Number.MAX_SAFE_INTEGER])(
"should not wrap a large table listing limit %s",
async (limit) => {
await db.createTable("a", [{ id: 1 }]);
await db.createTable("b", [{ id: 2 }]);
for (const names of [
db.tableNames({ limit }),
db.tableNames([], { limit }),
]) {
await expect(names).resolves.toEqual(["a", "b"]);
}
for (const namespacePath of [undefined, []]) {
expect(() =>
db.listTables(namespacePath, { pageLimit: limit }),
).toThrow(/limit/);
}
},
);
it.each([
-1,
1.5,
Number.NaN,
Number.POSITIVE_INFINITY,
Number.NEGATIVE_INFINITY,
Number.MAX_SAFE_INTEGER + 1,
])("should reject an invalid table listing limit %s", async (limit) => {
await expect(db.tableNames({ limit })).rejects.toThrow(/limit/);
await expect(db.tableNames([], { limit })).rejects.toThrow(/limit/);
for (const namespacePath of [undefined, []]) {
expect(() => db.listTables(namespacePath, { pageLimit: limit })).toThrow(
/limit/,
);
}
});
it("should preserve zero, small and unsigned-32-bit table listing limits", async () => {
await db.createTable("a", [{ id: 1 }]);
await db.createTable("b", [{ id: 2 }]);
for (const [limit, names] of [
[0, []],
[1, ["a"]],
[2 ** 32 - 1, ["a", "b"]],
] as const) {
await expect(db.tableNames({ limit })).resolves.toEqual(names);
}
await expect(db.tableNames()).resolves.toEqual(["a", "b"]);
for (const namespacePath of [undefined, []]) {
for (const pageLimit of [1, 2 ** 31 - 1]) {
expect(
await collectNames(db.listTables(namespacePath, { pageLimit })),
).toEqual(["a", "b"]);
}
for (const pageLimit of [0, 2 ** 31, 2 ** 32 - 1]) {
expect(() => db.listTables(namespacePath, { pageLimit })).toThrow(
/limit/,
);
}
}
expect(await collectNames(db.listTables())).toEqual(["a", "b"]);
});
it("should list all names and use a default page size of 100", async () => {
@@ -191,20 +398,17 @@ describe("given a connection", () => {
}
await expect(db.tableNames()).resolves.toEqual(names);
await expect(db.tableNames({ limit: 0 })).resolves.toEqual([]);
const first = await db.listTables();
expect(first.tables).toEqual(names.slice(0, 100));
expect(first.pageToken).toBeDefined();
const second = await db.listTables({ pageToken: first.pageToken });
expect(second.tables).toEqual(names.slice(100));
expect(second.pageToken).toBeUndefined();
await expect(db.listTables({ limit: 200 })).resolves.toEqual({
tables: names,
pageToken: undefined,
});
await expect(db.listTables({ limit: 0 })).resolves.toEqual({
tables: [],
pageToken: undefined,
});
const first = db.listTables();
const firstPage: string[] = [];
for (let i = 0; i < 100; i++) firstPage.push((await first.next()).value!);
expect(firstPage).toEqual(names.slice(0, 100));
expect(first.pageToken()).toBeDefined();
expect(first.numPageResults()).toBe(0);
expect(await collectNames(first)).toEqual(names.slice(100));
expect(await collectNames(db.listTables([], { pageLimit: 200 }))).toEqual(
names,
);
expect(() => db.listTables([], { pageLimit: 0 })).toThrow(/limit/);
});
it("should list tables in a namespace", async () => {
@@ -215,12 +419,8 @@ describe("given a connection", () => {
await db.createNamespace(["child"]);
await db.createTable("nested", [{ id: 1 }], ["child"]);
await expect(db.listTables(["child"])).resolves.toEqual(
expect.objectContaining({ tables: ["nested"] }),
);
await expect(db.listTables()).resolves.toEqual(
expect.objectContaining({ tables: [] }),
);
expect(await collectNames(db.listTables(["child"]))).toEqual(["nested"]);
expect(await collectNames(db.listTables())).toEqual([]);
});
it("should create tables in v2 mode", async () => {
@@ -293,6 +493,69 @@ describe("given a connection", () => {
});
});
describe.each(["createTable", "createEmptyTable"] as const)(
"%s storage options",
(method) => {
const schema = new Schema([new Field("id", new Float64(), true)]);
it("should accept frozen creation options without storage options", async () => {
const options = Object.freeze({
dataStorageVersion: "stable",
enableV2ManifestPaths: true,
});
const table =
method === "createTable"
? await db.createTable("frozen_options", [{ id: 1 }], options)
: await db.createEmptyTable("frozen_options", schema, options);
expect(await (table as LocalTable).usesV2ManifestPaths()).toBe(true);
expect(options).not.toHaveProperty("storageOptions");
});
it("should accept frozen storage options and preserve override precedence", async () => {
const storageOptions = Object.freeze({
newTableDataStorageVersion: "legacy",
newTableEnableV2ManifestPaths: "false",
});
const options = Object.freeze({
storageOptions,
dataStorageVersion: "stable",
enableV2ManifestPaths: true,
});
const table =
method === "createTable"
? await db.createTable("frozen_storage", [{ id: 1 }], options)
: await db.createEmptyTable("frozen_storage", schema, options);
expect(await (table as LocalTable).usesV2ManifestPaths()).toBe(true);
expect(options.storageOptions).toBe(storageOptions);
expect(storageOptions).toEqual({
newTableDataStorageVersion: "legacy",
newTableEnableV2ManifestPaths: "false",
});
});
it("should not retain a previous override when reusing storage options", async () => {
const storageOptions = { newTableEnableV2ManifestPaths: "false" };
const create = async (name: string, enableV2ManifestPaths?: boolean) =>
method === "createTable"
? db.createTable(name, [{ id: 1 }], {
storageOptions,
enableV2ManifestPaths,
})
: db.createEmptyTable(name, schema, {
storageOptions,
enableV2ManifestPaths,
});
const first = await create("override", true);
expect(await (first as LocalTable).usesV2ManifestPaths()).toBe(true);
const second = await create("without_override");
expect(await (second as LocalTable).usesV2ManifestPaths()).toBe(false);
expect(storageOptions).toEqual({
newTableEnableV2ManifestPaths: "false",
});
});
},
);
it("should be able to migrate tables to the V2 manifest paths", async () => {
const db = await connect(tmpDir.name);
const table = (await db.createEmptyTable(
@@ -451,22 +714,35 @@ describe("namespaces", () => {
it("should list namespaces created at the root", async () => {
await db.createNamespace(["alpha"]);
await db.createNamespace(["beta"]);
const list = await db.listNamespaces();
expect(list.namespaces).toEqual(expect.arrayContaining(["alpha", "beta"]));
const list = await collectNames(db.listNamespaces());
expect(list).toEqual(expect.arrayContaining(["alpha", "beta"]));
});
it("should list child namespaces under a parent", async () => {
await db.createNamespace(["parent"]);
await db.createNamespace(["parent", "child"]);
const list = await db.listNamespaces(["parent"]);
expect(list.namespaces).toContain("child");
const list = await collectNames(db.listNamespaces(["parent"]));
expect(list).toContain("child");
});
it("should resume a namespace listing from its page token", async () => {
await db.createNamespace(["alpha"]);
await db.createNamespace(["beta"]);
const first = db.listNamespaces([], { pageLimit: 1 });
expect(first.numPageResults()).toBe(0);
expect((await first.next()).value).toBe("alpha");
const second = db.listNamespaces([], {
pageLimit: 1,
pageToken: first.pageToken(),
});
expect(await collectNames(second)).toEqual(["beta"]);
});
it("should drop a namespace", async () => {
await db.createNamespace(["ephemeral"]);
await db.dropNamespace(["ephemeral"]);
const list = await db.listNamespaces();
expect(list.namespaces).not.toContain("ephemeral");
const list = await collectNames(db.listNamespaces());
expect(list).not.toContain("ephemeral");
});
it("should raise an error on any namespace op after close", async () => {
@@ -474,7 +750,7 @@ describe("namespaces", () => {
await expect(db.describeNamespace(["foo"])).rejects.toThrow(
"Connection is closed",
);
await expect(db.listNamespaces()).rejects.toThrow("Connection is closed");
expect(() => db.listNamespaces()).toThrow("Connection is closed");
await expect(db.createNamespace(["foo"])).rejects.toThrow(
"Connection is closed",
);
@@ -0,0 +1,66 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
const assert = require("node:assert/strict");
const {
TransformersEmbeddingFunction,
} = require("../../dist/embedding/transformers");
const { getRegistry } = require("../../dist/embedding/registry");
async function main() {
const transformers = await import("@huggingface/transformers");
// Never fetch a real model, including if a loader mock stops intercepting.
transformers.env.allowRemoteModels = false;
transformers.env.allowLocalModels = false;
const modelCalls = [];
const tokenizerCalls = [];
const testCase = Number(process.argv[2]);
const defaults = "Xenova/all-MiniLM-L6-v2";
const cases = [
[{}, defaults, defaults],
[{ model: "test/model" }, "test/model", "test/model"],
[
{ model: "test/model", tokenizer: "test/tokenizer" },
"test/model",
"test/tokenizer",
],
[
{ model: "test/model", tokenizer: "$var:tokenizer" },
"test/model",
"test/tokenizer",
],
[
{ model: "test/model", tokenizer: "test/tokenizer" },
"test/model",
"test/tokenizer",
],
];
assert.ok(
Number.isInteger(testCase) && testCase >= 0 && testCase < cases.length,
);
transformers.AutoModel.from_pretrained = async (...args) => {
modelCalls.push(args);
return {};
};
transformers.AutoTokenizer.from_pretrained = async (...args) => {
tokenizerCalls.push(args);
if (testCase === 4) throw new Error("missing tokenizer");
return {};
};
getRegistry().setVar("tokenizer", "test/tokenizer");
const [options, model, tokenizer] = cases[testCase];
const fn = new TransformersEmbeddingFunction(options);
if (testCase === 4) {
await assert.rejects(fn.init(), /error loading tokenizer test\/tokenizer/);
} else {
await fn.init();
}
assert.deepEqual(modelCalls, [[model, { dtype: "fp32" }]]);
assert.deepEqual(tokenizerCalls, [[tokenizer]]);
assert.deepEqual(fn.toJSON(), options);
}
main().catch((error) => {
console.error(error);
process.exitCode = 1;
});
+47
View File
@@ -0,0 +1,47 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
const {
checkNativeBinary,
findForbiddenUndefinedSymbols,
} = require("../scripts/check-native-symbols.js");
test("detects the unresolved AVX-512 symbol from broken musl binaries", () => {
const output = [
" U napi_create_function",
" U sum_4bit_dist_table_32bytes_batch_avx512",
" U strlen",
].join("\n");
expect(findForbiddenUndefinedSymbols(output)).toEqual([
"sum_4bit_dist_table_32bytes_batch_avx512",
]);
});
test("accepts native binaries without unresolved internal Lance symbols", () => {
const runNm = jest.fn(() => ({
status: 0,
stdout:
" U napi_create_function\n U strlen\n",
stderr: "",
}));
expect(() => checkNativeBinary("lancedb.node", runNm)).not.toThrow();
expect(runNm).toHaveBeenCalledWith(
"nm",
["-D", "--undefined-only", "lancedb.node"],
{ encoding: "utf8" },
);
});
test("rejects native binaries with the unresolved AVX-512 symbol", () => {
const runNm = jest.fn(() => ({
status: 0,
stdout: " U sum_4bit_dist_table_32bytes_batch_avx512\n",
stderr: "",
}));
expect(() => checkNativeBinary("lancedb.node", runNm)).toThrow(
"lancedb.node contains unresolved internal Lance symbols: sum_4bit_dist_table_32bytes_batch_avx512",
);
});
+327
View File
@@ -0,0 +1,327 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import * as tmp from "tmp";
import { type Table, blob, connect, isBlobField } from "../lancedb";
import {
DataType,
Field,
Int64,
List,
Schema,
Struct,
Utf8,
} from "../lancedb/arrow";
describe('nested blobs with blobMode "bytes"', () => {
let tmpDir: tmp.DirResult;
let table: Table;
const caption = new Field(
"caption",
new Utf8(),
false,
new Map([["label", "caption"]]),
);
const info = new Field(
"info",
new Struct([
blob("image"),
caption,
new Field("preview", new Struct([blob("image")]), true),
]),
true,
new Map([["label", "info"]]),
);
const schema = new Schema(
[new Field("id", new Int64(), false), info, blob("cover")],
new Map([["source", "nested-blobs"]]),
);
beforeEach(async () => {
tmpDir = tmp.dirSync({ unsafeCleanup: true });
const db = await connect(tmpDir.name);
table = await db.createTable(
"nested_blobs",
[
{
id: 1n,
info: {
image: Buffer.from("hello"),
caption: "first",
preview: { image: Buffer.from("preview") },
},
cover: Buffer.from("cover"),
},
{ id: 2n, info: null, cover: null },
{
id: 3n,
info: { image: null, caption: "third", preview: null },
cover: null,
},
{
id: 4n,
info: {
image: Buffer.alloc(0),
caption: "fourth",
preview: { image: null },
},
cover: Buffer.alloc(0),
},
],
{ schema },
);
});
afterEach(() => tmpDir.removeCallback());
it.each([false, true])(
"reads struct leaves across batches (select parent: %s)",
async (selectParent) => {
const query = table
.query()
.orderBy({ columnName: "id", ascending: true });
if (selectParent) {
query.select(["info"]);
}
const result = await query.toArrow({
blobMode: "bytes",
maxBatchLength: 1,
});
expect(result.batches.length).toBeGreaterThan(1);
expect(result.schema.fields.map((field) => field.name)).toEqual(
selectParent ? ["info"] : ["id", "info", "cover"],
);
expect(result.schema.metadata).toEqual(schema.metadata);
const infoField = result.schema.fields.find(
(field) => field.name === "info",
)!;
expect(infoField.nullable).toBe(true);
expect(infoField.metadata).toEqual(info.metadata);
const fields = (infoField.type as Struct).children;
expect(DataType.isLargeBinary(fields[0].type)).toBe(true);
expect(isBlobField(fields[0])).toBe(false);
expect(fields[0].metadata.size).toBe(0);
expect(fields[1]).toEqual(caption);
expect(
DataType.isLargeBinary((fields[2].type as Struct).children[0].type),
).toBe(true);
const rows = result.toArray();
expect(Buffer.from(rows[0].info.image).toString()).toBe("hello");
expect(rows[0].info.caption).toBe("first");
expect(Buffer.from(rows[0].info.preview.image).toString()).toBe(
"preview",
);
expect(rows[1].info).toBeNull();
expect(rows[2].info.image).toBeNull();
expect(rows[2].info.caption).toBe("third");
expect(rows[2].info.preview).toBeNull();
expect(Buffer.from(rows[3].info.image)).toEqual(Buffer.alloc(0));
expect(rows[3].info.preview.image).toBeNull();
if (!selectParent) {
expect(Buffer.from(rows[0].cover).toString()).toBe("cover");
expect(rows[1].cover).toBeNull();
}
},
);
it("reads direct leaf and struct aliases", async () => {
const rows = await table
.query()
.orderBy({ columnName: "id", ascending: true })
.select({
image: "info.image",
details: "info",
preview: "info.preview",
caption: "info.caption",
})
.toArray({ blobMode: "bytes", maxBatchLength: 1 });
expect(Object.keys(rows[0])).toEqual([
"image",
"details",
"preview",
"caption",
]);
expect(Buffer.from(rows[0].image).toString()).toBe("hello");
expect(Buffer.from(rows[0].details.image).toString()).toBe("hello");
expect(Buffer.from(rows[0].details.preview.image).toString()).toBe(
"preview",
);
expect(Buffer.from(rows[0].preview.image).toString()).toBe("preview");
expect(rows[0].caption).toBe("first");
expect(rows[1].details).toBeNull();
expect(rows[1].image).toBeNull();
expect(rows[2].image).toBeNull();
expect(Buffer.from(rows[3].image)).toEqual(Buffer.alloc(0));
});
it.each([false, true])(
"preserves a literal scalar alias matching a nested blob path (reverse: %s)",
async (reverse) => {
const projection: [string, string][] = [
["info", "info"],
["info.image", "id"],
];
if (reverse) {
projection.reverse();
}
const [row] = await table
.query()
.where("id = 1")
.select(new Map(projection))
.toArray({ blobMode: "bytes" });
expect(row["info.image"]).toBe(1n);
expect(Buffer.from(row.info.image).toString()).toBe("hello");
},
);
it.each([false, true])(
"materializes every overlapping blob projection (reverse: %s)",
async (reverse) => {
const columns = ["info.image", "info"];
if (reverse) {
columns.reverse();
}
const result = await table
.query()
.orderBy({ columnName: "id", ascending: true })
.select(columns)
.toArrow({ blobMode: "bytes", maxBatchLength: 1 });
expect(result.schema.fields.map((field) => field.name)).toEqual(columns);
expect(DataType.isLargeBinary(result.getChild("info.image")!.type)).toBe(
true,
);
expect(
DataType.isLargeBinary(
result.getChild("info")!.getChild("image")!.type,
),
).toBe(true);
const rows = result.toArray();
expect(Buffer.from(rows[0]["info.image"]).toString()).toBe("hello");
expect(Buffer.from(rows[0].info.image).toString()).toBe("hello");
expect(rows[0].info.caption).toBe("first");
expect(rows[1]["info.image"]).toBeNull();
expect(rows[1].info).toBeNull();
expect(rows[2]["info.image"]).toBeNull();
expect(rows[2].info.image).toBeNull();
expect(Buffer.from(rows[3]["info.image"])).toEqual(Buffer.alloc(0));
expect(Buffer.from(rows[3].info.image)).toEqual(Buffer.alloc(0));
},
);
it("materializes dotted aliases independently of nested fields", async () => {
const [row] = await table
.query()
.where("id = 1")
.select({
details: "info",
"details.image": "info.image",
"details.preview": "info.preview",
"details.preview.image": "info.preview.image",
})
.toArray({ blobMode: "bytes" });
expect(Buffer.from(row.details.image).toString()).toBe("hello");
expect(Buffer.from(row["details.image"]).toString()).toBe("hello");
expect(Buffer.from(row.details.preview.image).toString()).toBe("preview");
expect(Buffer.from(row["details.preview"].image).toString()).toBe(
"preview",
);
expect(Buffer.from(row["details.preview.image"]).toString()).toBe(
"preview",
);
});
it.each([false, true])(
"converts the empty result schema (alias: %s)",
async (alias) => {
const query = table.query().where("id < 0");
if (alias) {
query.select({ image: "info.image", details: "info" });
}
const result = await query.toArrow({ blobMode: "bytes" });
expect(result.numRows).toBe(0);
expect(result.schema.fields.map((field) => field.name)).toEqual(
alias ? ["image", "details"] : ["id", "info", "cover"],
);
const fields = (
result.schema.fields.find(
(field) => field.name === (alias ? "details" : "info"),
)!.type as Struct
).children;
expect(DataType.isLargeBinary(fields[0].type)).toBe(true);
expect(isBlobField(fields[0])).toBe(false);
expect(fields[1]).toEqual(caption);
expect(
DataType.isLargeBinary((fields[2].type as Struct).children[0].type),
).toBe(true);
if (alias) {
expect(DataType.isLargeBinary(result.schema.fields[0].type)).toBe(true);
}
},
);
it("preserves requested row IDs and leaves subsequent descriptors intact", async () => {
const query = table.query().where("id = 1").select(["info"]);
const [bytes] = await query.toArray({ blobMode: "bytes" });
expect(Object.keys(bytes)).toEqual(["info"]);
const [description] = await query.toArray();
expect(Object.keys(description)).toEqual(["info"]);
expect(Number(description.info.image.size)).toBe(5);
const [withRowId] = await query.withRowId().toArray({ blobMode: "bytes" });
expect(typeof withRowId._rowid).toBe("bigint");
expect(await table.fetchBlobs("info.image", [withRowId._rowid])).toEqual([
Buffer.from(withRowId.info.image),
]);
});
it("keeps descriptors in description mode", async () => {
const [description] = await table
.query()
.where("id = 1")
.toArray({ blobMode: "descriptions" });
expect(Number(description.info.image.size)).toBe(5);
});
it("keeps list elements as descriptors alongside struct blob bytes", async () => {
const db = await connect(tmpDir.name);
const table = await db.createTable(
"list_blobs",
[
{
info: { image: Buffer.from("hello"), images: [Buffer.from("list")] },
},
],
{
schema: new Schema([
new Field(
"info",
new Struct([
blob("image"),
new Field("images", new List(blob("item")), true),
]),
true,
),
]),
},
);
const [bytes] = await table.query().toArray({ blobMode: "bytes" });
expect(Buffer.from(bytes.info.image).toString()).toBe("hello");
expect(Number(bytes.info.images.get(0).size)).toBe(4);
const [row] = await table
.query()
.select(["info.images"])
.toArray({ blobMode: "bytes" });
expect(Number(row["info.images"].get(0).size)).toBe(4);
expect(Object.keys(row)).toEqual(["info.images"]);
const result = await table.query().toArrow({ blobMode: "bytes" });
const fields = (result.schema.fields[0].type as Struct).children;
const descriptions = await table.query().toArrow();
expect(fields[1]).toEqual(
(descriptions.schema.fields[0].type as Struct).children[1],
);
expect(DataType.isStruct((fields[1].type as List).children[0].type)).toBe(
true,
);
});
});
+218 -47
View File
@@ -3314,51 +3314,64 @@ describe("when dealing with blob columns", () => {
).rejects.toThrow(/blobMode must be/);
});
it("reruns the query when the table changes before the version check", async () => {
const { table, native, update } = await openVersionedBlobTable();
const liveVersion = native.version.bind(native);
native.version = async () => {
native.version = liveVersion;
await update(2n, "new");
return liveVersion();
};
const [row] = await table.query().toArray({ blobMode: "bytes" });
expect(row.generation).toBe(2n);
expect(Buffer.from(row.image).toString()).toBe("new");
});
it.each([false, true])(
"reruns the query when the table changes before the version check (nested: %s)",
async (nested) => {
const { table, native, update } = await openVersionedBlobTable(nested);
const liveVersion = native.version.bind(native);
native.version = async () => {
native.version = liveVersion;
await update(2n, "new");
return liveVersion();
};
const [row] = await table.query().toArray({ blobMode: "bytes" });
expect(row.generation).toBe(2n);
expect(Buffer.from(nested ? row.info.image : row.image).toString()).toBe(
"new",
);
},
);
it("reads bytes from the query's version when a write lands before the fetch", async () => {
const { table, native, update } = await openVersionedBlobTable();
const liveVersion = native.version.bind(native);
native.version = async () => {
native.version = liveVersion;
const version = await liveVersion();
await update(2n, "new");
return version;
};
const [row] = await table.query().toArray({ blobMode: "bytes" });
expect(row.generation).toBe(1n);
expect(Buffer.from(row.image).toString()).toBe("old");
});
it.each([false, true])(
"reads bytes from the query's version when a write lands before the fetch (nested: %s)",
async (nested) => {
const { table, native, update } = await openVersionedBlobTable(nested);
const liveVersion = native.version.bind(native);
native.version = async () => {
native.version = liveVersion;
const version = await liveVersion();
await update(2n, "new");
return version;
};
const [row] = await table.query().toArray({ blobMode: "bytes" });
expect(row.generation).toBe(1n);
expect(Buffer.from(nested ? row.info.image : row.image).toString()).toBe(
"old",
);
},
);
it("fails when the table changes during every attempt", async () => {
const { table, native, update } = await openVersionedBlobTable();
const liveVersion = native.version.bind(native);
let generation = 1n;
native.version = async () => {
generation += 1n;
await update(generation, `generation ${generation}`);
return liveVersion();
};
try {
await expect(
table.query().toArray({ blobMode: "bytes" }),
).rejects.toThrow(/changed during each of 3 attempts/);
expect(generation).toBe(4n);
} finally {
native.version = liveVersion;
}
});
it.each([false, true])(
"fails when the table changes during every attempt (nested: %s)",
async (nested) => {
const { table, native, update } = await openVersionedBlobTable(nested);
const liveVersion = native.version.bind(native);
let generation = 1n;
native.version = async () => {
generation += 1n;
await update(generation, `generation ${generation}`);
return liveVersion();
};
try {
await expect(
table.query().toArray({ blobMode: "bytes" }),
).rejects.toThrow(/changed during each of 3 attempts/);
expect(generation).toBe(4n);
} finally {
native.version = liveVersion;
}
},
);
it("rejects fetchBlobs on a non-blob column", async () => {
const { table, rowIds } = await openBlobTable();
@@ -3385,6 +3398,106 @@ describe("when dealing with blob columns", () => {
expect(bytes[0]!.equals(payload)).toBe(true);
});
it("returns bytes for blobs nested in structs", async () => {
const { table, alpha, empty } = await openNestedBlobTable();
const rows = await table.query().toArray({ blobMode: "bytes" });
const byId = new Map(rows.map((row) => [Number(row.id), row]));
expect(Buffer.from(byId.get(1)!.info.image)).toEqual(alpha);
expect(byId.get(1)!.info.caption).toBe("one");
expect(byId.get(2)!.info.image).toBeNull();
expect(byId.get(2)!.info.caption).toBe("two");
expect(byId.get(3)!.info).toBeNull();
expect(Buffer.from(byId.get(4)!.info.image)).toEqual(empty);
});
it("lines up nested rows and bytes across single-row batches", async () => {
const { table, alpha, empty } = await openNestedBlobTable();
const result = await table
.query()
.toArrow({ blobMode: "bytes", maxBatchLength: 1 });
expect(result.batches.length).toBe(4);
const byId = new Map(result.toArray().map((row) => [Number(row.id), row]));
expect(Buffer.from(byId.get(1)!.info.image)).toEqual(alpha);
expect(byId.get(1)!.info.caption).toBe("one");
expect(byId.get(2)!.info.image).toBeNull();
expect(byId.get(2)!.info.caption).toBe("two");
expect(byId.get(3)!.info).toBeNull();
expect(Buffer.from(byId.get(4)!.info.image)).toEqual(empty);
});
it("handles struct fields named like Object prototypes", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
new Field(
"info",
new Struct([
new Field("constructor", new Struct([blob("image")]), true),
]),
true,
),
]);
const payload = Buffer.from("proto");
const table = await db.createTable(
"prototype_named_blobs",
[{ id: 1n, info: { constructor: { image: payload } } }],
{ schema },
);
expect(await table.blobColumns()).toEqual(["info.constructor.image"]);
const [row] = await table.query().toArray({ blobMode: "bytes" });
// Arrow's StructRow shadows the `constructor` property, so read the
// field through the JSON form instead of property access.
const info = JSON.parse(JSON.stringify(row.info));
expect(Buffer.from(Object.values<number>(info.constructor.image))).toEqual(
payload,
);
});
it("returns nested blob bytes for a struct selection", async () => {
const { table, alpha } = await openNestedBlobTable();
const [row] = await table
.query()
.where("id = 1")
.select(["info"])
.toArray({ blobMode: "bytes" });
expect(Object.keys(row)).toEqual(["info"]);
expect(Buffer.from(row.info.image)).toEqual(alpha);
expect(row.info.caption).toBe("one");
});
it("returns nested blob bytes through aliases", async () => {
const { table, alpha } = await openNestedBlobTable();
const [row] = await table
.query()
.where("id = 1")
.select(
new Map([
["picture", "info.image"],
["details", "info"],
]),
)
.toArray({ blobMode: "bytes" });
expect(Buffer.from(row.picture)).toEqual(alpha);
expect(Buffer.from(row.details.image)).toEqual(alpha);
expect(row.details.caption).toBe("one");
});
it("returns a nested bytes schema for an empty result", async () => {
const { table } = await openNestedBlobTable();
const result = await table
.query()
.where("id > 100")
.toArrow({ blobMode: "bytes" });
expect(result.numRows).toBe(0);
const field = result.schema.fields.find((f) => f.name === "info")!;
expect(arrow.DataType.isStruct(field.type)).toBe(true);
const image = (field.type as Struct).children.find(
(f) => f.name === "image",
)!;
expect(arrow.DataType.isLargeBinary(image.type)).toBe(true);
expect(isBlobField(image)).toBe(false);
});
it("creates list blobs without advertising unfetchable elements", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([
@@ -3542,18 +3655,28 @@ describe("when dealing with blob columns", () => {
).map((value) => (value == null ? null : Number(value.size)));
}
async function openVersionedBlobTable() {
async function openVersionedBlobTable(nested = false) {
const db = await connect(tmpDir.name, {
storageOptions: { newTableEnableStableRowIds: "true" },
});
const schema = new Schema([
new Field("id", new Int64(), false),
new Field("generation", new Int64(), true),
blob("image"),
nested
? new Field("info", new Struct([blob("image")]), true)
: blob("image"),
]);
const table = await db.createTable(
"versions",
[{ id: 1n, generation: 1n, image: Buffer.from("old") }],
[
{
id: 1n,
generation: 1n,
...(nested
? { info: { image: Buffer.from("old") } }
: { image: Buffer.from("old") }),
},
],
{ schema },
);
const native = (
@@ -3563,7 +3686,15 @@ describe("when dealing with blob columns", () => {
await table
.mergeInsert("id")
.whenMatchedUpdateAll()
.execute([{ id: 1n, generation, image: Buffer.from(image) }]);
.execute([
{
id: 1n,
generation,
...(nested
? { info: { image: Buffer.from(image) } }
: { image: Buffer.from(image) }),
},
]);
};
return { table, native, update };
}
@@ -3593,6 +3724,31 @@ describe("when dealing with blob columns", () => {
return { table, rowIds, alpha, beta };
}
async function openNestedBlobTable() {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
new Field(
"info",
new Struct([blob("image"), new Field("caption", new Utf8(), true)]),
true,
),
]);
const alpha = Buffer.from("alpha");
const empty = Buffer.alloc(0);
const table = await db.createTable(
"nested_blobs",
[{ id: 1n, info: { image: alpha, caption: "one" } }],
{ schema },
);
await table.add([
{ id: 2n, info: { image: null, caption: "two" } },
{ id: 3n, info: null },
{ id: 4n, info: { image: empty, caption: "four" } },
]);
return { table, alpha, empty };
}
async function openExternalBlobTable() {
const db = await connect(path.join(tmpDir.name, "db"));
const payload = Buffer.from("external-payload");
@@ -3742,6 +3898,21 @@ describe("when optimizing a dataset", () => {
expect(stats.prune.oldVersionsRemoved).toBe(2);
});
it("retains tagged versions older than the cleanup cutoff", async () => {
await (await table.tags()).create("keep", 1);
await table.add([{ id: 3 }]);
const cutoff = await nextMillisecond();
const stats = await table.optimize({ cleanupOlderThan: cutoff });
// Versions 2 and 3 are pruned, tagged version 1 is retained
expect(stats.prune.oldVersionsRemoved).toBe(2);
await table.checkout("keep");
expect(await table.countRows()).toBe(1);
await table.checkoutLatest();
expect(await table.countRows()).toBe(3);
});
it("delete unverified", async () => {
const version = await table.version();
const versionFile = `${tmpDir.name}/${table.name}.lance/_versions/${String(
+26
View File
@@ -0,0 +1,26 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import { execFileSync } from "node:child_process";
import { resolve } from "node:path";
describe("Transformers embedding tokenizer selection", () => {
// Run init's native ESM import outside Jest's CommonJS sandbox, like a user.
// The fixture mocks model loading and forbids local/remote model access.
test.each([
["default model", 0],
["default tokenizer", 1],
["custom tokenizer", 2],
["tokenizer registry variable", 3],
["tokenizer loading error", 4],
])("handles %s", (_name, testCase) => {
execFileSync(
process.execPath,
[
resolve(__dirname, "fixtures", "transformers_tokenizer.cjs"),
String(testCase),
],
{ stdio: "pipe", timeout: 15_000 },
);
});
});
+1
View File
@@ -61,6 +61,7 @@ import {
import { inferSchema } from "./schema";
export * from "apache-arrow";
export type { Data as ArrowData } from "apache-arrow";
export type SchemaLike =
| Schema
| {
+58 -118
View File
@@ -30,16 +30,12 @@ import type {
DescribeNamespaceResponse,
DropNamespaceResponse,
JobInfo,
ListNamespacesResponse,
ListTablesResponse,
} from "./native";
import { ViewDescription, viewDescriptionFromNative } from "./view";
export type {
CreateNamespaceResponse,
DescribeNamespaceResponse,
DropNamespaceResponse,
ListNamespacesResponse,
ListTablesResponse,
};
import { sanitizeTable } from "./sanitize";
import { LocalTable, Table } from "./table";
@@ -53,8 +49,11 @@ export interface CreateTableOptions {
* happen. Any provided data will be ignored.
*
* If this is set to "overwrite" then any existing table will be replaced.
*
* If this is set to "exist_ok" then an existing table is opened and any
* provided data is ignored.
*/
mode: "create" | "overwrite";
mode: "create" | "overwrite" | "exist_ok";
/**
* If this is true and the table already exists and the mode is "create"
* then no error will be raised.
@@ -138,7 +137,7 @@ export interface OpenTableOptions {
}
/**
* @deprecated Use {@link ListTablesOptions} with {@link Connection.listTables}
* @deprecated Use {@link ListingOptions} with {@link Connection.listTables}
* instead.
*/
export interface TableNamesOptions {
@@ -154,32 +153,6 @@ export interface TableNamesOptions {
limit?: number;
}
export interface ListTablesOptions {
/**
* Token from a previous response, to resume listing where it left off.
*
* The token is opaque: it carries whatever the database needs to resume, and
* callers should not construct or interpret one.
*/
pageToken?: string;
/**
* An upper bound on how many tables to return.
* Defaults to 100. Zero returns an empty page without a continuation token.
*
* A page may hold fewer than this and still not be the last one, so keep
* going while the response carries a page token rather than while pages are
* full.
*/
limit?: number;
}
export interface ListNamespacesOptions {
/** Token from a previous response for pagination. */
pageToken?: string;
/** An optional limit to the number of results to return. */
limit?: number;
}
export interface CreateNamespaceOptions {
/** Creation mode. */
mode?: "create" | "exist_ok" | "overwrite";
@@ -275,45 +248,18 @@ export abstract class Connection {
options?: Partial<TableNamesOptions>,
): Promise<string[]>;
/**
* List a page of the tables in this database.
*
* To retrieve the tables after the page, pass the `pageToken` the response
* carries back in. A page can be shorter than `limit` without being the last
* one, so walk until a response carries no page token:
*
/** Lazily list tables in a namespace, requesting 100 per page by default.
* @example
* ```ts
* const names = [];
* let pageToken = undefined;
* do {
* const page = await conn.listTables({ pageToken, limit: 100 });
* names.push(...page.tables);
* pageToken = page.pageToken;
* } while (pageToken);
* for await (const name of conn.listTables([], { pageLimit: 20 })) {
* console.log(name);
* }
* ```
*
* @param {Partial<ListTablesOptions>} options - Pagination options
* (`pageToken`, `limit`).
* @returns {Promise<ListTablesResponse>} A page of table names and an
* optional token for the tables after it.
*/
abstract listTables(
options?: Partial<ListTablesOptions>,
): Promise<ListTablesResponse>;
/**
* List a page of the tables in this database.
*
* @param {string[]} namespacePath - The namespace path to list tables from
* (defaults to root namespace)
* @param {Partial<ListTablesOptions>} options - Pagination options
* (`pageToken`, `limit`).
* @returns {Promise<ListTablesResponse>} A page of table names and an
* optional token for the tables after it.
*/
abstract listTables(
namespacePath?: string[],
options?: Partial<ListTablesOptions>,
): Promise<ListTablesResponse>;
options?: ListingOptions,
): Listing<string>;
/**
* Open a table in the database.
@@ -539,23 +485,11 @@ export abstract class Connection {
namespacePath: string[],
): Promise<DescribeNamespaceResponse>;
/**
* List the immediate child namespaces under the given parent.
*
* Results may be paginated. To retrieve subsequent pages, pass the
* `pageToken` returned by a previous call.
*
* @param {string[]} namespacePath - The parent namespace path. Defaults
* to the root namespace if omitted.
* @param {Partial<ListNamespacesOptions>} options - Pagination options
* (`pageToken`, `limit`).
* @returns {Promise<ListNamespacesResponse>} Child namespace names and
* an optional token for fetching the next page.
*/
/** Lazily list immediate child namespaces, 100 per page by default. */
abstract listNamespaces(
namespacePath?: string[],
options?: Partial<ListNamespacesOptions>,
): Promise<ListNamespacesResponse>;
options?: ListingOptions,
): Listing<string>;
/**
* Create a new namespace at the given path.
@@ -717,13 +651,13 @@ export class LocalConnection extends Connection {
} else {
// First argument is options object (backwards compatibility)
namespacePath = undefined;
tableNamesOptions = namespacePathOrOptions;
tableNamesOptions = namespacePathOrOptions ?? options;
}
return this.inner.tableNames(
namespacePath ?? [],
tableNamesOptions?.startAfter,
tableNamesOptions?.limit,
normalizeTableListingLimit(tableNamesOptions?.limit),
);
}
@@ -818,22 +752,17 @@ export class LocalConnection extends Connection {
);
}
async listTables(
namespacePathOrOptions?: string[] | Partial<ListTablesOptions>,
options?: Partial<ListTablesOptions>,
): Promise<ListTablesResponse> {
// Detect if first argument is namespacePath array or options object
const namespacePath = Array.isArray(namespacePathOrOptions)
? namespacePathOrOptions
: undefined;
const listTablesOptions = Array.isArray(namespacePathOrOptions)
? options
: namespacePathOrOptions;
return this.inner.listTables(
namespacePath ?? [],
listTablesOptions?.pageToken,
listTablesOptions?.limit,
listTables(
namespacePath?: string[],
options: ListingOptions = {},
): Listing<string> {
validateListingOptions(options);
return new Listing(
this.inner.listTables(
namespacePath ?? [],
options.pageToken,
options.pageLimit,
),
);
}
@@ -890,23 +819,24 @@ export class LocalConnection extends Connection {
private getStorageOptions(
options?: Partial<CreateTableOptions>,
): Record<string, string> | undefined {
let storageOptions = options?.storageOptions;
if (options?.dataStorageVersion !== undefined) {
if (options.storageOptions === undefined) {
options.storageOptions = {};
}
options.storageOptions["newTableDataStorageVersion"] =
options.dataStorageVersion;
storageOptions = {
...storageOptions,
newTableDataStorageVersion: options.dataStorageVersion,
};
}
if (options?.enableV2ManifestPaths !== undefined) {
if (options.storageOptions === undefined) {
options.storageOptions = {};
}
options.storageOptions["newTableEnableV2ManifestPaths"] =
options.enableV2ManifestPaths ? "true" : "false";
storageOptions = {
...storageOptions,
newTableEnableV2ManifestPaths: options.enableV2ManifestPaths
? "true"
: "false",
};
}
return cleanseStorageOptions(options?.storageOptions);
return cleanseStorageOptions(storageOptions);
}
async createTable(
@@ -939,7 +869,7 @@ export class LocalConnection extends Connection {
} else {
// Third argument is options object (backwards compatibility)
namespacePath = undefined;
createOptions = namespacePathOrOptions;
createOptions = namespacePathOrOptions ?? options;
}
return this._createTableImpl(name, data, namespacePath, createOptions);
@@ -986,7 +916,7 @@ export class LocalConnection extends Connection {
} else {
// Third argument is options object (backwards compatibility)
namespacePath = undefined;
createOptions = namespacePathOrOptions;
createOptions = namespacePathOrOptions ?? options;
}
let mode: string = createOptions?.mode ?? "create";
@@ -1035,12 +965,15 @@ export class LocalConnection extends Connection {
listNamespaces(
namespacePath?: string[],
options?: Partial<ListNamespacesOptions>,
): Promise<ListNamespacesResponse> {
return this.inner.listNamespaces(
namespacePath ?? [],
options?.pageToken,
options?.limit,
options: ListingOptions = {},
): Listing<string> {
validateListingOptions(options);
return new Listing(
this.inner.listNamespaces(
namespacePath ?? [],
options.pageToken,
options.pageLimit,
),
);
}
@@ -1103,6 +1036,13 @@ export class LocalConnection extends Connection {
}
}
function normalizeTableListingLimit(limit?: number): number | undefined {
validateNonNegativeInteger(limit, "limit");
// The native binding takes u32: larger JS numbers otherwise wrap to zero or
// a smaller page size. A listing limit is an upper bound, so saturate it.
return limit === undefined ? undefined : Math.min(limit, 0xffffffff);
}
/**
* Takes storage options and makes all the keys snake case.
*/
+4 -2
View File
@@ -38,6 +38,7 @@ export class TransformersEmbeddingFunction extends EmbeddingFunction<
#model?: import("@huggingface/transformers").PreTrainedModel;
#tokenizer?: import("@huggingface/transformers").PreTrainedTokenizer;
#modelName: XenovaTransformerOptions["model"];
#tokenizerName: string;
#initialized = false;
#tokenizerOptions: XenovaTransformerOptions["tokenizerOptions"];
#ndims?: number;
@@ -58,6 +59,7 @@ export class TransformersEmbeddingFunction extends EmbeddingFunction<
this.#ndims = options.ndims;
this.#modelName = modelName;
this.#tokenizerName = options.tokenizer ?? modelName;
}
async init() {
@@ -86,11 +88,11 @@ export class TransformersEmbeddingFunction extends EmbeddingFunction<
}
try {
this.#tokenizer = await transformers.AutoTokenizer.from_pretrained(
this.#modelName,
this.#tokenizerName,
);
} catch (e) {
throw new Error(
`error loading tokenizer for ${this.#modelName}. Make sure you are using a wasm compatible model:\nReason: ${e}`,
`error loading tokenizer ${this.#tokenizerName}. Make sure you are using a wasm compatible model:\nReason: ${e}`,
);
}
this.#initialized = true;
+8 -13
View File
@@ -97,13 +97,9 @@ export {
Connection,
CreateTableOptions,
TableNamesOptions,
ListTablesOptions,
OpenTableOptions,
ListNamespacesOptions,
CreateNamespaceOptions,
DropNamespaceOptions,
ListNamespacesResponse,
ListTablesResponse,
CreateNamespaceResponse,
DropNamespaceResponse,
DescribeNamespaceResponse,
@@ -420,10 +416,10 @@ export async function connect(
throw new Error("uri is required");
}
finalOptions = (finalOptions as ConnectionOptions) ?? {};
(<ConnectionOptions>finalOptions).storageOptions = cleanseStorageOptions(
(<ConnectionOptions>finalOptions).storageOptions,
);
finalOptions = {
...finalOptions,
storageOptions: cleanseStorageOptions(finalOptions.storageOptions),
};
// Create native header provider if one was provided
let nativeProvider: NativeJsHeaderProvider | undefined;
@@ -630,11 +626,10 @@ export async function connectNamespace(
properties = configOrProperties as Record<string, string>;
}
const finalOptions: ConnectNamespaceOptions = (options ??
{}) as ConnectNamespaceOptions;
finalOptions.storageOptions = cleanseStorageOptions(
finalOptions.storageOptions,
);
const finalOptions: ConnectNamespaceOptions = {
...options,
storageOptions: cleanseStorageOptions(options?.storageOptions),
};
const nativeConn = await LanceDbConnection.newWithNamespace(
implName,
+108 -37
View File
@@ -1,8 +1,10 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import type { Data as ArrowData } from "apache-arrow";
import {
Table as ArrowTable,
DataType,
Field,
type IntoVector,
LargeBinary,
@@ -12,7 +14,6 @@ import {
extractVectorBuffer,
fromBufferToRecordBatch,
fromRecordBatchToBuffer,
makeData,
tableFromIPC,
vectorFromArray,
} from "./arrow";
@@ -66,20 +67,42 @@ export interface QueryBlobContext {
const BLOB_BYTES_ATTEMPTS = 3;
interface BlobSource {
// Output names are literal: an alias containing dots is still one field.
output: string;
path: string[];
source: string;
}
function blobSources(
blobColumns: string[],
projection: [string, string][] | undefined,
): Map<string, string> {
const columns = new Set(blobColumns);
const pairs =
projection ??
blobColumns.map((column): [string, string] => [column, column]);
return new Map(pairs.filter(([, source]) => columns.has(source)));
): BlobSource[] {
if (projection === undefined) {
return blobColumns.map((source) => {
const [output, ...path] = source.split(".");
return { output, path, source };
});
}
const sources: BlobSource[] = [];
for (const [output, source] of projection) {
for (const column of blobColumns) {
if (column === source || column.startsWith(`${source}.`)) {
sources.push({
output,
path:
column === source ? [] : column.slice(source.length + 1).split("."),
source: column,
});
}
}
}
return sources;
}
async function readBlobBytes(
table: ArrowTable,
sources: Map<string, string>,
sources: BlobSource[],
snapshot: NativeTable,
dropRowId: boolean,
): Promise<ArrowTable> {
@@ -91,18 +114,25 @@ async function readBlobBytes(
rowIdColumn === null
? []
: Array.from(rowIdColumn.toArray() as BigUint64Array);
const replacements = new Map<number, (Buffer | null)[]>();
for (const [output, source] of sources) {
const index = table.schema.fields.findIndex(
(field) => field.name === output,
);
if (index < 0) {
const outputFields = new Map(
table.schema.fields.map((field) => [field.name, field]),
);
const replacements = new Map<Field, (Buffer | null)[]>();
for (const { output, path, source } of sources) {
let field = outputFields.get(output);
for (const name of path) {
field =
field !== undefined && DataType.isStruct(field.type)
? field.type.children.find((child) => child.name === name)
: undefined;
}
if (field === undefined) {
continue;
}
const values =
rowIds.length === 0 ? [] : await snapshot.fetchBlobs(source, rowIds);
replacements.set(
index,
field,
values.map((value) => value ?? null),
);
}
@@ -117,37 +147,76 @@ async function readBlobBytes(
);
}
// A fresh field, not a clone: the descriptor field's metadata must not follow
// the bytes.
function withBinaryColumns(
table: ArrowTable,
replacements: Map<number, (Buffer | null)[]>,
replacements: Map<Field, (Buffer | null)[]>,
): ArrowTable {
if (replacements.size === 0) {
return table;
}
const fields = table.schema.fields.map((field, index) =>
replacements.has(index)
? new Field(field.name, new LargeBinary(), true)
: field,
);
const replaceField = (field: Field): Field => {
if (replacements.has(field)) {
// The descriptor's extension and encoding metadata must not follow bytes.
return new Field(field.name, new LargeBinary(), true);
}
if (DataType.isStruct(field.type)) {
const children = field.type.children.map(replaceField);
if (children.some((child, i) => child !== field.type.children[i])) {
return field.clone({ type: new Struct(children) });
}
}
return field;
};
const fields = table.schema.fields.map(replaceField);
const schema = new Schema(fields, table.schema.metadata);
let offset = 0;
const batches = table.batches.map((batch) => {
const children = batch.data.children.map((child, index) => {
const values = replacements.get(index);
if (values === undefined) {
return child;
const replaceData = (
data: ArrowData,
field: Field,
updated: Field,
): ArrowData => {
if (field === updated) {
return data;
}
return vectorFromArray(
values.slice(offset, offset + batch.numRows),
new LargeBinary(),
).data[0];
});
const values = replacements.get(field);
if (values !== undefined) {
return vectorFromArray(
values.slice(offset, offset + batch.numRows),
new LargeBinary(),
).data[0];
}
const children = data.children.map((child, index) =>
replaceData(
child,
field.type.children[index],
updated.type.children[index],
),
);
// Keep the struct's validity bitmap and offset, including null parents.
return data.clone(
updated.type,
undefined,
undefined,
undefined,
undefined,
children,
);
};
const children = batch.data.children.map((child, index) =>
replaceData(child, table.schema.fields[index], fields[index]),
);
offset += batch.numRows;
return new RecordBatch(
schema,
makeData({ type: new Struct(fields), length: batch.numRows, children }),
batch.data.clone(
new Struct(fields),
undefined,
undefined,
undefined,
undefined,
children,
),
);
});
return new ArrowTable(schema, batches);
@@ -175,12 +244,14 @@ export interface QueryExecutionOptions {
*
* - `"descriptions"` (default): each blob is a descriptor. Read the bytes
* with {@link Table.fetchBlobs} or {@link Table.fetchBlobFiles}.
* - `"bytes"`: each top-level blob column holds the blob bytes as
* `LargeBinary`. The query also reads `_rowid` to fetch the bytes, and
* - `"bytes"`: each blob column, including blobs nested in structs, holds
* the bytes as `LargeBinary`. The query also reads `_rowid` to fetch them, and
* leaves it out of the result unless {@link QueryBase.withRowId} was
* called. All the bytes are held in memory.
*
* Blobs nested in a struct or a list keep their descriptors.
* Selecting a parent struct or directly aliasing a blob or struct also returns
* bytes. Blobs inside lists keep their descriptors. Arbitrary projection
* expressions are not materialized as bytes.
*
* The rows and the bytes come from the same table version. If the table
* changes while the query runs, the query runs again, up to three times.
@@ -433,7 +504,7 @@ export class QueryBase<
await snapshot.blobColumns(),
context.projection,
);
if (sources.size === 0) {
if (sources.length === 0) {
return this.collect(options, false);
}
const addRowId = !context.withRowId;
+5 -1
View File
@@ -989,7 +989,11 @@ export abstract class Table {
*
* Model-backed tokenizers such as `jieba/*` and `lindera/*` are rebuilt in
* the client process from index metadata. For remote tables, this means the
* same tokenizer model files must also exist locally.
* same tokenizer model files must also exist locally (under
* `LANCE_LANGUAGE_MODEL_HOME`) and must match the files the server uses. A
* mismatch cannot be detected: the returned tokens may differ from what the
* server's index produces. If the model files are missing, an error naming
* the tokenizer is raised.
*/
abstract tokenize(
query: string,
+2 -1
View File
@@ -90,7 +90,8 @@
"postdocs": "node typedoc_post_process.js",
"lint": "biome check . && biome format .",
"lint-fix": "biome check --write . && biome format --write .",
"prepublishOnly": "napi prepublish -t npm",
"check:native-symbols": "node scripts/check-native-symbols.js",
"prepublishOnly": "pnpm check:native-symbols && napi prepublish -t npm",
"test": "jest --verbose",
"integration": "S3_TEST=1 pnpm test",
"universal": "napi universalize",
+77
View File
@@ -0,0 +1,77 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
const { spawnSync } = require("node:child_process");
const { existsSync } = require("node:fs");
const path = require("node:path");
const FORBIDDEN_UNDEFINED_SYMBOLS = new Set([
"sum_4bit_dist_table_32bytes_batch_avx512",
]);
function findForbiddenUndefinedSymbols(output) {
const found = new Set();
for (const line of output.split(/\r?\n/)) {
const columns = line.trim().split(/\s+/);
const symbol = columns.at(-1)?.split("@")[0];
if (symbol && FORBIDDEN_UNDEFINED_SYMBOLS.has(symbol)) {
found.add(symbol);
}
}
return [...found].sort();
}
function checkNativeBinary(binaryPath, runNm = spawnSync) {
const result = runNm("nm", ["-D", "--undefined-only", binaryPath], {
encoding: "utf8",
});
if (result.error) {
throw new Error(`Unable to inspect ${binaryPath}: ${result.error.message}`);
}
if (result.status !== 0) {
throw new Error(
`Unable to inspect ${binaryPath}: nm exited with status ${result.status}\n${result.stderr}`,
);
}
const forbidden = findForbiddenUndefinedSymbols(result.stdout);
if (forbidden.length > 0) {
throw new Error(
`${binaryPath} contains unresolved internal Lance symbols: ${forbidden.join(
", ",
)}`,
);
}
}
function main() {
const binaryPath = path.resolve(
__dirname,
"..",
"npm",
"linux-x64-musl",
"lancedb.linux-x64-musl.node",
);
if (!existsSync(binaryPath)) {
throw new Error(
`Missing ${binaryPath}; assemble the native artifacts before publishing`,
);
}
checkNativeBinary(binaryPath);
}
if (require.main === module) {
try {
main();
} catch (error) {
console.error(error instanceof Error ? error.message : error);
process.exitCode = 1;
}
}
module.exports = { checkNativeBinary, findForbiddenUndefinedSymbols };
+19 -52
View File
@@ -16,8 +16,7 @@ use crate::table::Table;
use lancedb::connection::{ConnectBuilder, Connection as LanceDBConnection, connect_namespace};
use lance_namespace::models::{
CreateNamespaceRequest, DescribeNamespaceRequest, DropNamespaceRequest, ListNamespacesRequest,
ListTablesRequest,
CreateNamespaceRequest, DescribeNamespaceRequest, DropNamespaceRequest,
};
use lancedb::ipc::{ipc_file_to_batches, ipc_file_to_schema};
@@ -31,18 +30,6 @@ pub struct DescribeNamespaceResponse {
pub properties: Option<HashMap<String, String>>,
}
#[napi(object)]
pub struct ListNamespacesResponse {
pub namespaces: Vec<String>,
pub page_token: Option<String>,
}
#[napi(object)]
pub struct ListTablesResponse {
pub tables: Vec<String>,
pub page_token: Option<String>,
}
#[napi(object)]
pub struct CreateNamespaceResponse {
pub properties: Option<HashMap<String, String>>,
@@ -239,31 +226,20 @@ impl Connection {
op.execute().await.default_error()
}
/// List a page of tables in the database.
/// Lazily list tables in the database.
#[napi(catch_unwind)]
pub async fn list_tables(
pub fn list_tables(
&self,
namespace_path: Option<Vec<String>>,
page_token: Option<String>,
limit: Option<u32>,
) -> napi::Result<ListTablesResponse> {
let request = ListTablesRequest {
// The root namespace is an empty path, not an absent one: a namespace-backed
// database rejects a request that names no namespace.
id: Some(namespace_path.unwrap_or_default()),
page_token,
limit: limit.map(|limit| i32::try_from(limit).unwrap_or(i32::MAX)),
..Default::default()
};
let response = self
.get_inner()?
.list_tables(request)
.await
.default_error()?;
Ok(ListTablesResponse {
tables: response.tables,
page_token: response.page_token,
})
) -> napi::Result<crate::listing::NameListing> {
Ok(crate::listing::NameListing::new(
self.get_inner()?.list_tables(
&namespace_path.unwrap_or_default(),
crate::listing::options(page_token, limit),
),
))
}
/// Create table from a Apache Arrow IPC (file) buffer.
@@ -663,28 +639,19 @@ impl Connection {
}
#[napi(catch_unwind)]
/// List child namespaces under the given namespace path
pub async fn list_namespaces(
/// Lazily list child namespaces under the given namespace path
pub fn list_namespaces(
&self,
namespace_path: Option<Vec<String>>,
page_token: Option<String>,
limit: Option<u32>,
) -> napi::Result<ListNamespacesResponse> {
let req = ListNamespacesRequest {
id: namespace_path,
page_token,
limit: limit.map(|l| l as i32),
..Default::default()
};
let resp = self
.get_inner()?
.list_namespaces(req)
.await
.default_error()?;
Ok(ListNamespacesResponse {
namespaces: resp.namespaces,
page_token: resp.page_token,
})
) -> napi::Result<crate::listing::NameListing> {
Ok(crate::listing::NameListing::new(
self.get_inner()?.list_namespaces(
&namespace_path.unwrap_or_default(),
crate::listing::options(page_token, limit),
),
))
}
#[napi(catch_unwind)]