chore: merge main into blob concurrency repair

Preserve descriptor-based blob sizing and version pinning together with shared request permits and finite cursor ranges. Adapt probe-only regression tests to the upstream probe helper.
This commit is contained in:
Gatefixer
2026-10-03 01:44:09 +00:00
76 changed files with 4730 additions and 2018 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
name = "lancedb-nodejs"
edition.workspace = true
version = "0.40.0-beta.11"
version = "0.40.0-beta.12"
publish = false
license.workspace = true
description.workspace = true
+8
View File
@@ -275,6 +275,7 @@
[@types/node@20.16.10](https://github.com/DefinitelyTyped/DefinitelyTyped) - MIT
[@types/node@20.17.9](https://github.com/DefinitelyTyped/DefinitelyTyped) - MIT
[@types/node@22.7.4](https://github.com/DefinitelyTyped/DefinitelyTyped) - MIT
[@types/node@25.9.8](https://github.com/DefinitelyTyped/DefinitelyTyped) - MIT
[@types/semver@7.5.6](https://github.com/DefinitelyTyped/DefinitelyTyped) - MIT
[@types/stack-utils@2.0.3](https://github.com/DefinitelyTyped/DefinitelyTyped) - MIT
[@types/tmp@0.2.6](https://github.com/DefinitelyTyped/DefinitelyTyped) - MIT
@@ -306,6 +307,9 @@
[apache-arrow@16.0.0](https://github.com/apache/arrow) - Apache-2.0
[apache-arrow@17.0.0](https://github.com/apache/arrow) - Apache-2.0
[apache-arrow@18.0.0](https://github.com/apache/arrow) - Apache-2.0
[apache-arrow@19.0.1](https://github.com/apache/arrow) - Apache-2.0
[apache-arrow@20.0.0](https://github.com/apache/arrow) - Apache-2.0
[apache-arrow@21.2.0](https://github.com/apache/arrow-js) - Apache-2.0
[argparse@1.0.10](https://github.com/nodeca/argparse) - MIT
[argparse@2.0.1](https://github.com/nodeca/argparse) - Python-2.0
[array-back@3.1.0](https://github.com/75lb/array-back) - MIT
@@ -348,6 +352,7 @@
[color@4.2.3](https://github.com/Qix-/color) - MIT
[combined-stream@1.0.8](https://github.com/felixge/node-combined-stream) - MIT
[command-line-args@5.2.1](https://github.com/75lb/command-line-args) - MIT
[command-line-args@6.0.2](https://github.com/75lb/command-line-args) - MIT
[command-line-usage@7.0.1](https://github.com/75lb/command-line-usage) - MIT
[concat-map@0.0.1](https://github.com/substack/node-concat-map) - MIT
[convert-source-map@2.0.0](https://github.com/thlorenz/convert-source-map) - MIT
@@ -398,12 +403,14 @@
[file-entry-cache@6.0.1](https://github.com/royriojas/file-entry-cache) - MIT
[fill-range@7.1.1](https://github.com/jonschlinkert/fill-range) - MIT
[find-replace@3.0.0](https://github.com/75lb/find-replace) - MIT
[find-replace@5.0.2](https://github.com/75lb/find-replace) - MIT
[find-up@4.1.0](https://github.com/sindresorhus/find-up) - MIT
[find-up@5.0.0](https://github.com/sindresorhus/find-up) - MIT
[flat-cache@3.2.0](https://github.com/jaredwray/flat-cache) - MIT
[flatbuffers@1.12.0](https://github.com/google/flatbuffers) - Apache*
[flatbuffers@23.5.26](https://github.com/google/flatbuffers) - Apache*
[flatbuffers@24.3.25](https://github.com/google/flatbuffers) - Apache-2.0
[flatbuffers@25.9.23](https://github.com/google/flatbuffers) - Apache-2.0
[flatted@3.2.9](https://github.com/WebReflection/flatted) - ISC
[follow-redirects@1.15.6](https://github.com/follow-redirects/follow-redirects) - MIT
[foreground-child@3.3.0](https://github.com/tapjs/foreground-child) - ISC
@@ -644,6 +651,7 @@
[uc.micro@2.1.0](https://github.com/markdown-it/uc.micro) - MIT
[undici-types@5.26.5](https://github.com/nodejs/undici) - MIT
[undici-types@6.19.8](https://github.com/nodejs/undici) - MIT
[undici-types@7.24.6](https://github.com/nodejs/undici) - MIT
[update-browserslist-db@1.0.13](https://github.com/browserslist/update-db) - MIT
[uri-js@4.4.1](https://github.com/garycourt/uri-js) - BSD-2-Clause
[uuid@9.0.1](https://github.com/uuidjs/uuid) - MIT
+1415 -1456
View File
File diff suppressed because it is too large Load Diff
+39
View File
@@ -0,0 +1,39 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
// Every Apache Arrow release the package accepts as a peer dependency, as
// separately installed copies so the tests can feed objects built by one
// Arrow version into a LanceDB running on another. When a new Arrow major
// is released, add the alias to `package.json` and extend this list; the
// test files pick the change up from here.
import * as arrow15 from "apache-arrow-15";
import * as arrow16 from "apache-arrow-16";
import * as arrow17 from "apache-arrow-17";
import * as arrow18 from "apache-arrow-18";
import * as arrow19 from "apache-arrow-19";
import * as arrow20 from "apache-arrow-20";
import * as arrow21 from "apache-arrow-21";
export { arrow15, arrow16, arrow17, arrow18, arrow19, arrow20, arrow21 };
export type ApacheArrow =
| typeof arrow15
| typeof arrow16
| typeof arrow17
| typeof arrow18
| typeof arrow19
| typeof arrow20
| typeof arrow21;
export const arrowVersions: ApacheArrow[] = [
arrow15,
arrow16,
arrow17,
arrow18,
arrow19,
arrow20,
arrow21,
];
/** The newest supported Arrow release, for tests that only need one. */
export const latestArrow = arrow21;
+5 -8
View File
@@ -3,13 +3,10 @@
import * as apiArrow from "apache-arrow";
import * as arrow15 from "apache-arrow-15";
import * as arrow16 from "apache-arrow-16";
import * as arrow17 from "apache-arrow-17";
import * as arrow18 from "apache-arrow-18";
import * as tmp from "tmp";
import { arrowVersions } from "./arrow_versions";
import { connect } from "../lancedb";
import {
EmbeddingFunction,
@@ -18,7 +15,7 @@ import {
} from "../lancedb/embedding";
import { getRegistry, register } from "../lancedb/embedding/registry";
describe.each([arrow15, arrow16, arrow17, arrow18])("LanceSchema", (arrow) => {
describe.each(arrowVersions)("LanceSchema", (arrow) => {
test("should preserve input order", async () => {
const schema = LanceSchema({
id: new arrow.Int32(),
@@ -29,7 +26,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])("LanceSchema", (arrow) => {
});
});
describe.each([arrow15, arrow16, arrow17, arrow18])("Registry", (arrow) => {
describe.each(arrowVersions)("Registry", (arrow) => {
let tmpDir: tmp.DirResult;
beforeEach(() => {
tmpDir = tmp.dirSync({ unsafeCleanup: true });
@@ -240,7 +237,7 @@ describe("Registry.setVar", () => {
return data.map(() => [1, 2, 3]);
}
embeddingDataType() {
return new arrow18.Float32() as apiArrow.Float;
return new apiArrow.Float32();
}
protected getSensitiveKeys() {
return ["secretKey"];
+113 -1
View File
@@ -2,7 +2,13 @@
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import * as arrow from "../lancedb/arrow";
import { sanitizeField, sanitizeMap, sanitizeType } from "../lancedb/sanitize";
import {
sanitizeField,
sanitizeMap,
sanitizeSchema,
sanitizeType,
} from "../lancedb/sanitize";
import { arrow21 } from "./arrow_versions";
describe("sanitize", function () {
describe("sanitizeType function", function () {
@@ -59,6 +65,8 @@ describe("sanitize", function () {
["durationmicrosecond", new arrow.DurationMicrosecond()],
["durationmillisecond", new arrow.DurationMillisecond()],
["durationsecond", new arrow.DurationSecond()],
["largebinary", new arrow.LargeBinary()],
["largeutf8", new arrow.LargeUtf8()],
] as const;
it.each(allTypeNameTestCases)(
@@ -89,6 +97,110 @@ describe("sanitize", function () {
"Unrecognized type name in schema: invalid_type",
);
});
// These types only exist in Arrow 20 (IntervalMonthDayNano) and 21 (the
// rest), so the objects come from a second copy of the newest release. The
// library itself may be running on an older Arrow (see the CI matrix), in
// which case sanitizing them has to fail with a clear message instead.
const hostArrow = arrow as unknown as Record<string, unknown>;
const hostSupportsNewerTypes = typeof hostArrow.Utf8View === "function";
const whenSupported = hostSupportsNewerTypes ? describe : describe.skip;
const whenUnsupported = hostSupportsNewerTypes ? describe.skip : describe;
const newerTypeTestCases = [
["Utf8View", () => new arrow21.Utf8View()],
["BinaryView", () => new arrow21.BinaryView()],
["IntervalMonthDayNano", () => new arrow21.IntervalMonthDayNano()],
[
"LargeList",
() =>
new arrow21.LargeList(new arrow21.Field("item", new arrow21.Utf8())),
],
] as const;
whenSupported("types added in Arrow 20 and 21", function () {
it.each(newerTypeTestCases)(
"should sanitize a %s type from another copy of Arrow",
function (name, makeForeign) {
const foreign = makeForeign();
const result = sanitizeType(foreign);
// Interval subclasses share `Type.Interval`, so the sanitized type
// is the generic `Interval` carrying the same unit, as for the
// other units.
const expected = name === "IntervalMonthDayNano" ? "Interval" : name;
expect(result).toBeInstanceOf(hostArrow[expected] as () => unknown);
expect(result.typeId).toBe(foreign.typeId);
expect(result.toString()).toBe(foreign.toString());
},
);
it.each([
["binaryview", "BinaryView"],
["utf8view", "Utf8View"],
["intervalmonthdaynano", "Interval"],
])('should map type name "%s" to %s', function (name, expected) {
expect(sanitizeType(name)).toBeInstanceOf(
hostArrow[expected] as () => unknown,
);
});
it("should sanitize the child field of a LargeList", function () {
const foreign = new arrow21.LargeList(
new arrow21.Field("item", new arrow21.Int32(), false),
);
const result = sanitizeType(foreign) as arrow.List;
expect(result.children[0]).toBeInstanceOf(arrow.Field);
expect(result.children[0].name).toBe("item");
expect(result.children[0].nullable).toBe(false);
const child = result.children[0].type;
expect(arrow.DataType.isInt(child)).toBe(true);
expect((child as arrow.Int).bitWidth).toBe(32);
expect((child as arrow.Int).isSigned).toBe(true);
});
});
whenUnsupported("types newer than the installed Arrow", function () {
it.each(
newerTypeTestCases.filter(([name]) => name !== "IntervalMonthDayNano"),
)(
"should explain that %s needs a newer apache-arrow",
function (name, makeForeign) {
expect(() => sanitizeType(makeForeign())).toThrow(
`The Arrow type ${name} requires apache-arrow`,
);
},
);
it("should sanitize IntervalMonthDayNano to a generic Interval", () => {
// The subclass shares `Type.Interval`, which every supported Arrow
// release has; only the unit is new.
const result = sanitizeType(new arrow21.IntervalMonthDayNano());
expect(result).toBeInstanceOf(arrow.Interval);
expect((result as arrow.Interval).unit).toBe(
arrow21.IntervalUnit.MONTH_DAY_NANO,
);
});
});
it("should rebuild a schema from a second copy of the same Arrow version", function () {
// Arrow 21 makes `instanceof Schema` succeed across copies of the library
// but not `instanceof <DataType>`; the schema must still be rebuilt.
const foreign = new arrow21.Schema([
new arrow21.Field("x", new arrow21.Float32(), true),
]);
// Only Arrow 21+ overrides `Symbol.hasInstance` this way.
const crossCopyInstanceof = Object.getOwnPropertySymbols(
arrow.Schema,
).includes(Symbol.hasInstance);
expect(foreign instanceof arrow.Schema).toBe(crossCopyInstanceof);
const result = sanitizeSchema(foreign);
expect(result).not.toBe(foreign);
const type = result.fields[0].type;
expect(type).toBeInstanceOf(arrow.DataType);
expect(arrow.DataType.isFloat(type)).toBe(true);
expect((type as arrow.Float).precision).toBe(arrow.Precision.SINGLE);
expect(() => arrow.makeBuilder({ type })).not.toThrow();
});
});
describe("sanitizeField function", function () {
+313 -12
View File
@@ -2,15 +2,11 @@
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import * as fs from "fs";
import { getEventListeners } from "node:events";
import { pathToFileURL } from "node:url";
import * as path from "path";
import * as tmp from "tmp";
import * as arrow15 from "apache-arrow-15";
import * as arrow16 from "apache-arrow-16";
import * as arrow17 from "apache-arrow-17";
import * as arrow18 from "apache-arrow-18";
import {
AutoQuery,
Connection,
@@ -55,19 +51,21 @@ import {
instanceOfFullTextQuery,
} from "../lancedb/query";
import { LocalTable } from "../lancedb/table";
import {
type ApacheArrow,
arrow18,
arrowVersions,
latestArrow,
} from "./arrow_versions";
describe.each([arrow15, arrow16, arrow17, arrow18])(
describe.each(arrowVersions)(
"Given a table",
// biome-ignore lint/suspicious/noExplicitAny: <explanation>
(arrow: any) => {
let tmpDir: tmp.DirResult;
let table: Table;
const schema:
| import("apache-arrow-15").Schema
| import("apache-arrow-16").Schema
| import("apache-arrow-17").Schema
| import("apache-arrow-18").Schema = new arrow.Schema([
const schema: InstanceType<ApacheArrow["Schema"]> = new arrow.Schema([
new arrow.Field("id", new arrow.Float64(), true),
]);
@@ -92,6 +90,38 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
await expect(table.countRows()).resolves.toBe(3);
});
it("creates and adds Arrow tables with large string and binary columns", async () => {
const data = new arrow.Table({
text: arrow.vectorFromArray(["alpha", "beta"], new arrow.LargeUtf8()),
bytes: arrow.vectorFromArray(
[Buffer.from("one"), Buffer.from("two")],
new arrow.LargeBinary(),
),
});
const conn = await connect(tmpDir.name);
const largeTable = await conn.createTable("large_columns", data);
await largeTable.add(data);
const rows = await largeTable.query().toArray();
expect(rows.map((row) => row.text)).toEqual([
"alpha",
"beta",
"alpha",
"beta",
]);
expect(rows.map((row) => Buffer.from(row.bytes).toString())).toEqual([
"one",
"two",
"one",
"two",
]);
const fields = (await largeTable.schema()).fields;
expect(fields.map((field) => field.type.typeId)).toEqual([
arrow.Type.LargeUtf8,
arrow.Type.LargeBinary,
]);
});
it("should support a foreign Float64 vector schema end to end", async () => {
const conn = await connect(tmpDir.name);
const schema = new arrow.Schema([
@@ -709,6 +739,114 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
);
// https://github.com/lancedb/lancedb/issues/1963
// These types were added in Arrow 21, so they are built with a second copy of
// that release rather than the per-version matrix above, and reading them back
// needs the library's own Arrow to know them too.
const hostArrowSupportsViewTypes =
typeof (arrow as unknown as Record<string, unknown>).Utf8View === "function";
(hostArrowSupportsViewTypes ? it : it.skip)(
"creates and adds tables with LargeList, Utf8View and BinaryView columns",
async () => {
const tmpDir = tmp.dirSync({ unsafeCleanup: true });
const db = await connect(tmpDir.name);
const makeData = () =>
new latestArrow.Table({
id: latestArrow.vectorFromArray([1, 2], new latestArrow.Int32()),
tags: latestArrow.vectorFromArray(
[["a", "b"], ["c"]],
new latestArrow.LargeList(
new latestArrow.Field("item", new latestArrow.Utf8(), true),
),
),
text: latestArrow.vectorFromArray(
["short", "a string that is longer than the twelve inline bytes"],
new latestArrow.Utf8View(),
),
bytes: latestArrow.vectorFromArray(
[new Uint8Array([1, 2, 3]), new Uint8Array(20).fill(7)],
new latestArrow.BinaryView(),
),
});
const table = await db.createTable("views", makeData());
await table.add(makeData());
const schema = await table.schema();
// Take the enum from the Arrow 21 copy: the host Arrow may predate
// `Type.LargeList`, and this file still has to type-check there.
expect(schema.fields.map((f) => f.type.typeId)).toEqual([
latestArrow.Type.Int,
latestArrow.Type.LargeList,
// Lance stores the view types as their offset-based equivalents.
latestArrow.Type.Utf8,
latestArrow.Type.Binary,
]);
const rows = await table.query().toArrow();
expect(rows.numRows).toBe(4);
const row = rows.get(1)!.toJSON();
expect(row.id).toBe(2);
expect(row.tags.toJSON()).toEqual(["c"]);
expect(row.text).toBe(
"a string that is longer than the twelve inline bytes",
);
expect(Array.from(row.bytes)).toEqual(Array(20).fill(7));
// Arrow JS 21 serializes a *sliced* view column with a truncated views
// buffer; the library rebuilds view columns as Utf8/Binary so this works.
const sliced = makeData().slice(1, 2);
expect(sliced.getChild("text")!.data[0].offset).toBe(1);
const slicedTable = await db.createTable("views_sliced", sliced);
await slicedTable.add(sliced);
const slicedRows = await slicedTable.query().toArrow();
expect(slicedRows.numRows).toBe(2);
for (const i of [0, 1]) {
const r = slicedRows.get(i)!.toJSON();
expect(r.id).toBe(2);
expect(r.text).toBe(
"a string that is longer than the twelve inline bytes",
);
expect(Array.from(r.bytes)).toEqual(Array(20).fill(7));
}
// Regression: a field with an integer-like name. JavaScript enumerates
// such keys first, so a key-ordered rebuild would pair "1" with the text
// column's values. The batches are built explicitly to pin field order.
const numericSchema = new latestArrow.Schema([
new latestArrow.Field("text", new latestArrow.Utf8View(), true),
new latestArrow.Field("1", new latestArrow.Utf8View(), true),
new latestArrow.Field("n", new latestArrow.Int32(), true),
]);
const column = (
values: unknown[],
type: InstanceType<typeof latestArrow.DataType>,
) => latestArrow.vectorFromArray(values, type).data[0];
const numericBatch = new latestArrow.RecordBatch(
numericSchema,
latestArrow.makeData({
type: new latestArrow.Struct(numericSchema.fields),
length: 3,
nullCount: 0,
children: [
column(["t0", "t1", "t2"], new latestArrow.Utf8View()),
column(["one0", "one1", "one2"], new latestArrow.Utf8View()),
column([0, 1, 2], new latestArrow.Int32()),
],
}),
);
const numericData = new latestArrow.Table(numericSchema, [numericBatch]);
const numericTable = await db.createTable("views_numeric", numericData);
await numericTable.add(numericData.slice(1));
const numericRows = (await numericTable.query().toArrow()).toArray();
expect(numericRows.map((r) => r.toJSON())).toEqual([
{ text: "t0", "1": "one0", n: 0 },
{ text: "t1", "1": "one1", n: 1 },
{ text: "t2", "1": "one2", n: 2 },
{ text: "t1", "1": "one1", n: 1 },
{ text: "t2", "1": "one2", n: 2 },
]);
tmpDir.removeCallback();
},
);
it("should query documents with LangChain PDF metadata", async () => {
const tmpDir = tmp.dirSync({ unsafeCleanup: true });
try {
@@ -2601,6 +2739,27 @@ describe("when dealing with blob columns", () => {
expect(Buffer.from(await files[1]!.read()).toString()).toBe("beta");
});
it("adds a LargeBinary Arrow column to a blob table", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([new Field("id", new Int64()), blob("image")]);
const table = await db.createTable(
"blobs",
[{ id: 1n, image: Buffer.from("alpha") }],
{ schema },
);
const payload = Buffer.from("beta");
const data = new arrow18.Table({
id: arrow18.vectorFromArray([2n], new arrow18.Int64()),
image: arrow18.vectorFromArray([payload], new arrow18.LargeBinary()),
});
await table.add(data);
const rows = await table.query().withRowId().toArray();
const rowId = rows.find((row) => row.id === 2n)!._rowid as bigint;
const [actual] = await table.fetchBlobs("image", [rowId]);
expect(actual).toEqual(payload);
});
it("reads a half-open range", async () => {
const { table, rowIds } = await openBlobTable();
const files = await table.fetchBlobFiles("image", rowIds);
@@ -2625,6 +2784,148 @@ describe("when dealing with blob columns", () => {
).rejects.toThrow(/exceeds blob size/);
});
it("reads at most maxBytes and advances the cursor", async () => {
const { table, rowIds } = await openBlobTable();
const [handle] = await table.fetchBlobFiles("image", rowIds);
expect((await handle!.read(2n)).toString()).toBe("al");
expect(await handle!.tell()).toBe(2n);
expect((await handle!.read(2n)).toString()).toBe("ph");
expect((await handle!.read(10n)).toString()).toBe("a");
expect(await handle!.tell()).toBe(5n);
expect(await handle!.read(1n)).toEqual(Buffer.alloc(0));
});
it("seeks the cursor", async () => {
const { table, rowIds, alpha } = await openBlobTable();
const [handle] = await table.fetchBlobFiles("image", rowIds);
await handle!.seek(3n);
expect(await handle!.tell()).toBe(3n);
expect((await handle!.read()).toString()).toBe("ha");
await handle!.seek(0n);
expect(await handle!.read()).toEqual(alpha);
});
it("rejects a negative seek position", async () => {
const { table, rowIds } = await openBlobTable();
const [handle] = await table.fetchBlobFiles("image", rowIds);
await expect(handle!.seek(-1n)).rejects.toThrow(/cannot be negative/);
});
it("readRanges returns one buffer per range in order", async () => {
const { table, rowIds } = await openBlobTable();
const [handle] = await table.fetchBlobFiles("image", rowIds);
const buffers = await handle!.readRanges([
{ start: 3n, end: 5n },
{ start: 0n, end: 2n },
{ start: 3n, end: 5n },
]);
expect(buffers.map((b) => b.toString())).toEqual(["ha", "al", "ha"]);
expect(await handle!.tell()).toBe(0n);
});
it("fails when a readRanges end is past the blob size", async () => {
const { table, rowIds, alpha } = await openBlobTable();
const [handle] = await table.fetchBlobFiles("image", rowIds);
await expect(
handle!.readRanges([
{ start: 0n, end: 1n },
{ start: 0n, end: BigInt(alpha.length + 1) },
]),
).rejects.toThrow(/exceeds blob size/);
});
it("fails reads after close and closes once", async () => {
const { table, rowIds } = await openBlobTable();
const [handle] = await table.fetchBlobFiles("image", rowIds);
expect(handle!.isClosed()).toBe(false);
await handle!.close();
expect(handle!.isClosed()).toBe(true);
await expect(handle!.close()).resolves.toBeUndefined();
await expect(handle!.read()).rejects.toThrow(/already closed/);
await expect(handle!.read(1n)).rejects.toThrow(/already closed/);
await expect(handle!.readRange(0n, 1n)).rejects.toThrow(/already closed/);
await expect(handle!.readRanges([{ start: 0n, end: 1n }])).rejects.toThrow(
/already closed/,
);
await expect(handle!.seek(0n)).rejects.toThrow(/already closed/);
await expect(handle!.tell()).rejects.toThrow(/already closed/);
expect(handle!.size()).toBe(5n);
});
it("rejects reads with an already aborted signal", async () => {
const { table, rowIds, alpha } = await openBlobTable();
const [handle] = await table.fetchBlobFiles("image", rowIds);
const signal = AbortSignal.abort();
const abortError = { name: "AbortError" };
await expect(handle!.read(undefined, { signal })).rejects.toMatchObject(
abortError,
);
await expect(handle!.read(2n, { signal })).rejects.toMatchObject(
abortError,
);
await expect(handle!.readRange(0n, 1n, { signal })).rejects.toMatchObject(
abortError,
);
await expect(
handle!.readRanges([{ start: 0n, end: 1n }], { signal }),
).rejects.toMatchObject(abortError);
await expect(
table.fetchBlobs("image", rowIds, { signal }),
).rejects.toMatchObject(abortError);
await expect(
table.fetchBlobFiles("image", rowIds, { signal }),
).rejects.toMatchObject(abortError);
expect(await handle!.read()).toEqual(alpha);
});
it("rejects with the abort reason", async () => {
const { table, rowIds } = await openBlobTable();
const [handle] = await table.fetchBlobFiles("image", rowIds);
const reason = new Error("caller gave up");
await expect(
handle!.readRange(0n, 1n, { signal: AbortSignal.abort(reason) }),
).rejects.toBe(reason);
});
it("reads with a signal and removes its abort listener", async () => {
const { table, rowIds, alpha, beta } = await openBlobTable();
const controller = new AbortController();
const { signal } = controller;
const [handle] = await table.fetchBlobFiles("image", rowIds, { signal });
expect((await handle!.readRange(0n, 2n, { signal })).toString()).toBe("al");
expect(await table.fetchBlobs("image", rowIds, { signal })).toEqual([
alpha,
beta,
null,
]);
expect(getEventListeners(signal, "abort")).toHaveLength(0);
controller.abort();
expect(await handle!.read()).toEqual(alpha);
});
it("cancels an in-flight read", async () => {
const db = await connect(tmpDir.name);
const schema = new Schema([
new Field("id", new Int64(), true),
blob("payload", { inlineSizeThreshold: 0, dedicatedSizeThreshold: 1 }),
]);
const payload = Buffer.alloc(32 * 1024 * 1024, 7);
const table = await db.createTable("large_blobs", [{ id: 1n, payload }], {
schema,
});
const [row] = await table.query().withRowId().toArray();
const [handle] = await table.fetchBlobFiles("payload", [
row._rowid as bigint,
]);
const controller = new AbortController();
const pending = handle!.readRange(0n, handle!.size(), {
signal: controller.signal,
});
controller.abort();
await expect(pending).rejects.toMatchObject({ name: "AbortError" });
expect(await handle!.readRange(0n, 4n)).toEqual(payload.subarray(0, 4));
});
it("rejects fetchBlobs on a non-blob column", async () => {
const { table, rowIds } = await openBlobTable();
await expect(table.fetchBlobs("id", rowIds)).rejects.toThrow(/blob/i);
@@ -3007,7 +3308,7 @@ it("passes cleanupOlderThan to the native binding as an absolute timestamp", asy
expect(optimize).toHaveBeenCalledWith(cutoff.getTime(), true);
});
describe.each([arrow15, arrow16, arrow17, arrow18])(
describe.each(arrowVersions)(
"when optimizing a dataset",
// biome-ignore lint/suspicious/noExplicitAny: <explanation>
(arrow: any) => {
+83 -4
View File
@@ -66,11 +66,20 @@ export type SchemaLike =
export type FieldLike =
| Field
| {
type: string;
type: string | DataTypeLike;
name: string;
nullable: boolean;
metadata?: Map<string, string>;
};
/**
* A `DataType` from any copy or version of apache-arrow.
*
* Arrow 21 brands its classes with `unique symbol` properties, so a type
* object from a second copy of the library no longer satisfies the `DataType`
* type of this one even though it is structurally identical. Inputs that only
* need to be sanitized accept this looser shape instead.
*/
export type DataTypeLike = DataType | { readonly typeId: number };
/**
* Create an Arrow field backed by LanceDB's JSON extension type.
@@ -1059,6 +1068,71 @@ export async function fromRecordsToStreamBuffer(
return Buffer.from(await writer.toUint8Array());
}
// `Type.Utf8View` / `Type.BinaryView` as numbers: the enum members only exist
// in Arrow 21+, and this module compiles against every supported release.
const UTF8_VIEW_TYPE_ID = 24;
const BINARY_VIEW_TYPE_ID = 23;
/**
* Copy a Utf8View / BinaryView `Data` into a single `Data` of `type`.
*/
function materializeViewData(child: ArrowData, type: DataType): ArrowData {
const builder = makeBuilder({ type, nullValues: [null] });
for (const value of new Vector([child])) {
builder.append(value);
}
return builder.finish().flush();
}
/**
* Rebuild any top-level Utf8View / BinaryView column as Utf8 / Binary.
*
* Lance stores the view types as their offset-based equivalents anyway, so
* nothing is lost. Doing it here also sidesteps an Arrow JS 21 bug: its IPC
* writer emits a truncated views buffer for a *sliced* view array, which the
* Rust reader rejects with "Need at least N bytes in buffers[0]".
*
* The record batches are rebuilt positionally rather than through a
* `Record<string, Vector>`: JavaScript enumerates integer-like keys first, so
* a field named e.g. `"1"` would otherwise be paired with the wrong column.
*
* Tables without view columns are returned as-is.
*/
function materializeViewColumns(table: ArrowTable): ArrowTable {
const replacements = new Map<number, DataType>();
table.schema.fields.forEach((field, i) => {
if (field.type.typeId === UTF8_VIEW_TYPE_ID) {
replacements.set(i, new Utf8());
} else if (field.type.typeId === BINARY_VIEW_TYPE_ID) {
replacements.set(i, new Binary());
}
});
if (replacements.size === 0) {
return table;
}
const fields = table.schema.fields.map((field, i) => {
const type = replacements.get(i);
return type === undefined
? field
: new Field(field.name, type, field.nullable, field.metadata);
});
const schema = new Schema(fields, table.schema.metadata);
const batches = table.batches.map((batch) => {
const children = batch.data.children.map((child, i) => {
const type = replacements.get(i);
return type === undefined ? child : materializeViewData(child, type);
});
const data = makeData({
type: new Struct(fields),
length: batch.numRows,
nullCount: 0,
children,
});
return new RecordBatch(schema, data);
});
return new ArrowTable(schema, batches);
}
/**
* Serialize an Arrow Table into a buffer using the Arrow IPC File serialization
*
@@ -1075,7 +1149,9 @@ export async function fromTableToBuffer(
if (schema !== undefined && schema !== null) {
schema = sanitizeSchema(schema);
}
const tableWithEmbeddings = await applyEmbeddings(table, embeddings, schema);
const tableWithEmbeddings = materializeViewColumns(
await applyEmbeddings(table, embeddings, schema),
);
validateBlobSchema(tableWithEmbeddings.schema);
const writer = RecordBatchFileWriter.writeAll(tableWithEmbeddings);
return Buffer.from(await writer.toUint8Array());
@@ -1146,7 +1222,8 @@ export async function fromRecordBatchToBuffer(
export async function fromRecordBatchToStreamBuffer(
batch: RecordBatch,
): Promise<Buffer> {
const writer = RecordBatchStreamWriter.writeAll([batch]);
const table = materializeViewColumns(new ArrowTable([batch]));
const writer = RecordBatchStreamWriter.writeAll(table);
return Buffer.from(await writer.toUint8Array());
}
@@ -1163,7 +1240,9 @@ export async function fromTableToStreamBuffer(
embeddings?: EmbeddingFunctionConfig,
schema?: SchemaLike,
): Promise<Buffer> {
const tableWithEmbeddings = await applyEmbeddings(table, embeddings, schema);
const tableWithEmbeddings = materializeViewColumns(
await applyEmbeddings(table, embeddings, schema),
);
const writer = RecordBatchStreamWriter.writeAll(tableWithEmbeddings);
return Buffer.from(await writer.toUint8Array());
}
+95 -7
View File
@@ -141,13 +141,16 @@ export class BlobFile {
}
/**
* Reads from the cursor to the end and advances the cursor.
* Reads from the cursor and advances the cursor.
*
* A second call returns an empty buffer. {@link BlobFile.readRange} does
* not move the cursor.
* Reads to the end when `maxBytes` is omitted, or at most `maxBytes` bytes
* otherwise. Returns an empty buffer at the end of the blob.
* {@link BlobFile.readRange} does not move the cursor.
*/
read(): Promise<Buffer> {
return this.inner.read();
read(maxBytes?: bigint, options?: BlobReadOptions): Promise<Buffer> {
return runWithSignal(options?.signal, (signal) =>
this.inner.read(maxBytes, signal),
);
}
/**
@@ -155,11 +158,96 @@ export class BlobFile {
*
* Fails when `end` is past the blob size. Does not move the cursor.
*/
readRange(start: bigint, end: bigint): Promise<Buffer> {
return this.inner.readRange(start, end);
readRange(
start: bigint,
end: bigint,
options?: BlobReadOptions,
): Promise<Buffer> {
return runWithSignal(options?.signal, (signal) =>
this.inner.readRange(start, end, signal),
);
}
/**
* Reads several half-open byte ranges. Returns one buffer per range, in
* the order given.
*
* Fails when any `end` is past the blob size. Does not move the cursor.
*/
readRanges(
ranges: BlobRange[],
options?: BlobReadOptions,
): Promise<Buffer[]> {
return runWithSignal(options?.signal, (signal) =>
this.inner.readRanges(ranges, signal),
);
}
/** Moves the cursor to `position`, in bytes from the start of the blob. */
seek(position: bigint): Promise<void> {
return this.inner.seek(position);
}
/** Returns the cursor position in bytes. */
tell(): Promise<bigint> {
return this.inner.tell();
}
/**
* Releases the handle. Reads after `close()` fail. Calling it again does
* nothing.
*/
close(): Promise<void> {
return this.inner.close();
}
/** Returns true after {@link BlobFile.close}. */
isClosed(): boolean {
return this.inner.isClosed();
}
}
/** Options for blob reads. */
export type BlobReadOptions = {
/**
* Cancels the read. The call rejects with `signal.reason`, and the native
* read stops, including in-flight requests to a remote table.
*/
signal?: AbortSignal;
};
/** @ignore */
export async function runWithSignal<T>(
signal: AbortSignal | undefined,
run: (signal: AbortSignal | undefined) => Promise<T>,
): Promise<T> {
if (signal === undefined) {
return run(undefined);
}
signal.throwIfAborted();
// napi-rs replaces `onabort` on the signal it receives, so pass a private
// signal and forward the caller's abort to it.
const controller = new AbortController();
const onAbort = () => controller.abort();
signal.addEventListener("abort", onAbort, { once: true });
try {
return await run(controller.signal);
} catch (err) {
if (signal.aborted) {
throw signal.reason;
}
throw err;
} finally {
signal.removeEventListener("abort", onAbort);
}
}
/** A half-open byte range `[start, end)` for {@link BlobFile.readRanges}. */
export type BlobRange = {
start: bigint;
end: bigint;
};
export function coerceBlobValue(value: unknown): BlobInput | null {
if (value == null) {
return null;
+2 -1
View File
@@ -84,7 +84,7 @@ export {
} from "./arrow";
export { blob, isBlobField, BlobFile } from "./blob";
export type { BlobOptions } from "./blob";
export type { BlobOptions, BlobRange, BlobReadOptions } from "./blob";
export {
Connection,
@@ -195,6 +195,7 @@ export * as rerankers from "./rerankers";
export {
SchemaLike,
TableLike,
DataTypeLike,
FieldLike,
RecordBatchLike,
DataLike,
+137 -15
View File
@@ -9,6 +9,7 @@
// comes from the exact same library instance. This is not always the case
// and so we must sanitize the input to ensure that it is compatible.
import * as arrowModule from "apache-arrow";
import { BufferType, Data, Vector } from "apache-arrow";
import type { IntBitWidth, TKeys, TimeBitWidth } from "apache-arrow/type";
import {
@@ -43,6 +44,8 @@ import {
Interval,
IntervalDayTime,
IntervalYearMonth,
LargeBinary,
LargeUtf8,
List,
Map_,
Null,
@@ -74,6 +77,66 @@ import {
Utf8,
} from "./arrow";
/**
* Whether `value` was created by *this* copy of apache-arrow's `ctor`.
*
* Arrow 21 overrides `Symbol.hasInstance` on `Schema`, `Field`, `Table`,
* `RecordBatch`, `Vector` and `Data` so that `instanceof` also matches objects
* built by a different copy of the library. Those are exactly the objects
* this module has to rebuild: the `DataType` instances nested inside them are
* deliberately *not* given that treatment upstream, so Arrow's own builders and
* writers reject them. Walk the real prototype chain instead of trusting
* `instanceof`.
*/
function isOwnInstance<T>(
value: unknown,
// biome-ignore lint/suspicious/noExplicitAny: matches any class constructor
ctor: abstract new (...args: any[]) => T,
): value is T {
return (
typeof value === "object" &&
value !== null &&
Object.prototype.isPrototypeOf.call(ctor.prototype, value)
);
}
/**
* Types added to Arrow JS after release 18, keyed by the `Type` enum value
* they use. The library compiles and runs against every Arrow release in the
* peer range, so these are looked up by name at runtime instead of imported:
* an older Arrow simply does not have the class.
*/
type NewerTypeName =
| "IntervalMonthDayNano"
| "LargeList"
| "BinaryView"
| "Utf8View";
const NEWER_TYPES: ReadonlyArray<{
name: NewerTypeName;
typeId: number;
since: number;
}> = [
{ name: "IntervalMonthDayNano", typeId: -31, since: 20 },
{ name: "LargeList", typeId: 21, since: 21 },
{ name: "BinaryView", typeId: 23, since: 21 },
{ name: "Utf8View", typeId: 24, since: 21 },
];
// biome-ignore lint/suspicious/noExplicitAny: constructor arity varies by type
type DataTypeConstructor = new (...args: any[]) => DataType;
function newerTypeConstructor(name: NewerTypeName): DataTypeConstructor {
const ctor = (arrowModule as unknown as Record<string, unknown>)[name];
if (typeof ctor !== "function") {
const since = NEWER_TYPES.find((t) => t.name === name)?.since;
throw new Error(
`The Arrow type ${name} requires apache-arrow ${since} or newer, ` +
"but the installed apache-arrow does not provide it",
);
}
return ctor as DataTypeConstructor;
}
type SanitizationContext = {
types: WeakMap<object, DataType>;
vectors: WeakMap<object, Vector>;
@@ -210,19 +273,37 @@ export function sanitizeList(typeLike: object) {
return sanitizeListWithContext(typeLike, createSanitizationContext());
}
function sanitizeListChildWithContext(
typeLike: object,
context: SanitizationContext,
typeName: string,
): Field {
if (!("children" in typeLike) || !Array.isArray(typeLike.children)) {
throw Error(
`Expected a ${typeName} type to have an array-like \`children\` property`,
);
}
if (typeLike.children.length !== 1) {
throw Error(`Expected a ${typeName} type to have exactly one child`);
}
return sanitizeFieldWithContext(typeLike.children[0], context);
}
function sanitizeListWithContext(
typeLike: object,
context: SanitizationContext,
) {
if (!("children" in typeLike) || !Array.isArray(typeLike.children)) {
throw Error(
"Expected a List type to have an array-like `children` property",
);
}
if (typeLike.children.length !== 1) {
throw Error("Expected a List type to have exactly one child");
}
return new List(sanitizeFieldWithContext(typeLike.children[0], context));
return new List(sanitizeListChildWithContext(typeLike, context, "List"));
}
function sanitizeLargeListWithContext(
typeLike: object,
context: SanitizationContext,
) {
const LargeList = newerTypeConstructor("LargeList");
return new LargeList(
sanitizeListChildWithContext(typeLike, context, "LargeList"),
);
}
export function sanitizeStruct(typeLike: object) {
@@ -449,11 +530,30 @@ function sanitizeTypeWithContext(
return type;
}
function sanitizeNewerTypeById(
typeLike: object,
typeId: number,
context: SanitizationContext,
): DataType | undefined {
const entry = NEWER_TYPES.find((t) => t.typeId === typeId);
if (entry === undefined) {
return undefined;
}
if (entry.name === "LargeList") {
return sanitizeLargeListWithContext(typeLike, context);
}
return new (newerTypeConstructor(entry.name))();
}
function sanitizeTypeById(
typeLike: object,
typeId: Type,
context: SanitizationContext,
): DataType {
const newer = sanitizeNewerTypeById(typeLike, typeId, context);
if (newer !== undefined) {
return newer;
}
switch (typeId) {
case Type.NONE:
throw Error("Received a Type with a typeId of NONE");
@@ -465,8 +565,12 @@ function sanitizeTypeById(
return sanitizeFloat(typeLike);
case Type.Binary:
return new Binary();
case Type.LargeBinary:
return new LargeBinary();
case Type.Utf8:
return new Utf8();
case Type.LargeUtf8:
return new LargeUtf8();
case Type.Bool:
return new Bool();
case Type.Decimal:
@@ -566,7 +670,7 @@ function sanitizeFieldWithContext(
fieldLike: unknown,
context: SanitizationContext,
): Field {
if (fieldLike instanceof Field) {
if (isOwnInstance(fieldLike, Field)) {
return fieldLike;
}
if (typeof fieldLike !== "object" || fieldLike === null) {
@@ -620,7 +724,7 @@ function sanitizeSchemaWithContext(
schemaLike: SchemaLike,
context: SanitizationContext,
): Schema {
if (schemaLike instanceof Schema) {
if (isOwnInstance(schemaLike, Schema)) {
return schemaLike;
}
if (typeof schemaLike !== "object" || schemaLike === null) {
@@ -647,7 +751,7 @@ function sanitizeSchemaWithContext(
}
export function sanitizeTable(tableLike: TableLike): Table {
if (tableLike instanceof Table) {
if (isOwnInstance(tableLike, Table)) {
return tableLike;
}
if (typeof tableLike !== "object" || tableLike === null) {
@@ -675,7 +779,7 @@ function sanitizeRecordBatch(
batchLike: RecordBatchLike,
context: SanitizationContext,
): RecordBatch {
if (batchLike instanceof RecordBatch) {
if (isOwnInstance(batchLike, RecordBatch)) {
return batchLike;
}
if (typeof batchLike !== "object" || batchLike === null) {
@@ -704,11 +808,17 @@ type DictionaryDataLike = DataLike & {
dictionary?: DictionaryVectorLike;
};
// Arrow 21 added `variadicBuffers` to `Data` for the Utf8View / BinaryView
// types. Older Arrow versions do not have the property, so it is optional.
type VariadicDataLike = DataLike & {
variadicBuffers?: ReadonlyArray<Uint8Array>;
};
function sanitizeData(
dataLike: DataLike,
context: SanitizationContext,
): Data<DataType> {
if (dataLike instanceof Data) {
if (isOwnInstance(dataLike, Data)) {
return dataLike;
}
const cachedData = context.data.get(dataLike);
@@ -726,7 +836,12 @@ function sanitizeData(
context.vectors.set(dictionaryLike, dictionary);
}
}
const data = new Data(
// Arrow 21 added an eighth constructor parameter (`variadicBuffers`); older
// releases ignore the extra argument but their typings do not declare it.
const DataCtor = Data as unknown as new (
...args: unknown[]
) => Data<DataType>;
const data = new DataCtor(
sanitizeTypeWithContext(dataLike.type, context),
dataLike.offset,
dataLike.length,
@@ -739,6 +854,7 @@ function sanitizeData(
},
dataLike.children.map((child) => sanitizeData(child, context)),
dictionary,
(dataLike as VariadicDataLike).variadicBuffers,
);
context.data.set(dataLike, data);
return data;
@@ -747,7 +863,11 @@ function sanitizeData(
const constructorsByTypeName = {
null: () => new Null(),
binary: () => new Binary(),
largebinary: () => new LargeBinary(),
binaryview: () => new (newerTypeConstructor("BinaryView"))(),
utf8: () => new Utf8(),
largeutf8: () => new LargeUtf8(),
utf8view: () => new (newerTypeConstructor("Utf8View"))(),
bool: () => new Bool(),
int8: () => new Int8(),
int16: () => new Int16(),
@@ -768,6 +888,8 @@ const constructorsByTypeName = {
timesecond: () => new TimeSecond(),
intervaldaytime: () => new IntervalDayTime(),
intervalyearmonth: () => new IntervalYearMonth(),
intervalmonthdaynano: () =>
new (newerTypeConstructor("IntervalMonthDayNano"))(),
durationnanosecond: () => new DurationNanosecond(),
durationmicrosecond: () => new DurationMicrosecond(),
durationmillisecond: () => new DurationMillisecond(),
+10 -5
View File
@@ -17,7 +17,7 @@ import {
tableFromIPC,
} from "./arrow";
import { BlobFile } from "./blob";
import { BlobFile, BlobReadOptions, runWithSignal } from "./blob";
import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry";
import { IndexOptions } from "./indices";
import { Job } from "./job";
@@ -545,6 +545,7 @@ export abstract class Table {
abstract fetchBlobs(
column: string,
rowIds: readonly (bigint | number)[],
options?: BlobReadOptions,
): Promise<(Buffer | null)[]>;
/**
@@ -557,6 +558,7 @@ export abstract class Table {
abstract fetchBlobFiles(
column: string,
rowIds: readonly (bigint | number)[],
options?: BlobReadOptions,
): Promise<(BlobFile | null)[]>;
/**
@@ -1248,8 +1250,11 @@ export class LocalTable extends Table {
async fetchBlobs(
column: string,
rowIds: readonly (bigint | number)[],
options?: BlobReadOptions,
): Promise<(Buffer | null)[]> {
const values = await this.inner.fetchBlobs(column, rowIdsToBigInts(rowIds));
const values = await runWithSignal(options?.signal, (signal) =>
this.inner.fetchBlobs(column, rowIdsToBigInts(rowIds), signal),
);
// N-API Option maps missing values to undefined. Collapse those to null.
return values.map((value) => value ?? null);
}
@@ -1257,10 +1262,10 @@ export class LocalTable extends Table {
async fetchBlobFiles(
column: string,
rowIds: readonly (bigint | number)[],
options?: BlobReadOptions,
): Promise<(BlobFile | null)[]> {
const files = await this.inner.fetchBlobFiles(
column,
rowIdsToBigInts(rowIds),
const files = await runWithSignal(options?.signal, (signal) =>
this.inner.fetchBlobFiles(column, rowIdsToBigInts(rowIds), signal),
);
// N-API Option maps missing values to undefined. Collapse those to null.
return files.map((file) =>
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-darwin-arm64",
"version": "0.40.0-beta.11",
"version": "0.40.0-beta.12",
"os": ["darwin"],
"cpu": ["arm64"],
"main": "lancedb.darwin-arm64.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-gnu",
"version": "0.40.0-beta.11",
"version": "0.40.0-beta.12",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-musl",
"version": "0.40.0-beta.11",
"version": "0.40.0-beta.12",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-gnu",
"version": "0.40.0-beta.11",
"version": "0.40.0-beta.12",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-musl",
"version": "0.40.0-beta.11",
"version": "0.40.0-beta.12",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-arm64-msvc",
"version": "0.40.0-beta.11",
"version": "0.40.0-beta.12",
"os": [
"win32"
],
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-x64-msvc",
"version": "0.40.0-beta.11",
"version": "0.40.0-beta.12",
"os": ["win32"],
"cpu": ["x64"],
"main": "lancedb.win32-x64-msvc.node",
+6 -2
View File
@@ -11,7 +11,7 @@
"ann"
],
"private": false,
"version": "0.40.0-beta.11",
"version": "0.40.0-beta.12",
"main": "dist/index.js",
"exports": {
".": "./dist/index.js",
@@ -49,10 +49,14 @@
"@types/jest": "^29.1.2",
"@types/node": "22.7.4",
"@types/tmp": "^0.2.6",
"apache-arrow": "21.2.0",
"apache-arrow-15": "npm:apache-arrow@15.0.0",
"apache-arrow-16": "npm:apache-arrow@16.0.0",
"apache-arrow-17": "npm:apache-arrow@17.0.0",
"apache-arrow-18": "npm:apache-arrow@18.0.0",
"apache-arrow-19": "npm:apache-arrow@19.0.1",
"apache-arrow-20": "npm:apache-arrow@20.0.0",
"apache-arrow-21": "npm:apache-arrow@21.2.0",
"eslint": "^8.57.0",
"jest": "^29.7.0",
"shx": "^0.3.4",
@@ -102,7 +106,7 @@
},
"peerDependencies": {
"@types/node": ">=22",
"apache-arrow": ">=15.0.0 <=18.1.0"
"apache-arrow": ">=15.0.0 <22.0.0"
},
"peerDependenciesMeta": {
"@types/node": {
+91 -7
View File
@@ -14,9 +14,6 @@ importers:
'@opentelemetry/api':
specifier: ^1.9.0
version: 1.9.1
apache-arrow:
specifier: '>=15.0.0 <=18.1.0'
version: 18.1.0
reflect-metadata:
specifier: ^0.2.2
version: 0.2.2
@@ -54,6 +51,9 @@ importers:
'@types/tmp':
specifier: ^0.2.6
version: 0.2.6
apache-arrow:
specifier: 21.2.0
version: 21.2.0
apache-arrow-15:
specifier: npm:apache-arrow@15.0.0
version: apache-arrow@15.0.0
@@ -66,6 +66,15 @@ importers:
apache-arrow-18:
specifier: npm:apache-arrow@18.0.0
version: apache-arrow@18.0.0
apache-arrow-19:
specifier: npm:apache-arrow@19.0.1
version: apache-arrow@19.0.1
apache-arrow-20:
specifier: npm:apache-arrow@20.0.0
version: apache-arrow@20.0.0
apache-arrow-21:
specifier: npm:apache-arrow@21.2.0
version: apache-arrow@21.2.0
eslint:
specifier: ^8.57.0
version: 8.57.1
@@ -1703,6 +1712,9 @@ packages:
'@types/node@22.7.4':
resolution: {integrity: sha512-y+NPi1rFzDs1NdQHHToqeiX2TIS79SWEAw9GYhkkx8bD0ChpfqC+n2j5OXOCpzfojBEBt6DnEnnG9MY0zk1XLg==}
'@types/node@25.9.8':
resolution: {integrity: sha512-VfMrScDmMhUJQmd5hArdQnFvK0OIeD36uN2Va1FcpYaLB/BgkgM8Ulc50XtYISPMz7APJ30+N0T5EM0jlcdfRw==}
'@types/stack-utils@2.0.3':
resolution: {integrity: sha512-9aEbYZ3TbYMznPdcdr3SmIrLXwC/AKZXQeCf9Pgao5CKb8CyHuEX5jzWPTkvregvhRJHcpRO6BFoGW9ycaOkYw==}
@@ -1840,8 +1852,16 @@ packages:
resolution: {integrity: sha512-gFlPaqN9osetbB83zC29AbbZqGiCuFH1vyyPseJ+B7SIbfBtESV62mMT/CkiIt77W6ykC/nTWFzTXFs0Uldg4g==}
hasBin: true
apache-arrow@18.1.0:
resolution: {integrity: sha512-v/ShMp57iBnBp4lDgV8Jx3d3Q5/Hac25FWmQ98eMahUiHPXcvwIMKJD0hBIgclm/FCG+LwPkAKtkRO1O/W0YGg==}
apache-arrow@19.0.1:
resolution: {integrity: sha512-APmMLzS4qbTivLrPdQXexGM4JRr+0g62QDaobzEvip/FdQIrv2qLy0mD5Qdmw4buydtVJgbFeKR8f59I6PPGDg==}
hasBin: true
apache-arrow@20.0.0:
resolution: {integrity: sha512-JUeK0jFRUd7rbmrhhzR3O2KXjLaZ4YYYFOptyUfxOsMIoZCPi6bZR58gVi/xi3HTBMPseXm9PXyQ2V916930pA==}
hasBin: true
apache-arrow@21.2.0:
resolution: {integrity: sha512-Hxe6Agq26gQOM954qpzYSllJBPJl+e16U5CkfuMUhLrNba+5nKkttIVlflaovN6oaTratqMGAO8H5u/aNhmHWQ==}
hasBin: true
argparse@1.0.10:
@@ -2033,6 +2053,15 @@ packages:
resolution: {integrity: sha512-H4UfQhZyakIjC74I9d34fGYDwk3XpSr17QhEd0Q3I9Xq1CETHo4Hcuo87WyWHpAF1aSLjLRf5lD9ZGX2qStUvg==}
engines: {node: '>=4.0.0'}
command-line-args@6.0.2:
resolution: {integrity: sha512-AIjYVxrV9X752LmPDLbVYv8aMCuHPSLZJXEo2qo/xJfv+NYhaZ4sMSF01rM+gHPaMgvPM0l5D/F+Qx+i2WfSmQ==}
engines: {node: '>=12.20'}
peerDependencies:
'@75lb/nature': latest
peerDependenciesMeta:
'@75lb/nature':
optional: true
command-line-usage@7.0.4:
resolution: {integrity: sha512-85UdvzTNx/+s5CkSgBm/0hzP80RFHAa7PsfeADE5ezZF3uHz3/Tqj9gIKGT9PTtpycc3Ua64T0oVulGfKxzfqg==}
engines: {node: '>=12.20.0'}
@@ -2281,6 +2310,15 @@ packages:
resolution: {integrity: sha512-6Tb2myMioCAgv5kfvP5/PkZZ/ntTpVK39fHY7WkWBgvbeE+VHd/tZuZ4mrC+bxh4cfOZeYKVPaJIZtZXV7GNCQ==}
engines: {node: '>=4.0.0'}
find-replace@5.0.2:
resolution: {integrity: sha512-Y45BAiE3mz2QsrN2fb5QEtO4qb44NcS7en/0y9PEVsg351HsLeVclP8QPMH79Le9sH3rs5RSwJu99W0WPZO43Q==}
engines: {node: '>=14'}
peerDependencies:
'@75lb/nature': latest
peerDependenciesMeta:
'@75lb/nature':
optional: true
find-up@4.1.0:
resolution: {integrity: sha512-PpOwAdQ/YlXQ2vj8a3h8IipDuYRi3wceVQQGYWxNINccq40Anw7BlsEXCMbt1Zt+OLA6Fq9suIpIWD0OsnISlw==}
engines: {node: '>=8'}
@@ -2302,6 +2340,9 @@ packages:
flatbuffers@24.12.23:
resolution: {integrity: sha512-dLVCAISd5mhls514keQzmEG6QHmUUsNuWsb4tFafIUwvvgDjXhtfAYSKOzt5SWOy+qByV5pbsDZ+Vb7HUOBEdA==}
flatbuffers@25.9.23:
resolution: {integrity: sha512-MI1qs7Lo4Syw0EOzUl0xjs2lsoeqFku44KpngfIduHBYvzm8h2+7K8YMQh1JtVVVrUvhLpNwqVi4DERegUJhPQ==}
flatted@3.4.2:
resolution: {integrity: sha512-PjDse7RzhcPkIJwy5t7KPWQSZ9cAbzQXcafsetQoD7sOJRQlGikNbx7yZp2OotDnJyrDcbyRq3Ttb18iYOqkxA==}
@@ -3345,6 +3386,9 @@ packages:
undici-types@6.21.0:
resolution: {integrity: sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==}
undici-types@7.24.6:
resolution: {integrity: sha512-WRNW+sJgj5OBN4/0JpHFqtqzhpbnV0GuB+OozA9gCL7a993SmU+1JBZCzLNxYsbMfIeDL+lTsphD5jN5N+n0zg==}
unist-util-is@6.0.1:
resolution: {integrity: sha512-LsiILbtBETkDz8I9p1dQ0uyRUWuaQzd/cuEeS1hoRSyW5E5XGmTzlwY1OrNzzakGowI9Dr/I8HVaw4hTtnxy8g==}
@@ -5505,6 +5549,10 @@ snapshots:
dependencies:
undici-types: 6.19.8
'@types/node@25.9.8':
dependencies:
undici-types: 7.24.6
'@types/stack-utils@2.0.3': {}
'@types/tmp@0.2.6': {}
@@ -5694,17 +5742,40 @@ snapshots:
json-bignum: 0.0.3
tslib: 2.8.1
apache-arrow@18.1.0:
apache-arrow@19.0.1:
dependencies:
'@swc/helpers': 0.5.21
'@types/command-line-args': 5.2.3
'@types/command-line-usage': 5.0.4
'@types/node': 20.19.41
command-line-args: 5.2.1
command-line-args: 6.0.2
command-line-usage: 7.0.4
flatbuffers: 24.12.23
json-bignum: 0.0.3
tslib: 2.8.1
transitivePeerDependencies:
- '@75lb/nature'
apache-arrow@20.0.0:
dependencies:
'@swc/helpers': 0.5.21
'@types/command-line-args': 5.2.3
'@types/command-line-usage': 5.0.4
'@types/node': 20.19.41
command-line-args: 6.0.2
command-line-usage: 7.0.4
flatbuffers: 25.9.23
json-bignum: 0.0.3
tslib: 2.8.1
transitivePeerDependencies:
- '@75lb/nature'
apache-arrow@21.2.0:
dependencies:
'@types/node': 25.9.8
flatbuffers: 25.9.23
json-with-bigint: 3.5.8
tslib: 2.8.1
argparse@1.0.10:
dependencies:
@@ -5906,6 +5977,13 @@ snapshots:
lodash.camelcase: 4.3.0
typical: 4.0.0
command-line-args@6.0.2:
dependencies:
array-back: 6.2.3
find-replace: 5.0.2
lodash.camelcase: 4.3.0
typical: 7.3.0
command-line-usage@7.0.4:
dependencies:
array-back: 6.2.3
@@ -6178,6 +6256,8 @@ snapshots:
dependencies:
array-back: 3.1.0
find-replace@5.0.2: {}
find-up@4.1.0:
dependencies:
locate-path: 5.0.0
@@ -6201,6 +6281,8 @@ snapshots:
flatbuffers@24.12.23: {}
flatbuffers@25.9.23: {}
flatted@3.4.2: {}
follow-redirects@1.16.0: {}
@@ -7405,6 +7487,8 @@ snapshots:
undici-types@6.21.0: {}
undici-types@7.24.6: {}
unist-util-is@6.0.1:
dependencies:
'@types/unist': 3.0.3
+133 -10
View File
@@ -1,25 +1,45 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
use std::future::Future;
use std::ops::Range;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use arrow_array::{Array, LargeBinaryArray};
use futures::future::abortable;
use lancedb::blob::BlobFile as LanceBlobFile;
use napi::Env;
use napi::bindgen_prelude::*;
use napi_derive::napi;
use crate::error::convert_error;
#[napi(object)]
pub struct BlobRange {
pub start: BigInt,
pub end: BigInt,
}
#[napi]
pub struct BlobFile {
inner: Arc<LanceBlobFile>,
closed: AtomicBool,
}
impl BlobFile {
pub(crate) fn new(inner: LanceBlobFile) -> Self {
Self {
inner: Arc::new(inner),
closed: AtomicBool::new(false),
}
}
fn ensure_open(&self) -> napi::Result<()> {
if self.closed.load(Ordering::Acquire) {
Err(napi::Error::from_reason("blob file is already closed"))
} else {
Ok(())
}
}
}
@@ -32,21 +52,124 @@ impl BlobFile {
}
#[napi]
pub async fn read(&self) -> napi::Result<Buffer> {
let bytes = self.inner.read().await.map_err(|err| convert_error(&err))?;
Ok(Buffer::from(bytes.as_ref()))
pub fn read<'env>(
&self,
env: &'env Env,
max_bytes: Option<BigInt>,
signal: Option<AbortSignal>,
) -> napi::Result<PromiseRaw<'env, Buffer>> {
self.ensure_open()?;
let max_bytes = max_bytes
.map(|max_bytes| {
usize::try_from(parse_u64(max_bytes, "maxBytes")?)
.map_err(|_| napi::Error::from_reason("maxBytes is too large"))
})
.transpose()?;
let inner = self.inner.clone();
spawn_abortable(env, signal, async move {
let bytes = match max_bytes {
None => inner.read().await,
Some(max_bytes) => inner.read_up_to(max_bytes).await,
}
.map_err(|err| convert_error(&err))?;
Ok(Buffer::from(bytes.as_ref()))
})
}
#[napi]
pub async fn read_range(&self, start: BigInt, end: BigInt) -> napi::Result<Buffer> {
pub fn read_range<'env>(
&self,
env: &'env Env,
start: BigInt,
end: BigInt,
signal: Option<AbortSignal>,
) -> napi::Result<PromiseRaw<'env, Buffer>> {
self.ensure_open()?;
let range = bigint_range(start, end)?;
let bytes = self
.inner
.read_range(range)
.await
.map_err(|err| convert_error(&err))?;
Ok(Buffer::from(bytes.as_ref()))
let inner = self.inner.clone();
spawn_abortable(env, signal, async move {
let bytes = inner
.read_range(range)
.await
.map_err(|err| convert_error(&err))?;
Ok(Buffer::from(bytes.as_ref()))
})
}
#[napi]
pub fn read_ranges<'env>(
&self,
env: &'env Env,
ranges: Vec<BlobRange>,
signal: Option<AbortSignal>,
) -> napi::Result<PromiseRaw<'env, Vec<Buffer>>> {
self.ensure_open()?;
let ranges = ranges
.into_iter()
.map(|range| bigint_range(range.start, range.end))
.collect::<napi::Result<Vec<_>>>()?;
let inner = self.inner.clone();
spawn_abortable(env, signal, async move {
let buffers = inner
.read_ranges(&ranges)
.await
.map_err(|err| convert_error(&err))?;
Ok(buffers
.iter()
.map(|bytes| Buffer::from(bytes.as_ref()))
.collect())
})
}
#[napi]
pub async fn seek(&self, position: BigInt) -> napi::Result<()> {
self.ensure_open()?;
let position = parse_u64(position, "position")?;
self.inner
.seek(position)
.await
.map_err(|err| convert_error(&err))
}
#[napi]
pub async fn tell(&self) -> napi::Result<BigInt> {
self.ensure_open()?;
let position = self.inner.tell().await.map_err(|err| convert_error(&err))?;
Ok(BigInt::from(position))
}
#[napi]
pub async fn close(&self) -> napi::Result<()> {
if self.closed.swap(true, Ordering::AcqRel) {
return Ok(());
}
self.inner.close().await.map_err(|err| convert_error(&err))
}
#[napi]
pub fn is_closed(&self) -> bool {
self.closed.load(Ordering::Acquire)
}
}
/// Runs `fut` on the napi runtime. Aborting `signal` drops the future, which
/// cancels in-flight reads.
pub fn spawn_abortable<'env, T>(
env: &'env Env,
signal: Option<AbortSignal>,
fut: impl Future<Output = napi::Result<T>> + Send + 'static,
) -> napi::Result<PromiseRaw<'env, T>>
where
T: ToNapiValue + Send + 'static,
{
let (fut, handle) = abortable(fut);
if let Some(signal) = signal {
signal.on_abort(move || handle.abort());
}
env.spawn_future(async move {
fut.await
.unwrap_or_else(|_| Err(napi::Error::from_reason("blob read aborted")))
})
}
fn bigint_range(start: BigInt, end: BigInt) -> napi::Result<Range<u64>> {
+1
View File
@@ -658,6 +658,7 @@ impl JsFullTextQuery {
FtsQuery::Phrase(_) => "phrase".to_string(),
FtsQuery::Boost(_) => "boost".to_string(),
FtsQuery::MultiMatch(_) => "multi_match".to_string(),
FtsQuery::CombinedFields(_) => "combined_fields".to_string(),
FtsQuery::Boolean(_) => "boolean".to_string(),
}
}
+29 -20
View File
@@ -11,11 +11,12 @@ use lancedb::table::{
FieldMetadataUpdate as LanceFieldMetadataUpdate, FtsToken as LanceDbFtsToken,
NewColumnTransform, OptimizeAction, OptimizeOptions, Ref, Table as LanceDbTable,
};
use napi::Env;
use napi::bindgen_prelude::*;
use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode};
use napi_derive::napi;
use crate::blob::{BlobFile, copy_blob_buffers, parse_row_ids};
use crate::blob::{BlobFile, copy_blob_buffers, parse_row_ids, spawn_abortable};
use crate::error::NapiErrorExt;
use crate::index::Index;
use crate::merge::NativeMergeInsertBuilder;
@@ -340,36 +341,44 @@ impl Table {
}
#[napi(catch_unwind)]
pub async fn fetch_blobs(
pub fn fetch_blobs<'env>(
&self,
env: &'env Env,
column: String,
row_ids: Vec<BigInt>,
) -> napi::Result<Vec<Option<Buffer>>> {
signal: Option<AbortSignal>,
) -> napi::Result<PromiseRaw<'env, Vec<Option<Buffer>>>> {
let row_ids = parse_row_ids(row_ids)?;
let array = self
.inner_ref()?
.fetch_blobs(column.as_str(), &row_ids)
.await
.default_error()?;
Ok(copy_blob_buffers(array))
let table = self.inner_ref()?.clone();
spawn_abortable(env, signal, async move {
let array = table
.fetch_blobs(column.as_str(), &row_ids)
.await
.default_error()?;
Ok(copy_blob_buffers(array))
})
}
#[napi(catch_unwind)]
pub async fn fetch_blob_files(
pub fn fetch_blob_files<'env>(
&self,
env: &'env Env,
column: String,
row_ids: Vec<BigInt>,
) -> napi::Result<Vec<Option<BlobFile>>> {
signal: Option<AbortSignal>,
) -> napi::Result<PromiseRaw<'env, Vec<Option<BlobFile>>>> {
let row_ids = parse_row_ids(row_ids)?;
let files = self
.inner_ref()?
.fetch_blob_files(column.as_str(), &row_ids)
.await
.default_error()?;
Ok(files
.into_iter()
.map(|file| file.map(BlobFile::new))
.collect())
let table = self.inner_ref()?.clone();
spawn_abortable(env, signal, async move {
let files = table
.fetch_blob_files(column.as_str(), &row_ids)
.await
.default_error()?;
Ok(files
.into_iter()
.map(|file| file.map(BlobFile::new))
.collect())
})
}
#[napi(catch_unwind)]
+4 -1
View File
@@ -10,7 +10,10 @@
"resolveJsonModule": true,
"emitDecoratorMetadata": true,
"experimentalDecorators": true,
"moduleResolution": "Node"
"moduleResolution": "Node",
// flatbuffers 25 (via apache-arrow 21) ships declarations that need
// TypeScript 5.7+ generic typed arrays; skip checking third-party .d.ts.
"skipLibCheck": true
},
"exclude": ["./dist/*", "./examples/*"],
"typedocOptions": {