Compare commits

..

1 Commits

Author SHA1 Message Date
lancedb automation 9123725493 chore: update lance dependency to v12.0.0-beta.6 2026-08-31 21:43:23 +00:00
35 changed files with 155 additions and 1519 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
[tool.bumpversion]
current_version = "0.39.0-beta.0"
current_version = "0.38.0"
parse = """(?x)
(?P<major>0|[1-9]\\d*)\\.
(?P<minor>0|[1-9]\\d*)\\.
Generated
+44 -44
View File
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]]
name = "fsst"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-array",
"rand 0.9.5",
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
[[package]]
name = "lance"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arc-swap",
"arrow",
@@ -4888,8 +4888,8 @@ dependencies = [
[[package]]
name = "lance-arrow"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4911,7 +4911,7 @@ dependencies = [
[[package]]
name = "lance-arrow-scalar"
version = "58.0.0"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4925,7 +4925,7 @@ dependencies = [
[[package]]
name = "lance-arrow-stats"
version = "58.0.0"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -4934,8 +4934,8 @@ dependencies = [
[[package]]
name = "lance-bitpacking"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrayref",
"crunchy",
@@ -4945,8 +4945,8 @@ dependencies = [
[[package]]
name = "lance-core"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4983,8 +4983,8 @@ dependencies = [
[[package]]
name = "lance-datafusion"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow",
"arrow-array",
@@ -5013,8 +5013,8 @@ dependencies = [
[[package]]
name = "lance-datagen"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow",
"arrow-array",
@@ -5031,8 +5031,8 @@ dependencies = [
[[package]]
name = "lance-derive"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"proc-macro2",
"quote",
@@ -5041,8 +5041,8 @@ dependencies = [
[[package]]
name = "lance-encoding"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -5075,8 +5075,8 @@ dependencies = [
[[package]]
name = "lance-file"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -5107,8 +5107,8 @@ dependencies = [
[[package]]
name = "lance-index"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arc-swap",
"arrow",
@@ -5172,8 +5172,8 @@ dependencies = [
[[package]]
name = "lance-index-core"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5195,8 +5195,8 @@ dependencies = [
[[package]]
name = "lance-io"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow",
"arrow-array",
@@ -5236,8 +5236,8 @@ dependencies = [
[[package]]
name = "lance-linalg"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5251,8 +5251,8 @@ dependencies = [
[[package]]
name = "lance-namespace"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow",
"async-trait",
@@ -5264,8 +5264,8 @@ dependencies = [
[[package]]
name = "lance-namespace-impls"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow",
"arrow-ipc",
@@ -5304,9 +5304,9 @@ dependencies = [
[[package]]
name = "lance-namespace-reqwest-client"
version = "0.11.0"
version = "0.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0a030196da1c994b63a96a4f0bf5b0cfa459fe6dadc9e962320246ca328da22a"
checksum = "1d06b1fbb5d41f93bc652b61e2872af92e8a6c5f6b4ce8839a8ecfa05365d359"
dependencies = [
"reqwest 0.12.28",
"serde",
@@ -5318,8 +5318,8 @@ dependencies = [
[[package]]
name = "lance-select"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5333,8 +5333,8 @@ dependencies = [
[[package]]
name = "lance-table"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow",
"arrow-array",
@@ -5374,8 +5374,8 @@ dependencies = [
[[package]]
name = "lance-testing"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5388,8 +5388,8 @@ dependencies = [
[[package]]
name = "lance-tokenizer"
version = "12.0.0-beta.5"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
version = "12.0.0-beta.6"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.6#667cb492c92f7cdb9f4c87218cd0e5e8071b5a5d"
dependencies = [
"frostem",
"icu_segmenter",
+14 -14
View File
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
rust-version = "1.91.0"
[workspace.dependencies]
lance = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
lance = { "version" = "=12.0.0-beta.6", default-features = false, "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=12.0.0-beta.6", default-features = false, "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=12.0.0-beta.6", default-features = false, "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=12.0.0-beta.6", "tag" = "v12.0.0-beta.6", "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
+1 -1
View File
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
<dependency>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-core</artifactId>
<version>0.39.0-beta.0</version>
<version>0.38.0</version>
</dependency>
```
@@ -50,16 +50,6 @@ projections: [string, string][];
***
### sourceNamespace
```ts
sourceNamespace: string[];
```
Namespace holding the source table; empty is the root namespace.
***
### sourceTable
```ts
+1 -1
View File
@@ -8,7 +8,7 @@
<parent>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.39.0-beta.0</version>
<version>0.38.0-final.0</version>
<relativePath>../pom.xml</relativePath>
</parent>
+2 -2
View File
@@ -6,7 +6,7 @@
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.39.0-beta.0</version>
<version>0.38.0-final.0</version>
<packaging>pom</packaging>
<name>${project.artifactId}</name>
<description>LanceDB Java SDK Parent POM</description>
@@ -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.5</lance-core.version>
<lance-core.version>12.0.0-beta.6</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
View File
@@ -1,7 +1,7 @@
[package]
name = "lancedb-nodejs"
edition.workspace = true
version = "0.39.0-beta.0"
version = "0.38.0"
publish = false
license.workspace = true
description.workspace = true
-22
View File
@@ -48,28 +48,6 @@ describe("materialized views", () => {
expect(definitionFromMetadata(safe, "v").limit).toBe(42);
});
it("reads the namespaced select kind and refuses unknown kinds", () => {
// "namespaced_select" is the namespaced form of "select": same shape, a
// separate kind so readers that predate it refuse instead of resolving
// the source at the root.
const namespaced = new Map([
[
DEFINITION_META_KEY,
'{"kind":"namespaced_select","source_table":"people","source_namespace":["ns"]}',
],
]);
const definition = definitionFromMetadata(namespaced, "v");
expect(definition.sourceTable).toBe("people");
expect(definition.sourceNamespace).toEqual(["ns"]);
const unknown = new Map([
[DEFINITION_META_KEY, '{"kind":"select_v3","source_table":"people"}'],
]);
expect(() => definitionFromMetadata(unknown, "v")).toThrow(
/cannot refresh/,
);
});
it("creates, refreshes and queries a view", async () => {
const view = await db.createMaterializedView("adults", "people", {
select: ["name", ["shout", "upper(name)"]],
+1 -5
View File
@@ -19,8 +19,6 @@ export interface MaterializedViewDefinition {
limit?: number;
/** Source columns the projections and filter read. */
inputs: string[];
/** Namespace holding the source table; empty is the root namespace. */
sourceNamespace: string[];
}
/**
@@ -80,8 +78,7 @@ export function definitionFromMetadata(
}
// biome-ignore lint/suspicious/noExplicitAny: raw JSON
const value: any = JSON.parse(raw);
// "namespaced_select" keeps older readers from resolving the source at root.
if (value.kind !== "select" && value.kind !== "namespaced_select") {
if (value.kind !== "select") {
throw new Error(
`materialized view '${name}' is defined by '${value.kind}', which this ` +
"version of lancedb cannot refresh",
@@ -106,7 +103,6 @@ export function definitionFromMetadata(
filter: value.filter ?? undefined,
limit,
inputs: value.inputs ?? [],
sourceNamespace: value.source_namespace ?? [],
};
}
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-darwin-arm64",
"version": "0.39.0-beta.0",
"version": "0.38.0",
"os": ["darwin"],
"cpu": ["arm64"],
"main": "lancedb.darwin-arm64.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-gnu",
"version": "0.39.0-beta.0",
"version": "0.38.0",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-musl",
"version": "0.39.0-beta.0",
"version": "0.38.0",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-gnu",
"version": "0.39.0-beta.0",
"version": "0.38.0",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-musl",
"version": "0.39.0-beta.0",
"version": "0.38.0",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-arm64-msvc",
"version": "0.39.0-beta.0",
"version": "0.38.0",
"os": [
"win32"
],
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-x64-msvc",
"version": "0.39.0-beta.0",
"version": "0.38.0",
"os": ["win32"],
"cpu": ["x64"],
"main": "lancedb.win32-x64-msvc.node",
+1 -1
View File
@@ -11,7 +11,7 @@
"ann"
],
"private": false,
"version": "0.39.0-beta.0",
"version": "0.38.0",
"main": "dist/index.js",
"exports": {
".": "./dist/index.js",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.39.0-beta.0"
version = "0.38.0"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
-1
View File
@@ -607,7 +607,6 @@ class FullTextQuery:
class PyQueryRequest:
limit: Optional[int]
offset: Optional[int]
take_offsets: Optional[List[int]]
filter: Optional[Union[str, bytes]]
full_text_search: Optional[FullTextQuery]
select: Optional[Union[str, List[str]]]
+1 -5
View File
@@ -42,8 +42,6 @@ class MaterializedViewDefinition:
"""Cap on the number of rows the view holds."""
inputs: List[str] = field(default_factory=list)
"""Source columns the projections and filter read."""
source_namespace: List[str] = field(default_factory=list)
"""Namespace holding the source table; empty is the root namespace."""
def _definition_from_schema(
@@ -55,8 +53,7 @@ def _definition_from_schema(
raise ValueError(f"Table '{name}' is not a materialized view")
value = json.loads(raw)
kind = value.get("kind")
# "namespaced_select" keeps older readers from resolving the source at root.
if kind not in ("select", "namespaced_select"):
if kind != "select":
raise NotImplementedError(
f"materialized view '{name}' is defined by '{kind}', which this "
"version of lancedb cannot refresh"
@@ -69,7 +66,6 @@ def _definition_from_schema(
filter=value.get("filter"),
limit=value.get("limit"),
inputs=value.get("inputs", []),
source_namespace=value.get("source_namespace", []),
)
-6
View File
@@ -109,7 +109,6 @@ def _query_is_plain_scan(query: Query) -> bool:
return (
query.vector is None
and query.full_text_query is None
and query.take_offsets is None
and not query.postfilter
and not query.order_by
)
@@ -805,10 +804,6 @@ class Query(pydantic.BaseModel):
# offset to start fetching results from
offset: Optional[int] = None
# Dataset offsets whose duplicate occurrences must be restored after lookup.
# This is populated when a take query is converted to this serializable form.
take_offsets: Optional[List[int]] = None
# if true, will only search the indexed data
fast_search: Optional[bool] = None
@@ -830,7 +825,6 @@ class Query(pydantic.BaseModel):
query = cls()
query.limit = req.limit
query.offset = req.offset
query.take_offsets = req.take_offsets
query.filter = req.filter
query.full_text_query = req.full_text_search
query.columns = req.select
+4 -25
View File
@@ -1678,9 +1678,9 @@ class Table(ABC):
Offsets are mostly useful for sampling as the set of all valid offsets is easily
known in advance to be [0, len(table)).
No guarantees are made regarding the order in which results are returned.
Repeated offsets produce repeated rows, which makes this method suitable for
sampling with replacement.
No guarantees are made regarding the order in which results are returned. If
you desire an output order that matches the order of the given offsets, you will
need to add the row offset column to the output and align it yourself.
Parameters
----------
@@ -4090,7 +4090,6 @@ class LanceTable(Table):
)
and not self._route_pushdown_to_rust
and self.current_branch() is None
and query.take_offsets is None
):
from lancedb.namespace import _execute_server_side_query
@@ -5984,23 +5983,7 @@ class AsyncTable:
def _sync_query_to_async(
self, query: Query
) -> (
AsyncHybridQuery
| AsyncFTSQuery
| AsyncVectorQuery
| AsyncQuery
| AsyncTakeQuery
):
if query.take_offsets is not None:
take_query = self.take_offsets(query.take_offsets)
if query.columns:
take_query = take_query.select(query.columns)
if query.use_lsm is not None:
take_query = take_query.use_lsm(query.use_lsm)
if query.with_row_id:
take_query = take_query.with_row_id()
return take_query
) -> AsyncHybridQuery | AsyncFTSQuery | AsyncVectorQuery | AsyncQuery:
async_query = self.query()
if query.limit is not None:
async_query = async_query.limit(query.limit)
@@ -6065,7 +6048,6 @@ class AsyncTable:
self._namespace_client, self._pushdown_operations
)
and not self._route_pushdown_to_rust
and query.take_offsets is None
):
from lancedb.namespace import _execute_server_side_query
@@ -6563,9 +6545,6 @@ class AsyncTable:
Offsets are mostly useful for sampling as the set of all valid offsets is easily
known in advance to be [0, len(table)).
No guarantees are made regarding the order in which results are returned.
Repeated offsets produce repeated rows.
Parameters
----------
offsets: list[int]
@@ -266,38 +266,3 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
handle._route_pushdown_to_rust == through_namespace._route_pushdown_to_rust
)
assert handle._namespace_path == through_namespace._namespace_path
def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused():
import json
import pyarrow as pa
from lancedb.materialized_view import _definition_from_schema
def schema_with(definition: dict) -> pa.Schema:
return pa.schema([pa.field("id", pa.int32())]).with_metadata(
{b"mv.definition": json.dumps(definition).encode()}
)
# "namespaced_select" is the namespaced form of "select": same shape,
# a separate kind so readers that predate it refuse instead of
# resolving the source at the root.
definition = _definition_from_schema(
schema_with(
{
"kind": "namespaced_select",
"source_table": "people",
"source_namespace": ["ns"],
"projections": [{"output": "name", "expression": "name"}],
}
),
"v",
)
assert definition.source_table == "people"
assert definition.source_namespace == ["ns"]
with pytest.raises(NotImplementedError, match="cannot refresh"):
_definition_from_schema(
schema_with({"kind": "select_v3", "source_table": "people"}), "v"
)
-15
View File
@@ -1923,21 +1923,6 @@ def test_take_queries(tmp_path):
17,
]
# Duplicate offsets are occurrences, not set members. Ordering is unspecified.
assert sorted(table.take_offsets([5, 2, 5, 17]).to_pandas()["idx"].to_list()) == [
2,
5,
5,
17,
]
# Converting a take builder to its serializable query representation must
# retain occurrence metadata and execute with the same multiplicity.
query = table.take_offsets([5, 2, 5, 17]).select(["idx"]).to_query_object()
assert query.take_offsets == [5, 2, 5, 17]
converted = table._execute_query(query).read_all()
assert sorted(converted["idx"].to_pylist()) == [2, 5, 5, 17]
# Take by row id
assert list(
sorted(table.take_row_ids([5, 2, 17]).to_pandas()["idx"].to_list())
+5 -30
View File
@@ -479,49 +479,24 @@ def test_remote_permutation_is_picklable():
match = re.search(
r"_rowoffset\s+in\s+\((.*?)\)", body["filter"], re.IGNORECASE
)
offsets = list(
dict.fromkeys(int(o.strip()) for o in match.group(1).split(","))
)
offsets = [int(o.strip()) for o in match.group(1).split(",")]
else:
offsets = list(range(len(rows)))
columns = body.get("columns") or ["a"]
table = pa.table(
{
column: (
[rows[offset] for offset in offsets]
if column == "a"
else offsets
)
for column in columns
}
)
table = pa.table({"a": [rows[offset] for offset in offsets]})
request.send_response(200)
request.send_header("Content-Type", "application/vnd.apache.arrow.file")
request.end_headers()
with pa.ipc.new_file(request.wfile, schema=table.schema) as writer:
writer.write_table(table, max_chunksize=2)
writer.write_table(table)
else:
request.send_response(404)
request.end_headers()
with mock_lancedb_connection(handler) as db:
table = db.open_table("test")
assert table.take_offsets([0, 2, 0, 4]).to_list() == [
{"a": 0},
{"a": 0},
{"a": 2},
{"a": 4},
]
permutation = Permutation.identity(table)
permutation = Permutation.identity(db.open_table("test"))
restored = pickle.loads(pickle.dumps(permutation))
assert restored.__getitems__([0, 2, 0, 4]) == [
{"a": 0},
{"a": 2},
{"a": 0},
{"a": 4},
]
assert restored.__getitems__([0, 2, 4]) == [{"a": 0}, {"a": 2}, {"a": 4}]
def test_create_table_exist_ok():
-3
View File
@@ -323,7 +323,6 @@ impl<'py> IntoPyObject<'py> for PyQueryVectors {
pub struct PyQueryRequest {
pub limit: Option<usize>,
pub offset: Option<usize>,
pub take_offsets: Option<Vec<u64>>,
pub filter: Option<PyQueryFilter>,
pub full_text_search: Option<PyLanceDB<FtsQuery>>,
pub select: PySelect,
@@ -354,7 +353,6 @@ impl From<AnyQuery> for PyQueryRequest {
AnyQuery::Query(query_request) => Self {
limit: query_request.limit,
offset: query_request.offset,
take_offsets: query_request.take_offsets,
filter: query_request.filter.map(PyQueryFilter),
full_text_search: query_request
.full_text_search
@@ -383,7 +381,6 @@ impl From<AnyQuery> for PyQueryRequest {
AnyQuery::VectorQuery(vector_query) => Self {
limit: vector_query.base.limit,
offset: vector_query.base.offset,
take_offsets: vector_query.base.take_offsets,
filter: vector_query.base.filter.map(PyQueryFilter),
full_text_search: None,
select_source_columns: PySelect::source_columns(&vector_query.base.select),
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb"
version = "0.39.0-beta.0"
version = "0.38.0"
edition.workspace = true
description = "LanceDB: A serverless, low-latency vector database for AI applications"
license.workspace = true
@@ -31,7 +31,7 @@ use lance::io::RecordBatchStream;
use lance_arrow::RecordBatchExt;
use lance_core::ROW_ID;
use lance_core::error::LanceOptionExt;
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use std::sync::Arc;
/// Reads a permutation of a source table based on row IDs stored in a separate table
@@ -234,14 +234,7 @@ impl PermutationReader {
.expect_ok()?
.values();
let mut unique_row_ids = HashSet::with_capacity(num_rows);
let in_list: Vec<Expr> = row_ids
.iter()
.copied()
.filter(|row_id| unique_row_ids.insert(*row_id))
.map(lit)
.collect();
let num_unique_row_ids = unique_row_ids.len();
let in_list: Vec<Expr> = row_ids.iter().map(|id| lit(*id)).collect();
let base_query = QueryRequest {
filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
@@ -254,7 +247,7 @@ impl PermutationReader {
.query(
&AnyQuery::Query(base_query),
QueryExecutionOptions {
max_batch_length: num_unique_row_ids as u32,
max_batch_length: num_rows as u32,
..Default::default()
},
)
@@ -269,9 +262,9 @@ impl PermutationReader {
});
}
if batches.iter().map(|b| b.num_rows()).sum::<usize>() != num_unique_row_ids {
if batches.iter().map(|b| b.num_rows()).sum::<usize>() != num_rows {
return Err(Error::InvalidInput {
message: "Base table returned a different number of rows than the number of unique row IDs"
message: "Base table returned different number of rows than the number of row IDs"
.to_string(),
});
}
@@ -511,7 +504,6 @@ impl PermutationReader {
let table = Table::from(self.base_table.clone());
let batches = table
.take_offsets(offsets.to_vec())
.preserve_order()
.select(selection.clone())
.execute()
.await?
@@ -811,10 +803,10 @@ mod tests {
.unwrap();
// Take offsets in reverse order and verify returned rows match that order
let offsets = vec![5, 3, 5, 1, 0];
let offsets = vec![5, 3, 1, 0];
let batch = reader.take_offsets(&offsets, Select::All).await.unwrap();
assert_eq!(batch.num_rows(), 5);
assert_eq!(batch.num_rows(), 4);
let idx_values = batch
.column(0)
@@ -828,52 +820,6 @@ mod tests {
assert_eq!(idx_values, expected);
}
#[tokio::test]
async fn test_take_offsets_preserves_repeated_rows_in_permutation() {
let base_table = lance_datagen::gen_batch()
.col("idx", lance_datagen::array::step::<Int32Type>())
.into_mem_table("tbl", RowCount::from(5), BatchCount::from(1))
.await;
let base_row_ids = collect_column::<UInt64Type>(&base_table, "_rowid").await;
let permutation_row_ids = vec![
base_row_ids[3],
base_row_ids[1],
base_row_ids[3],
base_row_ids[2],
];
let permutation_batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("row_id", DataType::UInt64, false),
Field::new(SPLIT_ID_COLUMN, DataType::UInt64, false),
])),
vec![
Arc::new(UInt64Array::from(permutation_row_ids)),
Arc::new(UInt64Array::from(vec![0; 4])),
],
)
.unwrap();
let permutation_table = virtual_table("row_ids", &permutation_batch).await;
let reader = PermutationReader::try_from_tables(
base_table.base_table().clone(),
permutation_table.base_table().clone(),
0,
)
.await
.unwrap();
let batch = reader
.take_offsets(&[0, 1, 2, 3], Select::All)
.await
.unwrap();
let idx_values = batch
.column(0)
.as_primitive::<Int32Type>()
.values()
.to_vec();
assert_eq!(idx_values, vec![3, 1, 3, 2]);
}
#[tokio::test]
async fn test_take_offsets_with_column_selection() {
let (base_table, row_ids_table, row_ids) = setup_permutation_tables(10).await;
@@ -937,17 +883,17 @@ mod tests {
.unwrap();
// With no permutation table, take_offsets uses the base table directly
let offsets = vec![0, 2, 0, 4, 6];
let offsets = vec![0, 2, 4, 6];
let batch = reader.take_offsets(&offsets, Select::All).await.unwrap();
assert_eq!(batch.num_rows(), 5);
assert_eq!(batch.num_rows(), 4);
let idx_values = batch
.column(0)
.as_primitive::<Int32Type>()
.values()
.to_vec();
assert_eq!(idx_values, vec![0, 2, 0, 4, 6]);
assert_eq!(idx_values, vec![0, 2, 4, 6]);
}
#[tokio::test]
+47 -207
View File
@@ -74,15 +74,8 @@ const EMBEDDING_FUNCTIONS_META_KEY: &str = "embedding_functions";
const COLUMN_DEFINITIONS_META_KEY: &str = "lancedb::column_definitions";
/// Value of the definition's `kind` tag for the projected `select` form.
/// Reserved for root-namespace sources; see [`NAMESPACED_SELECT_KIND`].
pub const SELECT_KIND: &str = "select";
/// The `select` form over a namespaced source: its own kind, because released
/// readers drop unknown fields and resolve a `select` source at the root, so
/// this routes them to the [`MaterializedViewKind::Unrecognized`] refusal
/// instead of a wrong-table refresh.
pub const NAMESPACED_SELECT_KIND: &str = "namespaced_select";
/// Which view outputs each source column is projected to directly. A column
/// may be projected more than once, so each carries every name the view gives
/// it, in projection order.
@@ -102,10 +95,6 @@ pub struct ViewProjection {
pub struct MaterializedViewDefinition {
/// Name of the source table, in the same database as the view.
pub source_table: String,
/// Namespace path holding the source table; empty is the root namespace.
/// A definition written before namespaced sources reads as root.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub source_namespace: Vec<String>,
/// The projected output columns, in view schema order.
pub projections: Vec<ViewProjection>,
/// SQL predicate selecting the source rows the view holds.
@@ -140,12 +129,7 @@ pub(crate) fn definition_to_metadata(definition: &MaterializedViewDefinition) ->
let mut value = serde_json::to_value(definition).map_err(|e| Error::Runtime {
message: format!("failed to serialize view definition: {e}"),
})?;
let kind = if definition.source_namespace.is_empty() {
SELECT_KIND
} else {
NAMESPACED_SELECT_KIND
};
value["kind"] = serde_json::Value::String(kind.to_string());
value["kind"] = serde_json::Value::String(SELECT_KIND.to_string());
Ok(value.to_string())
}
@@ -166,21 +150,12 @@ pub fn materialized_view_kind(
.get("kind")
.and_then(|k| k.as_str())
.ok_or_else(|| unreadable(&"missing kind tag"))?;
if kind != SELECT_KIND && kind != NAMESPACED_SELECT_KIND {
if kind != SELECT_KIND {
return Ok(Some(MaterializedViewKind::Unrecognized {
kind: kind.to_string(),
}));
}
let kind = kind.to_string();
let definition: MaterializedViewDefinition =
serde_json::from_value(value).map_err(|e| unreadable(&e))?;
// No correct writer produces a kind that disagrees with its namespace.
if (kind == SELECT_KIND) != definition.source_namespace.is_empty() {
return Err(unreadable(&format!(
"kind '{kind}' does not match its source namespace {:?}",
definition.source_namespace
)));
}
let definition = serde_json::from_value(value).map_err(|e| unreadable(&e))?;
Ok(Some(MaterializedViewKind::Select(definition)))
}
@@ -191,7 +166,6 @@ pub fn materialized_view_kind(
pub(crate) fn plan(
source_schema: SchemaRef,
source_table: &str,
source_namespace: &[String],
projections: &[(String, String)],
filter: Option<&str>,
limit: Option<u64>,
@@ -345,7 +319,6 @@ pub(crate) fn plan(
let definition = MaterializedViewDefinition {
source_table: source_table.to_string(),
source_namespace: source_namespace.to_vec(),
projections: projections
.into_iter()
.map(|(output, expression)| ViewProjection { output, expression })
@@ -629,7 +602,7 @@ pub struct PreparedDeclaration {
definition: MaterializedViewDefinition,
/// The source's own database: the only place
/// [`PreparedDeclaration::create`] will put the view, because refresh
/// resolves the recorded source coordinate through the view's database.
/// resolves the recorded source name through the view's database.
database: Arc<dyn Database>,
}
@@ -649,21 +622,10 @@ impl PreparedDeclaration {
/// Create the view table and verify it, consuming the declaration.
///
/// The view goes at the root of the source's own database, where refresh
/// resolves the recorded source coordinate. Stable row ids are requested
/// at both levels and verified rather than trusted; nothing is rolled
/// back on failure.
/// The view goes in the source's own database, where refresh resolves the
/// recorded source name. Stable row ids are requested at both levels and
/// verified rather than trusted; nothing is rolled back on failure.
pub async fn create(self, name: &str) -> Result<MaterializedView> {
self.create_in(&[], name).await
}
/// Create the view in `namespace_path`, empty for the root namespace.
/// Otherwise [`PreparedDeclaration::create`].
pub async fn create_in(
self,
namespace_path: &[String],
name: &str,
) -> Result<MaterializedView> {
let empty: Vec<std::result::Result<arrow_array::RecordBatch, arrow_schema::ArrowError>> =
vec![];
// Minted here, not at preparation: a declaration can be cloned and
@@ -678,7 +640,6 @@ impl PreparedDeclaration {
let reader: Box<dyn arrow_array::RecordBatchReader + Send> =
Box::new(arrow_array::RecordBatchIterator::new(empty, schema));
let mut request = CreateTableRequest::new(name.to_string(), Box::new(reader));
request.namespace_path = namespace_path.to_vec();
let write_params = request
.write_options
.lance_write_params
@@ -719,8 +680,8 @@ impl PreparedDeclaration {
/// Validate a view declaration against its live source and hold what its
/// creation needs. The declaration is canonicalized through the coordinate a
/// refresh will resolve -- name and namespace both -- so a handle that does
/// not resolve back to itself is rejected. Same creation-time checks as
/// refresh will resolve, so a handle that does not resolve back to itself is
/// rejected, as is a namespaced source. Same creation-time checks as
/// [`Connection::create_materialized_view`].
///
/// ```no_run
@@ -749,9 +710,17 @@ pub async fn prepare_declaration(
message: "materialized views are supported only on local databases".into(),
});
};
// Refresh resolves the source at exactly this coordinate, so the
// definition records the namespace alongside the name.
let source_namespace = source.namespace().to_vec();
// The definition records the source by bare name; any other source
// form would be recorded as a name its refresh cannot resolve.
if !source.namespace().is_empty() {
return Err(Error::NotSupported {
message: format!(
"a namespaced source cannot be recorded in a view definition; \
'{}' must be a root-namespace table",
source.name()
),
});
}
let database = source
.database_opt()
.ok_or_else(|| Error::InvalidInput {
@@ -765,7 +734,7 @@ pub async fn prepare_declaration(
let resolved = database
.open_table(OpenTableRequest {
name: source.name().to_string(),
namespace_path: source_namespace.clone(),
namespace_path: vec![],
index_cache_size: None,
lance_read_params: None,
location: None,
@@ -811,7 +780,6 @@ pub async fn prepare_declaration(
let (definition, mut fields, lineage) = plan(
source_schema.clone(),
resolved.name(),
&source_namespace,
projections,
filter,
limit,
@@ -871,9 +839,7 @@ fn ensure_local(connection: &Connection) -> Result<()> {
pub struct CreateMaterializedViewBuilder {
connection: Connection,
name: String,
namespace: Vec<String>,
source: String,
source_namespace: Vec<String>,
projections: Vec<(String, String)>,
filter: Option<String>,
limit: Option<u64>,
@@ -884,28 +850,13 @@ impl CreateMaterializedViewBuilder {
Self {
connection,
name,
namespace: Vec::new(),
source,
source_namespace: Vec::new(),
projections: Vec::new(),
filter: None,
limit: None,
}
}
/// The namespace to create the view in. Defaults to the root namespace.
pub fn namespace(mut self, namespace_path: Vec<String>) -> Self {
self.namespace = namespace_path;
self
}
/// The namespace holding the source table; recorded in the definition
/// for refresh to resolve. Defaults to the root namespace.
pub fn source_namespace(mut self, namespace_path: Vec<String>) -> Self {
self.source_namespace = namespace_path;
self
}
/// The view's columns, as `(name, SQL expression)` pairs. Not calling
/// this selects every source column, expanded at creation time.
pub fn select(
@@ -936,12 +887,7 @@ impl CreateMaterializedViewBuilder {
/// provenance across compaction, and cannot be enabled later.
pub async fn execute(self) -> Result<MaterializedView> {
ensure_local(&self.connection)?;
let source = self
.connection
.open_table(&self.source)
.namespace(self.source_namespace.clone())
.execute()
.await?;
let source = self.connection.open_table(&self.source).execute().await?;
let prepared = prepare_declaration(
&source,
&self.projections,
@@ -949,7 +895,7 @@ impl CreateMaterializedViewBuilder {
self.limit,
)
.await?;
prepared.create_in(&self.namespace, &self.name).await
prepared.create(&self.name).await
}
}
@@ -1206,7 +1152,6 @@ mod tests {
view.definition(),
&MaterializedViewDefinition {
source_table: "people".into(),
source_namespace: Vec::new(),
projections: vec![
ViewProjection {
output: "name".into(),
@@ -2138,138 +2083,33 @@ mod tests {
.await
.unwrap_err();
assert!(err.to_string().contains("custom_loc"), "{err}");
}
/// A view declared over a namespaced source records that namespace, and
/// refresh resolves the source through it -- the coordinate round-trips.
#[tokio::test]
async fn a_namespaced_source_round_trips_through_refresh() {
use lance_namespace::models::CreateNamespaceRequest;
let tmp = tempfile::tempdir().unwrap();
let mut properties = std::collections::HashMap::new();
properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string());
let conn = crate::connect_namespace("dir", properties)
.execute()
.await
.unwrap();
conn.create_namespace(CreateNamespaceRequest {
id: Some(vec!["ns".into()]),
..Default::default()
})
// A namespaced source cannot be recorded in the definition: the
// bare name refresh resolves would reach a different table or none.
let namespaced = crate::table::NativeTable::create(
"memory://ns_src",
"ns_src",
vec!["ns".to_string()],
Box::new(arrow_array::RecordBatchIterator::new(
vec![],
std::sync::Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
"id",
arrow_schema::DataType::Int32,
true,
)])),
)) as Box<dyn arrow_array::RecordBatchReader + Send>,
None,
None,
None,
None,
std::collections::HashSet::new(),
)
.await
.unwrap();
let batch = record_batch!(
("name", Utf8, ["ada", "grace", "alan"]),
("age", Int32, [36, 85, 41])
)
.unwrap();
conn.create_table("people", batch)
.namespace(vec!["ns".to_string()])
.write_options(stable_row_ids())
.execute()
let namespaced = Table::new(std::sync::Arc::new(namespaced), conn.database().clone());
let err = prepare_declaration(&namespaced, &[], None, None)
.await
.unwrap();
// A decoy of the same name at the root: resolving the source at the
// wrong namespace materializes one row here instead of three.
let decoy = record_batch!(("name", Utf8, ["mallory"]), ("age", Int32, [42])).unwrap();
conn.create_table("people", decoy)
.write_options(stable_row_ids())
.execute()
.await
.unwrap();
let view = conn
.create_materialized_view("adults", "people")
.namespace(vec!["ns".to_string()])
.source_namespace(vec!["ns".to_string()])
.select([("name", "name")])
.only_if("age >= 18")
.execute()
.await
.unwrap();
assert_eq!(view.definition().source_table, "people");
assert_eq!(view.definition().source_namespace, vec!["ns".to_string()]);
assert_eq!(view.table().namespace(), &["ns"]);
// Refresh resolves the source at the recorded namespace, not at root.
let result = view.refresh().execute().await.unwrap();
assert_eq!(result.rows_written, 3);
}
/// A definition stored before namespaced sources existed carries no
/// namespace key and must read as the root namespace.
#[test]
fn a_definition_without_a_namespace_reads_as_root() {
let stored =
r#"{"source_table":"people","projections":[{"output":"name","expression":"name"}]}"#;
let definition: MaterializedViewDefinition = serde_json::from_str(stored).unwrap();
assert!(definition.source_namespace.is_empty());
}
fn definition(source_namespace: Vec<String>) -> MaterializedViewDefinition {
MaterializedViewDefinition {
source_table: "people".to_string(),
source_namespace,
projections: vec![ViewProjection {
output: "name".to_string(),
expression: "name".to_string(),
}],
filter: None,
limit: None,
inputs: vec!["name".to_string()],
}
}
/// A root definition keeps the pre-namespace `select` form byte-stably;
/// a namespaced one moves off `select`, which sends pre-namespace readers
/// to the `Unrecognized` refusal instead of a root resolve.
#[test]
fn a_namespaced_definition_is_refused_by_the_pre_namespace_reader() {
let root = definition_to_metadata(&definition(Vec::new())).unwrap();
let root: serde_json::Value = serde_json::from_str(&root).unwrap();
assert_eq!(root["kind"], "select");
assert!(
root.get("source_namespace").is_none(),
"a root definition must not grow new keys: {root}"
);
let stored = definition_to_metadata(&definition(vec!["ns".to_string()])).unwrap();
let value: serde_json::Value = serde_json::from_str(&stored).unwrap();
// The pre-namespace discriminator is `kind == "select"`; anything
// else lands in its Unrecognized refusal rather than in a root open.
assert_eq!(value["kind"], "namespaced_select");
// The current reader round-trips the coordinate.
let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), stored)]);
match materialized_view_kind(&metadata).unwrap() {
Some(MaterializedViewKind::Select(read)) => {
assert_eq!(read.source_namespace, vec!["ns".to_string()])
}
other => panic!("expected the namespaced select form, got {other:?}"),
}
}
/// A kind that disagrees with its namespace is an error, not a view:
/// under `select` it is the shape old readers would resolve at the root.
#[test]
fn a_kind_namespace_mismatch_is_refused() {
for (kind, namespace) in [
(SELECT_KIND, vec!["ns".to_string()]),
(NAMESPACED_SELECT_KIND, Vec::new()),
] {
let mut value = serde_json::to_value(definition(namespace)).unwrap();
value["kind"] = serde_json::Value::String(kind.to_string());
let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), value.to_string())]);
let err = materialized_view_kind(&metadata).unwrap_err();
assert!(
err.to_string()
.contains("does not match its source namespace"),
"kind '{kind}': {err}"
);
}
.unwrap_err();
assert!(err.to_string().contains("namespaced source"), "{err}");
}
}
@@ -170,7 +170,6 @@ pub(crate) async fn execute_refresh(
let (replanned, mut planned_fields, _renames) = super::plan(
source_schema,
&definition.source_table,
&definition.source_namespace,
&projections,
definition.filter.as_deref(),
definition.limit,
@@ -591,7 +590,7 @@ async fn open_source(view: &Table, definition: &MaterializedViewDefinition) -> R
let source = database
.open_table(OpenTableRequest {
name: definition.source_table.clone(),
namespace_path: definition.source_namespace.clone(),
namespace_path: Vec::new(),
index_cache_size: None,
lance_read_params: None,
location: None,
@@ -2920,7 +2919,6 @@ mod tests {
let replacement = crate::materialized_view::MaterializedViewDefinition {
source_table: "src".into(),
source_namespace: Vec::new(),
projections: vec![
crate::materialized_view::ViewProjection {
output: "x".into(),
@@ -2960,7 +2958,6 @@ mod tests {
let narrower = crate::materialized_view::MaterializedViewDefinition {
source_table: "src".into(),
source_namespace: Vec::new(),
projections: vec![crate::materialized_view::ViewProjection {
output: "x".into(),
expression: "x".into(),
+5 -835
View File
@@ -1,37 +1,21 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
use std::collections::{HashMap, HashSet};
use std::pin::Pin;
use std::sync::Arc;
use std::{future::Future, time::Duration};
use arrow::compute::concat_batches;
use arrow_array::{
Array, Float16Array, Float32Array, Float64Array, RecordBatch, UInt64Array,
cast::AsArray,
make_array,
types::{Int64Type, UInt64Type},
};
use arrow_array::{Array, Float16Array, Float32Array, Float64Array, RecordBatch, make_array};
use arrow_schema::{DataType, SchemaRef};
use datafusion_common::{DataFusionError, Result as DataFusionResult};
use datafusion_execution::TaskContext;
use datafusion_expr::{Expr, col, lit};
use datafusion_physical_expr::{EquivalenceProperties, Partitioning};
use datafusion_physical_plan::{
DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties,
coalesce_partitions::CoalescePartitionsExec,
execution_plan::{Boundedness, EmissionType},
limit::GlobalLimitExec,
stream::RecordBatchStreamAdapter,
};
use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, stream, try_join};
use datafusion_physical_plan::ExecutionPlan;
use futures::{FutureExt, TryFutureExt, TryStreamExt, stream, try_join};
use half::f16;
/// Re-export Lance ColumnOrdering type for use in query ordering
pub use lance::dataset::scanner::ColumnOrdering;
use lance::dataset::{ROW_ID, scanner::DatasetRecordBatchStream};
use lance_arrow::RecordBatchExt;
use lance_datafusion::exec::{execute_plan, format_plan as format_analyzed_plan};
use lance_datafusion::exec::execute_plan;
use lance_index::scalar::FullTextSearchQuery;
use lance_index::scalar::inverted::SCORE_COL;
use lance_index::vector::DIST_COL;
@@ -841,14 +825,6 @@ pub struct QueryRequest {
/// Offset of the query.
pub offset: Option<usize>,
/// Dataset offsets whose occurrence multiplicity must be restored after
/// executing the physical lookup represented by this request.
///
/// This is client-side execution metadata used when a [`TakeQuery`] is
/// converted into a request. It is not sent to remote services.
#[doc(hidden)]
pub take_offsets: Option<Vec<u64>>,
/// Apply filter to the returned rows.
pub filter: Option<QueryFilter>,
@@ -917,7 +893,6 @@ impl Default for QueryRequest {
Self {
limit: None,
offset: None,
take_offsets: None,
filter: None,
filter_error: None,
full_text_search: None,
@@ -1554,302 +1529,6 @@ impl HasQuery for VectorQuery {
}
}
fn take_occurrences(offsets: &[u64]) -> HashMap<u64, usize> {
let mut occurrences = HashMap::with_capacity(offsets.len());
for offset in offsets {
*occurrences.entry(*offset).or_insert(0) += 1;
}
occurrences
}
fn restore_take_batch_with_occurrences(
batch: RecordBatch,
offsets: &[u64],
occurrences: &HashMap<u64, usize>,
ordering_column: &str,
drop_ordering_column: bool,
preserve_order: bool,
) -> Result<RecordBatch> {
let actual_offsets = batch
.column_by_name(ordering_column)
.ok_or_else(|| Error::Schema {
message: format!(
"take query result did not include ordering column '{ordering_column}'"
),
})?;
let actual_offsets = match actual_offsets.data_type() {
DataType::UInt64 => actual_offsets
.as_primitive::<UInt64Type>()
.values()
.to_vec(),
DataType::Int64 => actual_offsets
.as_primitive::<Int64Type>()
.values()
.iter()
.map(|offset| {
u64::try_from(*offset).map_err(|_| Error::Schema {
message: format!(
"take query ordering column '{ordering_column}' contained a negative offset"
),
})
})
.collect::<Result<Vec<_>>>()?,
data_type => {
return Err(Error::Schema {
message: format!(
"take query ordering column '{ordering_column}' had unsupported type {data_type}"
),
});
}
};
let mut desired_order = Vec::with_capacity(offsets.len());
if preserve_order {
let ordering = actual_offsets
.iter()
.copied()
.enumerate()
.map(|(index, offset)| (offset, index as u64))
.collect::<HashMap<_, _>>();
// Missing offsets retain the filter-based behavior of returning no row.
desired_order.extend(
offsets
.iter()
.filter_map(|offset| ordering.get(offset).copied()),
);
} else {
// Public take queries do not guarantee output order. Preserve the lookup's
// existing order and only restore the multiplicity of each matching row.
for (index, offset) in actual_offsets.iter().enumerate() {
if let Some(count) = occurrences.get(offset) {
desired_order.extend(std::iter::repeat_n(index as u64, *count));
}
}
}
let mut ordered_batch = if desired_order.len() == batch.num_rows()
&& desired_order
.iter()
.enumerate()
.all(|(index, desired)| *desired == index as u64)
{
batch
} else {
arrow_select::take::take_record_batch(&batch, &UInt64Array::from(desired_order))?
};
if drop_ordering_column {
ordered_batch = ordered_batch.drop_column(ordering_column)?;
}
Ok(ordered_batch)
}
#[cfg(test)]
fn restore_take_batch(
batch: RecordBatch,
offsets: &[u64],
ordering_column: &str,
drop_ordering_column: bool,
preserve_order: bool,
) -> Result<RecordBatch> {
restore_take_batch_with_occurrences(
batch,
offsets,
&take_occurrences(offsets),
ordering_column,
drop_ordering_column,
preserve_order,
)
}
/// Restores the logical offset occurrence sequence above the physical lookup plan.
///
/// The lookup plan returns each matching row at most once. For ordinary unordered
/// takes this operator expands each input batch incrementally and preserves the
/// lookup's partitioning. The explicitly ordered reader path collects one coalesced
/// input before restoring requested order. Pagination must remain above this operator
/// so it applies to occurrences.
#[derive(Debug)]
struct TakeRestoreExec {
input: Arc<dyn ExecutionPlan>,
offsets: Vec<u64>,
occurrences: Arc<HashMap<u64, usize>>,
ordering_column: String,
drop_ordering_column: bool,
preserve_order: bool,
schema: SchemaRef,
properties: Arc<PlanProperties>,
}
impl TakeRestoreExec {
fn try_new(
input: Arc<dyn ExecutionPlan>,
offsets: Vec<u64>,
ordering_column: String,
drop_ordering_column: bool,
preserve_order: bool,
) -> Result<Self> {
let schema = if drop_ordering_column {
RecordBatch::new_empty(input.schema())
.drop_column(&ordering_column)?
.schema()
} else {
input.schema()
};
let partition_count = if preserve_order {
1
} else {
input.output_partitioning().partition_count()
};
let emission_type = if preserve_order {
EmissionType::Final
} else {
EmissionType::Incremental
};
let properties = Arc::new(PlanProperties::new(
EquivalenceProperties::new(schema.clone()),
Partitioning::UnknownPartitioning(partition_count),
emission_type,
Boundedness::Bounded,
));
Ok(Self {
input,
occurrences: Arc::new(take_occurrences(&offsets)),
offsets,
ordering_column,
drop_ordering_column,
preserve_order,
schema,
properties,
})
}
}
impl DisplayAs for TakeRestoreExec {
fn fmt_as(
&self,
_display_type: DisplayFormatType,
formatter: &mut std::fmt::Formatter<'_>,
) -> std::fmt::Result {
write!(
formatter,
"TakeRestoreExec: occurrences={}",
self.offsets.len()
)
}
}
impl ExecutionPlan for TakeRestoreExec {
fn name(&self) -> &str {
"TakeRestoreExec"
}
fn properties(&self) -> &Arc<PlanProperties> {
&self.properties
}
fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
vec![&self.input]
}
fn maintains_input_order(&self) -> Vec<bool> {
vec![!self.preserve_order]
}
fn benefits_from_input_partitioning(&self) -> Vec<bool> {
vec![false]
}
fn with_new_children(
self: Arc<Self>,
children: Vec<Arc<dyn ExecutionPlan>>,
) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
if children.len() != 1 {
return Err(DataFusionError::Internal(format!(
"TakeRestoreExec expected one child, got {}",
children.len()
)));
}
let child = children.into_iter().next().unwrap();
let plan = Self::try_new(
child,
self.offsets.clone(),
self.ordering_column.clone(),
self.drop_ordering_column,
self.preserve_order,
)
.map_err(|error| DataFusionError::External(Box::new(error)))?;
Ok(Arc::new(plan))
}
fn execute(
&self,
partition: usize,
context: Arc<TaskContext>,
) -> DataFusionResult<datafusion_physical_plan::SendableRecordBatchStream> {
let partition_count = self.input.output_partitioning().partition_count();
if partition >= partition_count || (self.preserve_order && partition != 0) {
return Err(DataFusionError::Internal(format!(
"TakeRestoreExec cannot execute partition {partition}; input has {partition_count} partitions"
)));
}
let input = self.input.execute(partition, context)?;
let output_schema = self.schema.clone();
let offsets = self.offsets.clone();
let occurrences = self.occurrences.clone();
let ordering_column = self.ordering_column.clone();
let drop_ordering_column = self.drop_ordering_column;
let preserve_order = self.preserve_order;
let stream: Pin<Box<dyn futures::Stream<Item = DataFusionResult<RecordBatch>> + Send>> =
if preserve_order {
let input_schema = input.schema();
Box::pin(stream::once(async move {
let batches = input.try_collect::<Vec<_>>().await?;
let batch = if batches.is_empty() {
RecordBatch::new_empty(input_schema.clone())
} else {
concat_batches(&input_schema, &batches)?
};
restore_take_batch_with_occurrences(
batch,
&offsets,
&occurrences,
&ordering_column,
drop_ordering_column,
true,
)
.map_err(|error| DataFusionError::External(Box::new(error)))
}))
} else {
Box::pin(input.map(move |batch| {
batch.and_then(|batch| {
restore_take_batch_with_occurrences(
batch,
&offsets,
&occurrences,
&ordering_column,
drop_ordering_column,
false,
)
.map_err(|error| DataFusionError::External(Box::new(error)))
})
}))
};
Ok(Box::pin(RecordBatchStreamAdapter::new(
output_schema,
stream,
)))
}
fn supports_limit_pushdown(&self) -> bool {
false
}
}
/// A builder for LanceDB take queries.
///
/// See [`crate::Table::query`] for more details on queries
@@ -1866,8 +1545,6 @@ impl ExecutionPlan for TakeRestoreExec {
pub struct TakeQuery {
parent: Arc<dyn BaseTable>,
request: QueryRequest,
offsets: Option<Vec<u64>>,
preserve_order: bool,
}
impl TakeQuery {
@@ -1875,24 +1552,15 @@ impl TakeQuery {
///
/// See [`crate::Table::take_offsets`] for more details.
pub fn from_offsets(parent: Arc<dyn BaseTable>, offsets: Vec<u64>) -> Self {
let mut seen = HashSet::with_capacity(offsets.len());
let in_list: Vec<Expr> = offsets
.iter()
.copied()
.filter(|offset| seen.insert(*offset))
.map(lit)
.collect();
let in_list: Vec<Expr> = offsets.iter().map(|o| lit(*o)).collect();
Self {
parent,
request: QueryRequest {
filter: Some(QueryFilter::Datafusion(
col("_rowoffset").in_list(in_list, false),
)),
take_offsets: Some(offsets.clone()),
..Default::default()
},
offsets: Some(offsets),
preserve_order: false,
}
}
@@ -1907,181 +1575,9 @@ impl TakeQuery {
filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
..Default::default()
},
offsets: None,
preserve_order: false,
}
}
/// Preserve the requested offset order when restoring duplicate occurrences.
///
/// This is reserved for readers whose API explicitly guarantees ordering.
pub(crate) fn preserve_order(mut self) -> Self {
debug_assert!(self.offsets.is_some());
self.preserve_order = true;
self
}
async fn request_with_row_offset(
parent: &dyn BaseTable,
request: &QueryRequest,
) -> Result<(QueryRequest, String, bool)> {
const ROW_OFFSET: &str = "_rowoffset";
const INTERNAL_ROW_OFFSET: &str = "__lancedb_take_row_offset";
let mut request = request.clone();
// The physical lookup must not recursively restore occurrences. The
// wrapper above this request owns that logical operation.
request.take_offsets = None;
let (ordering_column, drop_ordering_column) = match &mut request.select {
Select::All => {
let mut columns = parent
.schema()
.await?
.fields()
.iter()
.map(|field| field.name().clone())
.collect::<Vec<_>>();
columns.push(ROW_OFFSET.to_string());
request.select = Select::Columns(columns);
(ROW_OFFSET.to_string(), true)
}
Select::Columns(columns) => {
if columns.iter().any(|column| column == ROW_OFFSET) {
(ROW_OFFSET.to_string(), false)
} else {
columns.push(ROW_OFFSET.to_string());
(ROW_OFFSET.to_string(), true)
}
}
Select::Dynamic(columns) => {
let mut ordering_column = INTERNAL_ROW_OFFSET.to_string();
while columns.iter().any(|(name, _)| name == &ordering_column) {
ordering_column.push('_');
}
columns.push((ordering_column.clone(), ROW_OFFSET.to_string()));
(ordering_column, true)
}
Select::Expr(columns) => {
let mut ordering_column = INTERNAL_ROW_OFFSET.to_string();
while columns.iter().any(|(name, _)| name == &ordering_column) {
ordering_column.push('_');
}
columns.push((ordering_column.clone(), col(ROW_OFFSET)));
(ordering_column, true)
}
};
Ok((request, ordering_column, drop_ordering_column))
}
async fn prepare_offsets_lookup(
parent: &dyn BaseTable,
request: &QueryRequest,
) -> Result<(QueryRequest, String, bool, usize, Option<usize>)> {
let (mut request, ordering_column, drop_ordering_column) =
Self::request_with_row_offset(parent, request).await?;
// The lookup operates on distinct physical rows. Pagination is a logical
// operation over occurrences and must be applied only after restoration.
let output_offset = request.offset.take().unwrap_or_default();
let output_limit = request.limit.take();
Ok((
request,
ordering_column,
drop_ordering_column,
output_offset,
output_limit,
))
}
fn wrap_offsets_plan(
lookup: Arc<dyn ExecutionPlan>,
offsets: &[u64],
ordering_column: String,
drop_ordering_column: bool,
output_offset: usize,
output_limit: Option<usize>,
preserve_order: bool,
) -> Result<Arc<dyn ExecutionPlan>> {
let lookup = if preserve_order {
Arc::new(CoalescePartitionsExec::new(lookup)) as Arc<dyn ExecutionPlan>
} else {
lookup
};
let restored: Arc<dyn ExecutionPlan> = Arc::new(TakeRestoreExec::try_new(
lookup,
offsets.to_vec(),
ordering_column,
drop_ordering_column,
preserve_order,
)?);
if output_offset > 0 || output_limit.is_some() {
Ok(Arc::new(GlobalLimitExec::new(
restored,
output_offset,
output_limit,
)))
} else {
Ok(restored)
}
}
fn wrap_offsets_explanation(
lookup: &str,
occurrence_count: usize,
output_offset: usize,
output_limit: Option<usize>,
preserve_order: bool,
) -> String {
fn indent(plan: &str, spaces: usize) -> String {
let indentation = " ".repeat(spaces);
plan.lines()
.map(|line| format!("{indentation}{line}"))
.collect::<Vec<_>>()
.join("\n")
}
let restored = if preserve_order {
format!(
"TakeRestoreExec: occurrences={occurrence_count}\n CoalescePartitionsExec\n{}",
indent(lookup, 4)
)
} else {
format!(
"TakeRestoreExec: occurrences={occurrence_count}\n{}",
indent(lookup, 2)
)
};
if output_offset > 0 || output_limit.is_some() {
let fetch = output_limit
.map(|limit| limit.to_string())
.unwrap_or_else(|| "None".to_string());
format!(
"GlobalLimitExec: skip={output_offset}, fetch={fetch}\n{}",
indent(&restored, 2)
)
} else {
restored
}
}
async fn create_offsets_plan(
&self,
offsets: &[u64],
options: QueryExecutionOptions,
) -> Result<Arc<dyn ExecutionPlan>> {
create_take_offsets_plan(
self.parent.as_ref(),
&self.request,
offsets,
options,
self.preserve_order,
)
.await
}
/// Convert the `TakeQuery` into a `QueryRequest`.
pub fn into_request(self) -> QueryRequest {
self.request
@@ -2126,63 +1622,6 @@ impl TakeQuery {
}
}
pub(crate) async fn create_take_offsets_plan(
parent: &dyn BaseTable,
request: &QueryRequest,
offsets: &[u64],
options: QueryExecutionOptions,
preserve_order: bool,
) -> Result<Arc<dyn ExecutionPlan>> {
let (request, ordering_column, drop_ordering_column, output_offset, output_limit) =
TakeQuery::prepare_offsets_lookup(parent, request).await?;
let lookup_options = if preserve_order {
options.without_output_batch_length_limit()
} else {
options
};
let lookup = parent
.create_plan(&AnyQuery::Query(request), lookup_options)
.await?;
TakeQuery::wrap_offsets_plan(
lookup,
offsets,
ordering_column,
drop_ordering_column,
output_offset,
output_limit,
preserve_order,
)
}
pub(crate) async fn explain_take_offsets_plan(
parent: &dyn BaseTable,
request: &QueryRequest,
offsets: &[u64],
verbose: bool,
) -> Result<String> {
let (request, _, _, output_offset, output_limit) =
TakeQuery::prepare_offsets_lookup(parent, request).await?;
let lookup = parent
.explain_plan(&AnyQuery::Query(request), verbose)
.await?;
Ok(TakeQuery::wrap_offsets_explanation(
&lookup,
offsets.len(),
output_offset,
output_limit,
false,
))
}
pub(crate) async fn prepare_take_offsets_request(
parent: &dyn BaseTable,
request: &QueryRequest,
) -> Result<QueryRequest> {
let (request, _, _, _, _) = TakeQuery::prepare_offsets_lookup(parent, request).await?;
Ok(request)
}
impl HasQuery for TakeQuery {
fn mut_query(&mut self) -> &mut QueryRequest {
&mut self.request
@@ -2191,10 +1630,6 @@ impl HasQuery for TakeQuery {
impl ExecutableQuery for TakeQuery {
async fn create_plan(&self, options: QueryExecutionOptions) -> Result<Arc<dyn ExecutionPlan>> {
if let Some(offsets) = &self.offsets {
return self.create_offsets_plan(offsets, options).await;
}
let req = AnyQuery::Query(self.request.clone());
self.parent.clone().create_plan(&req, options).await
}
@@ -2203,18 +1638,6 @@ impl ExecutableQuery for TakeQuery {
&self,
options: QueryExecutionOptions,
) -> Result<SendableRecordBatchStream> {
if self.offsets.is_some() {
let plan = self.create_plan(options.clone()).await?;
let inner = execute_plan(plan, Default::default())?;
let inner = MaxBatchLengthStream::new_boxed(inner, options.max_batch_length as usize);
let inner = if let Some(timeout) = options.timeout {
TimeoutStream::new_boxed(inner, timeout)
} else {
inner
};
return Ok(DatasetRecordBatchStream::new(inner).into());
}
let query = AnyQuery::Query(self.request.clone());
Ok(SendableRecordBatchStream::from(
self.parent.clone().query(&query, options).await?,
@@ -2222,51 +1645,11 @@ impl ExecutableQuery for TakeQuery {
}
async fn explain_plan(&self, verbose: bool) -> Result<String> {
if let Some(offsets) = &self.offsets {
let (request, _, _, output_offset, output_limit) =
Self::prepare_offsets_lookup(self.parent.as_ref(), &self.request).await?;
// Ask the backend to explain only the distinct-row lookup. This keeps
// remote explanation non-executing while still showing the client-side
// operators that create_plan and execution place above that lookup.
let lookup = self
.parent
.explain_plan(&AnyQuery::Query(request), verbose)
.await?;
return Ok(Self::wrap_offsets_explanation(
&lookup,
offsets.len(),
output_offset,
output_limit,
self.preserve_order,
));
}
let query = AnyQuery::Query(self.request.clone());
self.parent.explain_plan(&query, verbose).await
}
async fn analyze_plan_with_options(&self, options: QueryExecutionOptions) -> Result<String> {
if self.offsets.is_some() {
if self.parent.analyze_plan_is_remote() {
let (request, _, _, _, _) =
Self::prepare_offsets_lookup(self.parent.as_ref(), &self.request).await?;
// Remote analysis is owned by the service. The current wire
// request represents only the distinct-row lookup, so return
// the service report unchanged instead of fabricating metrics
// for client-side restoration operators.
return self
.parent
.analyze_plan(&AnyQuery::Query(request), options)
.await;
}
let plan = self.create_plan(options).await?;
execute_plan(plan.clone(), Default::default())?
.try_collect::<Vec<_>>()
.await?;
return Ok(format_analyzed_plan(plan));
}
let query = AnyQuery::Query(self.request.clone());
self.parent.analyze_plan(&query, options).await
}
@@ -2287,7 +1670,6 @@ mod tests {
StringArray, cast::AsArray, types::Float32Type,
};
use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
use datafusion_physical_plan::display::DisplayableExecutionPlan;
use futures::{StreamExt, TryStreamExt};
use lance_testing::datagen::{BatchGenerator, IncrementingInt32, RandomVector};
use rand::seq::IndexedRandom;
@@ -3542,218 +2924,6 @@ mod tests {
assert_eq!(results[0].num_columns(), 1);
}
#[tokio::test]
async fn test_take_offsets_preserves_duplicate_multiplicity() {
let tmp_dir = tempdir().unwrap();
let table = make_test_table(&tmp_dir).await;
let results = table
.take_offsets(vec![5, 1, 5, 17])
.select(Select::Columns(vec!["id".to_string()]))
.execute_with_options(QueryExecutionOptions {
max_batch_length: 2,
..Default::default()
})
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(results.len(), 2);
assert!(results.iter().all(|batch| batch.num_columns() == 1));
let mut ids = results
.iter()
.flat_map(|batch| {
batch
.column_by_name("id")
.unwrap()
.as_primitive::<Int32Type>()
.values()
.to_vec()
})
.collect::<Vec<_>>();
ids.sort_unstable();
assert_eq!(ids, vec![1, 5, 5, 17]);
}
#[tokio::test]
async fn test_take_offsets_plan_is_incremental() {
let tmp_dir = tempdir().unwrap();
let table = make_test_table(&tmp_dir).await;
let plan = table
.take_offsets(vec![5, 1, 17])
.create_plan(QueryExecutionOptions {
max_batch_length: 1,
..Default::default()
})
.await
.unwrap();
assert_eq!(plan.properties().emission_type, EmissionType::Incremental);
let displayed = DisplayableExecutionPlan::new(plan.as_ref())
.indent(false)
.to_string();
assert!(displayed.contains("TakeRestoreExec"));
assert!(!displayed.contains("CoalescePartitionsExec"));
}
#[tokio::test]
async fn test_take_into_request_preserves_duplicate_multiplicity() {
let tmp_dir = tempdir().unwrap();
let table = make_test_table(&tmp_dir).await;
let request = table.take_offsets(vec![5, 5]).into_request();
assert_eq!(request.take_offsets, Some(vec![5, 5]));
let batches = table
.base_table()
.query(&AnyQuery::Query(request), QueryExecutionOptions::default())
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
}
#[test]
fn test_restore_take_batch_only_reorders_when_requested() {
let batch = RecordBatch::try_from_iter([
(
"id",
Arc::new(Int32Array::from(vec![17, 5, 1])) as Arc<dyn Array>,
),
(
"_rowoffset",
Arc::new(UInt64Array::from(vec![17, 5, 1])) as Arc<dyn Array>,
),
])
.unwrap();
let restored =
restore_take_batch(batch.clone(), &[5, 1, 5, 17], "_rowoffset", true, false).unwrap();
assert_eq!(
restored
.column_by_name("id")
.unwrap()
.as_primitive::<Int32Type>()
.values(),
&[17, 5, 5, 1]
);
let ordered = restore_take_batch(batch, &[5, 1, 5, 17], "_rowoffset", true, true).unwrap();
assert_eq!(
ordered
.column_by_name("id")
.unwrap()
.as_primitive::<Int32Type>()
.values(),
&[5, 1, 5, 17]
);
}
#[tokio::test]
async fn test_take_offsets_applies_pagination_after_restoration() {
let tmp_dir = tempdir().unwrap();
let table = make_test_table(&tmp_dir).await;
let limited = table
.take_offsets(vec![0, 1, 0, 2])
.select(Select::Columns(vec!["id".to_string()]))
.limit(3)
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
let limited = concat_batches(&limited[0].schema(), &limited).unwrap();
assert_eq!(limited.num_rows(), 3);
assert!(
limited
.column_by_name("id")
.unwrap()
.as_primitive::<Int32Type>()
.values()
.iter()
.all(|id| [0, 1, 2].contains(id))
);
let offset = table
.take_offsets(vec![5, 1, 5, 17])
.select(Select::Columns(vec!["id".to_string()]))
.offset(1)
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
let offset = concat_batches(&offset[0].schema(), &offset).unwrap();
assert_eq!(offset.num_rows(), 3);
assert!(
offset
.column_by_name("id")
.unwrap()
.as_primitive::<Int32Type>()
.values()
.iter()
.all(|id| [1, 5, 17].contains(id))
);
}
#[tokio::test]
async fn test_take_offsets_create_plan_restores_occurrences() {
let tmp_dir = tempdir().unwrap();
let table = make_test_table(&tmp_dir).await;
let take = table
.take_offsets(vec![5, 1, 5, 17])
.select(Select::Columns(vec!["id".to_string()]));
let plan = take
.create_plan(QueryExecutionOptions::default())
.await
.unwrap();
assert_eq!(plan.schema().fields().len(), 1);
assert_eq!(plan.schema().field(0).name(), "id");
let planned = execute_plan(plan, Default::default())
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
let planned = concat_batches(&planned[0].schema(), &planned).unwrap();
let mut ids = planned
.column_by_name("id")
.unwrap()
.as_primitive::<Int32Type>()
.values()
.to_vec();
ids.sort_unstable();
assert_eq!(ids, vec![1, 5, 5, 17]);
}
#[tokio::test]
async fn test_take_offsets_plan_introspection_shows_restoration() {
let tmp_dir = tempdir().unwrap();
let table = make_test_table(&tmp_dir).await;
let take = table
.take_offsets(vec![0, 1, 0, 2])
.select(Select::Columns(vec!["id".to_string()]))
.limit(3);
let explained = take.explain_plan(false).await.unwrap();
assert!(explained.contains("GlobalLimitExec"));
assert!(explained.contains("TakeRestoreExec"));
assert!(!explained.contains("CoalescePartitionsExec"));
let analyzed = take.analyze_plan().await.unwrap();
assert!(analyzed.contains("GlobalLimitExec"));
assert!(analyzed.contains("TakeRestoreExec"));
assert!(!analyzed.contains("CoalescePartitionsExec"));
}
#[tokio::test]
async fn test_take_row_ids() {
let tmp_dir = tempdir().unwrap();
+3 -159
View File
@@ -40,8 +40,8 @@ use crate::table::{
use crate::table::{AnyQuery, Filter, Predicate, PreprocessingOutput, TableStatistics};
use crate::utils::background_cache::BackgroundCache;
use crate::utils::{
MaxBatchLengthStream, TimeoutStream, resolve_arrow_field_path, resolve_arrow_fts_field_path,
supported_btree_data_type, supported_vector_data_type,
resolve_arrow_field_path, resolve_arrow_fts_field_path, supported_btree_data_type,
supported_vector_data_type,
};
use crate::{DistanceType, Error};
use crate::{
@@ -2022,9 +2022,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
fn as_any(&self) -> &dyn std::any::Any {
self
}
fn analyze_plan_is_remote(&self) -> bool {
true
}
fn name(&self) -> &str {
&self.name
}
@@ -2597,13 +2594,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
query: &AnyQuery,
options: QueryExecutionOptions,
) -> Result<Arc<dyn ExecutionPlan>> {
if let AnyQuery::Query(request) = query
&& let Some(offsets) = &request.take_offsets
{
return crate::query::create_take_offsets_plan(self, request, offsets, options, false)
.await;
}
let streams = self.execute_query(query, &options).await?;
if streams.len() == 1 {
let stream = streams.into_iter().next().unwrap();
@@ -2622,27 +2612,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
query: &AnyQuery,
options: QueryExecutionOptions,
) -> Result<DatasetRecordBatchStream> {
if let AnyQuery::Query(request) = query
&& let Some(offsets) = &request.take_offsets
{
let plan = crate::query::create_take_offsets_plan(
self,
request,
offsets,
options.clone(),
false,
)
.await?;
let inner = execute_plan(plan, Default::default())?;
let inner = MaxBatchLengthStream::new_boxed(inner, options.max_batch_length as usize);
let inner = if let Some(timeout) = options.timeout {
TimeoutStream::new_boxed(inner, timeout)
} else {
inner
};
return Ok(DatasetRecordBatchStream::new(inner));
}
let streams = self.execute_query(query, &options).await?;
if streams.len() == 1 {
@@ -2680,12 +2649,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
}
async fn explain_plan(&self, query: &AnyQuery, verbose: bool) -> Result<String> {
if let AnyQuery::Query(request) = query
&& let Some(offsets) = &request.take_offsets
{
return crate::query::explain_take_offsets_plan(self, request, offsets, verbose).await;
}
let base_request = self
.client
.post(&format!("/v1/table/{}/explain_plan/", self.identifier));
@@ -2738,17 +2701,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
query: &AnyQuery,
options: QueryExecutionOptions,
) -> Result<String> {
let prepared_query = if let AnyQuery::Query(request) = query
&& request.take_offsets.is_some()
{
Some(AnyQuery::Query(
crate::query::prepare_take_offsets_request(self, request).await?,
))
} else {
None
};
let query = prepared_query.as_ref().unwrap_or(query);
let mut request = self
.client
.post(&format!("/v1/table/{}/analyze_plan/", self.identifier));
@@ -3738,7 +3690,7 @@ mod tests {
};
use arrow_schema::{DataType, Field, Schema};
use chrono::{DateTime, Utc};
use futures::{StreamExt, TryFutureExt, TryStreamExt, future::BoxFuture};
use futures::{StreamExt, TryFutureExt, future::BoxFuture};
use lance_index::scalar::inverted::{DocumentGranularity, query::MatchQuery};
use lance_index::scalar::{FullTextSearchQuery, InvertedIndexParams};
use reqwest::Body;
@@ -5659,114 +5611,6 @@ mod tests {
assert_eq!(result, "analyzed plan");
}
#[tokio::test]
async fn test_take_offsets_explain_plan_does_not_execute_query() {
let table = Table::new_with_handler("my_table", |request| {
assert_eq!(request.method(), "POST");
assert_eq!(request.url().path(), "/v1/table/my_table/explain_plan/");
http::Response::builder()
.status(200)
.body(r#""RemoteLookupExec""#)
.unwrap()
});
let explained = table
.take_offsets(vec![0, 1, 0, 2])
.select(crate::query::Select::columns(&["id"]))
.limit(3)
.explain_plan(false)
.await
.unwrap();
assert!(explained.contains("GlobalLimitExec"));
assert!(explained.contains("TakeRestoreExec"));
assert!(!explained.contains("CoalescePartitionsExec"));
assert!(explained.contains("RemoteLookupExec"));
}
#[tokio::test]
async fn test_converted_take_request_restores_remote_occurrences() {
let table = Table::new_with_handler("my_table", |request| {
assert_eq!(request.method(), "POST");
assert_eq!(request.url().path(), "/v1/table/my_table/query/");
let body: serde_json::Value =
serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
assert_eq!(body["columns"], json!(["id", "_rowoffset"]));
let data = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int32, false),
Field::new("_rowoffset", DataType::UInt64, false),
])),
vec![
Arc::new(Int32Array::from(vec![5])),
Arc::new(arrow_array::UInt64Array::from(vec![5])),
],
)
.unwrap();
http::Response::builder()
.status(200)
.header(CONTENT_TYPE, ARROW_FILE_CONTENT_TYPE)
.body(write_ipc_file(&data))
.unwrap()
});
let request = table
.take_offsets(vec![5, 5])
.select(crate::query::Select::columns(&["id"]))
.into_request();
let batches = table
.base_table()
.query(&AnyQuery::Query(request), QueryExecutionOptions::default())
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
assert!(
batches
.iter()
.all(|batch| batch.schema().fields().len() == 1)
);
}
#[tokio::test]
async fn test_take_offsets_analyze_plan_delegates_to_remote() {
let table = Table::new_with_handler("my_table", |request| {
assert_eq!(request.method(), "POST");
assert_eq!(request.url().path(), "/v1/table/my_table/analyze_plan/");
assert_eq!(
request
.url()
.query_pairs()
.find(|(key, _)| key == "distributed_metrics"),
Some(("distributed_metrics".into(), "per_worker".into()))
);
http::Response::builder()
.status(200)
.body(r#""Remote analyzed plan: worker metrics""#)
.unwrap()
});
let analyzed = table
.take_offsets(vec![0, 1, 0, 2])
.select(crate::query::Select::columns(&["id"]))
.limit(3)
.analyze_plan_with_options(QueryExecutionOptions {
analyze_plan_distributed_metrics: AnalyzePlanDistributedMetrics::PerWorker,
..Default::default()
})
.await
.unwrap();
assert_eq!(analyzed, "Remote analyzed plan: worker metrics");
}
#[tokio::test]
async fn test_query_structured_fts() {
let table =
+3 -11
View File
@@ -595,14 +595,6 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
query: &AnyQuery,
options: QueryExecutionOptions,
) -> Result<String>;
/// Whether [`BaseTable::analyze_plan`] is provided by a remote service.
///
/// Client-side query wrappers use this to preserve backend metrics and
/// distributed-analysis options instead of replacing them with a local plan.
#[doc(hidden)]
fn analyze_plan_is_remote(&self) -> bool {
false
}
/// Add new records to the table.
async fn add(&self, add: AddDataBuilder) -> Result<AddResult>;
@@ -1660,9 +1652,9 @@ impl Table {
/// Offsets are useful for sampling as the set of all valid offsets is easily
/// known in advance to be [0, len(table)).
///
/// No guarantees are made regarding the order in which results are returned.
/// Repeated offsets produce repeated rows, which makes this method suitable for
/// sampling with replacement.
/// No guarantees are made regarding the order in which results are returned. If you
/// desire an output order that matches the order of the given offsets, you will need
/// to add the row offset column to the output and align it yourself.
///
/// Parameters
/// ----------
+1 -8
View File
@@ -110,7 +110,7 @@ fn requires_local_namespace_execution(query: &AnyQuery) -> bool {
// pushing these down would silently ignore the user's setting. For use_lsm that
// is worse than a tuning miss: MemWAL read routing lives only in `create_plan`,
// so a pushed-down query would return stale base-only data with no error.
if query.base().use_lsm.is_some() || query.base().take_offsets.is_some() {
if query.base().use_lsm.is_some() {
return true;
}
matches!(
@@ -154,13 +154,6 @@ pub async fn create_plan(
options: QueryExecutionOptions,
) -> Result<Arc<dyn ExecutionPlan>> {
let query = query.canonicalized()?;
if let AnyQuery::Query(request) = &query
&& let Some(offsets) = &request.take_offsets
{
return crate::query::create_take_offsets_plan(table, request, offsets, options, false)
.await;
}
let query = match query {
AnyQuery::VectorQuery(query) => query,
AnyQuery::Query(query) => VectorQueryRequest::from_plain_query(query),