mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-23 20:22:37 +00:00
fix(node): materialize nested struct blobs in bytes mode (#4611)
`blobMode: "bytes"` left blobs inside structs as descriptors because projection matching ignored parent structs and byte replacement only rebuilt top-level fields. Expand direct struct projections and aliases to their blob leaves, then recursively replace the corresponding Arrow fields and data with `LargeBinary`. Track literal output names separately from descendant paths so scalar aliases stay intact and every overlapping blob projection is converted. Preserve null structs, null and empty blobs, sibling fields, schema metadata, and the existing row ID and snapshot behavior. Blobs inside lists retain their descriptors. Update the API docs to describe struct support. Regression coverage includes full and parent-struct projections, leaf and struct aliases, dotted output names, overlapping projections in both orders, nested structs, nulls, empty blobs and results, multiple batches, list siblings, query reuse, row IDs, and concurrent writes. Regression tests reproduce the nested-blob and projection-collision failures before the corresponding changes. Validation: - `pnpm build` - `pnpm lint` - `pnpm run docs` - `pnpm test __test__/nested_blob.test.ts __test__/blob.test.ts __test__/query.test.ts --runInBand --verbose=false` — 142 passed - `pnpm test __test__/table.test.ts --runInBand --verbose=false -t 'when dealing with blob columns'` — 60 passed Fixes #4609 <!-- lance-gatekeeper-fix:v1 agent=76edaac77d044c468d569b58efa2104a generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>
This commit is contained in:
lancedb-gatefixer[bot]
co-authored by
Xuanwo
Xuanwo
parent
e5873d9763
commit
f5f4463b15
@@ -0,0 +1,327 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import * as tmp from "tmp";
|
||||
import { type Table, blob, connect, isBlobField } from "../lancedb";
|
||||
import {
|
||||
DataType,
|
||||
Field,
|
||||
Int64,
|
||||
List,
|
||||
Schema,
|
||||
Struct,
|
||||
Utf8,
|
||||
} from "../lancedb/arrow";
|
||||
|
||||
describe('nested blobs with blobMode "bytes"', () => {
|
||||
let tmpDir: tmp.DirResult;
|
||||
let table: Table;
|
||||
|
||||
const caption = new Field(
|
||||
"caption",
|
||||
new Utf8(),
|
||||
false,
|
||||
new Map([["label", "caption"]]),
|
||||
);
|
||||
const info = new Field(
|
||||
"info",
|
||||
new Struct([
|
||||
blob("image"),
|
||||
caption,
|
||||
new Field("preview", new Struct([blob("image")]), true),
|
||||
]),
|
||||
true,
|
||||
new Map([["label", "info"]]),
|
||||
);
|
||||
const schema = new Schema(
|
||||
[new Field("id", new Int64(), false), info, blob("cover")],
|
||||
new Map([["source", "nested-blobs"]]),
|
||||
);
|
||||
|
||||
beforeEach(async () => {
|
||||
tmpDir = tmp.dirSync({ unsafeCleanup: true });
|
||||
const db = await connect(tmpDir.name);
|
||||
table = await db.createTable(
|
||||
"nested_blobs",
|
||||
[
|
||||
{
|
||||
id: 1n,
|
||||
info: {
|
||||
image: Buffer.from("hello"),
|
||||
caption: "first",
|
||||
preview: { image: Buffer.from("preview") },
|
||||
},
|
||||
cover: Buffer.from("cover"),
|
||||
},
|
||||
{ id: 2n, info: null, cover: null },
|
||||
{
|
||||
id: 3n,
|
||||
info: { image: null, caption: "third", preview: null },
|
||||
cover: null,
|
||||
},
|
||||
{
|
||||
id: 4n,
|
||||
info: {
|
||||
image: Buffer.alloc(0),
|
||||
caption: "fourth",
|
||||
preview: { image: null },
|
||||
},
|
||||
cover: Buffer.alloc(0),
|
||||
},
|
||||
],
|
||||
{ schema },
|
||||
);
|
||||
});
|
||||
|
||||
afterEach(() => tmpDir.removeCallback());
|
||||
|
||||
it.each([false, true])(
|
||||
"reads struct leaves across batches (select parent: %s)",
|
||||
async (selectParent) => {
|
||||
const query = table
|
||||
.query()
|
||||
.orderBy({ columnName: "id", ascending: true });
|
||||
if (selectParent) {
|
||||
query.select(["info"]);
|
||||
}
|
||||
const result = await query.toArrow({
|
||||
blobMode: "bytes",
|
||||
maxBatchLength: 1,
|
||||
});
|
||||
expect(result.batches.length).toBeGreaterThan(1);
|
||||
expect(result.schema.fields.map((field) => field.name)).toEqual(
|
||||
selectParent ? ["info"] : ["id", "info", "cover"],
|
||||
);
|
||||
expect(result.schema.metadata).toEqual(schema.metadata);
|
||||
const infoField = result.schema.fields.find(
|
||||
(field) => field.name === "info",
|
||||
)!;
|
||||
expect(infoField.nullable).toBe(true);
|
||||
expect(infoField.metadata).toEqual(info.metadata);
|
||||
const fields = (infoField.type as Struct).children;
|
||||
expect(DataType.isLargeBinary(fields[0].type)).toBe(true);
|
||||
expect(isBlobField(fields[0])).toBe(false);
|
||||
expect(fields[0].metadata.size).toBe(0);
|
||||
expect(fields[1]).toEqual(caption);
|
||||
expect(
|
||||
DataType.isLargeBinary((fields[2].type as Struct).children[0].type),
|
||||
).toBe(true);
|
||||
|
||||
const rows = result.toArray();
|
||||
expect(Buffer.from(rows[0].info.image).toString()).toBe("hello");
|
||||
expect(rows[0].info.caption).toBe("first");
|
||||
expect(Buffer.from(rows[0].info.preview.image).toString()).toBe(
|
||||
"preview",
|
||||
);
|
||||
expect(rows[1].info).toBeNull();
|
||||
expect(rows[2].info.image).toBeNull();
|
||||
expect(rows[2].info.caption).toBe("third");
|
||||
expect(rows[2].info.preview).toBeNull();
|
||||
expect(Buffer.from(rows[3].info.image)).toEqual(Buffer.alloc(0));
|
||||
expect(rows[3].info.preview.image).toBeNull();
|
||||
if (!selectParent) {
|
||||
expect(Buffer.from(rows[0].cover).toString()).toBe("cover");
|
||||
expect(rows[1].cover).toBeNull();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("reads direct leaf and struct aliases", async () => {
|
||||
const rows = await table
|
||||
.query()
|
||||
.orderBy({ columnName: "id", ascending: true })
|
||||
.select({
|
||||
image: "info.image",
|
||||
details: "info",
|
||||
preview: "info.preview",
|
||||
caption: "info.caption",
|
||||
})
|
||||
.toArray({ blobMode: "bytes", maxBatchLength: 1 });
|
||||
expect(Object.keys(rows[0])).toEqual([
|
||||
"image",
|
||||
"details",
|
||||
"preview",
|
||||
"caption",
|
||||
]);
|
||||
expect(Buffer.from(rows[0].image).toString()).toBe("hello");
|
||||
expect(Buffer.from(rows[0].details.image).toString()).toBe("hello");
|
||||
expect(Buffer.from(rows[0].details.preview.image).toString()).toBe(
|
||||
"preview",
|
||||
);
|
||||
expect(Buffer.from(rows[0].preview.image).toString()).toBe("preview");
|
||||
expect(rows[0].caption).toBe("first");
|
||||
expect(rows[1].details).toBeNull();
|
||||
expect(rows[1].image).toBeNull();
|
||||
expect(rows[2].image).toBeNull();
|
||||
expect(Buffer.from(rows[3].image)).toEqual(Buffer.alloc(0));
|
||||
});
|
||||
|
||||
it.each([false, true])(
|
||||
"preserves a literal scalar alias matching a nested blob path (reverse: %s)",
|
||||
async (reverse) => {
|
||||
const projection: [string, string][] = [
|
||||
["info", "info"],
|
||||
["info.image", "id"],
|
||||
];
|
||||
if (reverse) {
|
||||
projection.reverse();
|
||||
}
|
||||
const [row] = await table
|
||||
.query()
|
||||
.where("id = 1")
|
||||
.select(new Map(projection))
|
||||
.toArray({ blobMode: "bytes" });
|
||||
expect(row["info.image"]).toBe(1n);
|
||||
expect(Buffer.from(row.info.image).toString()).toBe("hello");
|
||||
},
|
||||
);
|
||||
|
||||
it.each([false, true])(
|
||||
"materializes every overlapping blob projection (reverse: %s)",
|
||||
async (reverse) => {
|
||||
const columns = ["info.image", "info"];
|
||||
if (reverse) {
|
||||
columns.reverse();
|
||||
}
|
||||
const result = await table
|
||||
.query()
|
||||
.orderBy({ columnName: "id", ascending: true })
|
||||
.select(columns)
|
||||
.toArrow({ blobMode: "bytes", maxBatchLength: 1 });
|
||||
expect(result.schema.fields.map((field) => field.name)).toEqual(columns);
|
||||
expect(DataType.isLargeBinary(result.getChild("info.image")!.type)).toBe(
|
||||
true,
|
||||
);
|
||||
expect(
|
||||
DataType.isLargeBinary(
|
||||
result.getChild("info")!.getChild("image")!.type,
|
||||
),
|
||||
).toBe(true);
|
||||
const rows = result.toArray();
|
||||
expect(Buffer.from(rows[0]["info.image"]).toString()).toBe("hello");
|
||||
expect(Buffer.from(rows[0].info.image).toString()).toBe("hello");
|
||||
expect(rows[0].info.caption).toBe("first");
|
||||
expect(rows[1]["info.image"]).toBeNull();
|
||||
expect(rows[1].info).toBeNull();
|
||||
expect(rows[2]["info.image"]).toBeNull();
|
||||
expect(rows[2].info.image).toBeNull();
|
||||
expect(Buffer.from(rows[3]["info.image"])).toEqual(Buffer.alloc(0));
|
||||
expect(Buffer.from(rows[3].info.image)).toEqual(Buffer.alloc(0));
|
||||
},
|
||||
);
|
||||
|
||||
it("materializes dotted aliases independently of nested fields", async () => {
|
||||
const [row] = await table
|
||||
.query()
|
||||
.where("id = 1")
|
||||
.select({
|
||||
details: "info",
|
||||
"details.image": "info.image",
|
||||
"details.preview": "info.preview",
|
||||
"details.preview.image": "info.preview.image",
|
||||
})
|
||||
.toArray({ blobMode: "bytes" });
|
||||
expect(Buffer.from(row.details.image).toString()).toBe("hello");
|
||||
expect(Buffer.from(row["details.image"]).toString()).toBe("hello");
|
||||
expect(Buffer.from(row.details.preview.image).toString()).toBe("preview");
|
||||
expect(Buffer.from(row["details.preview"].image).toString()).toBe(
|
||||
"preview",
|
||||
);
|
||||
expect(Buffer.from(row["details.preview.image"]).toString()).toBe(
|
||||
"preview",
|
||||
);
|
||||
});
|
||||
|
||||
it.each([false, true])(
|
||||
"converts the empty result schema (alias: %s)",
|
||||
async (alias) => {
|
||||
const query = table.query().where("id < 0");
|
||||
if (alias) {
|
||||
query.select({ image: "info.image", details: "info" });
|
||||
}
|
||||
const result = await query.toArrow({ blobMode: "bytes" });
|
||||
expect(result.numRows).toBe(0);
|
||||
expect(result.schema.fields.map((field) => field.name)).toEqual(
|
||||
alias ? ["image", "details"] : ["id", "info", "cover"],
|
||||
);
|
||||
const fields = (
|
||||
result.schema.fields.find(
|
||||
(field) => field.name === (alias ? "details" : "info"),
|
||||
)!.type as Struct
|
||||
).children;
|
||||
expect(DataType.isLargeBinary(fields[0].type)).toBe(true);
|
||||
expect(isBlobField(fields[0])).toBe(false);
|
||||
expect(fields[1]).toEqual(caption);
|
||||
expect(
|
||||
DataType.isLargeBinary((fields[2].type as Struct).children[0].type),
|
||||
).toBe(true);
|
||||
if (alias) {
|
||||
expect(DataType.isLargeBinary(result.schema.fields[0].type)).toBe(true);
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("preserves requested row IDs and leaves subsequent descriptors intact", async () => {
|
||||
const query = table.query().where("id = 1").select(["info"]);
|
||||
const [bytes] = await query.toArray({ blobMode: "bytes" });
|
||||
expect(Object.keys(bytes)).toEqual(["info"]);
|
||||
const [description] = await query.toArray();
|
||||
expect(Object.keys(description)).toEqual(["info"]);
|
||||
expect(Number(description.info.image.size)).toBe(5);
|
||||
const [withRowId] = await query.withRowId().toArray({ blobMode: "bytes" });
|
||||
expect(typeof withRowId._rowid).toBe("bigint");
|
||||
expect(await table.fetchBlobs("info.image", [withRowId._rowid])).toEqual([
|
||||
Buffer.from(withRowId.info.image),
|
||||
]);
|
||||
});
|
||||
|
||||
it("keeps descriptors in description mode", async () => {
|
||||
const [description] = await table
|
||||
.query()
|
||||
.where("id = 1")
|
||||
.toArray({ blobMode: "descriptions" });
|
||||
expect(Number(description.info.image.size)).toBe(5);
|
||||
});
|
||||
|
||||
it("keeps list elements as descriptors alongside struct blob bytes", async () => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const table = await db.createTable(
|
||||
"list_blobs",
|
||||
[
|
||||
{
|
||||
info: { image: Buffer.from("hello"), images: [Buffer.from("list")] },
|
||||
},
|
||||
],
|
||||
{
|
||||
schema: new Schema([
|
||||
new Field(
|
||||
"info",
|
||||
new Struct([
|
||||
blob("image"),
|
||||
new Field("images", new List(blob("item")), true),
|
||||
]),
|
||||
true,
|
||||
),
|
||||
]),
|
||||
},
|
||||
);
|
||||
const [bytes] = await table.query().toArray({ blobMode: "bytes" });
|
||||
expect(Buffer.from(bytes.info.image).toString()).toBe("hello");
|
||||
expect(Number(bytes.info.images.get(0).size)).toBe(4);
|
||||
const [row] = await table
|
||||
.query()
|
||||
.select(["info.images"])
|
||||
.toArray({ blobMode: "bytes" });
|
||||
expect(Number(row["info.images"].get(0).size)).toBe(4);
|
||||
expect(Object.keys(row)).toEqual(["info.images"]);
|
||||
const result = await table.query().toArrow({ blobMode: "bytes" });
|
||||
const fields = (result.schema.fields[0].type as Struct).children;
|
||||
const descriptions = await table.query().toArrow();
|
||||
expect(fields[1]).toEqual(
|
||||
(descriptions.schema.fields[0].type as Struct).children[1],
|
||||
);
|
||||
expect(DataType.isStruct((fields[1].type as List).children[0].type)).toBe(
|
||||
true,
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -3314,51 +3314,64 @@ describe("when dealing with blob columns", () => {
|
||||
).rejects.toThrow(/blobMode must be/);
|
||||
});
|
||||
|
||||
it("reruns the query when the table changes before the version check", async () => {
|
||||
const { table, native, update } = await openVersionedBlobTable();
|
||||
const liveVersion = native.version.bind(native);
|
||||
native.version = async () => {
|
||||
native.version = liveVersion;
|
||||
await update(2n, "new");
|
||||
return liveVersion();
|
||||
};
|
||||
const [row] = await table.query().toArray({ blobMode: "bytes" });
|
||||
expect(row.generation).toBe(2n);
|
||||
expect(Buffer.from(row.image).toString()).toBe("new");
|
||||
});
|
||||
it.each([false, true])(
|
||||
"reruns the query when the table changes before the version check (nested: %s)",
|
||||
async (nested) => {
|
||||
const { table, native, update } = await openVersionedBlobTable(nested);
|
||||
const liveVersion = native.version.bind(native);
|
||||
native.version = async () => {
|
||||
native.version = liveVersion;
|
||||
await update(2n, "new");
|
||||
return liveVersion();
|
||||
};
|
||||
const [row] = await table.query().toArray({ blobMode: "bytes" });
|
||||
expect(row.generation).toBe(2n);
|
||||
expect(Buffer.from(nested ? row.info.image : row.image).toString()).toBe(
|
||||
"new",
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it("reads bytes from the query's version when a write lands before the fetch", async () => {
|
||||
const { table, native, update } = await openVersionedBlobTable();
|
||||
const liveVersion = native.version.bind(native);
|
||||
native.version = async () => {
|
||||
native.version = liveVersion;
|
||||
const version = await liveVersion();
|
||||
await update(2n, "new");
|
||||
return version;
|
||||
};
|
||||
const [row] = await table.query().toArray({ blobMode: "bytes" });
|
||||
expect(row.generation).toBe(1n);
|
||||
expect(Buffer.from(row.image).toString()).toBe("old");
|
||||
});
|
||||
it.each([false, true])(
|
||||
"reads bytes from the query's version when a write lands before the fetch (nested: %s)",
|
||||
async (nested) => {
|
||||
const { table, native, update } = await openVersionedBlobTable(nested);
|
||||
const liveVersion = native.version.bind(native);
|
||||
native.version = async () => {
|
||||
native.version = liveVersion;
|
||||
const version = await liveVersion();
|
||||
await update(2n, "new");
|
||||
return version;
|
||||
};
|
||||
const [row] = await table.query().toArray({ blobMode: "bytes" });
|
||||
expect(row.generation).toBe(1n);
|
||||
expect(Buffer.from(nested ? row.info.image : row.image).toString()).toBe(
|
||||
"old",
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it("fails when the table changes during every attempt", async () => {
|
||||
const { table, native, update } = await openVersionedBlobTable();
|
||||
const liveVersion = native.version.bind(native);
|
||||
let generation = 1n;
|
||||
native.version = async () => {
|
||||
generation += 1n;
|
||||
await update(generation, `generation ${generation}`);
|
||||
return liveVersion();
|
||||
};
|
||||
try {
|
||||
await expect(
|
||||
table.query().toArray({ blobMode: "bytes" }),
|
||||
).rejects.toThrow(/changed during each of 3 attempts/);
|
||||
expect(generation).toBe(4n);
|
||||
} finally {
|
||||
native.version = liveVersion;
|
||||
}
|
||||
});
|
||||
it.each([false, true])(
|
||||
"fails when the table changes during every attempt (nested: %s)",
|
||||
async (nested) => {
|
||||
const { table, native, update } = await openVersionedBlobTable(nested);
|
||||
const liveVersion = native.version.bind(native);
|
||||
let generation = 1n;
|
||||
native.version = async () => {
|
||||
generation += 1n;
|
||||
await update(generation, `generation ${generation}`);
|
||||
return liveVersion();
|
||||
};
|
||||
try {
|
||||
await expect(
|
||||
table.query().toArray({ blobMode: "bytes" }),
|
||||
).rejects.toThrow(/changed during each of 3 attempts/);
|
||||
expect(generation).toBe(4n);
|
||||
} finally {
|
||||
native.version = liveVersion;
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("rejects fetchBlobs on a non-blob column", async () => {
|
||||
const { table, rowIds } = await openBlobTable();
|
||||
@@ -3642,18 +3655,28 @@ describe("when dealing with blob columns", () => {
|
||||
).map((value) => (value == null ? null : Number(value.size)));
|
||||
}
|
||||
|
||||
async function openVersionedBlobTable() {
|
||||
async function openVersionedBlobTable(nested = false) {
|
||||
const db = await connect(tmpDir.name, {
|
||||
storageOptions: { newTableEnableStableRowIds: "true" },
|
||||
});
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), false),
|
||||
new Field("generation", new Int64(), true),
|
||||
blob("image"),
|
||||
nested
|
||||
? new Field("info", new Struct([blob("image")]), true)
|
||||
: blob("image"),
|
||||
]);
|
||||
const table = await db.createTable(
|
||||
"versions",
|
||||
[{ id: 1n, generation: 1n, image: Buffer.from("old") }],
|
||||
[
|
||||
{
|
||||
id: 1n,
|
||||
generation: 1n,
|
||||
...(nested
|
||||
? { info: { image: Buffer.from("old") } }
|
||||
: { image: Buffer.from("old") }),
|
||||
},
|
||||
],
|
||||
{ schema },
|
||||
);
|
||||
const native = (
|
||||
@@ -3663,7 +3686,15 @@ describe("when dealing with blob columns", () => {
|
||||
await table
|
||||
.mergeInsert("id")
|
||||
.whenMatchedUpdateAll()
|
||||
.execute([{ id: 1n, generation, image: Buffer.from(image) }]);
|
||||
.execute([
|
||||
{
|
||||
id: 1n,
|
||||
generation,
|
||||
...(nested
|
||||
? { info: { image: Buffer.from(image) } }
|
||||
: { image: Buffer.from(image) }),
|
||||
},
|
||||
]);
|
||||
};
|
||||
return { table, native, update };
|
||||
}
|
||||
|
||||
+116
-143
@@ -1,8 +1,8 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import type { Data as ArrowData } from "apache-arrow";
|
||||
import {
|
||||
type ArrowData,
|
||||
Table as ArrowTable,
|
||||
DataType,
|
||||
Field,
|
||||
@@ -14,7 +14,6 @@ import {
|
||||
extractVectorBuffer,
|
||||
fromBufferToRecordBatch,
|
||||
fromRecordBatchToBuffer,
|
||||
makeData,
|
||||
tableFromIPC,
|
||||
vectorFromArray,
|
||||
} from "./arrow";
|
||||
@@ -68,84 +67,42 @@ export interface QueryBlobContext {
|
||||
|
||||
const BLOB_BYTES_ATTEMPTS = 3;
|
||||
|
||||
interface BlobSource {
|
||||
// Output names are literal: an alias containing dots is still one field.
|
||||
output: string;
|
||||
path: string[];
|
||||
source: string;
|
||||
}
|
||||
|
||||
function blobSources(
|
||||
blobColumns: string[],
|
||||
projection: [string, string][] | undefined,
|
||||
): Map<string, string> {
|
||||
const holds = (source: string) =>
|
||||
blobColumns.some(
|
||||
(column) => column === source || column.startsWith(`${source}.`),
|
||||
);
|
||||
const pairs =
|
||||
projection ??
|
||||
Array.from(new Set(blobColumns.map((column) => column.split(".")[0]))).map(
|
||||
(name): [string, string] => [name, name],
|
||||
);
|
||||
return new Map(pairs.filter(([, source]) => holds(source)));
|
||||
}
|
||||
|
||||
// Fetched bytes for the blobs under one output column. A blob column is a
|
||||
// leaf; a struct holds the blobs nested in it, keyed by field name. A Map
|
||||
// keeps struct field names from reaching shared JavaScript prototypes.
|
||||
type BlobBytes = (Buffer | null)[] | Map<string, BlobBytes>;
|
||||
|
||||
function isLeaf(bytes: BlobBytes): bytes is (Buffer | null)[] {
|
||||
return Array.isArray(bytes);
|
||||
}
|
||||
|
||||
function nestedFieldType(type: Struct, bytes: Map<string, BlobBytes>): Struct {
|
||||
return new Struct(
|
||||
type.children.map((field) => {
|
||||
const inner = bytes.get(field.name);
|
||||
if (inner === undefined) {
|
||||
return field;
|
||||
): BlobSource[] {
|
||||
if (projection === undefined) {
|
||||
return blobColumns.map((source) => {
|
||||
const [output, ...path] = source.split(".");
|
||||
return { output, path, source };
|
||||
});
|
||||
}
|
||||
const sources: BlobSource[] = [];
|
||||
for (const [output, source] of projection) {
|
||||
for (const column of blobColumns) {
|
||||
if (column === source || column.startsWith(`${source}.`)) {
|
||||
sources.push({
|
||||
output,
|
||||
path:
|
||||
column === source ? [] : column.slice(source.length + 1).split("."),
|
||||
source: column,
|
||||
});
|
||||
}
|
||||
if (isLeaf(inner)) {
|
||||
return new Field(field.name, new LargeBinary(), true);
|
||||
}
|
||||
return new Field(
|
||||
field.name,
|
||||
nestedFieldType(field.type as Struct, inner),
|
||||
field.nullable,
|
||||
);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
// Rebuilds a struct with the blobs under it as bytes. The struct keeps its own
|
||||
// validity, so a null struct stays null.
|
||||
function nestedData(
|
||||
data: ArrowData,
|
||||
bytes: Map<string, BlobBytes>,
|
||||
offset: number,
|
||||
): ArrowData {
|
||||
const type = data.type as Struct;
|
||||
const children = data.children.map((child, index) => {
|
||||
const inner = bytes.get(type.children[index].name);
|
||||
if (inner === undefined) {
|
||||
return child;
|
||||
}
|
||||
if (isLeaf(inner)) {
|
||||
return vectorFromArray(
|
||||
inner.slice(offset, offset + data.length),
|
||||
new LargeBinary(),
|
||||
).data[0];
|
||||
}
|
||||
return nestedData(child, inner, offset);
|
||||
});
|
||||
return makeData({
|
||||
type: nestedFieldType(type, bytes),
|
||||
length: data.length,
|
||||
nullCount: data.nullCount,
|
||||
nullBitmap: data.nullBitmap,
|
||||
children,
|
||||
});
|
||||
}
|
||||
return sources;
|
||||
}
|
||||
|
||||
async function readBlobBytes(
|
||||
table: ArrowTable,
|
||||
sources: Map<string, string>,
|
||||
blobColumns: string[],
|
||||
sources: BlobSource[],
|
||||
snapshot: NativeTable,
|
||||
dropRowId: boolean,
|
||||
): Promise<ArrowTable> {
|
||||
@@ -157,42 +114,27 @@ async function readBlobBytes(
|
||||
rowIdColumn === null
|
||||
? []
|
||||
: Array.from(rowIdColumn.toArray() as BigUint64Array);
|
||||
const fetch = async (column: string) =>
|
||||
(rowIds.length === 0 ? [] : await snapshot.fetchBlobs(column, rowIds)).map(
|
||||
(value) => value ?? null,
|
||||
const outputFields = new Map(
|
||||
table.schema.fields.map((field) => [field.name, field]),
|
||||
);
|
||||
const replacements = new Map<Field, (Buffer | null)[]>();
|
||||
for (const { output, path, source } of sources) {
|
||||
let field = outputFields.get(output);
|
||||
for (const name of path) {
|
||||
field =
|
||||
field !== undefined && DataType.isStruct(field.type)
|
||||
? field.type.children.find((child) => child.name === name)
|
||||
: undefined;
|
||||
}
|
||||
if (field === undefined) {
|
||||
continue;
|
||||
}
|
||||
const values =
|
||||
rowIds.length === 0 ? [] : await snapshot.fetchBlobs(source, rowIds);
|
||||
replacements.set(
|
||||
field,
|
||||
values.map((value) => value ?? null),
|
||||
);
|
||||
const replacements = new Map<number, BlobBytes>();
|
||||
for (const [output, source] of sources) {
|
||||
const index = table.schema.fields.findIndex(
|
||||
(field) => field.name === output,
|
||||
);
|
||||
if (index < 0) {
|
||||
continue;
|
||||
}
|
||||
if (blobColumns.includes(source)) {
|
||||
replacements.set(index, await fetch(source));
|
||||
continue;
|
||||
}
|
||||
if (!DataType.isStruct(table.schema.fields[index].type)) {
|
||||
continue;
|
||||
}
|
||||
const nested = new Map<string, BlobBytes>();
|
||||
for (const column of blobColumns.filter((c) =>
|
||||
c.startsWith(`${source}.`),
|
||||
)) {
|
||||
const path = column.slice(source.length + 1).split(".");
|
||||
let level = nested;
|
||||
for (const name of path.slice(0, -1)) {
|
||||
let next = level.get(name);
|
||||
if (next === undefined) {
|
||||
next = new Map<string, BlobBytes>();
|
||||
level.set(name, next);
|
||||
}
|
||||
level = next as Map<string, BlobBytes>;
|
||||
}
|
||||
level.set(path[path.length - 1], await fetch(column));
|
||||
}
|
||||
replacements.set(index, nested);
|
||||
}
|
||||
const result = withBinaryColumns(table, replacements);
|
||||
if (!dropRowId) {
|
||||
@@ -205,48 +147,76 @@ async function readBlobBytes(
|
||||
);
|
||||
}
|
||||
|
||||
// A fresh field, not a clone: the descriptor field's metadata must not follow
|
||||
// the bytes.
|
||||
function withBinaryColumns(
|
||||
table: ArrowTable,
|
||||
replacements: Map<number, BlobBytes>,
|
||||
replacements: Map<Field, (Buffer | null)[]>,
|
||||
): ArrowTable {
|
||||
if (replacements.size === 0) {
|
||||
return table;
|
||||
}
|
||||
const fields = table.schema.fields.map((field, index) => {
|
||||
const bytes = replacements.get(index);
|
||||
if (bytes === undefined) {
|
||||
return field;
|
||||
const replaceField = (field: Field): Field => {
|
||||
if (replacements.has(field)) {
|
||||
// The descriptor's extension and encoding metadata must not follow bytes.
|
||||
return new Field(field.name, new LargeBinary(), true);
|
||||
}
|
||||
return isLeaf(bytes)
|
||||
? new Field(field.name, new LargeBinary(), true)
|
||||
: new Field(
|
||||
field.name,
|
||||
nestedFieldType(field.type as Struct, bytes),
|
||||
field.nullable,
|
||||
);
|
||||
});
|
||||
if (DataType.isStruct(field.type)) {
|
||||
const children = field.type.children.map(replaceField);
|
||||
if (children.some((child, i) => child !== field.type.children[i])) {
|
||||
return field.clone({ type: new Struct(children) });
|
||||
}
|
||||
}
|
||||
return field;
|
||||
};
|
||||
const fields = table.schema.fields.map(replaceField);
|
||||
const schema = new Schema(fields, table.schema.metadata);
|
||||
let offset = 0;
|
||||
const batches = table.batches.map((batch) => {
|
||||
const children = batch.data.children.map((child, index) => {
|
||||
const bytes = replacements.get(index);
|
||||
if (bytes === undefined) {
|
||||
return child;
|
||||
const replaceData = (
|
||||
data: ArrowData,
|
||||
field: Field,
|
||||
updated: Field,
|
||||
): ArrowData => {
|
||||
if (field === updated) {
|
||||
return data;
|
||||
}
|
||||
if (!isLeaf(bytes)) {
|
||||
return nestedData(child, bytes, offset);
|
||||
const values = replacements.get(field);
|
||||
if (values !== undefined) {
|
||||
return vectorFromArray(
|
||||
values.slice(offset, offset + batch.numRows),
|
||||
new LargeBinary(),
|
||||
).data[0];
|
||||
}
|
||||
return vectorFromArray(
|
||||
bytes.slice(offset, offset + batch.numRows),
|
||||
new LargeBinary(),
|
||||
).data[0];
|
||||
});
|
||||
const children = data.children.map((child, index) =>
|
||||
replaceData(
|
||||
child,
|
||||
field.type.children[index],
|
||||
updated.type.children[index],
|
||||
),
|
||||
);
|
||||
// Keep the struct's validity bitmap and offset, including null parents.
|
||||
return data.clone(
|
||||
updated.type,
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
children,
|
||||
);
|
||||
};
|
||||
const children = batch.data.children.map((child, index) =>
|
||||
replaceData(child, table.schema.fields[index], fields[index]),
|
||||
);
|
||||
offset += batch.numRows;
|
||||
return new RecordBatch(
|
||||
schema,
|
||||
makeData({ type: new Struct(fields), length: batch.numRows, children }),
|
||||
batch.data.clone(
|
||||
new Struct(fields),
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
children,
|
||||
),
|
||||
);
|
||||
});
|
||||
return new ArrowTable(schema, batches);
|
||||
@@ -274,13 +244,14 @@ export interface QueryExecutionOptions {
|
||||
*
|
||||
* - `"descriptions"` (default): each blob is a descriptor. Read the bytes
|
||||
* with {@link Table.fetchBlobs} or {@link Table.fetchBlobFiles}.
|
||||
* - `"bytes"`: each blob column holds the blob bytes as `LargeBinary`,
|
||||
* including blobs nested in structs. The query also reads `_rowid` to
|
||||
* fetch the bytes, and leaves it out of the result unless
|
||||
* {@link QueryBase.withRowId} was called. All the bytes are held in
|
||||
* memory.
|
||||
* - `"bytes"`: each blob column, including blobs nested in structs, holds
|
||||
* the bytes as `LargeBinary`. The query also reads `_rowid` to fetch them, and
|
||||
* leaves it out of the result unless {@link QueryBase.withRowId} was
|
||||
* called. All the bytes are held in memory.
|
||||
*
|
||||
* Blobs nested in a list keep their descriptors.
|
||||
* Selecting a parent struct or directly aliasing a blob or struct also returns
|
||||
* bytes. Blobs inside lists keep their descriptors. Arbitrary projection
|
||||
* expressions are not materialized as bytes.
|
||||
*
|
||||
* The rows and the bytes come from the same table version. If the table
|
||||
* changes while the query runs, the query runs again, up to three times.
|
||||
@@ -529,15 +500,17 @@ export class QueryBase<
|
||||
// unchanged version afterwards proves it read the snapshot's version.
|
||||
const snapshot = await context.table.querySnapshot();
|
||||
const version = await snapshot.version();
|
||||
const blobColumns = await snapshot.blobColumns();
|
||||
const sources = blobSources(blobColumns, context.projection);
|
||||
if (sources.size === 0) {
|
||||
const sources = blobSources(
|
||||
await snapshot.blobColumns(),
|
||||
context.projection,
|
||||
);
|
||||
if (sources.length === 0) {
|
||||
return this.collect(options, false);
|
||||
}
|
||||
const addRowId = !context.withRowId;
|
||||
const table = await this.collect(options, addRowId);
|
||||
if ((await context.table.version()) === version) {
|
||||
return readBlobBytes(table, sources, blobColumns, snapshot, addRowId);
|
||||
return readBlobBytes(table, sources, snapshot, addRowId);
|
||||
}
|
||||
if (attempt === BLOB_BYTES_ATTEMPTS) {
|
||||
throw new Error(
|
||||
|
||||
Reference in New Issue
Block a user