feat(node): expose seek, tell, bounded read, readRanges and close on BlobFile (#4348)

Closes #4300.

The TypeScript `BlobFile` only had `size()`, `read()` and `readRange()`,
while the Rust `BlobFile` behind it also has `seek`, `tell`,
`read_up_to`, `read_ranges` and `close`. This exposes those on the Node
binding:

- `read(maxBytes?)` reads at most `maxBytes` from the cursor (Rust
`read_up_to`). Without the argument it reads to the end, as before.
- `readRanges([{ start, end }, ...])` reads several half-open ranges and
returns one buffer per range, in order. Like `readRange`, it does not
move the cursor.
- `seek(position)` and `tell()` move and report the cursor.
- `close()` releases the handle, and `isClosed()` reports it. Every
read, `seek` and `tell` after `close()` fails with "blob file is already
closed". A second `close()` does nothing. The binding keeps its own
`closed` flag, the same way the Python `BlobFile` does.

Offsets and lengths are `bigint`, like `readRange` and `size()`.
`BlobRange` is exported as a type.

The new tests in `table.test.ts` cover chunked reads with
`read(maxBytes)`, `seek` and `tell`, a negative seek position,
`readRanges` order and cursor, a `readRanges` end past the blob size,
and reads after `close()`. The Node docs under `docs/src/js` are
regenerated with `pnpm run docs`.

I left the `ReadableStream`/async iterator idea from the issue out of
this PR. The AbortSignal work in #4288 touches the same methods, so it
can follow on top of this.

docs/src/js/classes/BlobFile.md

@@ -10,16 +10,50 @@ A lazy handle to blob bytes. Create one with [Table.fetchBlobFiles](Table.md#fet
 
 ## Methods
 
+### close()
+
+```ts
+close(): Promise<void>
+```
+
+Releases the handle. Reads after `close()` fail. Calling it again does
+nothing.
+
+#### Returns
+
+`Promise`&lt;`void`&gt;
+
+***
+
+### isClosed()
+
+```ts
+isClosed(): boolean
+```
+
+Returns true after [BlobFile.close](BlobFile.md#close).
+
+#### Returns
+
+`boolean`
+
+***
+
 ### read()
 
 ```ts
-read(): Promise<Buffer>
+read(maxBytes?): Promise<Buffer>
 ```
 
-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. [BlobFile.readRange](BlobFile.md#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.
+[BlobFile.readRange](BlobFile.md#readrange) does not move the cursor.
+
+#### Parameters
+
+* **maxBytes?**: `bigint`
 
 #### Returns
 
@@ -49,6 +83,45 @@ Fails when `end` is past the blob size. Does not move the cursor.
 
 ***
 
+### readRanges()
+
+```ts
+readRanges(ranges): Promise<Buffer[]>
+```
+
+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.
+
+#### Parameters
+
+* **ranges**: [`BlobRange`](../type-aliases/BlobRange.md)[]
+
+#### Returns
+
+`Promise`&lt;`Buffer`[]&gt;
+
+***
+
+### seek()
+
+```ts
+seek(position): Promise<void>
+```
+
+Moves the cursor to `position`, in bytes from the start of the blob.
+
+#### Parameters
+
+* **position**: `bigint`
+
+#### Returns
+
+`Promise`&lt;`void`&gt;
+
+***
+
 ### size()
 
 ```ts
@@ -60,3 +133,17 @@ Returns the blob size in bytes.
 #### Returns
 
 `bigint`
+
+***
+
+### tell()
+
+```ts
+tell(): Promise<bigint>
+```
+
+Returns the cursor position in bytes.
+
+#### Returns
+
+`Promise`&lt;`bigint`&gt;

docs/src/js/globals.md

@@ -158,6 +158,7 @@
 - [AnalyzePlanDistributedMetrics](type-aliases/AnalyzePlanDistributedMetrics.md)
 - [BaseTokenizer](type-aliases/BaseTokenizer.md)
 - [BlobOptions](type-aliases/BlobOptions.md)
+- [BlobRange](type-aliases/BlobRange.md)
 - [Data](type-aliases/Data.md)
 - [DataLike](type-aliases/DataLike.md)
 - [FieldLike](type-aliases/FieldLike.md)

docs/src/js/type-aliases/BlobRange.md

@@ -0,0 +1,27 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / BlobRange
+
+# Type Alias: BlobRange
+
+```ts
+type BlobRange: object;
+```
+
+A half-open byte range `[start, end)` for [BlobFile.readRanges](../classes/BlobFile.md#readranges).
+
+## Type declaration
+
+### end
+
+```ts
+end: bigint;
+```
+
+### start
+
+```ts
+start: bigint;
+```

nodejs/__test__/table.test.ts

@@ -2625,6 +2625,74 @@ 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 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

@@ -141,13 +141,14 @@ 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): Promise<Buffer> {
+    return this.inner.read(maxBytes);
   }
 
   /**
@@ -158,8 +159,47 @@ export class BlobFile {
   readRange(start: bigint, end: bigint): Promise<Buffer> {
     return this.inner.readRange(start, end);
   }
+
+  /**
+   * 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[]): Promise<Buffer[]> {
+    return this.inner.readRanges(ranges);
+  }
+
+  /** 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();
+  }
 }
 
+/** 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;

nodejs/lancedb/index.ts

@@ -84,7 +84,7 @@ export {
 } from "./arrow";
 
 export { blob, isBlobField, BlobFile } from "./blob";
-export type { BlobOptions } from "./blob";
+export type { BlobOptions, BlobRange } from "./blob";
 
 export {
   Connection,

nodejs/src/blob.rs

@@ -3,6 +3,7 @@
 
 use std::ops::Range;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 
 use arrow_array::{Array, LargeBinaryArray};
 use lancedb::blob::BlobFile as LanceBlobFile;
@@ -11,15 +12,31 @@ 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,13 +49,23 @@ impl BlobFile {
     }
 
     #[napi]
-    pub async fn read(&self) -> napi::Result<Buffer> {
-        let bytes = self.inner.read().await.map_err(|err| convert_error(&err))?;
+    pub async fn read(&self, max_bytes: Option<BigInt>) -> napi::Result<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
+            }
+        }
+        .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> {
+        self.ensure_open()?;
         let range = bigint_range(start, end)?;
         let bytes = self
             .inner
@@ -47,6 +74,54 @@ impl BlobFile {
             .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>> {
+        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())
+    }
+
+    #[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)
+    }
 }
 
 fn bigint_range(start: BigInt, end: BigInt) -> napi::Result<Range<u64>> {