Merge main into gatekeeper/fix-4284-1

nodejs/__test__/table.test.ts

@@ -92,6 +92,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([
@@ -2613,6 +2645,27 @@ describe("when dealing with blob columns", () => {
     );
   });
 
+  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);
@@ -2637,6 +2690,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/lancedb/sanitize.ts

@@ -43,6 +43,8 @@ import {
   Interval,
   IntervalDayTime,
   IntervalYearMonth,
+  LargeBinary,
+  LargeUtf8,
   List,
   Map_,
   Null,
@@ -465,8 +467,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:

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>> {

nodejs/src/query.rs

@@ -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(),
         }
     }