From ef8e58d26f574e63030fd51417c772ed8adcba3b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 29 Sep 2026 17:40:43 +0000 Subject: [PATCH] fix(node): read each distinct Blob once in resolveBlobInputs Key the read cache by the Blob itself instead of the column value, so rows sharing a Blob through separate { data } structs read it once and share the bytes. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012U4KdHofT5edv3LcXHXbLv --- nodejs/__test__/blob.test.ts | 25 ++++++++++++++++++++++++ nodejs/lancedb/arrow.ts | 31 ++++++++++++++++++----------- nodejs/lancedb/blob.ts | 38 +++++++++++++++++++----------------- 3 files changed, 65 insertions(+), 29 deletions(-) diff --git a/nodejs/__test__/blob.test.ts b/nodejs/__test__/blob.test.ts index ce7de494c..db27ea78a 100644 --- a/nodejs/__test__/blob.test.ts +++ b/nodejs/__test__/blob.test.ts @@ -146,6 +146,31 @@ describe("resolveBlobInputs", () => { expect(data[0].images?.[0]).toBe(shared); }); + it("reads each distinct Blob once, bare or inside data structs", async () => { + class CountingBlob extends Blob { + calls = 0; + async arrayBuffer(): Promise { + this.calls++; + return super.arrayBuffer(); + } + } + const shared = new CountingBlob(["same"]); + const schema = new Schema([blob("image")]); + const resolved = await resolveBlobInputs( + [ + { image: { data: shared } }, + { image: { data: shared } }, + { image: shared }, + ], + schema, + ); + expect(shared.calls).toBe(1); + const first = (resolved[0].image as { data: Uint8Array }).data; + expect(Buffer.from(first).toString()).toBe("same"); + expect((resolved[1].image as { data: Uint8Array }).data).toBe(first); + expect(resolved[2].image).toBe(first); + }); + it("returns the input when there is nothing to read", async () => { const data = [{ image: Buffer.from("x") }]; const schema = new Schema([blob("image")]); diff --git a/nodejs/lancedb/arrow.ts b/nodejs/lancedb/arrow.ts index f009a089a..e5ad1a2b4 100644 --- a/nodejs/lancedb/arrow.ts +++ b/nodejs/lancedb/arrow.ts @@ -41,10 +41,10 @@ import { import { Buffers } from "apache-arrow/data"; import { typedArrayToArrowType } from "./arrow_type"; import { + blobToRead, coerceBlobValue, isBlobField, - needsBlobResolution, - resolveBlobInput, + withBlobBytes, } from "./blob"; import { type EmbeddingFunction } from "./embedding/embedding_function"; import { @@ -622,24 +622,33 @@ export async function resolveBlobInputs( return data; } const sanitized = sanitizeSchema(schema); - const pending = new Map>(); + // Keyed by the Blob itself, so rows sharing one Blob (bare or as + // `{ data }`) read it once and share the bytes. + const reads = new Map>(); mapBlobInputs(data, sanitized, (value) => { - if (needsBlobResolution(value) && !pending.has(value)) { - pending.set(value, resolveBlobInput(value)); + const source = blobToRead(value); + if (source !== undefined && !reads.has(source)) { + reads.set( + source, + source.arrayBuffer().then((buffer) => new Uint8Array(buffer)), + ); } return value; }); - if (pending.size === 0) { + if (reads.size === 0) { return data; } - const resolved = new Map( + const bytes = new Map( await Promise.all( - [...pending].map(async ([value, bytes]) => [value, await bytes] as const), + [...reads].map(async ([source, read]) => [source, await read] as const), ), ); - return mapBlobInputs(data, sanitized, (value) => - resolved.has(value) ? resolved.get(value) : value, - ); + return mapBlobInputs(data, sanitized, (value) => { + const source = blobToRead(value); + return source === undefined + ? value + : withBlobBytes(value, bytes.get(source) as Uint8Array); + }); } function isObject(value: unknown): value is Record { diff --git a/nodejs/lancedb/blob.ts b/nodejs/lancedb/blob.ts index f0afc0e82..ea593911f 100644 --- a/nodejs/lancedb/blob.ts +++ b/nodejs/lancedb/blob.ts @@ -292,10 +292,10 @@ export type BlobRange = { * `URL`) into `Uint8Array` and URI strings, keeping the value's shape. Other * values are returned unchanged and validated later by {@link coerceBlobValue}. * - * Throws on `Blob` / `File`, which need {@link resolveBlobInput} first. + * Throws on `Blob` / `File`, which must be read into bytes first. */ export function normalizeBlobInput(value: unknown): unknown { - if (needsBlobResolution(value)) { + if (blobToRead(value) !== undefined) { throw new Error( "Blob and File values must be read asynchronously. Pass them to " + "createTable, add, or mergeInsert, or convert them with " + @@ -324,26 +324,28 @@ export function normalizeBlobInput(value: unknown): unknown { return value; } -/** Whether {@link resolveBlobInput} has anything to read for this value. */ -export function needsBlobResolution(value: unknown): boolean { - return isBlobLike(value) || (isBlobStruct(value) && isBlobLike(value.data)); +/** + * Returns the `Blob` / `File` a blob value needs read, either the value itself + * or its `data` field, or `undefined` when there is nothing to read. + */ +export function blobToRead(value: unknown): Blob | undefined { + if (isBlobLike(value)) { + return value; + } + if (isBlobStruct(value) && isBlobLike(value.data)) { + return value.data; + } + return undefined; } /** - * Reads `Blob` / `File` values, at the top level or in `data`, into - * `Uint8Array`. Other values are returned unchanged. + * Replaces the `Blob` that {@link blobToRead} found in `value` with `bytes`, + * keeping the value's shape. */ -export async function resolveBlobInput(value: unknown): Promise { - if (isBlobLike(value)) { - return new Uint8Array(await value.arrayBuffer()); - } - if (isBlobStruct(value) && isBlobLike(value.data)) { - return { - ...value, - data: new Uint8Array(await value.data.arrayBuffer()), - }; - } - return value; +export function withBlobBytes(value: unknown, bytes: Uint8Array): unknown { + return isBlobLike(value) + ? bytes + : { ...(value as Record), data: bytes }; } export function coerceBlobValue(input: unknown): BlobValue | null {