From 0a1c1a04c78bb71cc8f3b4231694510201b3f70c Mon Sep 17 00:00:00 2001 From: Bilal Atique <74999164+BilalAtique@users.noreply.github.com> Date: Wed, 30 Sep 2026 00:01:17 +0500 Subject: [PATCH] feat(node): accept an AbortSignal to cancel blob reads (#4350) Closes #4288. Blob reads in Node can now be cancelled. `BlobFile.read`, `readRange` and `readRanges`, and `Table.fetchBlobs` and `fetchBlobFiles` take an optional `{ signal?: AbortSignal }` (new `BlobReadOptions` type). When the signal aborts, the call rejects with `signal.reason` (an `AbortError` unless the caller gave a reason) and the native future is dropped, so the read stops instead of running to completion. A signal that is already aborted rejects before any read starts. How it works: - On the Rust side, these five methods are now sync `#[napi]` methods that return `env.spawn_future(...)`. The future is wrapped in `futures::future::abortable`, and the abort handle is registered with napi's `AbortSignal::on_abort`. napi's `AbortSignal` is `Rc`-based, so it can't be an argument of an `async fn`, which is why the methods changed shape. The TypeScript signatures are unchanged apart from the new optional argument. - napi-rs installs its own `onabort` on the signal it receives, which would overwrite a handler the caller set. So the TypeScript wrapper passes a private `AbortController` signal to the native call for each request, forwards the caller's abort to it with `addEventListener(..., { once: true })`, and removes the listener when the call settles. Tests in `table.test.ts`: - an already aborted signal rejects on all five methods, and the handle still reads afterwards - a custom abort reason is what the call rejects with - a read with a signal that never aborts returns normally and leaves no `abort` listener on the signal - aborting a 32 MiB `readRange` right after it starts rejects with `AbortError`, and the same handle reads again afterwards The Node docs under `docs/src/js` are regenerated. I didn't add a test for where the `read()` cursor ends up after a mid-flight abort on a local table. The remote `BlobFile` documents that a cancelled read leaves the cursor unchanged, but I haven't confirmed the same for the native Lance handle, so the docs don't claim it. --- docs/src/js/classes/BlobFile.md | 15 ++- docs/src/js/classes/Table.md | 14 ++- docs/src/js/globals.md | 1 + docs/src/js/type-aliases/BlobReadOptions.md | 24 +++++ nodejs/__test__/table.test.ts | 75 ++++++++++++++ nodejs/lancedb/blob.ts | 60 ++++++++++-- nodejs/lancedb/index.ts | 2 +- nodejs/lancedb/table.ts | 15 ++- nodejs/src/blob.rs | 102 ++++++++++++++------ nodejs/src/table.rs | 49 ++++++---- 10 files changed, 293 insertions(+), 64 deletions(-) create mode 100644 docs/src/js/type-aliases/BlobReadOptions.md diff --git a/docs/src/js/classes/BlobFile.md b/docs/src/js/classes/BlobFile.md index ee9f9ad28..ee012be25 100644 --- a/docs/src/js/classes/BlobFile.md +++ b/docs/src/js/classes/BlobFile.md @@ -42,7 +42,7 @@ Returns true after [BlobFile.close](BlobFile.md#close). ### read() ```ts -read(maxBytes?): Promise +read(maxBytes?, options?): Promise ``` Reads from the cursor and advances the cursor. @@ -55,6 +55,8 @@ otherwise. Returns an empty buffer at the end of the blob. * **maxBytes?**: `bigint` +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) + #### Returns `Promise`<`Buffer`> @@ -64,7 +66,10 @@ otherwise. Returns an empty buffer at the end of the blob. ### readRange() ```ts -readRange(start, end): Promise +readRange( + start, + end, + options?): Promise ``` Reads the half-open byte range `[start, end)`. @@ -77,6 +82,8 @@ Fails when `end` is past the blob size. Does not move the cursor. * **end**: `bigint` +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) + #### Returns `Promise`<`Buffer`> @@ -86,7 +93,7 @@ Fails when `end` is past the blob size. Does not move the cursor. ### readRanges() ```ts -readRanges(ranges): Promise +readRanges(ranges, options?): Promise ``` Reads several half-open byte ranges. Returns one buffer per range, in @@ -98,6 +105,8 @@ Fails when any `end` is past the blob size. Does not move the cursor. * **ranges**: [`BlobRange`](../type-aliases/BlobRange.md)[] +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) + #### Returns `Promise`<`Buffer`[]> diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index 17e2d4f34..0d813ff1e 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -516,7 +516,10 @@ Drop an index from the table. ### fetchBlobFiles() ```ts -abstract fetchBlobFiles(column, rowIds): Promise<(null | BlobFile)[]> +abstract fetchBlobFiles( + column, + rowIds, + options?): Promise<(null | BlobFile)[]> ``` Opens lazy blob handles for `column` at the given row IDs using the @@ -531,6 +534,8 @@ See [Table.fetchBlobs](Table.md#fetchblobs) for row-ID validity across versions. * **rowIds**: readonly (`number` \| `bigint`)[] +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) + #### Returns `Promise`<(`null` \| [`BlobFile`](BlobFile.md))[]> @@ -540,7 +545,10 @@ See [Table.fetchBlobs](Table.md#fetchblobs) for row-ID validity across versions. ### fetchBlobs() ```ts -abstract fetchBlobs(column, rowIds): Promise<(null | Buffer)[]> +abstract fetchBlobs( + column, + rowIds, + options?): Promise<(null | Buffer)[]> ``` Bytes for `column` at row IDs from [Query.withRowId](Query.md#withrowid). @@ -559,6 +567,8 @@ memory; use [Table.fetchBlobFiles](Table.md#fetchblobfiles) for large values. * **rowIds**: readonly (`number` \| `bigint`)[] +* **options?**: [`BlobReadOptions`](../type-aliases/BlobReadOptions.md) + #### Returns `Promise`<(`null` \| `Buffer`)[]> diff --git a/docs/src/js/globals.md b/docs/src/js/globals.md index 9b9f76334..a82d9e334 100644 --- a/docs/src/js/globals.md +++ b/docs/src/js/globals.md @@ -159,6 +159,7 @@ - [BaseTokenizer](type-aliases/BaseTokenizer.md) - [BlobOptions](type-aliases/BlobOptions.md) - [BlobRange](type-aliases/BlobRange.md) +- [BlobReadOptions](type-aliases/BlobReadOptions.md) - [Data](type-aliases/Data.md) - [DataLike](type-aliases/DataLike.md) - [FieldLike](type-aliases/FieldLike.md) diff --git a/docs/src/js/type-aliases/BlobReadOptions.md b/docs/src/js/type-aliases/BlobReadOptions.md new file mode 100644 index 000000000..17264b0e4 --- /dev/null +++ b/docs/src/js/type-aliases/BlobReadOptions.md @@ -0,0 +1,24 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / BlobReadOptions + +# Type Alias: BlobReadOptions + +```ts +type BlobReadOptions: object; +``` + +Options for blob reads. + +## Type declaration + +### signal? + +```ts +optional signal: AbortSignal; +``` + +Cancels the read. The call rejects with `signal.reason`, and the native +read stops, including in-flight requests to a remote table. diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts index 665b0653a..904c282d4 100644 --- a/nodejs/__test__/table.test.ts +++ b/nodejs/__test__/table.test.ts @@ -2,6 +2,7 @@ // 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"; @@ -2746,6 +2747,80 @@ describe("when dealing with blob columns", () => { 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); diff --git a/nodejs/lancedb/blob.ts b/nodejs/lancedb/blob.ts index fc8c43196..e6c780ca9 100644 --- a/nodejs/lancedb/blob.ts +++ b/nodejs/lancedb/blob.ts @@ -147,8 +147,10 @@ export class BlobFile { * otherwise. Returns an empty buffer at the end of the blob. * {@link BlobFile.readRange} does not move the cursor. */ - read(maxBytes?: bigint): Promise { - return this.inner.read(maxBytes); + read(maxBytes?: bigint, options?: BlobReadOptions): Promise { + return runWithSignal(options?.signal, (signal) => + this.inner.read(maxBytes, signal), + ); } /** @@ -156,8 +158,14 @@ export class BlobFile { * * Fails when `end` is past the blob size. Does not move the cursor. */ - readRange(start: bigint, end: bigint): Promise { - return this.inner.readRange(start, end); + readRange( + start: bigint, + end: bigint, + options?: BlobReadOptions, + ): Promise { + return runWithSignal(options?.signal, (signal) => + this.inner.readRange(start, end, signal), + ); } /** @@ -166,8 +174,13 @@ export class BlobFile { * * Fails when any `end` is past the blob size. Does not move the cursor. */ - readRanges(ranges: BlobRange[]): Promise { - return this.inner.readRanges(ranges); + readRanges( + ranges: BlobRange[], + options?: BlobReadOptions, + ): Promise { + return runWithSignal(options?.signal, (signal) => + this.inner.readRanges(ranges, signal), + ); } /** Moves the cursor to `position`, in bytes from the start of the blob. */ @@ -194,6 +207,41 @@ export class BlobFile { } } +/** 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( + signal: AbortSignal | undefined, + run: (signal: AbortSignal | undefined) => Promise, +): Promise { + 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; diff --git a/nodejs/lancedb/index.ts b/nodejs/lancedb/index.ts index 2e444784b..a1776e3f4 100644 --- a/nodejs/lancedb/index.ts +++ b/nodejs/lancedb/index.ts @@ -84,7 +84,7 @@ export { } from "./arrow"; export { blob, isBlobField, BlobFile } from "./blob"; -export type { BlobOptions, BlobRange } from "./blob"; +export type { BlobOptions, BlobRange, BlobReadOptions } from "./blob"; export { Connection, diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index 132e8bc20..d4c5db509 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -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) => diff --git a/nodejs/src/blob.rs b/nodejs/src/blob.rs index 413bf6d97..e47c693d4 100644 --- a/nodejs/src/blob.rs +++ b/nodejs/src/blob.rs @@ -1,12 +1,15 @@ // 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; @@ -49,48 +52,73 @@ impl BlobFile { } #[napi] - pub async fn read(&self, max_bytes: Option) -> napi::Result { + pub fn read<'env>( + &self, + env: &'env Env, + max_bytes: Option, + signal: Option, + ) -> napi::Result> { self.ensure_open()?; - let bytes = match max_bytes { - None => self.inner.read().await, - Some(max_bytes) => { - let max_bytes = usize::try_from(parse_u64(max_bytes, "maxBytes")?) - .map_err(|_| napi::Error::from_reason("maxBytes is too large"))?; - self.inner.read_up_to(max_bytes).await + 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())) + .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 { + pub fn read_range<'env>( + &self, + env: &'env Env, + start: BigInt, + end: BigInt, + signal: Option, + ) -> napi::Result> { 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 async fn read_ranges(&self, ranges: Vec) -> napi::Result> { + pub fn read_ranges<'env>( + &self, + env: &'env Env, + ranges: Vec, + signal: Option, + ) -> napi::Result>> { self.ensure_open()?; let ranges = ranges .into_iter() .map(|range| bigint_range(range.start, range.end)) .collect::>>()?; - let buffers = self - .inner - .read_ranges(&ranges) - .await - .map_err(|err| convert_error(&err))?; - Ok(buffers - .iter() - .map(|bytes| Buffer::from(bytes.as_ref())) - .collect()) + 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] @@ -124,6 +152,26 @@ impl BlobFile { } } +/// Runs `fut` on the napi runtime. Aborting `signal` drops the future, which +/// cancels in-flight reads. +pub(crate) fn spawn_abortable<'env, T>( + env: &'env Env, + signal: Option, + fut: impl Future> + Send + 'static, +) -> napi::Result> +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> { let start = parse_u64(start, "start")?; let end = parse_u64(end, "end")?; diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index d381d98e5..f5c3b2987 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -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, - ) -> napi::Result>> { + signal: Option, + ) -> napi::Result>>> { 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, - ) -> napi::Result>> { + signal: Option, + ) -> napi::Result>>> { 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)]