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

@@ -42,7 +42,7 @@ Returns true after [BlobFile.close](BlobFile.md#close).
 ### read()
 
 ```ts
-read(maxBytes?): Promise<Buffer>
+read(maxBytes?, options?): Promise<Buffer>
 ```
 
 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`&lt;`Buffer`&gt;
@@ -64,7 +66,10 @@ otherwise. Returns an empty buffer at the end of the blob.
 ### readRange()
 
 ```ts
-readRange(start, end): Promise<Buffer>
+readRange(
+   start,
+   end,
+   options?): Promise<Buffer>
 ```
 
 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`&lt;`Buffer`&gt;
@@ -86,7 +93,7 @@ Fails when `end` is past the blob size. Does not move the cursor.
 ### readRanges()
 
 ```ts
-readRanges(ranges): Promise<Buffer[]>
+readRanges(ranges, options?): Promise<Buffer[]>
 ```
 
 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`&lt;`Buffer`[]&gt;

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`&lt;(`null` \| [`BlobFile`](BlobFile.md))[]&gt;
@@ -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`&lt;(`null` \| `Buffer`)[]&gt;

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)

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.

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);

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<Buffer> {
-    return this.inner.read(maxBytes);
+  read(maxBytes?: bigint, options?: BlobReadOptions): Promise<Buffer> {
+    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<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),
+    );
   }
 
   /**
@@ -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<Buffer[]> {
-    return this.inner.readRanges(ranges);
+  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. */
@@ -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<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;

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,

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) =>

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<BigInt>) -> napi::Result<Buffer> {
+    pub fn read<'env>(
+        &self,
+        env: &'env Env,
+        max_bytes: Option<BigInt>,
+        signal: Option<AbortSignal>,
+    ) -> napi::Result<PromiseRaw<'env, Buffer>> {
         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<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 async fn read_ranges(&self, ranges: Vec<BlobRange>) -> napi::Result<Vec<Buffer>> {
+    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 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<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>> {
     let start = parse_u64(start, "start")?;
     let end = parse_u64(end, "end")?;

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<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)]