mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-01 19:18:38 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2c427c5dd2 | |||
| 9123725493 |
+1
-1
@@ -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
+45
-46
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
|
||||
|
||||
[[package]]
|
||||
name = "fsst"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5000,7 +5000,6 @@ dependencies = [
|
||||
"datafusion-functions",
|
||||
"datafusion-physical-expr",
|
||||
"futures",
|
||||
"half",
|
||||
"jsonb",
|
||||
"lance-arrow",
|
||||
"lance-core",
|
||||
@@ -5014,8 +5013,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-datagen"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5032,8 +5031,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-derive"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5042,8 +5041,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-encoding"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5076,8 +5075,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-file"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5108,8 +5107,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-index"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5173,8 +5172,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-index-core"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5196,8 +5195,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-io"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5237,8 +5236,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-linalg"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5252,8 +5251,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5265,8 +5264,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace-impls"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5319,8 +5318,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-select"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5334,8 +5333,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-table"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5375,8 +5374,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-testing"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5389,8 +5388,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-tokenizer"
|
||||
version = "12.0.0-beta.9"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.9#6f93e3fe389f5a661a02aff00c14df6be7bf0505"
|
||||
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",
|
||||
@@ -5403,7 +5402,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lancedb"
|
||||
version = "0.39.0-beta.0"
|
||||
version = "0.38.0"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"anyhow",
|
||||
@@ -5491,7 +5490,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lancedb-nodejs"
|
||||
version = "0.39.0-beta.0"
|
||||
version = "0.38.0"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -5516,7 +5515,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lancedb-python"
|
||||
version = "0.39.0-beta.0"
|
||||
version = "0.38.0"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"async-trait",
|
||||
|
||||
+14
-14
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
|
||||
rust-version = "1.91.0"
|
||||
|
||||
[workspace.dependencies]
|
||||
lance = { "version" = "=12.0.0-beta.9", default-features = false, "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=12.0.0-beta.9", default-features = false, "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=12.0.0-beta.9", default-features = false, "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=12.0.0-beta.9", "tag" = "v12.0.0-beta.9", "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
|
||||
|
||||
@@ -446,15 +446,6 @@ paths:
|
||||
properties:
|
||||
column:
|
||||
type: string
|
||||
name:
|
||||
type: string
|
||||
description: Optional name for the created index.
|
||||
replace:
|
||||
type: boolean
|
||||
default: true
|
||||
description: |
|
||||
Whether to replace an existing index with the same resolved
|
||||
name. Defaults to true.
|
||||
metric_type:
|
||||
type: string
|
||||
nullable: false
|
||||
|
||||
@@ -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>
|
||||
```
|
||||
|
||||
|
||||
@@ -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
@@ -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.9</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
@@ -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
|
||||
|
||||
@@ -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,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,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,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,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,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-win32-arm64-msvc",
|
||||
"version": "0.39.0-beta.0",
|
||||
"version": "0.38.0",
|
||||
"os": [
|
||||
"win32"
|
||||
],
|
||||
|
||||
@@ -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
@@ -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
@@ -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"
|
||||
|
||||
@@ -150,7 +150,6 @@ class Connection(object):
|
||||
def job(self, job_id: str) -> Job: ...
|
||||
async def create_function_async(self, request_json: str) -> Job: ...
|
||||
async def get_function(self, name: str, version: str) -> str: ...
|
||||
async def list_functions(self) -> List[str]: ...
|
||||
async def drop_function(self, name: str, version: str) -> bool: ...
|
||||
async def list_jobs(self) -> List[JobInfo]: ...
|
||||
async def get_job(self, job_id: str) -> Optional[JobDescription]: ...
|
||||
@@ -608,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]]]
|
||||
|
||||
@@ -712,24 +712,6 @@ class DBConnection(EnforceOverrides):
|
||||
"Function catalog operations are not supported for this connection type"
|
||||
)
|
||||
|
||||
def list_functions(self) -> List[FunctionVersion]:
|
||||
"""List every published immutable Function version.
|
||||
|
||||
Results are ordered by Function name then version. Local connections
|
||||
raise ``NotImplementedError``.
|
||||
|
||||
Examples
|
||||
--------
|
||||
List the identities available to use in Function-backed columns:
|
||||
|
||||
```python
|
||||
[(function.name, function.version) for function in db.list_functions()]
|
||||
```
|
||||
"""
|
||||
raise NotImplementedError(
|
||||
"Function catalog operations are not supported for this connection type"
|
||||
)
|
||||
|
||||
def drop_function(self, name: str, *, version: str) -> bool:
|
||||
"""Drop one exact immutable Function version from the remote catalog.
|
||||
|
||||
@@ -1441,10 +1423,6 @@ class LanceDBConnection(DBConnection):
|
||||
def get_function(self, name: str, *, version: str) -> FunctionVersion:
|
||||
return LOOP.run(self._conn.get_function(name, version=version))
|
||||
|
||||
@override
|
||||
def list_functions(self) -> List[FunctionVersion]:
|
||||
return LOOP.run(self._conn.list_functions())
|
||||
|
||||
@override
|
||||
def drop_function(self, name: str, *, version: str) -> bool:
|
||||
return LOOP.run(self._conn.drop_function(name, version=version))
|
||||
@@ -2279,17 +2257,6 @@ class AsyncConnection(object):
|
||||
"""Open one exact immutable Function version from the remote catalog."""
|
||||
return FunctionVersion.from_json(await self._inner.get_function(name, version))
|
||||
|
||||
async def list_functions(self) -> List[FunctionVersion]:
|
||||
"""List every published immutable Function version.
|
||||
|
||||
Results are ordered by Function name then version. Local connections
|
||||
raise ``NotImplementedError``.
|
||||
"""
|
||||
return [
|
||||
FunctionVersion.from_json(value)
|
||||
for value in await self._inner.list_functions()
|
||||
]
|
||||
|
||||
async def drop_function(self, name: str, *, version: str) -> bool:
|
||||
"""Drop one exact immutable Function version from the remote catalog."""
|
||||
return await self._inner.drop_function(name, version)
|
||||
|
||||
@@ -521,12 +521,6 @@ class RefreshColumnResult(_RemoteValue):
|
||||
|
||||
_FUNCTION_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*$")
|
||||
_FUNCTION_BLOB_V2_TYPE = "blob_v2"
|
||||
_ARROW_EXTENSION_NAME_KEY = "ARROW:extension:name"
|
||||
_BLOB_V2_EXTENSION_NAME = "lance.blob.v2"
|
||||
_NESTED_BLOB_COLLECTION_ERROR = (
|
||||
"unsupported Arrow type for Function signature: Blob v2 fields nested under "
|
||||
"collection types are not supported"
|
||||
)
|
||||
|
||||
|
||||
_GRAMMAR_PRIMITIVES = (
|
||||
@@ -597,19 +591,6 @@ def _validate_exact_arrow_field(field: pa.Field) -> None:
|
||||
"unsupported Arrow type for Function signature: lance.blob.v2 "
|
||||
f"requires a supported Blob storage layout, got {field}"
|
||||
)
|
||||
metadata = {
|
||||
(key.decode() if isinstance(key, bytes) else key): (
|
||||
value.decode() if isinstance(value, bytes) else value
|
||||
)
|
||||
for key, value in (field.metadata or {}).items()
|
||||
}
|
||||
if metadata and metadata != {
|
||||
_ARROW_EXTENSION_NAME_KEY: _BLOB_V2_EXTENSION_NAME
|
||||
}:
|
||||
raise TypeError(
|
||||
"unsupported Arrow type for Function signature: lance.blob.v2 "
|
||||
"field metadata must contain only its canonical extension marker"
|
||||
)
|
||||
elif field.metadata:
|
||||
raise TypeError(
|
||||
"unsupported Arrow type for Function signature: field metadata "
|
||||
@@ -674,84 +655,23 @@ def _canonical_arrow_field(field: pa.Field) -> str:
|
||||
return _canonical_arrow_type(field.type)
|
||||
|
||||
|
||||
def _blob_storage_type(field: pa.Field) -> pa.DataType:
|
||||
data_type = field.type
|
||||
if isinstance(data_type, pa.ExtensionType):
|
||||
return data_type.storage_type
|
||||
return data_type
|
||||
|
||||
|
||||
def _exact_blob_storage_type(field: pa.Field) -> dict[str, Any]:
|
||||
storage = _blob_storage_type(field)
|
||||
if not pa.types.is_struct(storage):
|
||||
raise TypeError(
|
||||
"unsupported Arrow type for Function signature: lance.blob.v2 "
|
||||
"requires struct storage"
|
||||
)
|
||||
return {
|
||||
"type": "struct",
|
||||
"fields": [
|
||||
{
|
||||
"name": child.name,
|
||||
"nullable": child.nullable,
|
||||
"type": (
|
||||
{"type": "large_binary"}
|
||||
if pa.types.is_large_binary(child.type)
|
||||
else _exact_arrow_type(child.type)
|
||||
),
|
||||
}
|
||||
for child in storage
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _data_type_has_blob_v2(data_type: pa.DataType) -> bool:
|
||||
if pa.types.is_struct(data_type):
|
||||
return any(
|
||||
_is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
|
||||
for field in data_type
|
||||
)
|
||||
if (
|
||||
pa.types.is_list(data_type)
|
||||
or pa.types.is_large_list(data_type)
|
||||
or pa.types.is_fixed_size_list(data_type)
|
||||
):
|
||||
field = data_type.value_field
|
||||
return _is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
|
||||
if pa.types.is_map(data_type):
|
||||
return any(
|
||||
_is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
|
||||
for field in (data_type.key_field, data_type.item_field)
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
def _exact_arrow_field(
|
||||
field: pa.Field, *, inside_collection: bool = False
|
||||
) -> dict[str, Any]:
|
||||
def _exact_arrow_field(field: pa.Field) -> dict[str, Any]:
|
||||
_validate_exact_arrow_field(field)
|
||||
if _is_blob_v2_field(field):
|
||||
if inside_collection:
|
||||
raise TypeError(_NESTED_BLOB_COLLECTION_ERROR)
|
||||
return {
|
||||
"name": field.name,
|
||||
"nullable": field.nullable,
|
||||
"type": _exact_blob_storage_type(field),
|
||||
"metadata": {
|
||||
_ARROW_EXTENSION_NAME_KEY: _BLOB_V2_EXTENSION_NAME,
|
||||
},
|
||||
}
|
||||
raise TypeError(
|
||||
"unsupported Arrow type for Function signature: nested Blob v2 "
|
||||
"fields are not supported; declare Blob parameters or named result "
|
||||
"fields directly"
|
||||
)
|
||||
value = {
|
||||
"name": field.name,
|
||||
"nullable": field.nullable,
|
||||
"type": _exact_arrow_type(field.type, inside_collection=inside_collection),
|
||||
"type": _exact_arrow_type(field.type),
|
||||
}
|
||||
return value
|
||||
|
||||
|
||||
def _exact_arrow_type(
|
||||
data_type: pa.DataType, *, inside_collection: bool = False
|
||||
) -> dict[str, Any]:
|
||||
def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
|
||||
for candidate, name in _GRAMMAR_PRIMITIVES:
|
||||
if data_type == candidate:
|
||||
return {"type": name}
|
||||
@@ -765,10 +685,7 @@ def _exact_arrow_type(
|
||||
)
|
||||
return {
|
||||
"type": "struct",
|
||||
"fields": [
|
||||
_exact_arrow_field(field, inside_collection=inside_collection)
|
||||
for field in fields
|
||||
],
|
||||
"fields": [_exact_arrow_field(field) for field in fields],
|
||||
}
|
||||
if (
|
||||
pa.types.is_list(data_type)
|
||||
@@ -793,15 +710,11 @@ def _exact_arrow_type(
|
||||
if pa.types.is_large_list(data_type)
|
||||
else "fixed_size_list"
|
||||
),
|
||||
"fields": [
|
||||
_exact_arrow_field(data_type.value_field, inside_collection=True)
|
||||
],
|
||||
"fields": [_exact_arrow_field(data_type.value_field)],
|
||||
}
|
||||
if pa.types.is_fixed_size_list(data_type):
|
||||
value["length"] = data_type.list_size
|
||||
return value
|
||||
if pa.types.is_map(data_type) and _data_type_has_blob_v2(data_type):
|
||||
raise TypeError(_NESTED_BLOB_COLLECTION_ERROR)
|
||||
raise TypeError(f"unsupported Arrow type for Function signature: {data_type}")
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -749,10 +749,6 @@ class RemoteDBConnection(DBConnection):
|
||||
def get_function(self, name: str, *, version: str) -> FunctionVersion:
|
||||
return LOOP.run(self._conn.get_function(name, version=version))
|
||||
|
||||
@override
|
||||
def list_functions(self) -> List[FunctionVersion]:
|
||||
return LOOP.run(self._conn.list_functions())
|
||||
|
||||
@override
|
||||
def drop_function(self, name: str, *, version: str) -> bool:
|
||||
return LOOP.run(self._conn.drop_function(name, version=version))
|
||||
|
||||
@@ -548,7 +548,6 @@ class RemoteTable(Table):
|
||||
LOOP.run(
|
||||
self._table.create_index(
|
||||
column,
|
||||
replace=replace,
|
||||
config=config,
|
||||
wait_timeout=wait_timeout,
|
||||
name=name,
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -668,167 +668,6 @@ def test_blob_fields_use_the_scalar_function_semantic_type():
|
||||
assert signature.output.arrow_type == "blob_v2"
|
||||
|
||||
|
||||
def test_whole_named_struct_function_can_include_a_blob_result_field():
|
||||
@udf(
|
||||
input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
|
||||
output_schema=pa.field(
|
||||
"payload",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field("mime_type", pa.string(), nullable=False),
|
||||
lancedb.blob("image", nullable=False),
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
),
|
||||
)
|
||||
def inspect_blob(image):
|
||||
return {"mime_type": "image/png", "image": image}
|
||||
|
||||
output = inspect_blob.registration_request.signature.output
|
||||
assert output.kind == "named_struct"
|
||||
assert [(field.name, field.arrow_type) for field in output.fields] == [
|
||||
("mime_type", "utf8"),
|
||||
("image", "blob_v2"),
|
||||
]
|
||||
|
||||
|
||||
def test_struct_blob_signature_fields_preserve_exact_metadata_and_nullability():
|
||||
nested_input = pa.field(
|
||||
"payload",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field("mime_type", pa.string(), nullable=False),
|
||||
pa.field(
|
||||
"nested",
|
||||
pa.struct([lancedb.blob("image", nullable=True)]),
|
||||
nullable=True,
|
||||
),
|
||||
]
|
||||
),
|
||||
nullable=True,
|
||||
)
|
||||
nested_output = pa.field(
|
||||
"result",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field("mime_type", pa.string(), nullable=False),
|
||||
pa.field(
|
||||
"nested",
|
||||
pa.struct([lancedb.blob("image", nullable=True)]),
|
||||
nullable=False,
|
||||
),
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
)
|
||||
|
||||
@udf(input_schema=pa.schema([nested_input]), output_schema=nested_output)
|
||||
def copy_payload(payload):
|
||||
return payload
|
||||
|
||||
signature = copy_payload.registration_request.signature
|
||||
input_type = json.loads(signature.inputs[0].arrow_type)
|
||||
assert input_type["fields"][1]["nullable"] is True
|
||||
input_blob = input_type["fields"][1]["type"]["fields"][0]
|
||||
assert input_blob["nullable"] is True
|
||||
assert input_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
|
||||
|
||||
assert signature.output.kind == "named_struct"
|
||||
nested_result = next(
|
||||
field for field in signature.output.fields if field.name == "nested"
|
||||
)
|
||||
output_type = json.loads(nested_result.arrow_type)
|
||||
output_blob = output_type["fields"][0]
|
||||
assert output_blob["nullable"] is True
|
||||
assert output_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
|
||||
|
||||
|
||||
def test_struct_blob_signature_supports_multiple_struct_levels():
|
||||
recursive = pa.field(
|
||||
"value",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field(
|
||||
"level_1",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field(
|
||||
"level_2",
|
||||
pa.struct([lancedb.blob("image", nullable=False)]),
|
||||
nullable=False,
|
||||
)
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
)
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
)
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([recursive]),
|
||||
output_schema=pa.field("size", pa.int64(), nullable=False),
|
||||
)
|
||||
def blob_size(value):
|
||||
return len(value["level_1"]["level_2"]["image"])
|
||||
|
||||
encoded = json.loads(blob_size.registration_request.signature.inputs[0].arrow_type)
|
||||
blob = encoded["fields"][0]["type"]["fields"][0]["type"]["fields"][0]
|
||||
assert blob["metadata"]["ARROW:extension:name"] == "lance.blob.v2"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"data_type",
|
||||
[
|
||||
pa.list_(lancedb.blob("item", nullable=False)),
|
||||
pa.large_list(lancedb.blob("item", nullable=False)),
|
||||
pa.list_(lancedb.blob("item", nullable=False), 2),
|
||||
pa.map_(pa.string(), lancedb.blob("value", nullable=False).type),
|
||||
],
|
||||
)
|
||||
def test_blob_signature_rejects_collection_ancestors(data_type):
|
||||
with pytest.raises(
|
||||
TypeError,
|
||||
match="Blob v2 fields nested under collection types are not supported",
|
||||
):
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([pa.field("value", data_type, nullable=False)]),
|
||||
output_schema=pa.field("size", pa.int64(), nullable=False),
|
||||
)
|
||||
def blob_size(value):
|
||||
return len(value)
|
||||
|
||||
|
||||
def test_blob_signature_rejects_collection_below_a_struct():
|
||||
nested = pa.field(
|
||||
"value",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field(
|
||||
"images",
|
||||
pa.list_(lancedb.blob("item", nullable=False)),
|
||||
nullable=False,
|
||||
)
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
)
|
||||
with pytest.raises(
|
||||
TypeError,
|
||||
match="Blob v2 fields nested under collection types are not supported",
|
||||
):
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([nested]),
|
||||
output_schema=pa.field("size", pa.int64(), nullable=False),
|
||||
)
|
||||
def blob_size(value):
|
||||
return len(value["images"])
|
||||
|
||||
|
||||
def test_named_struct_function_can_include_a_blob_result_field():
|
||||
@udf(
|
||||
input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
|
||||
@@ -890,6 +729,22 @@ def test_blob_marker_rejects_invalid_storage_layout():
|
||||
return len(image)
|
||||
|
||||
|
||||
def test_nested_blob_signature_field_has_a_clear_error():
|
||||
nested = pa.field(
|
||||
"value",
|
||||
pa.struct([lancedb.blob("image", nullable=False)]),
|
||||
nullable=False,
|
||||
)
|
||||
with pytest.raises(TypeError, match="nested Blob v2 fields are not supported"):
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([nested]),
|
||||
output_schema=pa.field("size", pa.int64(), nullable=False),
|
||||
)
|
||||
def blob_size(value):
|
||||
return len(value["image"])
|
||||
|
||||
|
||||
def test_nested_non_blob_extension_is_not_silently_unwrapped():
|
||||
class TestExtension(pa.ExtensionType):
|
||||
def __init__(self):
|
||||
@@ -1162,8 +1017,6 @@ def test_local_function_catalog_operations_are_not_supported(tmp_path):
|
||||
db.create_function_async(normalize_score)
|
||||
with pytest.raises(NotImplementedError, match=message):
|
||||
db.get_function("normalize_score", version="fv_exact")
|
||||
with pytest.raises(NotImplementedError, match=message):
|
||||
db.list_functions()
|
||||
with pytest.raises(NotImplementedError, match=message):
|
||||
db.drop_function("normalize_score", version="fv_exact")
|
||||
|
||||
@@ -1211,22 +1064,6 @@ def _mock_remote_function_catalog():
|
||||
"version": "fv_exact",
|
||||
}
|
||||
response = state["version"]
|
||||
elif self.path == "/v1/functions/list":
|
||||
assert body["include_definition"] is True
|
||||
if "page_token" not in body:
|
||||
response = {
|
||||
"functions": [
|
||||
{
|
||||
"name": "normalize_score",
|
||||
"version": "fv_exact",
|
||||
"definition": state["version"],
|
||||
}
|
||||
],
|
||||
"page_token": "next",
|
||||
}
|
||||
else:
|
||||
assert body["page_token"] == "next"
|
||||
response = {"functions": []}
|
||||
elif self.path == "/v1/functions/drop":
|
||||
assert body == {
|
||||
"name": "normalize_score",
|
||||
@@ -1293,49 +1130,6 @@ def test_blocking_remote_registration_returns_function_version():
|
||||
]
|
||||
|
||||
|
||||
def test_remote_list_functions_paginates_and_returns_typed_versions():
|
||||
with _mock_remote_function_catalog() as (host, state):
|
||||
db = lancedb.connect(
|
||||
"db://dev",
|
||||
api_key="fake",
|
||||
host_override=host,
|
||||
client_config={"retry_config": {"retries": 0}},
|
||||
)
|
||||
created = db.create_function(normalize_score)
|
||||
state["requests"].clear()
|
||||
functions = db.list_functions()
|
||||
|
||||
assert functions == [created]
|
||||
assert state["requests"] == [
|
||||
("/v1/functions/list", {"include_definition": True}),
|
||||
(
|
||||
"/v1/functions/list",
|
||||
{"include_definition": True, "page_token": "next"},
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_remote_list_functions_returns_typed_versions():
|
||||
with _mock_remote_function_catalog() as (host, state):
|
||||
db = await lancedb.connect_async(
|
||||
"db://dev",
|
||||
api_key="fake",
|
||||
host_override=host,
|
||||
client_config={"retry_config": {"retries": 0}},
|
||||
)
|
||||
registration = await db.create_function_async(normalize_score)
|
||||
created = await registration.wait()
|
||||
state["requests"].clear()
|
||||
functions = await db.list_functions()
|
||||
|
||||
assert functions == [created]
|
||||
assert [path for path, _ in state["requests"]] == [
|
||||
"/v1/functions/list",
|
||||
"/v1/functions/list",
|
||||
]
|
||||
|
||||
|
||||
def test_remote_drop_function_sends_exact_version():
|
||||
with _mock_remote_function_catalog() as (host, state):
|
||||
db = lancedb.connect(
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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():
|
||||
@@ -820,13 +795,11 @@ def test_table_create_indices():
|
||||
scalar_req = received_requests[0]
|
||||
assert "name" in scalar_req
|
||||
assert scalar_req["name"] == "custom_scalar_idx"
|
||||
assert scalar_req["replace"] is False
|
||||
|
||||
# Check FTS index request has custom name
|
||||
fts_req = received_requests[1]
|
||||
assert "name" in fts_req
|
||||
assert fts_req["name"] == "custom_fts_idx"
|
||||
assert fts_req["replace"] is False
|
||||
assert fts_req["block_size"] == 256
|
||||
assert fts_req["custom_stop_words"] == ["cloud"]
|
||||
|
||||
@@ -834,7 +807,6 @@ def test_table_create_indices():
|
||||
vector_req = received_requests[2]
|
||||
assert "name" in vector_req
|
||||
assert vector_req["name"] == "custom_vector_idx"
|
||||
assert "replace" not in vector_req
|
||||
|
||||
table.wait_for_index(["custom_scalar_idx"], timedelta(seconds=2))
|
||||
table.wait_for_index(
|
||||
@@ -1107,9 +1079,6 @@ def test_remote_create_index_new_api():
|
||||
table.create_index("text", config=FTS(block_size=256))
|
||||
# IvfRq via new API
|
||||
table.create_index("vector", config=IvfRq(distance_type="l2"))
|
||||
table.create_index(
|
||||
"vector", config=IvfPq(distance_type="l2"), replace=False
|
||||
)
|
||||
|
||||
# Legacy index_type="IVF_RQ" routes to IvfRq config under the hood.
|
||||
with pytest.warns(DeprecationWarning, match="create_index"):
|
||||
@@ -1119,17 +1088,15 @@ def test_remote_create_index_new_api():
|
||||
num_partitions=8,
|
||||
)
|
||||
|
||||
assert len(received_requests) == 6
|
||||
assert len(received_requests) == 5
|
||||
assert [req["column"] for req in received_requests] == [
|
||||
"vector",
|
||||
"category",
|
||||
"text",
|
||||
"vector",
|
||||
"vector",
|
||||
"vector",
|
||||
]
|
||||
assert received_requests[2]["block_size"] == 256
|
||||
assert received_requests[4]["replace"] is False
|
||||
|
||||
|
||||
def test_table_wait_for_index_timeout():
|
||||
|
||||
@@ -629,19 +629,6 @@ impl Connection {
|
||||
})
|
||||
}
|
||||
|
||||
pub fn list_functions(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.get_inner()?.clone();
|
||||
future_into_py(self_.py(), async move {
|
||||
inner
|
||||
.list_functions()
|
||||
.await
|
||||
.infer_error()?
|
||||
.into_iter()
|
||||
.map(|function| function.to_canonical_json().infer_error())
|
||||
.collect::<PyResult<Vec<_>>>()
|
||||
})
|
||||
}
|
||||
|
||||
pub fn drop_function(
|
||||
self_: PyRef<'_, Self>,
|
||||
name: String,
|
||||
|
||||
@@ -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,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
|
||||
|
||||
@@ -523,28 +523,6 @@ impl Connection {
|
||||
.await
|
||||
}
|
||||
|
||||
/// List every published immutable Function version in the remote catalog.
|
||||
///
|
||||
/// Results are ordered by Function name then version. The client walks all
|
||||
/// server pages before returning. Local databases return
|
||||
/// [`Error::NotSupported`].
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```no_run
|
||||
/// # async fn list_functions(
|
||||
/// # connection: &lancedb::Connection,
|
||||
/// # ) -> Result<(), Box<dyn std::error::Error>> {
|
||||
/// for function in connection.list_functions().await? {
|
||||
/// println!("{} {}", function.name(), function.version());
|
||||
/// }
|
||||
/// # Ok(())
|
||||
/// # }
|
||||
/// ```
|
||||
pub async fn list_functions(&self) -> Result<Vec<crate::function::FunctionVersion>> {
|
||||
self.internal.list_functions().await
|
||||
}
|
||||
|
||||
/// Drop one exact immutable Function version from the remote catalog.
|
||||
///
|
||||
/// Returns `true` when the server appended a Dropped transition and
|
||||
|
||||
@@ -307,10 +307,6 @@ pub trait Database:
|
||||
) -> Result<crate::function::FunctionVersion> {
|
||||
function_catalog_not_supported()
|
||||
}
|
||||
/// List every published immutable Function version in the remote catalog.
|
||||
async fn list_functions(&self) -> Result<Vec<crate::function::FunctionVersion>> {
|
||||
function_catalog_not_supported()
|
||||
}
|
||||
/// Drop one exact immutable Function version from the remote catalog.
|
||||
async fn drop_function(&self, _name: &str, _version: &str) -> Result<bool> {
|
||||
function_catalog_not_supported()
|
||||
|
||||
@@ -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]
|
||||
|
||||
+5
-835
@@ -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();
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
@@ -533,19 +533,6 @@ struct RemoteListJobsResponse {
|
||||
page_token: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
struct RemoteListedFunctionVersion {
|
||||
definition: FunctionVersion,
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
struct RemoteListFunctionsResponse {
|
||||
#[serde(default)]
|
||||
functions: Vec<RemoteListedFunctionVersion>,
|
||||
#[serde(default)]
|
||||
page_token: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
struct RemoteDropFunctionResponse {
|
||||
dropped: bool,
|
||||
@@ -601,43 +588,6 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
|
||||
response.json().await.err_to_http(request_id)
|
||||
}
|
||||
|
||||
async fn list_functions(&self) -> Result<Vec<FunctionVersion>> {
|
||||
let mut functions = Vec::new();
|
||||
let mut page_token: Option<String> = None;
|
||||
let mut seen_page_tokens = HashSet::new();
|
||||
loop {
|
||||
let mut body = serde_json::json!({ "include_definition": true });
|
||||
if let Some(token) = &page_token {
|
||||
body["page_token"] = serde_json::Value::String(token.clone());
|
||||
}
|
||||
let req = self.client.post("/v1/functions/list").json(&body);
|
||||
let (request_id, response) = self.client.send(req).await?;
|
||||
let response = self.client.check_response(&request_id, response).await?;
|
||||
let status = response.status();
|
||||
let response: RemoteListFunctionsResponse =
|
||||
response.json().await.err_to_http(request_id.clone())?;
|
||||
functions.extend(
|
||||
response
|
||||
.functions
|
||||
.into_iter()
|
||||
.map(|listed| listed.definition),
|
||||
);
|
||||
let Some(next_page_token) = response.page_token.filter(|token| !token.is_empty())
|
||||
else {
|
||||
break;
|
||||
};
|
||||
if !seen_page_tokens.insert(next_page_token.clone()) {
|
||||
return Err(Error::Http {
|
||||
source: "Function listing response repeated a page_token".into(),
|
||||
request_id,
|
||||
status_code: Some(status),
|
||||
});
|
||||
}
|
||||
page_token = Some(next_page_token);
|
||||
}
|
||||
Ok(functions)
|
||||
}
|
||||
|
||||
async fn drop_function(&self, name: &str, version: &str) -> Result<bool> {
|
||||
let req = self
|
||||
.client
|
||||
@@ -2758,119 +2708,6 @@ mod tests {
|
||||
assert_eq!(version.version(), "fv_01K3EXACT");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_list_functions_requests_definitions_and_paginates() {
|
||||
const VERSION: &str = include_str!(
|
||||
"../../tests/fixtures/first_class_functions/v1/remote_function_version.canonical.json"
|
||||
);
|
||||
let version: serde_json::Value = serde_json::from_str(VERSION).unwrap();
|
||||
let page = Arc::new(AtomicUsize::new(0));
|
||||
let conn = Connection::new_with_handler(move |request| {
|
||||
assert_eq!(request.method(), &reqwest::Method::POST);
|
||||
assert_eq!(request.url().path(), "/v1/functions/list");
|
||||
let body: serde_json::Value =
|
||||
serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
|
||||
assert_eq!(body["include_definition"], true);
|
||||
match page.fetch_add(1, Ordering::SeqCst) {
|
||||
0 => {
|
||||
assert!(body.get("page_token").is_none());
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"functions": [], "page_token": "next"}"#.to_string())
|
||||
.unwrap()
|
||||
}
|
||||
_ => {
|
||||
assert_eq!(body["page_token"], "next");
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(
|
||||
serde_json::json!({
|
||||
"functions": [{
|
||||
"name": "embed",
|
||||
"version": "fv_01K3EXACT",
|
||||
"definition": version.clone(),
|
||||
}],
|
||||
})
|
||||
.to_string(),
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
}
|
||||
});
|
||||
let functions = conn.list_functions().await.unwrap();
|
||||
assert_eq!(functions.len(), 1);
|
||||
assert_eq!(functions[0].name(), "embed");
|
||||
assert_eq!(functions[0].version(), "fv_01K3EXACT");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_list_functions_stops_on_an_empty_page_token() {
|
||||
let requests = Arc::new(AtomicUsize::new(0));
|
||||
let seen = requests.clone();
|
||||
let conn = Connection::new_with_handler(move |request| {
|
||||
seen.fetch_add(1, Ordering::SeqCst);
|
||||
let body: serde_json::Value =
|
||||
serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
|
||||
assert!(body.get("page_token").is_none());
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"functions": [], "page_token": ""}"#)
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
let functions = conn.list_functions().await.unwrap();
|
||||
assert!(functions.is_empty());
|
||||
assert_eq!(requests.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_list_functions_rejects_a_page_token_cycle() {
|
||||
let page = Arc::new(AtomicUsize::new(0));
|
||||
let requests = page.clone();
|
||||
let conn = Connection::new_with_handler(move |request| {
|
||||
let body: serde_json::Value =
|
||||
serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
|
||||
let next_page_token = match page.fetch_add(1, Ordering::SeqCst) {
|
||||
0 => {
|
||||
assert!(body.get("page_token").is_none());
|
||||
"one"
|
||||
}
|
||||
1 => {
|
||||
assert_eq!(body["page_token"], "one");
|
||||
"two"
|
||||
}
|
||||
2 => {
|
||||
assert_eq!(body["page_token"], "two");
|
||||
"one"
|
||||
}
|
||||
page => panic!("unexpected page: {page}"),
|
||||
};
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(
|
||||
serde_json::json!({
|
||||
"functions": [],
|
||||
"page_token": next_page_token,
|
||||
})
|
||||
.to_string(),
|
||||
)
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
let error = conn.list_functions().await.unwrap_err();
|
||||
assert!(
|
||||
matches!(
|
||||
&error,
|
||||
Error::Http {
|
||||
status_code: Some(http::StatusCode::OK),
|
||||
..
|
||||
}
|
||||
),
|
||||
"got {error:?}"
|
||||
);
|
||||
assert_eq!(requests.load(Ordering::SeqCst), 3);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_drop_function_sends_exact_version_and_decodes_replay() {
|
||||
let conn = Connection::new_with_handler(|request| {
|
||||
|
||||
@@ -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::{
|
||||
@@ -527,10 +527,6 @@ impl<S: HttpSend> RemoteTable<S> {
|
||||
"column": canonical_column
|
||||
});
|
||||
|
||||
if !index.replace {
|
||||
body["replace"] = false.into();
|
||||
}
|
||||
|
||||
// Add name parameter if provided (for backwards compatibility, only include if Some)
|
||||
if let Some(ref name) = index.name {
|
||||
body["name"] = serde_json::Value::String(name.clone());
|
||||
@@ -2026,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
|
||||
}
|
||||
@@ -2601,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();
|
||||
@@ -2626,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 {
|
||||
@@ -2684,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));
|
||||
@@ -2742,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));
|
||||
@@ -3742,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;
|
||||
@@ -5663,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 =
|
||||
@@ -6237,40 +6077,6 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_create_index_forwards_replace_false_on_existing_route() {
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
assert_eq!(request.method(), "POST");
|
||||
match request.url().path() {
|
||||
"/v1/table/my_table/describe/" => {
|
||||
let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(describe_response(&schema))
|
||||
.unwrap()
|
||||
}
|
||||
"/v1/table/my_table/create_index/" => {
|
||||
let body = request.body().unwrap().as_bytes().unwrap();
|
||||
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
|
||||
assert_eq!(body["replace"], json!(false));
|
||||
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body("{}".to_string())
|
||||
.unwrap()
|
||||
}
|
||||
path => panic!("Unexpected path: {}", path),
|
||||
}
|
||||
});
|
||||
|
||||
table
|
||||
.create_index(&["a"], Index::BTree(Default::default()))
|
||||
.replace(false)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_create_index_returns_job() {
|
||||
let describe_calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
|
||||
@@ -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
|
||||
/// ----------
|
||||
|
||||
@@ -589,13 +589,16 @@ fn canonical_input_arrow_type(field: &JsonArrowField) -> Result<String> {
|
||||
.and_then(|metadata| metadata.get(ARROW_EXT_NAME_KEY))
|
||||
.map(String::as_str)
|
||||
== Some(BLOB_V2_EXT_NAME);
|
||||
if is_blob_v2 || field.r#type.fields.is_some() {
|
||||
if is_blob_v2 {
|
||||
let arrow_field = lance_namespace::schema::convert_json_arrow_field(field)
|
||||
.map_err(|e| invalid_function(format!("invalid Function input field: {e}")))?;
|
||||
validate_function_blob_nesting(&arrow_field, false)?;
|
||||
if is_blob_v2 {
|
||||
return Ok(FUNCTION_BLOB_V2_TYPE.to_string());
|
||||
if !has_supported_blob_v2_layout(&arrow_field) {
|
||||
return Err(invalid_function(format!(
|
||||
"Function input '{}' has an invalid Blob v2 storage layout",
|
||||
arrow_field.name()
|
||||
)));
|
||||
}
|
||||
return Ok(FUNCTION_BLOB_V2_TYPE.to_string());
|
||||
}
|
||||
if field.r#type.fields.is_none() && field.r#type.length.is_none() {
|
||||
Ok(field.r#type.r#type.clone())
|
||||
@@ -614,34 +617,6 @@ fn has_supported_blob_v2_layout(field: &ArrowField) -> bool {
|
||||
)
|
||||
}
|
||||
|
||||
fn validate_function_blob_nesting(field: &ArrowField, inside_collection: bool) -> Result<()> {
|
||||
if field.is_blob_v2() {
|
||||
if inside_collection {
|
||||
return Err(invalid_function(format!(
|
||||
"Function field '{}' nests Blob v2 under a collection, which Function signatures do not support",
|
||||
field.name()
|
||||
)));
|
||||
}
|
||||
if !has_supported_blob_v2_layout(field) {
|
||||
return Err(invalid_function(format!(
|
||||
"Function field '{}' has an invalid Blob v2 storage layout",
|
||||
field.name()
|
||||
)));
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
match field.data_type() {
|
||||
DataType::Struct(fields) => fields
|
||||
.iter()
|
||||
.try_for_each(|field| validate_function_blob_nesting(field, inside_collection)),
|
||||
DataType::List(field)
|
||||
| DataType::LargeList(field)
|
||||
| DataType::FixedSizeList(field, _)
|
||||
| DataType::Map(field, _) => validate_function_blob_nesting(field, true),
|
||||
_ => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
/// `fixed_size_list<item, size>` -> (`item`, `size`); the comma must sit outside
|
||||
/// any nested `<...>`.
|
||||
fn split_fixed_size_list(raw: &str) -> Option<(&str, i32)> {
|
||||
@@ -722,22 +697,21 @@ fn parse_output_arrow_type(raw: &str) -> Result<JsonArrowDataType> {
|
||||
}
|
||||
|
||||
fn function_output_field(name: &str, nullable: bool, raw: &str) -> Result<JsonArrowField> {
|
||||
let field = if raw == FUNCTION_BLOB_V2_TYPE {
|
||||
lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![crate::blob(
|
||||
name, nullable,
|
||||
)]))
|
||||
if raw == FUNCTION_BLOB_V2_TYPE {
|
||||
return lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![
|
||||
crate::blob(name, nullable),
|
||||
]))
|
||||
.map_err(|e| invalid_function(format!("could not encode Blob v2 output field: {e}")))?
|
||||
.fields
|
||||
.into_iter()
|
||||
.next()
|
||||
.ok_or_else(|| invalid_function("Blob v2 output field is missing"))?
|
||||
} else {
|
||||
JsonArrowField::new(name.to_string(), nullable, parse_output_arrow_type(raw)?)
|
||||
};
|
||||
let arrow_field = lance_namespace::schema::convert_json_arrow_field(&field)
|
||||
.map_err(|e| invalid_function(format!("invalid Function output field: {e}")))?;
|
||||
validate_function_blob_nesting(&arrow_field, false)?;
|
||||
Ok(field)
|
||||
.ok_or_else(|| invalid_function("Blob v2 output field is missing"));
|
||||
}
|
||||
Ok(JsonArrowField::new(
|
||||
name.to_string(),
|
||||
nullable,
|
||||
parse_output_arrow_type(raw)?,
|
||||
))
|
||||
}
|
||||
|
||||
fn function_output_field_matches(expected: &ArrowField, actual: &ArrowField) -> bool {
|
||||
@@ -2745,28 +2719,6 @@ mod tests {
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn exact_arrow_type(field: ArrowField) -> String {
|
||||
let json =
|
||||
lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![field])).unwrap();
|
||||
serde_json::to_string(json.fields[0].r#type.as_ref()).unwrap()
|
||||
}
|
||||
|
||||
fn single_input_application(path: &str) -> FunctionApplication {
|
||||
FunctionApplication::from_json(
|
||||
&serde_json::json!({
|
||||
"function": {"name": "inspect", "version": "fv_nested_blob"},
|
||||
"inputs": [{
|
||||
"parameter": "value",
|
||||
"kind": "column",
|
||||
"value": {"path": path}
|
||||
}],
|
||||
"output": {"kind": "scalar", "arrow_type": "int64", "nullable": false}
|
||||
})
|
||||
.to_string(),
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn binding_from_plan(plan: &FunctionDeclarationPlan) -> FunctionBinding {
|
||||
let inputs = plan
|
||||
.input_bindings
|
||||
@@ -3206,128 +3158,6 @@ mod tests {
|
||||
assert_eq!(fields[1].data_type(), &DataType::Int32);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_struct_blob_input_preserves_exact_schema_and_nullability() {
|
||||
let payload = ArrowField::new(
|
||||
"payload",
|
||||
DataType::Struct(Fields::from(vec![
|
||||
ArrowField::new("mime_type", DataType::Utf8, false),
|
||||
ArrowField::new(
|
||||
"nested",
|
||||
DataType::Struct(Fields::from(vec![crate::blob("image", true)])),
|
||||
true,
|
||||
),
|
||||
])),
|
||||
true,
|
||||
);
|
||||
let plan = plan_function_application(
|
||||
&ArrowSchema::new(vec![payload]),
|
||||
&single_input_application("payload"),
|
||||
Some("size"),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let declared: JsonArrowDataType =
|
||||
serde_json::from_str(&plan.input_bindings[0].arrow_type).unwrap();
|
||||
let DataType::Struct(fields) =
|
||||
lance_namespace::schema::convert_json_arrow_type(&declared).unwrap()
|
||||
else {
|
||||
panic!("expected a struct Function input")
|
||||
};
|
||||
assert!(fields[1].is_nullable());
|
||||
let DataType::Struct(nested) = fields[1].data_type() else {
|
||||
panic!("expected a recursive struct Function input")
|
||||
};
|
||||
assert!(nested[0].is_blob_v2());
|
||||
assert!(nested[0].is_nullable());
|
||||
|
||||
let exact = lance_namespace::schema::convert_json_arrow_schema(&plan.input_schema).unwrap();
|
||||
let DataType::Struct(fields) = exact.field(0).data_type() else {
|
||||
panic!("expected exact input schema to retain the struct")
|
||||
};
|
||||
let DataType::Struct(nested) = fields[1].data_type() else {
|
||||
panic!("expected exact input schema to retain the nested struct")
|
||||
};
|
||||
assert!(nested[0].is_blob_v2());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_recursive_blob_result_plans_one_whole_named_struct_column() {
|
||||
let details_type = exact_arrow_type(ArrowField::new(
|
||||
"details",
|
||||
DataType::Struct(Fields::from(vec![crate::blob("image", true)])),
|
||||
false,
|
||||
));
|
||||
let application = FunctionApplication::from_json(
|
||||
&serde_json::json!({
|
||||
"function": {"name": "inspect", "version": "fv_nested_blob"},
|
||||
"inputs": [],
|
||||
"output": {
|
||||
"kind": "named_struct",
|
||||
"fields": [
|
||||
{"name": "mime_type", "arrow_type": "utf8", "nullable": false},
|
||||
{"name": "details", "arrow_type": details_type, "nullable": false}
|
||||
]
|
||||
}
|
||||
})
|
||||
.to_string(),
|
||||
)
|
||||
.unwrap();
|
||||
let plan = plan_function_application(&ArrowSchema::empty(), &application, Some("payload"))
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(plan.outputs.len(), 1);
|
||||
assert_eq!(plan.outputs[0].result_field, WHOLE_RESULT_FIELD);
|
||||
let schema =
|
||||
lance_namespace::schema::convert_json_arrow_schema(&plan.output_schema).unwrap();
|
||||
assert_eq!(schema.field(0).name(), "payload");
|
||||
let DataType::Struct(fields) = schema.field(0).data_type() else {
|
||||
panic!("whole named result must be one struct column")
|
||||
};
|
||||
assert_eq!(
|
||||
fields.iter().map(|field| field.name()).collect::<Vec<_>>(),
|
||||
["mime_type", "details"]
|
||||
);
|
||||
let DataType::Struct(details) = fields[1].data_type() else {
|
||||
panic!("expected recursive result struct")
|
||||
};
|
||||
assert!(details[0].is_blob_v2());
|
||||
assert!(!fields.iter().any(|field| field.name() == "payload"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_blob_children_under_collections_are_rejected() {
|
||||
let collections = vec![
|
||||
DataType::List(Arc::new(crate::blob("item", false))),
|
||||
DataType::LargeList(Arc::new(crate::blob("item", false))),
|
||||
DataType::FixedSizeList(Arc::new(crate::blob("item", false)), 2),
|
||||
DataType::Map(
|
||||
Arc::new(ArrowField::new(
|
||||
"entries",
|
||||
DataType::Struct(Fields::from(vec![
|
||||
ArrowField::new("key", DataType::Utf8, false),
|
||||
crate::blob("value", false),
|
||||
])),
|
||||
false,
|
||||
)),
|
||||
false,
|
||||
),
|
||||
];
|
||||
for data_type in collections {
|
||||
let schema = ArrowSchema::new(vec![ArrowField::new("value", data_type, false)]);
|
||||
let error = plan_function_application(
|
||||
&schema,
|
||||
&single_input_application("value"),
|
||||
Some("size"),
|
||||
)
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
error.to_string().contains("under a collection"),
|
||||
"got: {error}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_blob_whole_struct_binding_accepts_full_logical_layout() {
|
||||
let input = crate::blob("image", false);
|
||||
|
||||
@@ -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),
|
||||
|
||||
Reference in New Issue
Block a user