mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-12 00:02:21 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
  Will JonesandClaude Opus 5
|
65e9a4b873 | ||
 lancedb automation
|
88442be843 |
Generated
+243
-254
File diff suppressed because it is too large
Load Diff
+15
-15
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
|
||||
rust-version = "1.91.0"
|
||||
|
||||
[workspace.dependencies]
|
||||
lance = { "version" = "=12.0.0-beta.17", default-features = false, "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=12.0.0-beta.17", default-features = false, "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=12.0.0-beta.17", default-features = false, "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=12.0.0-beta.17", "tag" = "v12.0.0-beta.17", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lancedb = { path = "rust/lancedb", default-features = false }
|
||||
ahash = "0.8"
|
||||
# Note that this one does not include pyarrow
|
||||
@@ -60,7 +60,7 @@ log = "0.4"
|
||||
metrics = "0.24"
|
||||
metrics-util = "0.19"
|
||||
moka = { version = "0.12", features = ["future"] }
|
||||
object_store = "0.14.1"
|
||||
object_store = "0.13.2"
|
||||
pin-project = "1.0.7"
|
||||
rand = "0.9"
|
||||
snafu = "0.8"
|
||||
|
||||
@@ -1,62 +0,0 @@
|
||||
[**@lancedb/lancedb**](../README.md) • **Docs**
|
||||
|
||||
***
|
||||
|
||||
[@lancedb/lancedb](../globals.md) / BlobFile
|
||||
|
||||
# Class: BlobFile
|
||||
|
||||
A lazy handle to blob bytes. Create one with [Table.fetchBlobFiles](Table.md#fetchblobfiles).
|
||||
|
||||
## Methods
|
||||
|
||||
### read()
|
||||
|
||||
```ts
|
||||
read(): Promise<Buffer>
|
||||
```
|
||||
|
||||
Reads from the cursor to the end and advances the cursor.
|
||||
|
||||
A second call returns an empty buffer. [BlobFile.readRange](BlobFile.md#readrange) does
|
||||
not move the cursor.
|
||||
|
||||
#### Returns
|
||||
|
||||
`Promise`<`Buffer`>
|
||||
|
||||
***
|
||||
|
||||
### readRange()
|
||||
|
||||
```ts
|
||||
readRange(start, end): Promise<Buffer>
|
||||
```
|
||||
|
||||
Reads the half-open byte range `[start, end)`.
|
||||
|
||||
Fails when `end` is past the blob size. Does not move the cursor.
|
||||
|
||||
#### Parameters
|
||||
|
||||
* **start**: `bigint`
|
||||
|
||||
* **end**: `bigint`
|
||||
|
||||
#### Returns
|
||||
|
||||
`Promise`<`Buffer`>
|
||||
|
||||
***
|
||||
|
||||
### size()
|
||||
|
||||
```ts
|
||||
size(): bigint
|
||||
```
|
||||
|
||||
Returns the blob size in bytes.
|
||||
|
||||
#### Returns
|
||||
|
||||
`bigint`
|
||||
@@ -137,20 +137,6 @@ containing the new version number of the table after altering the columns.
|
||||
|
||||
***
|
||||
|
||||
### blobColumns()
|
||||
|
||||
```ts
|
||||
abstract blobColumns(): Promise<string[]>
|
||||
```
|
||||
|
||||
Blob v2 columns, including nested dotted paths.
|
||||
|
||||
#### Returns
|
||||
|
||||
`Promise`<`string`[]>
|
||||
|
||||
***
|
||||
|
||||
### branches()
|
||||
|
||||
```ts
|
||||
@@ -513,54 +499,6 @@ Drop an index from the table.
|
||||
|
||||
***
|
||||
|
||||
### fetchBlobFiles()
|
||||
|
||||
```ts
|
||||
abstract fetchBlobFiles(column, rowIds): Promise<(null | BlobFile)[]>
|
||||
```
|
||||
|
||||
Opens lazy blob handles for `column` at the given row IDs using the
|
||||
table's current checkout.
|
||||
|
||||
Preserves input order, duplicates, and nulls. Use this for large payloads.
|
||||
See [Table.fetchBlobs](Table.md#fetchblobs) for row-ID validity across versions.
|
||||
|
||||
#### Parameters
|
||||
|
||||
* **column**: `string`
|
||||
|
||||
* **rowIds**: readonly (`number` \| `bigint`)[]
|
||||
|
||||
#### Returns
|
||||
|
||||
`Promise`<(`null` \| [`BlobFile`](BlobFile.md))[]>
|
||||
|
||||
***
|
||||
|
||||
### fetchBlobs()
|
||||
|
||||
```ts
|
||||
abstract fetchBlobs(column, rowIds): Promise<(null | Buffer)[]>
|
||||
```
|
||||
|
||||
Bytes for `column` at row IDs from [Query.withRowId](Query.md#withrowid).
|
||||
|
||||
Reads the table's current checkout. IDs from another version can fail after
|
||||
compaction unless stable row ids are enabled. Results keep input order and
|
||||
duplicates. Null blobs are `null`. Empty blobs are empty buffers.
|
||||
|
||||
#### Parameters
|
||||
|
||||
* **column**: `string`
|
||||
|
||||
* **rowIds**: readonly (`number` \| `bigint`)[]
|
||||
|
||||
#### Returns
|
||||
|
||||
`Promise`<(`null` \| `Buffer`)[]>
|
||||
|
||||
***
|
||||
|
||||
### flushLsm()
|
||||
|
||||
```ts
|
||||
|
||||
@@ -1,55 +0,0 @@
|
||||
[**@lancedb/lancedb**](../README.md) • **Docs**
|
||||
|
||||
***
|
||||
|
||||
[@lancedb/lancedb](../globals.md) / blob
|
||||
|
||||
# Function: blob()
|
||||
|
||||
```ts
|
||||
function blob(name, options): Field
|
||||
```
|
||||
|
||||
Declares a `lance.blob.v2` column.
|
||||
|
||||
Query results are descriptors, not payload bytes. Use [Table.fetchBlobs](../classes/Table.md#fetchblobs)
|
||||
or [Table.fetchBlobFiles](../classes/Table.md#fetchblobfiles) to read bytes.
|
||||
|
||||
## Parameters
|
||||
|
||||
* **name**: `string`
|
||||
|
||||
* **options**: [`BlobOptions`](../type-aliases/BlobOptions.md) = `{}`
|
||||
|
||||
## Returns
|
||||
|
||||
`Field`
|
||||
|
||||
## Example
|
||||
|
||||
```ts
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { Field, Int64, Schema } from "apache-arrow";
|
||||
import { blob, connect } from "@lancedb/lancedb";
|
||||
|
||||
const db = await connect("./data");
|
||||
const video = await readFile("clip.mp4");
|
||||
const table = await db.createTable(
|
||||
"videos",
|
||||
[{ id: 1n, video }],
|
||||
{
|
||||
schema: new Schema([
|
||||
new Field("id", new Int64()),
|
||||
blob("video"),
|
||||
]),
|
||||
},
|
||||
);
|
||||
|
||||
const rows = await table.query().select(["id"]).withRowId().toArray();
|
||||
const rowIds = rows.map((row) => row._rowid as bigint);
|
||||
const bytes = await table.fetchBlobs("video", rowIds);
|
||||
|
||||
const [handle] = await table.fetchBlobFiles("video", rowIds);
|
||||
const size = handle!.size();
|
||||
const header = await handle!.readRange(0n, size < 65536n ? size : 65536n);
|
||||
```
|
||||
@@ -1,22 +0,0 @@
|
||||
[**@lancedb/lancedb**](../README.md) • **Docs**
|
||||
|
||||
***
|
||||
|
||||
[@lancedb/lancedb](../globals.md) / isBlobField
|
||||
|
||||
# Function: isBlobField()
|
||||
|
||||
```ts
|
||||
function isBlobField(field): boolean
|
||||
```
|
||||
|
||||
Checks for the `lance.blob.v2` extension marker. Does not validate the
|
||||
field's storage type.
|
||||
|
||||
## Parameters
|
||||
|
||||
* **field**: `Field`<`any`>
|
||||
|
||||
## Returns
|
||||
|
||||
`boolean`
|
||||
@@ -19,7 +19,6 @@
|
||||
## Classes
|
||||
|
||||
- [AutoQuery](classes/AutoQuery.md)
|
||||
- [BlobFile](classes/BlobFile.md)
|
||||
- [BooleanQuery](classes/BooleanQuery.md)
|
||||
- [BoostQuery](classes/BoostQuery.md)
|
||||
- [BranchContents](classes/BranchContents.md)
|
||||
@@ -144,7 +143,6 @@
|
||||
|
||||
- [AnalyzePlanDistributedMetrics](type-aliases/AnalyzePlanDistributedMetrics.md)
|
||||
- [BaseTokenizer](type-aliases/BaseTokenizer.md)
|
||||
- [BlobOptions](type-aliases/BlobOptions.md)
|
||||
- [Data](type-aliases/Data.md)
|
||||
- [DataLike](type-aliases/DataLike.md)
|
||||
- [FieldLike](type-aliases/FieldLike.md)
|
||||
@@ -160,11 +158,9 @@
|
||||
## Functions
|
||||
|
||||
- [RecordBatchIterator](functions/RecordBatchIterator.md)
|
||||
- [blob](functions/blob.md)
|
||||
- [connect](functions/connect.md)
|
||||
- [connectNamespace](functions/connectNamespace.md)
|
||||
- [instrumentLanceDbMetrics](functions/instrumentLanceDbMetrics.md)
|
||||
- [isBlobField](functions/isBlobField.md)
|
||||
- [makeArrowTable](functions/makeArrowTable.md)
|
||||
- [packBits](functions/packBits.md)
|
||||
- [permutationBuilder](functions/permutationBuilder.md)
|
||||
|
||||
@@ -1,48 +0,0 @@
|
||||
[**@lancedb/lancedb**](../README.md) • **Docs**
|
||||
|
||||
***
|
||||
|
||||
[@lancedb/lancedb](../globals.md) / BlobOptions
|
||||
|
||||
# Type Alias: BlobOptions
|
||||
|
||||
```ts
|
||||
type BlobOptions: object;
|
||||
```
|
||||
|
||||
## Type declaration
|
||||
|
||||
### dedicatedSizeThreshold?
|
||||
|
||||
```ts
|
||||
optional dedicatedSizeThreshold: number;
|
||||
```
|
||||
|
||||
Max payload bytes stored in a packed sidecar before a dedicated file. Must
|
||||
be a positive safe integer.
|
||||
|
||||
### inlineSizeThreshold?
|
||||
|
||||
```ts
|
||||
optional inlineSizeThreshold: number;
|
||||
```
|
||||
|
||||
Max payload bytes kept inline in the data file. Zero is allowed. Must be a
|
||||
safe integer.
|
||||
|
||||
### nullable?
|
||||
|
||||
```ts
|
||||
optional nullable: boolean;
|
||||
```
|
||||
|
||||
Defaults to true.
|
||||
|
||||
### packFileSizeThreshold?
|
||||
|
||||
```ts
|
||||
optional packFileSizeThreshold: number;
|
||||
```
|
||||
|
||||
Max bytes in one packed sidecar before starting another. Must be a positive
|
||||
safe integer.
|
||||
+1
-1
@@ -28,7 +28,7 @@
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
<arrow.version>15.0.0</arrow.version>
|
||||
<lance-core.version>12.0.0-beta.17</lance-core.version>
|
||||
<lance-core.version>12.0.0-beta.16</lance-core.version>
|
||||
<spotless.skip>false</spotless.skip>
|
||||
<spotless.version>2.30.0</spotless.version>
|
||||
<spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
|
||||
|
||||
+1
-1
@@ -44,6 +44,6 @@ aws-lc-rs = "=1.16.3"
|
||||
napi-build = "2.3.1"
|
||||
|
||||
[features]
|
||||
default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/goosefs", "lancedb/metrics-otel"]
|
||||
default = ["remote", "lancedb/sql", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/goosefs", "lancedb/metrics-otel"]
|
||||
fp16kernels = ["lancedb/fp16kernels"]
|
||||
remote = ["lancedb/remote"]
|
||||
|
||||
@@ -1,185 +0,0 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import { Field, Int64, List, Schema, Struct, Utf8 } from "apache-arrow";
|
||||
import { makeArrowTable } from "../lancedb/arrow";
|
||||
import { BlobFile, blob, coerceBlobValue, isBlobField } from "../lancedb/blob";
|
||||
|
||||
describe("blob()", () => {
|
||||
it("marks the field as lance.blob.v2", () => {
|
||||
const field = blob("image", { nullable: false });
|
||||
expect(field.nullable).toBe(false);
|
||||
expect(isBlobField(field)).toBe(true);
|
||||
expect(field.metadata.get("ARROW:extension:name")).toBe("lance.blob.v2");
|
||||
});
|
||||
|
||||
it("writes encoding thresholds as field metadata", () => {
|
||||
const field = blob("video", {
|
||||
inlineSizeThreshold: 1024,
|
||||
dedicatedSizeThreshold: 2 * 1024 * 1024,
|
||||
packFileSizeThreshold: 64 * 1024 * 1024,
|
||||
});
|
||||
expect(
|
||||
field.metadata.get("lance-encoding:blob-inline-size-threshold"),
|
||||
).toBe("1024");
|
||||
expect(
|
||||
field.metadata.get("lance-encoding:blob-dedicated-size-threshold"),
|
||||
).toBe(String(2 * 1024 * 1024));
|
||||
expect(
|
||||
field.metadata.get("lance-encoding:blob-pack-file-size-threshold"),
|
||||
).toBe(String(64 * 1024 * 1024));
|
||||
});
|
||||
|
||||
it("rejects invalid thresholds", () => {
|
||||
expect(() => blob("image", { inlineSizeThreshold: -1 })).toThrow(
|
||||
/inlineSizeThreshold must be non-negative/,
|
||||
);
|
||||
expect(() => blob("image", { dedicatedSizeThreshold: 0 })).toThrow(
|
||||
/dedicatedSizeThreshold must be positive/,
|
||||
);
|
||||
expect(() => blob("image", { packFileSizeThreshold: 1.5 })).toThrow(
|
||||
/packFileSizeThreshold must be a safe integer/,
|
||||
);
|
||||
expect(() =>
|
||||
blob("image", { dedicatedSizeThreshold: Number.MAX_SAFE_INTEGER + 1 }),
|
||||
).toThrow(/dedicatedSizeThreshold must be a safe integer/);
|
||||
});
|
||||
});
|
||||
|
||||
describe("coerceBlobValue", () => {
|
||||
it.each([
|
||||
["Buffer", Buffer.from("x"), { data: Buffer.from("x"), uri: null }],
|
||||
[
|
||||
"Uint8Array",
|
||||
new Uint8Array([120]),
|
||||
{ data: new Uint8Array([120]), uri: null },
|
||||
],
|
||||
["URI string", "s3://bucket/key", { data: null, uri: "s3://bucket/key" }],
|
||||
[
|
||||
"data struct",
|
||||
{ data: Buffer.from("y") },
|
||||
{ data: Buffer.from("y"), uri: null },
|
||||
],
|
||||
[
|
||||
"uri struct",
|
||||
{ uri: "s3://bucket/key" },
|
||||
{ data: null, uri: "s3://bucket/key" },
|
||||
],
|
||||
["null", null, null],
|
||||
])("accepts %s", (_name, input, expected) => {
|
||||
expect(coerceBlobValue(input)).toEqual(expected);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["empty URI", "", /uri cannot be empty/],
|
||||
["object without data or uri", { position: 0 }, /data' or 'uri/],
|
||||
[
|
||||
"Int16Array",
|
||||
new Int16Array([1]),
|
||||
/Blob data must be Buffer or Uint8Array/,
|
||||
],
|
||||
[
|
||||
"both data and uri",
|
||||
{ data: Buffer.from("y"), uri: "s3://bucket/key" },
|
||||
/exactly one of 'data' or 'uri'/,
|
||||
],
|
||||
[
|
||||
"neither data nor uri",
|
||||
{ data: null, uri: null },
|
||||
/exactly one of 'data' or 'uri'/,
|
||||
],
|
||||
])("rejects %s", (_name, input, message) => {
|
||||
expect(() => coerceBlobValue(input)).toThrow(message);
|
||||
});
|
||||
});
|
||||
|
||||
describe("BlobFile", () => {
|
||||
it("rejects constructing BlobFile without a native handle", () => {
|
||||
expect(() => new (BlobFile as unknown as { new (): BlobFile })()).toThrow(
|
||||
/fetchBlobFiles/,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("makeArrowTable blob columns", () => {
|
||||
it("coerces Buffer input onto a blob field", () => {
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
blob("image"),
|
||||
]);
|
||||
const table = makeArrowTable([{ id: 1n, image: Buffer.from("hello") }], {
|
||||
schema,
|
||||
});
|
||||
expect(isBlobField(table.schema.fields[1])).toBe(true);
|
||||
const image = table.getChild("image")!;
|
||||
expect(image.nullCount).toBe(0);
|
||||
expect(image.getChild("uri")!.get(0)).toBeNull();
|
||||
expect(image.getChild("data")!.nullCount).toBe(0);
|
||||
expect(Buffer.from(image.getChild("data")!.get(0)!).toString()).toBe(
|
||||
"hello",
|
||||
);
|
||||
});
|
||||
|
||||
it("coerces Buffer elements inside a list and keeps null slots", () => {
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
new Field("images", new List(blob("image")), true),
|
||||
]);
|
||||
const table = makeArrowTable(
|
||||
[
|
||||
{ id: 1n, images: [Buffer.from("a"), Buffer.from("bb")] },
|
||||
{ id: 2n, images: null },
|
||||
{ id: 3n, images: [Buffer.from("c"), null] },
|
||||
{ id: 4n, images: [] },
|
||||
],
|
||||
{ schema },
|
||||
);
|
||||
const images = table.getChild("images")!;
|
||||
expect(images.nullCount).toBe(1);
|
||||
const rows = images.toArray();
|
||||
expect(rows[1]).toBeNull();
|
||||
expect(Array.from(rows[3] as Iterable<unknown>)).toHaveLength(0);
|
||||
const first = Array.from(rows[0] as Iterable<{ data: Uint8Array | null }>);
|
||||
expect(Buffer.from(first[0].data!).toString()).toBe("a");
|
||||
expect(Buffer.from(first[1].data!).toString()).toBe("bb");
|
||||
const third = Array.from(
|
||||
rows[2] as Iterable<{ data: Uint8Array | null } | null>,
|
||||
);
|
||||
expect(Buffer.from(third[0]!.data!).toString()).toBe("c");
|
||||
expect(third[1]).toBeNull();
|
||||
});
|
||||
|
||||
it("coerces Buffer fields inside list structs", () => {
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
new Field(
|
||||
"items",
|
||||
new List(
|
||||
new Field(
|
||||
"item",
|
||||
new Struct([new Field("name", new Utf8(), true), blob("image")]),
|
||||
true,
|
||||
),
|
||||
),
|
||||
true,
|
||||
),
|
||||
]);
|
||||
const table = makeArrowTable(
|
||||
[
|
||||
{
|
||||
id: 1n,
|
||||
items: [{ name: "one", image: Buffer.from("alpha") }],
|
||||
},
|
||||
],
|
||||
{ schema },
|
||||
);
|
||||
const items = Array.from(
|
||||
table.getChild("items")!.toArray()[0] as Iterable<{
|
||||
name: string;
|
||||
image: { data: Uint8Array | null };
|
||||
}>,
|
||||
);
|
||||
expect(items[0].name).toBe("one");
|
||||
expect(Buffer.from(items[0].image.data!).toString()).toBe("alpha");
|
||||
});
|
||||
});
|
||||
@@ -18,7 +18,6 @@ import {
|
||||
Query,
|
||||
Table,
|
||||
VectorQuery,
|
||||
blob,
|
||||
connect,
|
||||
tokenize,
|
||||
} from "../lancedb";
|
||||
@@ -2402,276 +2401,6 @@ describe("when dealing with versioning", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("when dealing with blob columns", () => {
|
||||
let tmpDir: tmp.DirResult;
|
||||
beforeEach(() => {
|
||||
tmpDir = tmp.dirSync({ unsafeCleanup: true });
|
||||
});
|
||||
afterEach(() => {
|
||||
tmpDir.removeCallback();
|
||||
});
|
||||
|
||||
it("discovers blob columns", async () => {
|
||||
const { table } = await openBlobTable();
|
||||
expect(await table.blobColumns()).toEqual(["image"]);
|
||||
});
|
||||
|
||||
it("preserves order, duplicates, and nulls", async () => {
|
||||
const { table, rowIds } = await openBlobTable();
|
||||
const [alphaId, betaId, nullId] = rowIds;
|
||||
const bytes = await table.fetchBlobs("image", [
|
||||
betaId,
|
||||
alphaId,
|
||||
betaId,
|
||||
nullId,
|
||||
]);
|
||||
expect(bytes.map((b) => (b == null ? null : b.toString()))).toEqual([
|
||||
"beta",
|
||||
"alpha",
|
||||
"beta",
|
||||
null,
|
||||
]);
|
||||
const files = await table.fetchBlobFiles("image", [
|
||||
betaId,
|
||||
nullId,
|
||||
alphaId,
|
||||
]);
|
||||
expect(files.map((f) => f == null)).toEqual([false, true, false]);
|
||||
});
|
||||
|
||||
it("reads full blob contents", async () => {
|
||||
const { table, rowIds, alpha, beta } = await openBlobTable();
|
||||
const bytes = await table.fetchBlobs("image", rowIds);
|
||||
expect(bytes[0]!.equals(alpha)).toBe(true);
|
||||
expect(bytes[1]!.equals(beta)).toBe(true);
|
||||
const files = await table.fetchBlobFiles("image", rowIds);
|
||||
expect(files[0]!.size()).toBe(BigInt(alpha.length));
|
||||
expect(Buffer.from(await files[0]!.read()).toString()).toBe("alpha");
|
||||
expect(Buffer.from(await files[1]!.read()).toString()).toBe("beta");
|
||||
});
|
||||
|
||||
it("reads a half-open range", async () => {
|
||||
const { table, rowIds } = await openBlobTable();
|
||||
const files = await table.fetchBlobFiles("image", rowIds);
|
||||
expect(Buffer.from(await files[0]!.readRange(0n, 2n)).toString()).toBe(
|
||||
"al",
|
||||
);
|
||||
});
|
||||
|
||||
it("readRange does not move the cursor", async () => {
|
||||
const { table, rowIds, alpha } = await openBlobTable();
|
||||
const [handle] = await table.fetchBlobFiles("image", rowIds);
|
||||
expect((await handle!.readRange(1n, 3n)).toString()).toBe("lp");
|
||||
expect(await handle!.read()).toEqual(alpha);
|
||||
expect(await handle!.read()).toEqual(Buffer.alloc(0));
|
||||
});
|
||||
|
||||
it("fails when readRange end is past the blob size", async () => {
|
||||
const { table, rowIds, alpha } = await openBlobTable();
|
||||
const files = await table.fetchBlobFiles("image", rowIds);
|
||||
await expect(
|
||||
files[0]!.readRange(0n, BigInt(alpha.length + 1)),
|
||||
).rejects.toThrow(/exceeds blob size/);
|
||||
});
|
||||
|
||||
it("rejects fetchBlobs on a non-blob column", async () => {
|
||||
const { table, rowIds } = await openBlobTable();
|
||||
await expect(table.fetchBlobs("id", rowIds)).rejects.toThrow(/blob/i);
|
||||
});
|
||||
|
||||
it("discovers and fetches nested blob columns", async () => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
new Field("info", new Struct([blob("image")]), true),
|
||||
]);
|
||||
const payload = Buffer.from("nested");
|
||||
const table = await db.createTable(
|
||||
"nested_blobs",
|
||||
[{ id: 1n, info: { image: payload } }],
|
||||
{ schema },
|
||||
);
|
||||
expect(await table.blobColumns()).toEqual(["info.image"]);
|
||||
const rows = await table.query().withRowId().toArray();
|
||||
const bytes = await table.fetchBlobs("info.image", [
|
||||
rows[0]._rowid as bigint,
|
||||
]);
|
||||
expect(bytes[0]!.equals(payload)).toBe(true);
|
||||
});
|
||||
|
||||
it("creates and adds list blob columns", async () => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
new Field("images", new List(blob("image")), true),
|
||||
]);
|
||||
const alpha = Buffer.from("alpha");
|
||||
const beta = Buffer.from("beta");
|
||||
const gamma = Buffer.from("gamma");
|
||||
const table = await db.createTable(
|
||||
"list_blobs",
|
||||
[{ id: 1n, images: [alpha, beta] }],
|
||||
{ schema },
|
||||
);
|
||||
await table.add([
|
||||
{ id: 2n, images: null },
|
||||
{ id: 3n, images: [gamma, null] },
|
||||
{ id: 4n, images: [] },
|
||||
]);
|
||||
expect(await table.blobColumns()).toEqual(["images.image"]);
|
||||
const rows = await table.query().toArray();
|
||||
const byId = new Map(rows.map((row) => [Number(row.id), row]));
|
||||
expect(descriptorSizes(byId.get(1)!.images)).toEqual([
|
||||
alpha.length,
|
||||
beta.length,
|
||||
]);
|
||||
expect(byId.get(2)!.images).toBeNull();
|
||||
expect(descriptorSizes(byId.get(3)!.images)).toEqual([gamma.length, null]);
|
||||
expect(Array.from(byId.get(4)!.images as Iterable<unknown>)).toHaveLength(
|
||||
0,
|
||||
);
|
||||
});
|
||||
|
||||
it("creates and adds list struct blob columns", async () => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
new Field(
|
||||
"items",
|
||||
new List(
|
||||
new Field(
|
||||
"item",
|
||||
new Struct([new Field("name", new Utf8(), true), blob("image")]),
|
||||
true,
|
||||
),
|
||||
),
|
||||
true,
|
||||
),
|
||||
]);
|
||||
const alpha = Buffer.from("nested-alpha");
|
||||
const beta = Buffer.from("nested-beta");
|
||||
const table = await db.createTable(
|
||||
"list_struct_blobs",
|
||||
[{ id: 1n, items: [{ name: "one", image: alpha }] }],
|
||||
{ schema },
|
||||
);
|
||||
await table.add([
|
||||
{
|
||||
id: 2n,
|
||||
items: [
|
||||
{ name: "two", image: beta },
|
||||
{ name: "three", image: null },
|
||||
],
|
||||
},
|
||||
]);
|
||||
const rows = await table.query().toArray();
|
||||
const byId = new Map(rows.map((row) => [Number(row.id), row]));
|
||||
expect(
|
||||
descriptorSizes(
|
||||
Array.from(byId.get(1)!.items as Iterable<{ image: unknown }>).map(
|
||||
(item) => item.image,
|
||||
),
|
||||
),
|
||||
).toEqual([alpha.length]);
|
||||
expect(
|
||||
descriptorSizes(
|
||||
Array.from(byId.get(2)!.items as Iterable<{ image: unknown }>).map(
|
||||
(item) => item.image,
|
||||
),
|
||||
),
|
||||
).toEqual([beta.length, null]);
|
||||
});
|
||||
|
||||
it("rejects blob fields inside a fixed-size list", async () => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
new Field("frames", new FixedSizeList(2, blob("frame")), true),
|
||||
]);
|
||||
await expect(
|
||||
db.createTable(
|
||||
"fsl_blobs",
|
||||
[{ id: 1n, frames: [Buffer.from("a"), Buffer.from("b")] }],
|
||||
{ schema },
|
||||
),
|
||||
).rejects.toThrow(
|
||||
"Blob fields inside FixedSizeList are not supported. Use List instead.",
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects blob fields inside a nested fixed-size list", async () => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
new Field(
|
||||
"clip",
|
||||
new Struct([
|
||||
new Field("frames", new FixedSizeList(2, blob("frame")), true),
|
||||
]),
|
||||
true,
|
||||
),
|
||||
]);
|
||||
await expect(
|
||||
db.createTable(
|
||||
"nested_fsl_blobs",
|
||||
[
|
||||
{
|
||||
id: 1n,
|
||||
clip: { frames: [Buffer.from("a"), Buffer.from("b")] },
|
||||
},
|
||||
],
|
||||
{ schema },
|
||||
),
|
||||
).rejects.toThrow(
|
||||
"Blob fields inside FixedSizeList are not supported. Use List instead.",
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects an Arrow table with blob fields inside a fixed-size list", async () => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
new Field("frames", new FixedSizeList(2, blob("frame")), true),
|
||||
]);
|
||||
await expect(
|
||||
db.createTable("fsl_blobs_ipc", new ArrowTable(schema)),
|
||||
).rejects.toThrow(
|
||||
"Blob fields inside FixedSizeList are not supported. Use List instead.",
|
||||
);
|
||||
});
|
||||
|
||||
function descriptorSizes(values: unknown): (number | null)[] {
|
||||
return Array.from(
|
||||
values as Iterable<{ size?: bigint | number } | null>,
|
||||
).map((value) => (value == null ? null : Number(value.size)));
|
||||
}
|
||||
|
||||
async function openBlobTable() {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = new Schema([
|
||||
new Field("id", new Int64(), true),
|
||||
blob("image"),
|
||||
]);
|
||||
const alpha = Buffer.from("alpha");
|
||||
const beta = Buffer.from("beta");
|
||||
const table = await db.createTable(
|
||||
"blobs",
|
||||
[
|
||||
{ id: 1n, image: alpha },
|
||||
{ id: 2n, image: beta },
|
||||
{ id: 3n, image: null },
|
||||
],
|
||||
{ schema },
|
||||
);
|
||||
const rows = await table.query().withRowId().toArray();
|
||||
const rowIdById = new Map(
|
||||
rows.map((r) => [Number(r.id), r._rowid as bigint]),
|
||||
);
|
||||
const rowIds = [1, 2, 3].map((id) => rowIdById.get(id)!);
|
||||
return { table, rowIds, alpha, beta };
|
||||
}
|
||||
});
|
||||
|
||||
describe("when dealing with tags", () => {
|
||||
let tmpDir: tmp.DirResult;
|
||||
beforeEach(() => {
|
||||
|
||||
+1
-102
@@ -40,7 +40,6 @@ import {
|
||||
} from "apache-arrow";
|
||||
import { Buffers } from "apache-arrow/data";
|
||||
import { typedArrayToArrowType } from "./arrow_type";
|
||||
import { coerceBlobValue, isBlobField } from "./blob";
|
||||
import { type EmbeddingFunction } from "./embedding/embedding_function";
|
||||
import {
|
||||
EmbeddingFunctionConfig,
|
||||
@@ -431,14 +430,12 @@ export function makeArrowTable(
|
||||
throw new Error("A schema must be provided if data is empty");
|
||||
} else {
|
||||
schema = new Schema(schema.fields, schemaMetadata);
|
||||
validateBlobSchema(schema);
|
||||
return new ArrowTable(schema);
|
||||
}
|
||||
}
|
||||
|
||||
let inferredSchema = inferSchema(data, schema, opt);
|
||||
inferredSchema = new Schema(inferredSchema.fields, schemaMetadata);
|
||||
validateBlobSchema(inferredSchema);
|
||||
|
||||
const finalColumns: Record<string, Vector> = {};
|
||||
for (const field of inferredSchema.fields) {
|
||||
@@ -448,35 +445,6 @@ export function makeArrowTable(
|
||||
return new ArrowTable(inferredSchema, finalColumns);
|
||||
}
|
||||
|
||||
function validateBlobSchema(schema: Schema): void {
|
||||
for (const field of schema.fields) {
|
||||
validateBlobField(field);
|
||||
}
|
||||
}
|
||||
|
||||
function validateBlobField(field: Field): void {
|
||||
if (
|
||||
isFixedSizeList(field.type) &&
|
||||
containsBlobField(field.type.children[0])
|
||||
) {
|
||||
throw new Error(
|
||||
"Blob fields inside FixedSizeList are not supported. Use List instead.",
|
||||
);
|
||||
}
|
||||
for (const child of field.type.children ?? []) {
|
||||
validateBlobField(child);
|
||||
}
|
||||
}
|
||||
|
||||
function containsBlobField(field: Field): boolean {
|
||||
if (isBlobField(field)) {
|
||||
return true;
|
||||
}
|
||||
return (field.type.children ?? []).some((child: Field) =>
|
||||
containsBlobField(child),
|
||||
);
|
||||
}
|
||||
|
||||
function isObject(value: unknown): value is Record<string, unknown> {
|
||||
return (
|
||||
typeof value === "object" &&
|
||||
@@ -512,32 +480,6 @@ function transposeData(
|
||||
path: string[] = [],
|
||||
): Vector {
|
||||
const valuesPath = [...path, field.name];
|
||||
if (isBlobField(field) && field.type instanceof Struct) {
|
||||
const blobRows = data.map((datum) =>
|
||||
coerceBlobValue(valueAtPath(datum, valuesPath)),
|
||||
);
|
||||
const childVectors = field.type.children.map((child) => {
|
||||
const values = blobRows.map((row) =>
|
||||
row == null ? null : (row[child.name as "data" | "uri"] ?? null),
|
||||
);
|
||||
return makeVector(values, child.type, undefined, child.nullable);
|
||||
});
|
||||
const nullCount = blobRows.filter((row) => row === null).length;
|
||||
const structData = makeData({
|
||||
type: field.type,
|
||||
length: blobRows.length,
|
||||
nullCount,
|
||||
nullBitmap:
|
||||
nullCount > 0
|
||||
? arrowUtil.packBools(blobRows.map((row) => row !== null))
|
||||
: undefined,
|
||||
children: childVectors.map((v) => v.data[0]),
|
||||
});
|
||||
return arrowMakeVector(structData);
|
||||
}
|
||||
if (isList(field.type) && containsBlobField(field.type.children[0])) {
|
||||
return transposeListData(data, field, valuesPath);
|
||||
}
|
||||
const values = data.map((datum) => valueAtPath(datum, valuesPath));
|
||||
if (field.type instanceof Struct) {
|
||||
const childFields = field.type.children;
|
||||
@@ -553,7 +495,7 @@ function transposeData(
|
||||
nullCount > 0
|
||||
? arrowUtil.packBools(values.map((value) => value !== null))
|
||||
: undefined,
|
||||
children: childVectors.map((v) => v.data[0]),
|
||||
children: childVectors as unknown as ArrowData<DataType>[],
|
||||
});
|
||||
return arrowMakeVector(structData);
|
||||
} else {
|
||||
@@ -561,48 +503,6 @@ function transposeData(
|
||||
}
|
||||
}
|
||||
|
||||
function transposeListData(
|
||||
data: Record<string, unknown>[],
|
||||
field: Field,
|
||||
valuesPath: string[],
|
||||
): Vector {
|
||||
const listType = field.type as List;
|
||||
const childField = listType.children[0];
|
||||
const lists = data.map((datum) => valueAtPath(datum, valuesPath));
|
||||
const flattened: Record<string, unknown>[] = [];
|
||||
const validity: boolean[] = [];
|
||||
const offsets: number[] = [0];
|
||||
|
||||
for (const list of lists) {
|
||||
if (list == null) {
|
||||
validity.push(false);
|
||||
offsets.push(flattened.length);
|
||||
continue;
|
||||
}
|
||||
if (!Array.isArray(list)) {
|
||||
throw new Error(`expected an array for list field '${field.name}'`);
|
||||
}
|
||||
validity.push(true);
|
||||
for (const element of list) {
|
||||
flattened.push({ [childField.name]: element });
|
||||
}
|
||||
offsets.push(flattened.length);
|
||||
}
|
||||
|
||||
const childVector = transposeData(flattened, childField, []);
|
||||
const nullCount = validity.filter((valid) => !valid).length;
|
||||
return arrowMakeVector(
|
||||
makeData({
|
||||
type: listType,
|
||||
length: lists.length,
|
||||
nullCount,
|
||||
nullBitmap: nullCount > 0 ? arrowUtil.packBools(validity) : undefined,
|
||||
valueOffsets: Int32Array.from(offsets),
|
||||
child: childVector.data[0],
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an empty Arrow table with the provided schema
|
||||
*/
|
||||
@@ -1052,7 +952,6 @@ export async function fromTableToBuffer(
|
||||
schema = sanitizeSchema(schema);
|
||||
}
|
||||
const tableWithEmbeddings = await applyEmbeddings(table, embeddings, schema);
|
||||
validateBlobSchema(tableWithEmbeddings.schema);
|
||||
const writer = RecordBatchFileWriter.writeAll(tableWithEmbeddings);
|
||||
return Buffer.from(await writer.toUint8Array());
|
||||
}
|
||||
|
||||
@@ -1,236 +0,0 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import { Field, LargeBinary, Struct, Utf8 } from "apache-arrow";
|
||||
import { BlobFile as NativeBlobFile } from "./native";
|
||||
|
||||
const BLOB_V2_EXTENSION_NAME = "lance.blob.v2";
|
||||
|
||||
const INLINE_SIZE_THRESHOLD_KEY = "lance-encoding:blob-inline-size-threshold";
|
||||
const DEDICATED_SIZE_THRESHOLD_KEY =
|
||||
"lance-encoding:blob-dedicated-size-threshold";
|
||||
const PACK_FILE_SIZE_THRESHOLD_KEY =
|
||||
"lance-encoding:blob-pack-file-size-threshold";
|
||||
|
||||
export type BlobInput = {
|
||||
data: Buffer | Uint8Array | null;
|
||||
uri: string | null;
|
||||
};
|
||||
|
||||
export type BlobOptions = {
|
||||
/** Defaults to true. */
|
||||
nullable?: boolean;
|
||||
/**
|
||||
* Max payload bytes kept inline in the data file. Zero is allowed. Must be a
|
||||
* safe integer.
|
||||
*/
|
||||
inlineSizeThreshold?: number;
|
||||
/**
|
||||
* Max payload bytes stored in a packed sidecar before a dedicated file. Must
|
||||
* be a positive safe integer.
|
||||
*/
|
||||
dedicatedSizeThreshold?: number;
|
||||
/**
|
||||
* Max bytes in one packed sidecar before starting another. Must be a positive
|
||||
* safe integer.
|
||||
*/
|
||||
packFileSizeThreshold?: number;
|
||||
};
|
||||
|
||||
/**
|
||||
* Declares a `lance.blob.v2` column.
|
||||
*
|
||||
* Query results are descriptors, not payload bytes. Use {@link Table.fetchBlobs}
|
||||
* or {@link Table.fetchBlobFiles} to read bytes.
|
||||
*
|
||||
* @example
|
||||
* ```ts
|
||||
* import { readFile } from "node:fs/promises";
|
||||
* import { Field, Int64, Schema } from "apache-arrow";
|
||||
* import { blob, connect } from "@lancedb/lancedb";
|
||||
*
|
||||
* const db = await connect("./data");
|
||||
* const video = await readFile("clip.mp4");
|
||||
* const table = await db.createTable(
|
||||
* "videos",
|
||||
* [{ id: 1n, video }],
|
||||
* {
|
||||
* schema: new Schema([
|
||||
* new Field("id", new Int64()),
|
||||
* blob("video"),
|
||||
* ]),
|
||||
* },
|
||||
* );
|
||||
*
|
||||
* const rows = await table.query().select(["id"]).withRowId().toArray();
|
||||
* const rowIds = rows.map((row) => row._rowid as bigint);
|
||||
* const bytes = await table.fetchBlobs("video", rowIds);
|
||||
*
|
||||
* const [handle] = await table.fetchBlobFiles("video", rowIds);
|
||||
* const size = handle!.size();
|
||||
* const header = await handle!.readRange(0n, size < 65536n ? size : 65536n);
|
||||
* ```
|
||||
*/
|
||||
export function blob(name: string, options: BlobOptions = {}): Field {
|
||||
const metadata = new Map<string, string>([
|
||||
["ARROW:extension:name", BLOB_V2_EXTENSION_NAME],
|
||||
]);
|
||||
setThreshold(
|
||||
metadata,
|
||||
INLINE_SIZE_THRESHOLD_KEY,
|
||||
"inlineSizeThreshold",
|
||||
options.inlineSizeThreshold,
|
||||
0,
|
||||
);
|
||||
setThreshold(
|
||||
metadata,
|
||||
DEDICATED_SIZE_THRESHOLD_KEY,
|
||||
"dedicatedSizeThreshold",
|
||||
options.dedicatedSizeThreshold,
|
||||
1,
|
||||
);
|
||||
setThreshold(
|
||||
metadata,
|
||||
PACK_FILE_SIZE_THRESHOLD_KEY,
|
||||
"packFileSizeThreshold",
|
||||
options.packFileSizeThreshold,
|
||||
1,
|
||||
);
|
||||
return new Field(
|
||||
name,
|
||||
new Struct([
|
||||
new Field("data", new LargeBinary(), true),
|
||||
new Field("uri", new Utf8(), true),
|
||||
]),
|
||||
options.nullable ?? true,
|
||||
metadata,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks for the `lance.blob.v2` extension marker. Does not validate the
|
||||
* field's storage type.
|
||||
*/
|
||||
export function isBlobField(field: Field): boolean {
|
||||
return field.metadata?.get("ARROW:extension:name") === BLOB_V2_EXTENSION_NAME;
|
||||
}
|
||||
|
||||
/**
|
||||
* A lazy handle to blob bytes. Create one with {@link Table.fetchBlobFiles}.
|
||||
*
|
||||
* @hideconstructor
|
||||
*/
|
||||
export class BlobFile {
|
||||
private readonly inner: NativeBlobFile;
|
||||
|
||||
private constructor(inner: NativeBlobFile) {
|
||||
if (!(inner instanceof NativeBlobFile)) {
|
||||
throw new Error("BlobFile handles come from Table.fetchBlobFiles");
|
||||
}
|
||||
this.inner = inner;
|
||||
}
|
||||
|
||||
/** @ignore */
|
||||
static fromNative(inner: NativeBlobFile): BlobFile {
|
||||
return new BlobFile(inner);
|
||||
}
|
||||
|
||||
/** Returns the blob size in bytes. */
|
||||
size(): bigint {
|
||||
return this.inner.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads from the cursor to the end and advances the cursor.
|
||||
*
|
||||
* A second call returns an empty buffer. {@link BlobFile.readRange} does
|
||||
* not move the cursor.
|
||||
*/
|
||||
read(): Promise<Buffer> {
|
||||
return this.inner.read();
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads the half-open byte range `[start, end)`.
|
||||
*
|
||||
* Fails when `end` is past the blob size. Does not move the cursor.
|
||||
*/
|
||||
readRange(start: bigint, end: bigint): Promise<Buffer> {
|
||||
return this.inner.readRange(start, end);
|
||||
}
|
||||
}
|
||||
|
||||
export function coerceBlobValue(value: unknown): BlobInput | null {
|
||||
if (value == null) {
|
||||
return null;
|
||||
}
|
||||
if (isBlobBytes(value)) {
|
||||
return { data: value, uri: null };
|
||||
}
|
||||
if (ArrayBuffer.isView(value)) {
|
||||
throw new Error("Blob data must be Buffer or Uint8Array");
|
||||
}
|
||||
if (typeof value === "string") {
|
||||
if (value === "") {
|
||||
throw new Error("Blob uri cannot be empty");
|
||||
}
|
||||
return { data: null, uri: value };
|
||||
}
|
||||
if (typeof value === "object") {
|
||||
const record = value as Record<string, unknown>;
|
||||
if (!("data" in record) && !("uri" in record)) {
|
||||
throw new Error(
|
||||
"Blob struct values must include a 'data' or 'uri' field",
|
||||
);
|
||||
}
|
||||
const uri = record.uri;
|
||||
if (uri === "") {
|
||||
throw new Error("Blob uri cannot be empty");
|
||||
}
|
||||
if (uri != null && typeof uri !== "string") {
|
||||
throw new Error(`Blob uri must be a string or null, got ${typeof uri}`);
|
||||
}
|
||||
const data = record.data;
|
||||
if (data != null && !isBlobBytes(data)) {
|
||||
throw new Error("Blob data must be Buffer, Uint8Array, or null");
|
||||
}
|
||||
const bytes = (data as Buffer | Uint8Array | null | undefined) ?? null;
|
||||
const uriValue = uri ?? null;
|
||||
if ((bytes == null) === (uriValue == null)) {
|
||||
throw new Error(
|
||||
"Blob struct values must set exactly one of 'data' or 'uri'",
|
||||
);
|
||||
}
|
||||
return { data: bytes, uri: uriValue };
|
||||
}
|
||||
throw new Error(
|
||||
"Blob column values must be Buffer, Uint8Array, a URI string, null, or { data?, uri? }",
|
||||
);
|
||||
}
|
||||
|
||||
function isBlobBytes(value: unknown): value is Buffer | Uint8Array {
|
||||
return Buffer.isBuffer(value) || value instanceof Uint8Array;
|
||||
}
|
||||
|
||||
function setThreshold(
|
||||
metadata: Map<string, string>,
|
||||
key: string,
|
||||
optionName: string,
|
||||
value: number | undefined,
|
||||
minimum: number,
|
||||
): void {
|
||||
if (value === undefined) {
|
||||
return;
|
||||
}
|
||||
if (!Number.isSafeInteger(value)) {
|
||||
throw new Error(`${optionName} must be a safe integer`);
|
||||
}
|
||||
if (value < minimum) {
|
||||
throw new Error(
|
||||
minimum <= 0
|
||||
? `${optionName} must be non-negative`
|
||||
: `${optionName} must be positive`,
|
||||
);
|
||||
}
|
||||
metadata.set(key, String(value));
|
||||
}
|
||||
@@ -77,9 +77,6 @@ export {
|
||||
VectorColumnOptions,
|
||||
} from "./arrow";
|
||||
|
||||
export { blob, isBlobField, BlobFile } from "./blob";
|
||||
export type { BlobOptions } from "./blob";
|
||||
|
||||
export {
|
||||
Connection,
|
||||
CreateTableOptions,
|
||||
|
||||
+16
-75
@@ -17,7 +17,6 @@ import {
|
||||
tableFromIPC,
|
||||
} from "./arrow";
|
||||
|
||||
import { BlobFile } from "./blob";
|
||||
import { EmbeddingFunctionConfig, getRegistry } from "./embedding/registry";
|
||||
import { IndexOptions } from "./indices";
|
||||
import { Job } from "./job";
|
||||
@@ -511,35 +510,6 @@ export abstract class Table {
|
||||
*/
|
||||
abstract takeRowIds(rowIds: readonly (bigint | number)[]): TakeQuery;
|
||||
|
||||
/**
|
||||
* Blob v2 columns, including nested dotted paths.
|
||||
*/
|
||||
abstract blobColumns(): Promise<string[]>;
|
||||
|
||||
/**
|
||||
* Bytes for `column` at row IDs from {@link Query.withRowId}.
|
||||
*
|
||||
* Reads the table's current checkout. IDs from another version can fail after
|
||||
* compaction unless stable row ids are enabled. Results keep input order and
|
||||
* duplicates. Null blobs are `null`. Empty blobs are empty buffers.
|
||||
*/
|
||||
abstract fetchBlobs(
|
||||
column: string,
|
||||
rowIds: readonly (bigint | number)[],
|
||||
): Promise<(Buffer | null)[]>;
|
||||
|
||||
/**
|
||||
* Opens lazy blob handles for `column` at the given row IDs using the
|
||||
* table's current checkout.
|
||||
*
|
||||
* Preserves input order, duplicates, and nulls. Use this for large payloads.
|
||||
* See {@link Table.fetchBlobs} for row-ID validity across versions.
|
||||
*/
|
||||
abstract fetchBlobFiles(
|
||||
column: string,
|
||||
rowIds: readonly (bigint | number)[],
|
||||
): Promise<(BlobFile | null)[]>;
|
||||
|
||||
/**
|
||||
* Create a search query to find the nearest neighbors
|
||||
* of the given query
|
||||
@@ -1190,34 +1160,23 @@ export class LocalTable extends Table {
|
||||
}
|
||||
|
||||
takeRowIds(rowIds: readonly (bigint | number)[]): TakeQuery {
|
||||
return new TakeQuery(this.inner.takeRowIds(rowIdsToBigInts(rowIds)));
|
||||
}
|
||||
const ids = rowIds.map((id) => {
|
||||
if (typeof id === "bigint") {
|
||||
return id;
|
||||
}
|
||||
if (!Number.isInteger(id)) {
|
||||
throw new Error("Row id must be an integer (or bigint)");
|
||||
}
|
||||
if (id < 0) {
|
||||
throw new Error("Row id cannot be negative");
|
||||
}
|
||||
if (!Number.isSafeInteger(id)) {
|
||||
throw new Error("Row id is too large for number; use bigint instead");
|
||||
}
|
||||
return BigInt(id);
|
||||
});
|
||||
|
||||
blobColumns(): Promise<string[]> {
|
||||
return this.inner.blobColumns();
|
||||
}
|
||||
|
||||
async fetchBlobs(
|
||||
column: string,
|
||||
rowIds: readonly (bigint | number)[],
|
||||
): Promise<(Buffer | null)[]> {
|
||||
const values = await this.inner.fetchBlobs(column, rowIdsToBigInts(rowIds));
|
||||
// N-API Option maps missing values to undefined. Collapse those to null.
|
||||
return values.map((value) => value ?? null);
|
||||
}
|
||||
|
||||
async fetchBlobFiles(
|
||||
column: string,
|
||||
rowIds: readonly (bigint | number)[],
|
||||
): Promise<(BlobFile | null)[]> {
|
||||
const files = await this.inner.fetchBlobFiles(
|
||||
column,
|
||||
rowIdsToBigInts(rowIds),
|
||||
);
|
||||
// N-API Option maps missing values to undefined. Collapse those to null.
|
||||
return files.map((file) =>
|
||||
file == null ? null : BlobFile.fromNative(file),
|
||||
);
|
||||
return new TakeQuery(this.inner.takeRowIds(ids));
|
||||
}
|
||||
|
||||
query(): Query {
|
||||
@@ -1774,21 +1733,3 @@ export class Branches {
|
||||
)) as unknown as CherryPickResult;
|
||||
}
|
||||
}
|
||||
|
||||
function rowIdsToBigInts(rowIds: readonly (bigint | number)[]): bigint[] {
|
||||
return rowIds.map((id) => {
|
||||
if (typeof id === "bigint") {
|
||||
return id;
|
||||
}
|
||||
if (!Number.isInteger(id)) {
|
||||
throw new Error("Row id must be an integer (or bigint)");
|
||||
}
|
||||
if (id < 0) {
|
||||
throw new Error("Row id cannot be negative");
|
||||
}
|
||||
if (!Number.isSafeInteger(id)) {
|
||||
throw new Error("Row id is too large for number; use bigint instead");
|
||||
}
|
||||
return BigInt(id);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1,95 +0,0 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
use std::ops::Range;
|
||||
use std::sync::Arc;
|
||||
|
||||
use arrow_array::{Array, LargeBinaryArray};
|
||||
use lancedb::blob::BlobFile as LanceBlobFile;
|
||||
use napi::bindgen_prelude::*;
|
||||
use napi_derive::napi;
|
||||
|
||||
use crate::error::convert_error;
|
||||
|
||||
#[napi]
|
||||
pub struct BlobFile {
|
||||
inner: Arc<LanceBlobFile>,
|
||||
}
|
||||
|
||||
impl BlobFile {
|
||||
pub(crate) fn new(inner: LanceBlobFile) -> Self {
|
||||
Self {
|
||||
inner: Arc::new(inner),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[napi]
|
||||
impl BlobFile {
|
||||
#[napi]
|
||||
pub fn size(&self) -> BigInt {
|
||||
BigInt::from(self.inner.size())
|
||||
}
|
||||
|
||||
#[napi]
|
||||
pub async fn read(&self) -> napi::Result<Buffer> {
|
||||
let bytes = self.inner.read().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> {
|
||||
let range = bigint_range(start, end)?;
|
||||
let bytes = self
|
||||
.inner
|
||||
.read_range(range)
|
||||
.await
|
||||
.map_err(|err| convert_error(&err))?;
|
||||
Ok(Buffer::from(bytes.as_ref()))
|
||||
}
|
||||
}
|
||||
|
||||
fn bigint_range(start: BigInt, end: BigInt) -> napi::Result<Range<u64>> {
|
||||
let start = parse_u64(start, "start")?;
|
||||
let end = parse_u64(end, "end")?;
|
||||
if start > end {
|
||||
return Err(napi::Error::from_reason(format!(
|
||||
"invalid blob range: start ({start}) > end ({end})"
|
||||
)));
|
||||
}
|
||||
Ok(start..end)
|
||||
}
|
||||
|
||||
pub(crate) fn parse_u64(value: BigInt, name: &str) -> napi::Result<u64> {
|
||||
let (negative, value, lossless) = value.get_u64();
|
||||
if negative {
|
||||
return Err(napi::Error::from_reason(format!(
|
||||
"{name} cannot be negative"
|
||||
)));
|
||||
}
|
||||
if !lossless {
|
||||
return Err(napi::Error::from_reason(format!(
|
||||
"{name} is too large to fit in u64"
|
||||
)));
|
||||
}
|
||||
Ok(value)
|
||||
}
|
||||
|
||||
pub(crate) fn parse_row_ids(row_ids: Vec<BigInt>) -> napi::Result<Vec<u64>> {
|
||||
row_ids
|
||||
.into_iter()
|
||||
.map(|id| parse_u64(id, "row id"))
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub(crate) fn copy_blob_buffers(array: LargeBinaryArray) -> Vec<Option<Buffer>> {
|
||||
(0..array.len())
|
||||
.map(|i| {
|
||||
if array.is_null(i) {
|
||||
None
|
||||
} else {
|
||||
Some(Buffer::from(array.value(i).to_vec()))
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
@@ -10,7 +10,6 @@ use std::collections::HashMap;
|
||||
use env_logger::Env;
|
||||
use napi_derive::*;
|
||||
|
||||
mod blob;
|
||||
mod connection;
|
||||
mod error;
|
||||
mod header;
|
||||
|
||||
@@ -15,7 +15,6 @@ use napi::bindgen_prelude::*;
|
||||
use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode};
|
||||
use napi_derive::napi;
|
||||
|
||||
use crate::blob::{BlobFile, copy_blob_buffers, parse_row_ids};
|
||||
use crate::error::NapiErrorExt;
|
||||
use crate::index::Index;
|
||||
use crate::merge::NativeMergeInsertBuilder;
|
||||
@@ -330,44 +329,6 @@ impl Table {
|
||||
))
|
||||
}
|
||||
|
||||
#[napi(catch_unwind)]
|
||||
pub async fn blob_columns(&self) -> napi::Result<Vec<String>> {
|
||||
self.inner_ref()?.blob_columns().await.default_error()
|
||||
}
|
||||
|
||||
#[napi(catch_unwind)]
|
||||
pub async fn fetch_blobs(
|
||||
&self,
|
||||
column: String,
|
||||
row_ids: Vec<BigInt>,
|
||||
) -> napi::Result<Vec<Option<Buffer>>> {
|
||||
let row_ids = parse_row_ids(row_ids)?;
|
||||
let array = self
|
||||
.inner_ref()?
|
||||
.fetch_blobs(column.as_str(), &row_ids)
|
||||
.await
|
||||
.default_error()?;
|
||||
Ok(copy_blob_buffers(array))
|
||||
}
|
||||
|
||||
#[napi(catch_unwind)]
|
||||
pub async fn fetch_blob_files(
|
||||
&self,
|
||||
column: String,
|
||||
row_ids: Vec<BigInt>,
|
||||
) -> napi::Result<Vec<Option<BlobFile>>> {
|
||||
let row_ids = parse_row_ids(row_ids)?;
|
||||
let files = self
|
||||
.inner_ref()?
|
||||
.fetch_blob_files(column.as_str(), &row_ids)
|
||||
.await
|
||||
.default_error()?;
|
||||
Ok(files
|
||||
.into_iter()
|
||||
.map(|file| file.map(BlobFile::new))
|
||||
.collect())
|
||||
}
|
||||
|
||||
#[napi(catch_unwind)]
|
||||
pub fn vector_search(&self, vector: Float32Array) -> napi::Result<VectorQuery> {
|
||||
self.query()?.nearest_to(vector)
|
||||
|
||||
+1
-1
@@ -47,6 +47,6 @@ libc = "0.2"
|
||||
pyo3-build-config = { version = "0.28", features = ["abi3-py310"] }
|
||||
|
||||
[features]
|
||||
default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/cos", "lancedb/goosefs", "lancedb/metrics-otel"]
|
||||
default = ["remote", "lancedb/sql", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/cos", "lancedb/goosefs", "lancedb/metrics-otel"]
|
||||
fp16kernels = ["lancedb/fp16kernels"]
|
||||
remote = ["lancedb/remote"]
|
||||
|
||||
@@ -120,7 +120,13 @@ pprof = { version = "0.14", features = ["flamegraph"] }
|
||||
|
||||
|
||||
[features]
|
||||
default = []
|
||||
default = ["sql"]
|
||||
# The SQL statement extension seam (`lancedb::sql`): the registry a host adds
|
||||
# dialect statements through, the access vocabulary those statements declare,
|
||||
# and the write-commit observer. It pulls in no dependency that is not already
|
||||
# required, so it is on by default; the flag exists so an embedder that does
|
||||
# not want the surface can opt out of it.
|
||||
sql = []
|
||||
aws = [
|
||||
"lance/aws",
|
||||
"lance-io/aws",
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
//! The result of a DML statement, as a one-row record batch.
|
||||
//!
|
||||
//! A DML statement has to answer over the same channel a query does, so its
|
||||
//! result is carried as an ordinary [`RecordBatch`] with a fixed schema. The
|
||||
//! round trip is lossless, which is what lets a caller recover the typed form
|
||||
//! after the batch has crossed a transport such as Arrow Flight.
|
||||
|
||||
use std::fmt;
|
||||
use std::sync::Arc;
|
||||
|
||||
use arrow_array::{Int64Array, RecordBatch, StringArray};
|
||||
use arrow_schema::{DataType, Field, Schema};
|
||||
|
||||
/// Which DML statement produced a result.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum DmlOperation {
|
||||
Insert,
|
||||
Update,
|
||||
Delete,
|
||||
}
|
||||
|
||||
impl DmlOperation {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Insert => "INSERT",
|
||||
Self::Update => "UPDATE",
|
||||
Self::Delete => "DELETE",
|
||||
}
|
||||
}
|
||||
|
||||
fn parse(s: &str) -> Option<Self> {
|
||||
match s {
|
||||
"INSERT" => Some(Self::Insert),
|
||||
"UPDATE" => Some(Self::Update),
|
||||
"DELETE" => Some(Self::Delete),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Display for DmlOperation {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.write_str(self.as_str())
|
||||
}
|
||||
}
|
||||
|
||||
/// What a DML statement did.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct DmlResult {
|
||||
pub table: String,
|
||||
pub operation: DmlOperation,
|
||||
pub rows_affected: i64,
|
||||
pub version: i64,
|
||||
}
|
||||
|
||||
/// The schema every [`DmlResult`] batch carries.
|
||||
pub fn dml_result_schema() -> Schema {
|
||||
Schema::new(vec![
|
||||
Field::new("table", DataType::Utf8, false),
|
||||
Field::new("operation", DataType::Utf8, false),
|
||||
Field::new("rows_affected", DataType::Int64, false),
|
||||
Field::new("version", DataType::Int64, false),
|
||||
])
|
||||
}
|
||||
|
||||
impl DmlResult {
|
||||
pub fn new(
|
||||
table: impl Into<String>,
|
||||
operation: DmlOperation,
|
||||
rows_affected: i64,
|
||||
version: i64,
|
||||
) -> Self {
|
||||
Self {
|
||||
table: table.into(),
|
||||
operation,
|
||||
rows_affected,
|
||||
version,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn to_record_batch(&self) -> RecordBatch {
|
||||
RecordBatch::try_new(
|
||||
Arc::new(dml_result_schema()),
|
||||
vec![
|
||||
Arc::new(StringArray::from(vec![self.table.as_str()])),
|
||||
Arc::new(StringArray::from(vec![self.operation.as_str()])),
|
||||
Arc::new(Int64Array::from(vec![self.rows_affected])),
|
||||
Arc::new(Int64Array::from(vec![self.version])),
|
||||
],
|
||||
)
|
||||
.expect("static schema")
|
||||
}
|
||||
|
||||
/// Recover a result from a batch, or `None` if the batch is not one.
|
||||
///
|
||||
/// A query result can arrive on the same channel, so this has to be able
|
||||
/// to say "not a DML result" rather than fail.
|
||||
pub fn try_from_batch(batch: &RecordBatch) -> Option<Self> {
|
||||
if *batch.schema().as_ref() != dml_result_schema() || batch.num_rows() != 1 {
|
||||
return None;
|
||||
}
|
||||
Some(Self {
|
||||
table: batch
|
||||
.column(0)
|
||||
.as_any()
|
||||
.downcast_ref::<StringArray>()?
|
||||
.value(0)
|
||||
.to_string(),
|
||||
operation: DmlOperation::parse(
|
||||
batch
|
||||
.column(1)
|
||||
.as_any()
|
||||
.downcast_ref::<StringArray>()?
|
||||
.value(0),
|
||||
)?,
|
||||
rows_affected: batch
|
||||
.column(2)
|
||||
.as_any()
|
||||
.downcast_ref::<Int64Array>()?
|
||||
.value(0),
|
||||
version: batch
|
||||
.column(3)
|
||||
.as_any()
|
||||
.downcast_ref::<Int64Array>()?
|
||||
.value(0),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn roundtrip() {
|
||||
let original = DmlResult::new("foo", DmlOperation::Insert, 1, 3);
|
||||
let batch = original.to_record_batch();
|
||||
assert_eq!(DmlResult::try_from_batch(&batch), Some(original));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_dml_returns_none() {
|
||||
let s = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
|
||||
let batch = RecordBatch::try_new(s, vec![Arc::new(Int64Array::from(vec![1]))]).unwrap();
|
||||
assert_eq!(DmlResult::try_from_batch(&batch), None);
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,35 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
//! Handles to SQL queries running on a remote database.
|
||||
//! SQL: handles to queries running on a remote database, and the seam an
|
||||
//! embedder extends the dialect through.
|
||||
//!
|
||||
//! The extension seam is behind the default-on `sql` feature. It lets a host
|
||||
//! add statements to the dialect from outside this crate: a statement brings
|
||||
//! its own grammar ([`CustomSqlHandler`]), and declares its audit label and
|
||||
//! the access it needs ([`SqlStatement`]), so the host's authorization and
|
||||
//! auditing do not have to know each statement by name.
|
||||
|
||||
#[cfg(feature = "sql")]
|
||||
mod dml;
|
||||
#[cfg(feature = "sql")]
|
||||
mod observer;
|
||||
#[cfg(feature = "sql")]
|
||||
mod parser;
|
||||
#[cfg(feature = "sql")]
|
||||
mod statement;
|
||||
|
||||
#[cfg(feature = "sql")]
|
||||
pub use dml::{DmlOperation, DmlResult, dml_result_schema};
|
||||
#[cfg(feature = "sql")]
|
||||
pub use observer::{CommittedWrite, DmlEventKind, WriteObserver, observe_write};
|
||||
#[cfg(feature = "sql")]
|
||||
pub use parser::route_custom_sql;
|
||||
#[cfg(feature = "sql")]
|
||||
pub use statement::{
|
||||
AccessRequirement, CreateKind, CustomSqlHandler, DatabaseScope, RelationKind,
|
||||
RequirementContext, SqlStatement, StatementRegistry, SystemScope, WriteMode,
|
||||
};
|
||||
|
||||
use std::{fmt, sync::Arc};
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
//! Notification of committed writes.
|
||||
//!
|
||||
//! A statement that writes rows often has to tell its host that it did, so
|
||||
//! that follow-up work can be scheduled. What it should *not* have to know is
|
||||
//! how the host represents that notification. [`WriteObserver`] is the seam:
|
||||
//! the statement reports what it wrote, and the host decides what that means
|
||||
//! -- an event on a bus, a metric, or nothing at all.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
/// Which DML operation committed.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum DmlEventKind {
|
||||
Insert,
|
||||
Update,
|
||||
Delete,
|
||||
}
|
||||
|
||||
/// A write that has already been made durable.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct CommittedWrite {
|
||||
/// The database holding the table.
|
||||
pub database: String,
|
||||
/// The schema the statement named the table through.
|
||||
pub schema: String,
|
||||
/// The table written.
|
||||
pub table: String,
|
||||
/// The table's storage location, when the statement resolved one.
|
||||
pub table_uri: Option<String>,
|
||||
/// Which operation committed.
|
||||
pub kind: DmlEventKind,
|
||||
}
|
||||
|
||||
/// Notified after a statement's write commits.
|
||||
///
|
||||
/// Implementations are best-effort by contract: the write is already durable
|
||||
/// when this is called, so an observer that fails must not fail the statement.
|
||||
/// That is why the method cannot report an error.
|
||||
#[async_trait]
|
||||
pub trait WriteObserver: Send + Sync {
|
||||
async fn write_committed(&self, write: CommittedWrite);
|
||||
}
|
||||
|
||||
/// Report a committed write, if anything is observing.
|
||||
pub async fn observe_write(
|
||||
observer: Option<&Arc<dyn WriteObserver>>,
|
||||
database: &str,
|
||||
schema: &str,
|
||||
table: &str,
|
||||
table_uri: Option<String>,
|
||||
kind: DmlEventKind,
|
||||
) {
|
||||
let Some(observer) = observer else {
|
||||
return;
|
||||
};
|
||||
observer
|
||||
.write_committed(CommittedWrite {
|
||||
database: database.to_string(),
|
||||
schema: schema.to_string(),
|
||||
table: table.to_string(),
|
||||
table_uri,
|
||||
kind,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
@@ -0,0 +1,174 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
//! Routing a statement to the grammar that owns it.
|
||||
|
||||
use datafusion::error::{DataFusionError, Result};
|
||||
use datafusion::logical_expr::LogicalPlan;
|
||||
use datafusion::sql::sqlparser::{
|
||||
dialect::GenericDialect,
|
||||
parser::{Parser, ParserError},
|
||||
tokenizer::{Token, Tokenizer, TokenizerError},
|
||||
};
|
||||
|
||||
use super::statement::StatementRegistry;
|
||||
|
||||
/// Route a statement through the registry's grammars.
|
||||
///
|
||||
/// Returns `Ok(None)` when no grammar claims the statement, which is the
|
||||
/// caller's cue to hand it to DataFusion's own planner.
|
||||
///
|
||||
/// The first grammar whose `matches` accepts the tokens is the only one given
|
||||
/// the statement: a grammar that matches and then returns `Ok(None)` declines
|
||||
/// the form rather than falling through to the next grammar. Registration
|
||||
/// order therefore decides reachability, which is why [`StatementRegistry`]
|
||||
/// fixes it explicitly.
|
||||
pub fn route_custom_sql(registry: &StatementRegistry, sql: &str) -> Result<Option<LogicalPlan>> {
|
||||
let dialect = GenericDialect {};
|
||||
let mut tokenizer = Tokenizer::new(&dialect, sql);
|
||||
let tokens = tokenizer.tokenize().map_err(|e: TokenizerError| {
|
||||
DataFusionError::SQL(Box::new(ParserError::TokenizerError(e.to_string())), None)
|
||||
})?;
|
||||
|
||||
// Handlers match on keywords, so layout must not change the decision.
|
||||
let word_tokens: Vec<&Token> = tokens
|
||||
.iter()
|
||||
.filter(|t| !matches!(t, Token::Whitespace(_)))
|
||||
.collect();
|
||||
|
||||
for handler in registry.parsers() {
|
||||
if handler.matches(&word_tokens) {
|
||||
// `Parser` takes ownership of the tokens, so it is built only once
|
||||
// a handler has claimed the statement.
|
||||
let mut parser = Parser::new(&dialect).with_tokens(tokens.clone());
|
||||
if let Some(plan) = handler.parse(&mut parser)? {
|
||||
return Ok(Some(plan));
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::{
|
||||
Arc,
|
||||
atomic::{AtomicUsize, Ordering},
|
||||
};
|
||||
|
||||
use datafusion::common::DFSchema;
|
||||
use datafusion::error::Result as DfResult;
|
||||
use datafusion::logical_expr::{EmptyRelation, LogicalPlan};
|
||||
use datafusion::sql::sqlparser::keywords::Keyword;
|
||||
|
||||
use super::*;
|
||||
use crate::sql::statement::CustomSqlHandler;
|
||||
|
||||
fn empty_plan() -> LogicalPlan {
|
||||
LogicalPlan::EmptyRelation(EmptyRelation {
|
||||
produce_one_row: false,
|
||||
schema: Arc::new(DFSchema::empty()),
|
||||
})
|
||||
}
|
||||
|
||||
/// Matches on a leading keyword, and reports whether it was asked to parse.
|
||||
struct Handler {
|
||||
keyword: Keyword,
|
||||
outcome: Outcome,
|
||||
parsed: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
enum Outcome {
|
||||
Plans,
|
||||
Declines,
|
||||
}
|
||||
|
||||
impl Handler {
|
||||
fn new(keyword: Keyword, outcome: Outcome) -> (Arc<Self>, Arc<AtomicUsize>) {
|
||||
let parsed = Arc::new(AtomicUsize::new(0));
|
||||
let handler = Arc::new(Self {
|
||||
keyword,
|
||||
outcome,
|
||||
parsed: parsed.clone(),
|
||||
});
|
||||
(handler, parsed)
|
||||
}
|
||||
}
|
||||
|
||||
impl CustomSqlHandler for Handler {
|
||||
fn matches(&self, tokens: &[&Token]) -> bool {
|
||||
matches!(tokens.first(), Some(Token::Word(w)) if w.keyword == self.keyword)
|
||||
}
|
||||
|
||||
fn parse(&self, _parser: &mut Parser) -> DfResult<Option<LogicalPlan>> {
|
||||
self.parsed.fetch_add(1, Ordering::SeqCst);
|
||||
Ok(match self.outcome {
|
||||
Outcome::Plans => Some(empty_plan()),
|
||||
Outcome::Declines => None,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_unclaimed_statement_is_left_for_datafusion() {
|
||||
let registry = StatementRegistry::new();
|
||||
assert!(route_custom_sql(®istry, "SELECT 1").unwrap().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn whitespace_does_not_change_which_handler_matches() {
|
||||
let (handler, parsed) = Handler::new(Keyword::EXPLAIN, Outcome::Plans);
|
||||
let mut registry = StatementRegistry::new();
|
||||
registry.register_parser(handler);
|
||||
|
||||
for sql in ["EXPLAIN t", " EXPLAIN\n\t t "] {
|
||||
assert!(route_custom_sql(®istry, sql).unwrap().is_some());
|
||||
}
|
||||
assert_eq!(parsed.load(Ordering::SeqCst), 2);
|
||||
}
|
||||
|
||||
/// Front-insertion is what lets an extension get ahead of a catch-all that
|
||||
/// would otherwise swallow the same keyword.
|
||||
#[test]
|
||||
fn the_last_registered_handler_is_consulted_first() {
|
||||
let (first, first_parsed) = Handler::new(Keyword::EXPLAIN, Outcome::Plans);
|
||||
let (second, second_parsed) = Handler::new(Keyword::EXPLAIN, Outcome::Plans);
|
||||
|
||||
let mut registry = StatementRegistry::new();
|
||||
registry.register_parser(first).register_parser(second);
|
||||
|
||||
assert!(route_custom_sql(®istry, "EXPLAIN t").unwrap().is_some());
|
||||
assert_eq!(second_parsed.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(first_parsed.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
|
||||
/// A handler that matches and declines vetoes the statement rather than
|
||||
/// letting a later handler see it. Shadowing is silent, which is why
|
||||
/// registration order is part of the contract.
|
||||
#[test]
|
||||
fn a_handler_that_declines_shadows_the_handlers_behind_it() {
|
||||
let (shadowed, shadowed_parsed) = Handler::new(Keyword::EXPLAIN, Outcome::Plans);
|
||||
let (decliner, decliner_parsed) = Handler::new(Keyword::EXPLAIN, Outcome::Declines);
|
||||
|
||||
let mut registry = StatementRegistry::new();
|
||||
registry.register_parser(shadowed).register_parser(decliner);
|
||||
|
||||
assert!(route_custom_sql(®istry, "EXPLAIN t").unwrap().is_none());
|
||||
assert_eq!(decliner_parsed.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(shadowed_parsed.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_parts_keeps_the_order_it_was_given() {
|
||||
let (first, first_parsed) = Handler::new(Keyword::EXPLAIN, Outcome::Plans);
|
||||
let (second, second_parsed) = Handler::new(Keyword::EXPLAIN, Outcome::Plans);
|
||||
|
||||
let registry = StatementRegistry::from_parts(vec![first, second], vec![]);
|
||||
|
||||
assert!(route_custom_sql(®istry, "EXPLAIN t").unwrap().is_some());
|
||||
assert_eq!(first_parsed.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(second_parsed.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,365 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
//! The statement registry: the seam between the SQL dialect and the behaviour
|
||||
//! an embedder adds to it.
|
||||
//!
|
||||
//! A statement owns three things that are otherwise easy to spread across
|
||||
//! parallel `downcast_ref` chains: the grammar that produces its node, the
|
||||
//! audit label it reports, and the access it requires. Keeping them in one
|
||||
//! place is what makes a statement addable from outside this crate.
|
||||
//!
|
||||
//! Two registries, because the axes differ. Grammar is matched against tokens
|
||||
//! before a node exists, and several statements can share one handler -- an
|
||||
//! `ALTER TABLE` handler may yield a different node per subcommand. A planned
|
||||
//! node, by contrast, is claimed by exactly one statement.
|
||||
|
||||
use std::any::Any;
|
||||
use std::sync::Arc;
|
||||
|
||||
use datafusion::common::{ResolvedTableReference, TableReference};
|
||||
use datafusion::error::Result as DfResult;
|
||||
use datafusion::logical_expr::LogicalPlan;
|
||||
use datafusion::sql::sqlparser::{parser::Parser, tokenizer::Token};
|
||||
|
||||
/// A pluggable handler for custom SQL statements.
|
||||
pub trait CustomSqlHandler: Send + Sync {
|
||||
/// Whether this handler wants to handle these tokens.
|
||||
///
|
||||
/// The tokens have had whitespace removed, so a handler can match on
|
||||
/// leading keywords without accounting for layout.
|
||||
fn matches(&self, tokens: &[&Token]) -> bool;
|
||||
|
||||
/// Parse the statement into a logical plan.
|
||||
///
|
||||
/// Returning `Ok(None)` declines a form this handler matched on; the
|
||||
/// statement then goes to DataFusion's own planner. See
|
||||
/// [`StatementRegistry`] for why that stops routing rather than falling
|
||||
/// through to the next handler.
|
||||
fn parse(&self, parser: &mut Parser) -> DfResult<Option<LogicalPlan>>;
|
||||
}
|
||||
|
||||
/// What kind of relation a requirement is about.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum RelationKind {
|
||||
Table,
|
||||
View,
|
||||
}
|
||||
|
||||
/// What kind of object a DDL statement brings into existence.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum CreateKind {
|
||||
Table,
|
||||
View,
|
||||
MaterializedView,
|
||||
}
|
||||
|
||||
/// What a statement needs authorized before it runs.
|
||||
///
|
||||
/// The vocabulary is deliberately generic: it names *what is being reached
|
||||
/// for*, not the privilege that grants it. An embedder maps these onto its own
|
||||
/// privilege model and audit labels, so no access-control concept has to live
|
||||
/// in the dialect.
|
||||
///
|
||||
/// The variants are finer-grained than a bare read/write split because the
|
||||
/// distinctions are load-bearing for that mapping -- appending to a table and
|
||||
/// redefining it are different grants, and collapsing them would silently
|
||||
/// widen what a statement is allowed to do.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum AccessRequirement {
|
||||
/// Read the contents of a relation.
|
||||
Read {
|
||||
relation: ResolvedTableReference,
|
||||
kind: RelationKind,
|
||||
},
|
||||
/// Change the rows of a relation.
|
||||
Write {
|
||||
relation: ResolvedTableReference,
|
||||
kind: RelationKind,
|
||||
mode: WriteMode,
|
||||
},
|
||||
/// Change a relation's definition, or anything about it other than its
|
||||
/// rows. Index and column changes land here.
|
||||
Own {
|
||||
relation: ResolvedTableReference,
|
||||
kind: RelationKind,
|
||||
},
|
||||
/// Bring a new relation into existence.
|
||||
CreateIn {
|
||||
relation: ResolvedTableReference,
|
||||
kind: CreateKind,
|
||||
},
|
||||
/// Reach the connected database itself rather than a relation in it.
|
||||
Database { name: String, scope: DatabaseScope },
|
||||
/// Reach a namespace's metadata.
|
||||
Namespace { database: String, namespace: String },
|
||||
/// Reach the deployment rather than any one database.
|
||||
System { scope: SystemScope },
|
||||
}
|
||||
|
||||
/// How an [`AccessRequirement::Write`] changes a relation's rows.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum WriteMode {
|
||||
/// Add rows.
|
||||
Append,
|
||||
/// Change existing rows.
|
||||
Modify,
|
||||
/// Take rows away.
|
||||
Remove,
|
||||
}
|
||||
|
||||
/// How far into a database an [`AccessRequirement::Database`] reaches.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum DatabaseScope {
|
||||
/// See that the database exists and list what is in it.
|
||||
Usage,
|
||||
/// Change what the database contains.
|
||||
Ownership,
|
||||
}
|
||||
|
||||
/// How far into the deployment an [`AccessRequirement::System`] reaches.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum SystemScope {
|
||||
/// Observe deployment-wide state.
|
||||
Usage,
|
||||
/// Act on deployment-wide state.
|
||||
Operate,
|
||||
}
|
||||
|
||||
impl AccessRequirement {
|
||||
/// The relation this requirement is about, for the relation-shaped
|
||||
/// variants.
|
||||
///
|
||||
/// An embedder's privilege mapping is written against the variants
|
||||
/// directly; this is the shortcut for the common case of needing the
|
||||
/// relation without caring which shape asked for it.
|
||||
pub fn relation(&self) -> Option<(&ResolvedTableReference, RelationKind)> {
|
||||
match self {
|
||||
Self::Read { relation, kind }
|
||||
| Self::Write { relation, kind, .. }
|
||||
| Self::Own { relation, kind } => Some((relation, *kind)),
|
||||
Self::CreateIn { .. }
|
||||
| Self::Database { .. }
|
||||
| Self::Namespace { .. }
|
||||
| Self::System { .. } => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Everything a statement needs in order to state its requirements, without
|
||||
/// reaching for the engine running it.
|
||||
pub struct RequirementContext<'a> {
|
||||
/// The database a bare relation name resolves against.
|
||||
pub default_database: &'a str,
|
||||
/// The schema a bare relation name resolves against.
|
||||
pub default_schema: &'a str,
|
||||
}
|
||||
|
||||
impl RequirementContext<'_> {
|
||||
/// Resolve a possibly-bare reference against the request's defaults.
|
||||
pub fn resolve(&self, relation: TableReference) -> ResolvedTableReference {
|
||||
relation.resolve(self.default_database, self.default_schema)
|
||||
}
|
||||
|
||||
/// Resolve a bare relation name against the request's defaults.
|
||||
pub fn resolve_bare(&self, name: impl Into<String>) -> ResolvedTableReference {
|
||||
self.resolve(TableReference::bare(name.into()))
|
||||
}
|
||||
}
|
||||
|
||||
/// One statement in the dialect: the node it plans to, what it is called in an
|
||||
/// audit log, and what it needs authorized.
|
||||
pub trait SqlStatement: Send + Sync {
|
||||
/// Whether this statement owns the given planned node.
|
||||
fn claims(&self, node: &dyn Any) -> bool;
|
||||
|
||||
/// The audit label for this statement.
|
||||
///
|
||||
/// This is an open string rather than an enum so that an embedder can add
|
||||
/// a statement -- and a label for it -- without changing this crate.
|
||||
fn audit_operation(&self) -> &'static str;
|
||||
|
||||
/// What must be authorized before the node runs.
|
||||
///
|
||||
/// Returning an empty set means the statement needs nothing beyond
|
||||
/// whatever the engine already collects from the plan's scans.
|
||||
fn access_requirements(
|
||||
&self,
|
||||
node: &dyn Any,
|
||||
context: &RequirementContext<'_>,
|
||||
) -> DfResult<Vec<AccessRequirement>>;
|
||||
}
|
||||
|
||||
/// The set of statements and grammars an engine knows about.
|
||||
///
|
||||
/// Ordering is load-bearing on the parse side and stays explicit. A handler
|
||||
/// may be a catch-all over its leading keyword -- erroring on any form of that
|
||||
/// keyword it does not recognize, or matching on the first token alone -- so a
|
||||
/// handler registered *after* such a one can never be reached for that
|
||||
/// keyword. Extensions are therefore consulted before whatever is already
|
||||
/// registered.
|
||||
///
|
||||
/// A handler that matches and then returns `Ok(None)` stops routing entirely
|
||||
/// rather than falling through to the next handler; the statement then goes to
|
||||
/// DataFusion's own planner. That veto is intentional -- it is how a handler
|
||||
/// declines a form it matched on -- but it means an overlapping handler
|
||||
/// registered later is shadowed rather than reported, which is the other
|
||||
/// reason ordering is explicit here.
|
||||
#[derive(Default)]
|
||||
pub struct StatementRegistry {
|
||||
parsers: Vec<Arc<dyn CustomSqlHandler>>,
|
||||
statements: Vec<Arc<dyn SqlStatement>>,
|
||||
}
|
||||
|
||||
impl StatementRegistry {
|
||||
/// An empty registry.
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// Build a registry from an explicit, already-ordered set.
|
||||
///
|
||||
/// The ordering is used as given -- unlike [`Self::register_parser`], this
|
||||
/// does not reverse anything. It is how an embedder that owns the whole
|
||||
/// dialect states the order once.
|
||||
pub fn from_parts(
|
||||
parsers: Vec<Arc<dyn CustomSqlHandler>>,
|
||||
statements: Vec<Arc<dyn SqlStatement>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
parsers,
|
||||
statements,
|
||||
}
|
||||
}
|
||||
|
||||
/// Add a grammar, consulted before every grammar already registered.
|
||||
///
|
||||
/// Registration is front-insertion because an existing catch-all handler
|
||||
/// would otherwise shadow anything added later; see the type docs.
|
||||
pub fn register_parser(&mut self, parser: Arc<dyn CustomSqlHandler>) -> &mut Self {
|
||||
self.parsers.insert(0, parser);
|
||||
self
|
||||
}
|
||||
|
||||
/// Add a statement, consulted before every statement already registered.
|
||||
pub fn register_statement(&mut self, statement: Arc<dyn SqlStatement>) -> &mut Self {
|
||||
self.statements.insert(0, statement);
|
||||
self
|
||||
}
|
||||
|
||||
/// The grammars, in the order they are consulted.
|
||||
pub fn parsers(&self) -> &[Arc<dyn CustomSqlHandler>] {
|
||||
&self.parsers
|
||||
}
|
||||
|
||||
/// The statement owning this planned node, if any.
|
||||
pub fn claim(&self, node: &dyn Any) -> Option<&Arc<dyn SqlStatement>> {
|
||||
self.statements.iter().find(|s| s.claims(node))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
struct Claimant {
|
||||
label: &'static str,
|
||||
claims_everything: bool,
|
||||
}
|
||||
|
||||
impl SqlStatement for Claimant {
|
||||
fn claims(&self, node: &dyn Any) -> bool {
|
||||
self.claims_everything && node.is::<u8>()
|
||||
}
|
||||
|
||||
fn audit_operation(&self) -> &'static str {
|
||||
self.label
|
||||
}
|
||||
|
||||
fn access_requirements(
|
||||
&self,
|
||||
_node: &dyn Any,
|
||||
_context: &RequirementContext<'_>,
|
||||
) -> DfResult<Vec<AccessRequirement>> {
|
||||
Ok(vec![])
|
||||
}
|
||||
}
|
||||
|
||||
fn claimant(label: &'static str, claims_everything: bool) -> Arc<dyn SqlStatement> {
|
||||
Arc::new(Claimant {
|
||||
label,
|
||||
claims_everything,
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_unclaimed_node_has_no_statement() {
|
||||
let mut registry = StatementRegistry::new();
|
||||
registry.register_statement(claimant("never", false));
|
||||
assert!(registry.claim(&0u8).is_none());
|
||||
}
|
||||
|
||||
/// Front-insertion on the claim side too: an extension must be able to
|
||||
/// take over a node shape that something already registered also claims.
|
||||
#[test]
|
||||
fn the_last_registered_statement_claims_first() {
|
||||
let mut registry = StatementRegistry::new();
|
||||
registry
|
||||
.register_statement(claimant("first", true))
|
||||
.register_statement(claimant("second", true));
|
||||
|
||||
assert_eq!(registry.claim(&0u8).unwrap().audit_operation(), "second");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_parts_keeps_the_claim_order_it_was_given() {
|
||||
let registry =
|
||||
StatementRegistry::from_parts(vec![], vec![claimant("a", true), claimant("b", true)]);
|
||||
assert_eq!(registry.claim(&0u8).unwrap().audit_operation(), "a");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_bare_name_resolves_against_the_request_defaults() {
|
||||
let context = RequirementContext {
|
||||
default_database: "db",
|
||||
default_schema: "public",
|
||||
};
|
||||
let resolved = context.resolve_bare("t");
|
||||
assert_eq!(&*resolved.catalog, "db");
|
||||
assert_eq!(&*resolved.schema, "public");
|
||||
assert_eq!(&*resolved.table, "t");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_qualified_name_keeps_its_own_parts() {
|
||||
let context = RequirementContext {
|
||||
default_database: "db",
|
||||
default_schema: "public",
|
||||
};
|
||||
let resolved = context.resolve(TableReference::partial("other", "t"));
|
||||
assert_eq!(&*resolved.catalog, "db");
|
||||
assert_eq!(&*resolved.schema, "other");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn only_the_relation_shaped_requirements_name_a_relation() {
|
||||
let relation = TableReference::bare("t").resolve("db", "public");
|
||||
|
||||
let read = AccessRequirement::Read {
|
||||
relation: relation.clone(),
|
||||
kind: RelationKind::Table,
|
||||
};
|
||||
assert_eq!(read.relation().unwrap().1, RelationKind::Table);
|
||||
|
||||
let create = AccessRequirement::CreateIn {
|
||||
relation,
|
||||
kind: CreateKind::MaterializedView,
|
||||
};
|
||||
assert!(create.relation().is_none());
|
||||
|
||||
let system = AccessRequirement::System {
|
||||
scope: SystemScope::Operate,
|
||||
};
|
||||
assert!(system.relation().is_none());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user