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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012U4KdHofT5edv3LcXHXbLv
This commit is contained in:
Claude
2026-09-29 19:39:36 +00:00
parent ad89422813
commit ef8e58d26f
3 changed files with 65 additions and 29 deletions
+25
View File
@@ -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<ArrayBuffer> {
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")]);
+20 -11
View File
@@ -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<unknown, Promise<unknown>>();
// 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<Blob, Promise<Uint8Array>>();
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<string, unknown> {
+20 -18
View File
@@ -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<unknown> {
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<string, unknown>), data: bytes };
}
export function coerceBlobValue(input: unknown): BlobValue | null {