mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-01 19:18:38 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b0f010ad83 | |||
| d82fe8520f |
+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
+47
-48
@@ -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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-ipc",
|
||||
@@ -5305,9 +5304,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace-reqwest-client"
|
||||
version = "0.11.1"
|
||||
version = "0.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1d06b1fbb5d41f93bc652b61e2872af92e8a6c5f6b4ce8839a8ecfa05365d359"
|
||||
checksum = "0a030196da1c994b63a96a4f0bf5b0cfa459fe6dadc9e962320246ca328da22a"
|
||||
dependencies = [
|
||||
"reqwest 0.12.28",
|
||||
"serde",
|
||||
@@ -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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.5#556637791d0048c2b4f1342dd84b67c8bbd65259"
|
||||
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.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=12.0.0-beta.5", default-features = false, "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=12.0.0-beta.5", "tag" = "v12.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" }
|
||||
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>
|
||||
```
|
||||
|
||||
|
||||
@@ -50,16 +50,6 @@ projections: [string, string][];
|
||||
|
||||
***
|
||||
|
||||
### sourceNamespace
|
||||
|
||||
```ts
|
||||
sourceNamespace: string[];
|
||||
```
|
||||
|
||||
Namespace holding the source table; empty is the root namespace.
|
||||
|
||||
***
|
||||
|
||||
### sourceTable
|
||||
|
||||
```ts
|
||||
|
||||
@@ -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.5</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
|
||||
|
||||
@@ -48,28 +48,6 @@ describe("materialized views", () => {
|
||||
expect(definitionFromMetadata(safe, "v").limit).toBe(42);
|
||||
});
|
||||
|
||||
it("reads the namespaced select kind and refuses unknown kinds", () => {
|
||||
// "namespaced_select" is the namespaced form of "select": same shape, a
|
||||
// separate kind so readers that predate it refuse instead of resolving
|
||||
// the source at the root.
|
||||
const namespaced = new Map([
|
||||
[
|
||||
DEFINITION_META_KEY,
|
||||
'{"kind":"namespaced_select","source_table":"people","source_namespace":["ns"]}',
|
||||
],
|
||||
]);
|
||||
const definition = definitionFromMetadata(namespaced, "v");
|
||||
expect(definition.sourceTable).toBe("people");
|
||||
expect(definition.sourceNamespace).toEqual(["ns"]);
|
||||
|
||||
const unknown = new Map([
|
||||
[DEFINITION_META_KEY, '{"kind":"select_v3","source_table":"people"}'],
|
||||
]);
|
||||
expect(() => definitionFromMetadata(unknown, "v")).toThrow(
|
||||
/cannot refresh/,
|
||||
);
|
||||
});
|
||||
|
||||
it("creates, refreshes and queries a view", async () => {
|
||||
const view = await db.createMaterializedView("adults", "people", {
|
||||
select: ["name", ["shout", "upper(name)"]],
|
||||
|
||||
@@ -987,6 +987,22 @@ describe("remote connection jobs surface", () => {
|
||||
res
|
||||
.writeHead(200, { "Content-Type": "application/json" })
|
||||
.end('{"job_id": "job-1"}');
|
||||
} else if (req.url === "/v1/jobs/pause") {
|
||||
if (payload["job_id"] !== "job-1") {
|
||||
res.writeHead(404).end("no such job");
|
||||
return;
|
||||
}
|
||||
res
|
||||
.writeHead(200, { "Content-Type": "application/json" })
|
||||
.end('{"job_id": "job-1", "paused": true}');
|
||||
} else if (req.url === "/v1/jobs/resume") {
|
||||
if (payload["job_id"] !== "job-1") {
|
||||
res.writeHead(404).end("no such job");
|
||||
return;
|
||||
}
|
||||
res
|
||||
.writeHead(200, { "Content-Type": "application/json" })
|
||||
.end('{"job_id": "job-1", "resumed": false, "still_pausing": true}');
|
||||
} else if (req.url === "/v1/jobs/query_events") {
|
||||
res
|
||||
.writeHead(200, {
|
||||
@@ -1015,6 +1031,9 @@ describe("remote connection jobs surface", () => {
|
||||
expect(await db.cancelJob("job-1")).toBe(true);
|
||||
expect(await db.cancelJob("missing")).toBe(false);
|
||||
|
||||
expect(await db.pauseJob("job-1")).toEqual("pausing");
|
||||
expect(await db.resumeJob("job-1")).toEqual("still_pausing");
|
||||
|
||||
const history = await db.jobHistory("job-1");
|
||||
expect(history.numRows).toEqual(2);
|
||||
|
||||
|
||||
@@ -583,6 +583,24 @@ export abstract class Connection {
|
||||
*/
|
||||
abstract cancelJob(jobId: string): Promise<boolean>;
|
||||
|
||||
/**
|
||||
* Pause a server-side job by id.
|
||||
*
|
||||
* The job's workers drain and it stays parked until resumed. Resolves to
|
||||
* "pausing", "already_paused", or "committing" -- a job finalizing its
|
||||
* results cannot be parked; retry shortly.
|
||||
*/
|
||||
abstract pauseJob(jobId: string): Promise<string>;
|
||||
|
||||
/**
|
||||
* Resume a paused server-side job by id.
|
||||
*
|
||||
* Its workers pick their work back up from checkpoints. Resolves to
|
||||
* "resumed", "still_pausing" -- the pause's worker drain is not confirmed
|
||||
* yet; retry shortly -- or "not_paused".
|
||||
*/
|
||||
abstract resumeJob(jobId: string): Promise<string>;
|
||||
|
||||
/**
|
||||
* The lifecycle event history of a server-side job, as an Arrow table.
|
||||
*
|
||||
@@ -944,6 +962,14 @@ export class LocalConnection extends Connection {
|
||||
return this.inner.cancelJob(jobId);
|
||||
}
|
||||
|
||||
async pauseJob(jobId: string): Promise<string> {
|
||||
return this.inner.pauseJob(jobId);
|
||||
}
|
||||
|
||||
async resumeJob(jobId: string): Promise<string> {
|
||||
return this.inner.resumeJob(jobId);
|
||||
}
|
||||
|
||||
async jobHistory(jobId?: string): Promise<ArrowTable> {
|
||||
const buf = await this.inner.jobHistory(jobId);
|
||||
if (buf.length === 0) {
|
||||
|
||||
@@ -19,8 +19,6 @@ export interface MaterializedViewDefinition {
|
||||
limit?: number;
|
||||
/** Source columns the projections and filter read. */
|
||||
inputs: string[];
|
||||
/** Namespace holding the source table; empty is the root namespace. */
|
||||
sourceNamespace: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -80,8 +78,7 @@ export function definitionFromMetadata(
|
||||
}
|
||||
// biome-ignore lint/suspicious/noExplicitAny: raw JSON
|
||||
const value: any = JSON.parse(raw);
|
||||
// "namespaced_select" keeps older readers from resolving the source at root.
|
||||
if (value.kind !== "select" && value.kind !== "namespaced_select") {
|
||||
if (value.kind !== "select") {
|
||||
throw new Error(
|
||||
`materialized view '${name}' is defined by '${value.kind}', which this ` +
|
||||
"version of lancedb cannot refresh",
|
||||
@@ -106,7 +103,6 @@ export function definitionFromMetadata(
|
||||
filter: value.filter ?? undefined,
|
||||
limit,
|
||||
inputs: value.inputs ?? [],
|
||||
sourceNamespace: value.source_namespace ?? [],
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -1,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",
|
||||
|
||||
@@ -477,6 +477,34 @@ impl Connection {
|
||||
self.get_inner()?.cancel_job(&job_id).await.default_error()
|
||||
}
|
||||
|
||||
/// Pause a server-side job by id: its workers drain and it stays parked
|
||||
/// until resumed. Returns "pausing", "already_paused", or "committing".
|
||||
#[napi(catch_unwind)]
|
||||
pub async fn pause_job(&self, job_id: String) -> napi::Result<String> {
|
||||
let status = self.get_inner()?.pause_job(&job_id).await.default_error()?;
|
||||
Ok(match status {
|
||||
lancedb::database::PauseJobStatus::Pausing => "pausing".to_string(),
|
||||
lancedb::database::PauseJobStatus::AlreadyPaused => "already_paused".to_string(),
|
||||
lancedb::database::PauseJobStatus::Committing => "committing".to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Resume a paused server-side job by id. Returns "resumed",
|
||||
/// "still_pausing", or "not_paused".
|
||||
#[napi(catch_unwind)]
|
||||
pub async fn resume_job(&self, job_id: String) -> napi::Result<String> {
|
||||
let status = self
|
||||
.get_inner()?
|
||||
.resume_job(&job_id)
|
||||
.await
|
||||
.default_error()?;
|
||||
Ok(match status {
|
||||
lancedb::database::ResumeJobStatus::Resumed => "resumed".to_string(),
|
||||
lancedb::database::ResumeJobStatus::StillPausing => "still_pausing".to_string(),
|
||||
lancedb::database::ResumeJobStatus::NotPaused => "not_paused".to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
/// The lifecycle event history of a server-side job (all jobs when
|
||||
/// `job_id` is null), as an Arrow IPC stream buffer. Empty when there is
|
||||
/// no history.
|
||||
|
||||
+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,11 +150,12 @@ 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]: ...
|
||||
async def cancel_job(self, job_id: str) -> bool: ...
|
||||
async def pause_job(self, job_id: str) -> str: ...
|
||||
async def resume_job(self, job_id: str) -> str: ...
|
||||
async def job_history(
|
||||
self, job_id: Optional[str] = None
|
||||
) -> List[pa.RecordBatch]: ...
|
||||
|
||||
+53
-33
@@ -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.
|
||||
|
||||
@@ -771,6 +753,26 @@ class DBConnection(EnforceOverrides):
|
||||
"cancel_job is not supported for this connection type"
|
||||
)
|
||||
|
||||
def pause_job(self, job_id: str) -> str:
|
||||
"""Pause a server-side job by id.
|
||||
|
||||
The job's workers drain and it stays parked until resumed. Returns
|
||||
"pausing", "already_paused", or "committing" -- a job finalizing its
|
||||
results cannot be parked; retry shortly.
|
||||
"""
|
||||
raise NotImplementedError("pause_job is not supported for this connection type")
|
||||
|
||||
def resume_job(self, job_id: str) -> str:
|
||||
"""Resume a paused server-side job by id.
|
||||
|
||||
Its workers pick their work back up from checkpoints. Returns
|
||||
"resumed", "still_pausing" -- the pause's worker drain is not
|
||||
confirmed yet; retry shortly -- or "not_paused".
|
||||
"""
|
||||
raise NotImplementedError(
|
||||
"resume_job is not supported for this connection type"
|
||||
)
|
||||
|
||||
def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
|
||||
"""The lifecycle event history of a server-side job, as Arrow batches.
|
||||
|
||||
@@ -1441,10 +1443,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))
|
||||
@@ -1472,6 +1470,22 @@ class LanceDBConnection(DBConnection):
|
||||
"""
|
||||
return LOOP.run(self._conn.cancel_job(job_id))
|
||||
|
||||
@override
|
||||
def pause_job(self, job_id: str) -> str:
|
||||
"""Pause a server-side job by id.
|
||||
|
||||
Returns "pausing", "already_paused", or "committing".
|
||||
"""
|
||||
return LOOP.run(self._conn.pause_job(job_id))
|
||||
|
||||
@override
|
||||
def resume_job(self, job_id: str) -> str:
|
||||
"""Resume a paused server-side job by id.
|
||||
|
||||
Returns "resumed", "still_pausing", or "not_paused".
|
||||
"""
|
||||
return LOOP.run(self._conn.resume_job(job_id))
|
||||
|
||||
@override
|
||||
def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
|
||||
"""The lifecycle event history of a server-side job, as Arrow batches.
|
||||
@@ -2279,17 +2293,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)
|
||||
@@ -2314,6 +2317,23 @@ class AsyncConnection(object):
|
||||
"""
|
||||
return await self._inner.cancel_job(job_id)
|
||||
|
||||
async def pause_job(self, job_id: str) -> str:
|
||||
"""Pause a server-side job by id.
|
||||
|
||||
The job's workers drain and it stays parked until resumed. Returns
|
||||
"pausing", "already_paused", or "committing" -- a job finalizing its
|
||||
results cannot be parked; retry shortly.
|
||||
"""
|
||||
return await self._inner.pause_job(job_id)
|
||||
|
||||
async def resume_job(self, job_id: str) -> str:
|
||||
"""Resume a paused server-side job by id.
|
||||
|
||||
Its workers pick their work back up from checkpoints. Returns
|
||||
"resumed", "still_pausing" -- retry shortly -- or "not_paused".
|
||||
"""
|
||||
return await self._inner.resume_job(job_id)
|
||||
|
||||
async def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
|
||||
"""The lifecycle event history of a server-side job, as Arrow batches.
|
||||
|
||||
|
||||
@@ -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}")
|
||||
|
||||
|
||||
|
||||
@@ -42,8 +42,6 @@ class MaterializedViewDefinition:
|
||||
"""Cap on the number of rows the view holds."""
|
||||
inputs: List[str] = field(default_factory=list)
|
||||
"""Source columns the projections and filter read."""
|
||||
source_namespace: List[str] = field(default_factory=list)
|
||||
"""Namespace holding the source table; empty is the root namespace."""
|
||||
|
||||
|
||||
def _definition_from_schema(
|
||||
@@ -55,8 +53,7 @@ def _definition_from_schema(
|
||||
raise ValueError(f"Table '{name}' is not a materialized view")
|
||||
value = json.loads(raw)
|
||||
kind = value.get("kind")
|
||||
# "namespaced_select" keeps older readers from resolving the source at root.
|
||||
if kind not in ("select", "namespaced_select"):
|
||||
if kind != "select":
|
||||
raise NotImplementedError(
|
||||
f"materialized view '{name}' is defined by '{kind}', which this "
|
||||
"version of lancedb cannot refresh"
|
||||
@@ -69,7 +66,6 @@ def _definition_from_schema(
|
||||
filter=value.get("filter"),
|
||||
limit=value.get("limit"),
|
||||
inputs=value.get("inputs", []),
|
||||
source_namespace=value.get("source_namespace", []),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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))
|
||||
@@ -780,6 +776,22 @@ class RemoteDBConnection(DBConnection):
|
||||
"""
|
||||
return LOOP.run(self._conn.cancel_job(job_id))
|
||||
|
||||
@override
|
||||
def pause_job(self, job_id: str) -> str:
|
||||
"""Pause a server-side job by id.
|
||||
|
||||
Returns "pausing", "already_paused", or "committing".
|
||||
"""
|
||||
return LOOP.run(self._conn.pause_job(job_id))
|
||||
|
||||
@override
|
||||
def resume_job(self, job_id: str) -> str:
|
||||
"""Resume a paused server-side job by id.
|
||||
|
||||
Returns "resumed", "still_pausing", or "not_paused".
|
||||
"""
|
||||
return LOOP.run(self._conn.resume_job(job_id))
|
||||
|
||||
@override
|
||||
def job_history(self, job_id: Optional[str] = None) -> List[pa.RecordBatch]:
|
||||
"""The lifecycle event history of a server-side job, as Arrow batches.
|
||||
|
||||
@@ -548,7 +548,6 @@ class RemoteTable(Table):
|
||||
LOOP.run(
|
||||
self._table.create_index(
|
||||
column,
|
||||
replace=replace,
|
||||
config=config,
|
||||
wait_timeout=wait_timeout,
|
||||
name=name,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -266,38 +266,3 @@ async def test_async_namespace_connection_materialized_views(tmp_path):
|
||||
handle._route_pushdown_to_rust == through_namespace._route_pushdown_to_rust
|
||||
)
|
||||
assert handle._namespace_path == through_namespace._namespace_path
|
||||
|
||||
|
||||
def test_namespaced_select_kind_is_read_and_unknown_kinds_are_refused():
|
||||
import json
|
||||
|
||||
import pyarrow as pa
|
||||
|
||||
from lancedb.materialized_view import _definition_from_schema
|
||||
|
||||
def schema_with(definition: dict) -> pa.Schema:
|
||||
return pa.schema([pa.field("id", pa.int32())]).with_metadata(
|
||||
{b"mv.definition": json.dumps(definition).encode()}
|
||||
)
|
||||
|
||||
# "namespaced_select" is the namespaced form of "select": same shape,
|
||||
# a separate kind so readers that predate it refuse instead of
|
||||
# resolving the source at the root.
|
||||
definition = _definition_from_schema(
|
||||
schema_with(
|
||||
{
|
||||
"kind": "namespaced_select",
|
||||
"source_table": "people",
|
||||
"source_namespace": ["ns"],
|
||||
"projections": [{"output": "name", "expression": "name"}],
|
||||
}
|
||||
),
|
||||
"v",
|
||||
)
|
||||
assert definition.source_table == "people"
|
||||
assert definition.source_namespace == ["ns"]
|
||||
|
||||
with pytest.raises(NotImplementedError, match="cannot refresh"):
|
||||
_definition_from_schema(
|
||||
schema_with({"kind": "select_v3", "source_table": "people"}), "v"
|
||||
)
|
||||
|
||||
@@ -820,13 +820,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 +832,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 +1104,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 +1113,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():
|
||||
@@ -2542,6 +2534,26 @@ def test_remote_connection_jobs_surface():
|
||||
request.send_header("Content-Type", "application/json")
|
||||
request.end_headers()
|
||||
request.wfile.write(b'{"job_id": "job-1"}')
|
||||
elif request.path == "/v1/jobs/pause":
|
||||
if payload["job_id"] != "job-1":
|
||||
request.send_response(404)
|
||||
request.end_headers()
|
||||
return
|
||||
request.send_response(200)
|
||||
request.send_header("Content-Type", "application/json")
|
||||
request.end_headers()
|
||||
request.wfile.write(b'{"job_id": "job-1", "paused": true}')
|
||||
elif request.path == "/v1/jobs/resume":
|
||||
if payload["job_id"] != "job-1":
|
||||
request.send_response(404)
|
||||
request.end_headers()
|
||||
return
|
||||
request.send_response(200)
|
||||
request.send_header("Content-Type", "application/json")
|
||||
request.end_headers()
|
||||
request.wfile.write(
|
||||
b'{"job_id": "job-1", "resumed": false, "still_pausing": true}'
|
||||
)
|
||||
elif request.path == "/v1/jobs/query_events":
|
||||
assert payload["job_id"] == "job-1"
|
||||
request.send_response(200)
|
||||
@@ -2570,6 +2582,9 @@ def test_remote_connection_jobs_surface():
|
||||
assert db.cancel_job("job-1") is True
|
||||
assert db.cancel_job("missing") is False
|
||||
|
||||
assert db.pause_job("job-1") == "pausing"
|
||||
assert db.resume_job("job-1") == "still_pausing"
|
||||
|
||||
batches = db.job_history("job-1")
|
||||
assert len(batches) == 1
|
||||
assert batches[0].num_rows == 2
|
||||
|
||||
+24
-13
@@ -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,
|
||||
@@ -679,6 +666,30 @@ impl Connection {
|
||||
})
|
||||
}
|
||||
|
||||
pub fn pause_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.get_inner()?.clone();
|
||||
future_into_py(self_.py(), async move {
|
||||
let status = inner.pause_job(&job_id).await.infer_error()?;
|
||||
Ok(match status {
|
||||
lancedb::database::PauseJobStatus::Pausing => "pausing",
|
||||
lancedb::database::PauseJobStatus::AlreadyPaused => "already_paused",
|
||||
lancedb::database::PauseJobStatus::Committing => "committing",
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
pub fn resume_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.get_inner()?.clone();
|
||||
future_into_py(self_.py(), async move {
|
||||
let status = inner.resume_job(&job_id).await.infer_error()?;
|
||||
Ok(match status {
|
||||
lancedb::database::ResumeJobStatus::Resumed => "resumed",
|
||||
lancedb::database::ResumeJobStatus::StillPausing => "still_pausing",
|
||||
lancedb::database::ResumeJobStatus::NotPaused => "not_paused",
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
#[pyo3(signature = (job_id=None))]
|
||||
pub fn job_history(
|
||||
self_: PyRef<'_, Self>,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -24,7 +24,7 @@ use crate::data::scannable::Scannable;
|
||||
use crate::database::listing::ListingDatabase;
|
||||
use crate::database::{
|
||||
CloneTableRequest, Database, DatabaseOptions, JobDescription, JobInfo, OpenTableRequest,
|
||||
ReadConsistency, TableNamesRequest,
|
||||
PauseJobStatus, ReadConsistency, ResumeJobStatus, TableNamesRequest,
|
||||
};
|
||||
use crate::embeddings::{EmbeddingRegistry, MemoryRegistry};
|
||||
use crate::error::{Error, Result};
|
||||
@@ -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
|
||||
@@ -612,6 +590,18 @@ impl Connection {
|
||||
self.internal.cancel_job(job_id.as_ref()).await
|
||||
}
|
||||
|
||||
/// Pause a server-side job by id. Its workers drain and it stays parked
|
||||
/// until resumed; see [`PauseJobStatus`] for the outcomes.
|
||||
pub async fn pause_job(&self, job_id: impl AsRef<str>) -> Result<PauseJobStatus> {
|
||||
self.internal.pause_job(job_id.as_ref()).await
|
||||
}
|
||||
|
||||
/// Resume a paused server-side job by id. Its workers pick their work
|
||||
/// back up from checkpoints; see [`ResumeJobStatus`] for the outcomes.
|
||||
pub async fn resume_job(&self, job_id: impl AsRef<str>) -> Result<ResumeJobStatus> {
|
||||
self.internal.resume_job(job_id.as_ref()).await
|
||||
}
|
||||
|
||||
/// The lifecycle event history of a server-side job (all jobs when
|
||||
/// `job_id` is `None`), as recorded Arrow batches.
|
||||
pub async fn job_history(&self, job_id: Option<&str>) -> Result<Vec<RecordBatch>> {
|
||||
|
||||
@@ -235,6 +235,29 @@ pub struct JobDescription {
|
||||
pub failure: Option<crate::error::JobFailure>,
|
||||
}
|
||||
|
||||
/// The server's answer to a pause request.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum PauseJobStatus {
|
||||
/// The pause was accepted; workers drain and the job stays parked.
|
||||
Pausing,
|
||||
/// The job was already paused, so a repeated pause changed nothing.
|
||||
AlreadyPaused,
|
||||
/// The job is finalizing its results and cannot be parked right now.
|
||||
/// The commit is the short tail of a long job; retry shortly.
|
||||
Committing,
|
||||
}
|
||||
|
||||
/// The server's answer to a resume request.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum ResumeJobStatus {
|
||||
/// The job re-entered the queue and will run again.
|
||||
Resumed,
|
||||
/// The pause's worker drain is not confirmed yet; retry shortly.
|
||||
StillPausing,
|
||||
/// The job was not paused, so there was nothing to resume.
|
||||
NotPaused,
|
||||
}
|
||||
|
||||
fn job_op_not_supported<T>(what: &str) -> Result<T> {
|
||||
Err(crate::error::Error::NotSupported {
|
||||
message: format!("{} is not supported by this database", what),
|
||||
@@ -307,10 +330,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()
|
||||
@@ -335,6 +354,16 @@ pub trait Database:
|
||||
async fn cancel_job(&self, _job_id: &str) -> Result<bool> {
|
||||
job_op_not_supported("cancel_job")
|
||||
}
|
||||
/// Pause a job by id. The job's workers drain and it stays parked until
|
||||
/// resumed; see [`PauseJobStatus`] for the outcomes.
|
||||
async fn pause_job(&self, _job_id: &str) -> Result<PauseJobStatus> {
|
||||
job_op_not_supported("pause_job")
|
||||
}
|
||||
/// Resume a paused job by id. It re-enters the queue and its workers pick
|
||||
/// their work back up from checkpoints; see [`ResumeJobStatus`].
|
||||
async fn resume_job(&self, _job_id: &str) -> Result<ResumeJobStatus> {
|
||||
job_op_not_supported("resume_job")
|
||||
}
|
||||
/// The lifecycle event history of a job (all jobs when `job_id` is
|
||||
/// `None`), as recorded Arrow batches.
|
||||
async fn job_history(&self, _job_id: Option<&str>) -> Result<Vec<RecordBatch>> {
|
||||
|
||||
@@ -74,15 +74,8 @@ const EMBEDDING_FUNCTIONS_META_KEY: &str = "embedding_functions";
|
||||
const COLUMN_DEFINITIONS_META_KEY: &str = "lancedb::column_definitions";
|
||||
|
||||
/// Value of the definition's `kind` tag for the projected `select` form.
|
||||
/// Reserved for root-namespace sources; see [`NAMESPACED_SELECT_KIND`].
|
||||
pub const SELECT_KIND: &str = "select";
|
||||
|
||||
/// The `select` form over a namespaced source: its own kind, because released
|
||||
/// readers drop unknown fields and resolve a `select` source at the root, so
|
||||
/// this routes them to the [`MaterializedViewKind::Unrecognized`] refusal
|
||||
/// instead of a wrong-table refresh.
|
||||
pub const NAMESPACED_SELECT_KIND: &str = "namespaced_select";
|
||||
|
||||
/// Which view outputs each source column is projected to directly. A column
|
||||
/// may be projected more than once, so each carries every name the view gives
|
||||
/// it, in projection order.
|
||||
@@ -102,10 +95,6 @@ pub struct ViewProjection {
|
||||
pub struct MaterializedViewDefinition {
|
||||
/// Name of the source table, in the same database as the view.
|
||||
pub source_table: String,
|
||||
/// Namespace path holding the source table; empty is the root namespace.
|
||||
/// A definition written before namespaced sources reads as root.
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub source_namespace: Vec<String>,
|
||||
/// The projected output columns, in view schema order.
|
||||
pub projections: Vec<ViewProjection>,
|
||||
/// SQL predicate selecting the source rows the view holds.
|
||||
@@ -140,12 +129,7 @@ pub(crate) fn definition_to_metadata(definition: &MaterializedViewDefinition) ->
|
||||
let mut value = serde_json::to_value(definition).map_err(|e| Error::Runtime {
|
||||
message: format!("failed to serialize view definition: {e}"),
|
||||
})?;
|
||||
let kind = if definition.source_namespace.is_empty() {
|
||||
SELECT_KIND
|
||||
} else {
|
||||
NAMESPACED_SELECT_KIND
|
||||
};
|
||||
value["kind"] = serde_json::Value::String(kind.to_string());
|
||||
value["kind"] = serde_json::Value::String(SELECT_KIND.to_string());
|
||||
Ok(value.to_string())
|
||||
}
|
||||
|
||||
@@ -166,21 +150,12 @@ pub fn materialized_view_kind(
|
||||
.get("kind")
|
||||
.and_then(|k| k.as_str())
|
||||
.ok_or_else(|| unreadable(&"missing kind tag"))?;
|
||||
if kind != SELECT_KIND && kind != NAMESPACED_SELECT_KIND {
|
||||
if kind != SELECT_KIND {
|
||||
return Ok(Some(MaterializedViewKind::Unrecognized {
|
||||
kind: kind.to_string(),
|
||||
}));
|
||||
}
|
||||
let kind = kind.to_string();
|
||||
let definition: MaterializedViewDefinition =
|
||||
serde_json::from_value(value).map_err(|e| unreadable(&e))?;
|
||||
// No correct writer produces a kind that disagrees with its namespace.
|
||||
if (kind == SELECT_KIND) != definition.source_namespace.is_empty() {
|
||||
return Err(unreadable(&format!(
|
||||
"kind '{kind}' does not match its source namespace {:?}",
|
||||
definition.source_namespace
|
||||
)));
|
||||
}
|
||||
let definition = serde_json::from_value(value).map_err(|e| unreadable(&e))?;
|
||||
Ok(Some(MaterializedViewKind::Select(definition)))
|
||||
}
|
||||
|
||||
@@ -191,7 +166,6 @@ pub fn materialized_view_kind(
|
||||
pub(crate) fn plan(
|
||||
source_schema: SchemaRef,
|
||||
source_table: &str,
|
||||
source_namespace: &[String],
|
||||
projections: &[(String, String)],
|
||||
filter: Option<&str>,
|
||||
limit: Option<u64>,
|
||||
@@ -345,7 +319,6 @@ pub(crate) fn plan(
|
||||
|
||||
let definition = MaterializedViewDefinition {
|
||||
source_table: source_table.to_string(),
|
||||
source_namespace: source_namespace.to_vec(),
|
||||
projections: projections
|
||||
.into_iter()
|
||||
.map(|(output, expression)| ViewProjection { output, expression })
|
||||
@@ -629,7 +602,7 @@ pub struct PreparedDeclaration {
|
||||
definition: MaterializedViewDefinition,
|
||||
/// The source's own database: the only place
|
||||
/// [`PreparedDeclaration::create`] will put the view, because refresh
|
||||
/// resolves the recorded source coordinate through the view's database.
|
||||
/// resolves the recorded source name through the view's database.
|
||||
database: Arc<dyn Database>,
|
||||
}
|
||||
|
||||
@@ -649,21 +622,10 @@ impl PreparedDeclaration {
|
||||
|
||||
/// Create the view table and verify it, consuming the declaration.
|
||||
///
|
||||
/// The view goes at the root of the source's own database, where refresh
|
||||
/// resolves the recorded source coordinate. Stable row ids are requested
|
||||
/// at both levels and verified rather than trusted; nothing is rolled
|
||||
/// back on failure.
|
||||
/// The view goes in the source's own database, where refresh resolves the
|
||||
/// recorded source name. Stable row ids are requested at both levels and
|
||||
/// verified rather than trusted; nothing is rolled back on failure.
|
||||
pub async fn create(self, name: &str) -> Result<MaterializedView> {
|
||||
self.create_in(&[], name).await
|
||||
}
|
||||
|
||||
/// Create the view in `namespace_path`, empty for the root namespace.
|
||||
/// Otherwise [`PreparedDeclaration::create`].
|
||||
pub async fn create_in(
|
||||
self,
|
||||
namespace_path: &[String],
|
||||
name: &str,
|
||||
) -> Result<MaterializedView> {
|
||||
let empty: Vec<std::result::Result<arrow_array::RecordBatch, arrow_schema::ArrowError>> =
|
||||
vec![];
|
||||
// Minted here, not at preparation: a declaration can be cloned and
|
||||
@@ -678,7 +640,6 @@ impl PreparedDeclaration {
|
||||
let reader: Box<dyn arrow_array::RecordBatchReader + Send> =
|
||||
Box::new(arrow_array::RecordBatchIterator::new(empty, schema));
|
||||
let mut request = CreateTableRequest::new(name.to_string(), Box::new(reader));
|
||||
request.namespace_path = namespace_path.to_vec();
|
||||
let write_params = request
|
||||
.write_options
|
||||
.lance_write_params
|
||||
@@ -719,8 +680,8 @@ impl PreparedDeclaration {
|
||||
|
||||
/// Validate a view declaration against its live source and hold what its
|
||||
/// creation needs. The declaration is canonicalized through the coordinate a
|
||||
/// refresh will resolve -- name and namespace both -- so a handle that does
|
||||
/// not resolve back to itself is rejected. Same creation-time checks as
|
||||
/// refresh will resolve, so a handle that does not resolve back to itself is
|
||||
/// rejected, as is a namespaced source. Same creation-time checks as
|
||||
/// [`Connection::create_materialized_view`].
|
||||
///
|
||||
/// ```no_run
|
||||
@@ -749,9 +710,17 @@ pub async fn prepare_declaration(
|
||||
message: "materialized views are supported only on local databases".into(),
|
||||
});
|
||||
};
|
||||
// Refresh resolves the source at exactly this coordinate, so the
|
||||
// definition records the namespace alongside the name.
|
||||
let source_namespace = source.namespace().to_vec();
|
||||
// The definition records the source by bare name; any other source
|
||||
// form would be recorded as a name its refresh cannot resolve.
|
||||
if !source.namespace().is_empty() {
|
||||
return Err(Error::NotSupported {
|
||||
message: format!(
|
||||
"a namespaced source cannot be recorded in a view definition; \
|
||||
'{}' must be a root-namespace table",
|
||||
source.name()
|
||||
),
|
||||
});
|
||||
}
|
||||
let database = source
|
||||
.database_opt()
|
||||
.ok_or_else(|| Error::InvalidInput {
|
||||
@@ -765,7 +734,7 @@ pub async fn prepare_declaration(
|
||||
let resolved = database
|
||||
.open_table(OpenTableRequest {
|
||||
name: source.name().to_string(),
|
||||
namespace_path: source_namespace.clone(),
|
||||
namespace_path: vec![],
|
||||
index_cache_size: None,
|
||||
lance_read_params: None,
|
||||
location: None,
|
||||
@@ -811,7 +780,6 @@ pub async fn prepare_declaration(
|
||||
let (definition, mut fields, lineage) = plan(
|
||||
source_schema.clone(),
|
||||
resolved.name(),
|
||||
&source_namespace,
|
||||
projections,
|
||||
filter,
|
||||
limit,
|
||||
@@ -871,9 +839,7 @@ fn ensure_local(connection: &Connection) -> Result<()> {
|
||||
pub struct CreateMaterializedViewBuilder {
|
||||
connection: Connection,
|
||||
name: String,
|
||||
namespace: Vec<String>,
|
||||
source: String,
|
||||
source_namespace: Vec<String>,
|
||||
projections: Vec<(String, String)>,
|
||||
filter: Option<String>,
|
||||
limit: Option<u64>,
|
||||
@@ -884,28 +850,13 @@ impl CreateMaterializedViewBuilder {
|
||||
Self {
|
||||
connection,
|
||||
name,
|
||||
namespace: Vec::new(),
|
||||
source,
|
||||
source_namespace: Vec::new(),
|
||||
projections: Vec::new(),
|
||||
filter: None,
|
||||
limit: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// The namespace to create the view in. Defaults to the root namespace.
|
||||
pub fn namespace(mut self, namespace_path: Vec<String>) -> Self {
|
||||
self.namespace = namespace_path;
|
||||
self
|
||||
}
|
||||
|
||||
/// The namespace holding the source table; recorded in the definition
|
||||
/// for refresh to resolve. Defaults to the root namespace.
|
||||
pub fn source_namespace(mut self, namespace_path: Vec<String>) -> Self {
|
||||
self.source_namespace = namespace_path;
|
||||
self
|
||||
}
|
||||
|
||||
/// The view's columns, as `(name, SQL expression)` pairs. Not calling
|
||||
/// this selects every source column, expanded at creation time.
|
||||
pub fn select(
|
||||
@@ -936,12 +887,7 @@ impl CreateMaterializedViewBuilder {
|
||||
/// provenance across compaction, and cannot be enabled later.
|
||||
pub async fn execute(self) -> Result<MaterializedView> {
|
||||
ensure_local(&self.connection)?;
|
||||
let source = self
|
||||
.connection
|
||||
.open_table(&self.source)
|
||||
.namespace(self.source_namespace.clone())
|
||||
.execute()
|
||||
.await?;
|
||||
let source = self.connection.open_table(&self.source).execute().await?;
|
||||
let prepared = prepare_declaration(
|
||||
&source,
|
||||
&self.projections,
|
||||
@@ -949,7 +895,7 @@ impl CreateMaterializedViewBuilder {
|
||||
self.limit,
|
||||
)
|
||||
.await?;
|
||||
prepared.create_in(&self.namespace, &self.name).await
|
||||
prepared.create(&self.name).await
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1206,7 +1152,6 @@ mod tests {
|
||||
view.definition(),
|
||||
&MaterializedViewDefinition {
|
||||
source_table: "people".into(),
|
||||
source_namespace: Vec::new(),
|
||||
projections: vec![
|
||||
ViewProjection {
|
||||
output: "name".into(),
|
||||
@@ -2138,138 +2083,33 @@ mod tests {
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(err.to_string().contains("custom_loc"), "{err}");
|
||||
}
|
||||
|
||||
/// A view declared over a namespaced source records that namespace, and
|
||||
/// refresh resolves the source through it -- the coordinate round-trips.
|
||||
#[tokio::test]
|
||||
async fn a_namespaced_source_round_trips_through_refresh() {
|
||||
use lance_namespace::models::CreateNamespaceRequest;
|
||||
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let mut properties = std::collections::HashMap::new();
|
||||
properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string());
|
||||
let conn = crate::connect_namespace("dir", properties)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
conn.create_namespace(CreateNamespaceRequest {
|
||||
id: Some(vec!["ns".into()]),
|
||||
..Default::default()
|
||||
})
|
||||
// A namespaced source cannot be recorded in the definition: the
|
||||
// bare name refresh resolves would reach a different table or none.
|
||||
let namespaced = crate::table::NativeTable::create(
|
||||
"memory://ns_src",
|
||||
"ns_src",
|
||||
vec!["ns".to_string()],
|
||||
Box::new(arrow_array::RecordBatchIterator::new(
|
||||
vec![],
|
||||
std::sync::Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
|
||||
"id",
|
||||
arrow_schema::DataType::Int32,
|
||||
true,
|
||||
)])),
|
||||
)) as Box<dyn arrow_array::RecordBatchReader + Send>,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
std::collections::HashSet::new(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let batch = record_batch!(
|
||||
("name", Utf8, ["ada", "grace", "alan"]),
|
||||
("age", Int32, [36, 85, 41])
|
||||
)
|
||||
.unwrap();
|
||||
conn.create_table("people", batch)
|
||||
.namespace(vec!["ns".to_string()])
|
||||
.write_options(stable_row_ids())
|
||||
.execute()
|
||||
let namespaced = Table::new(std::sync::Arc::new(namespaced), conn.database().clone());
|
||||
let err = prepare_declaration(&namespaced, &[], None, None)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// A decoy of the same name at the root: resolving the source at the
|
||||
// wrong namespace materializes one row here instead of three.
|
||||
let decoy = record_batch!(("name", Utf8, ["mallory"]), ("age", Int32, [42])).unwrap();
|
||||
conn.create_table("people", decoy)
|
||||
.write_options(stable_row_ids())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let view = conn
|
||||
.create_materialized_view("adults", "people")
|
||||
.namespace(vec!["ns".to_string()])
|
||||
.source_namespace(vec!["ns".to_string()])
|
||||
.select([("name", "name")])
|
||||
.only_if("age >= 18")
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(view.definition().source_table, "people");
|
||||
assert_eq!(view.definition().source_namespace, vec!["ns".to_string()]);
|
||||
assert_eq!(view.table().namespace(), &["ns"]);
|
||||
|
||||
// Refresh resolves the source at the recorded namespace, not at root.
|
||||
let result = view.refresh().execute().await.unwrap();
|
||||
assert_eq!(result.rows_written, 3);
|
||||
}
|
||||
|
||||
/// A definition stored before namespaced sources existed carries no
|
||||
/// namespace key and must read as the root namespace.
|
||||
#[test]
|
||||
fn a_definition_without_a_namespace_reads_as_root() {
|
||||
let stored =
|
||||
r#"{"source_table":"people","projections":[{"output":"name","expression":"name"}]}"#;
|
||||
let definition: MaterializedViewDefinition = serde_json::from_str(stored).unwrap();
|
||||
assert!(definition.source_namespace.is_empty());
|
||||
}
|
||||
|
||||
fn definition(source_namespace: Vec<String>) -> MaterializedViewDefinition {
|
||||
MaterializedViewDefinition {
|
||||
source_table: "people".to_string(),
|
||||
source_namespace,
|
||||
projections: vec![ViewProjection {
|
||||
output: "name".to_string(),
|
||||
expression: "name".to_string(),
|
||||
}],
|
||||
filter: None,
|
||||
limit: None,
|
||||
inputs: vec!["name".to_string()],
|
||||
}
|
||||
}
|
||||
|
||||
/// A root definition keeps the pre-namespace `select` form byte-stably;
|
||||
/// a namespaced one moves off `select`, which sends pre-namespace readers
|
||||
/// to the `Unrecognized` refusal instead of a root resolve.
|
||||
#[test]
|
||||
fn a_namespaced_definition_is_refused_by_the_pre_namespace_reader() {
|
||||
let root = definition_to_metadata(&definition(Vec::new())).unwrap();
|
||||
let root: serde_json::Value = serde_json::from_str(&root).unwrap();
|
||||
assert_eq!(root["kind"], "select");
|
||||
assert!(
|
||||
root.get("source_namespace").is_none(),
|
||||
"a root definition must not grow new keys: {root}"
|
||||
);
|
||||
|
||||
let stored = definition_to_metadata(&definition(vec!["ns".to_string()])).unwrap();
|
||||
let value: serde_json::Value = serde_json::from_str(&stored).unwrap();
|
||||
// The pre-namespace discriminator is `kind == "select"`; anything
|
||||
// else lands in its Unrecognized refusal rather than in a root open.
|
||||
assert_eq!(value["kind"], "namespaced_select");
|
||||
|
||||
// The current reader round-trips the coordinate.
|
||||
let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), stored)]);
|
||||
match materialized_view_kind(&metadata).unwrap() {
|
||||
Some(MaterializedViewKind::Select(read)) => {
|
||||
assert_eq!(read.source_namespace, vec!["ns".to_string()])
|
||||
}
|
||||
other => panic!("expected the namespaced select form, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// A kind that disagrees with its namespace is an error, not a view:
|
||||
/// under `select` it is the shape old readers would resolve at the root.
|
||||
#[test]
|
||||
fn a_kind_namespace_mismatch_is_refused() {
|
||||
for (kind, namespace) in [
|
||||
(SELECT_KIND, vec!["ns".to_string()]),
|
||||
(NAMESPACED_SELECT_KIND, Vec::new()),
|
||||
] {
|
||||
let mut value = serde_json::to_value(definition(namespace)).unwrap();
|
||||
value["kind"] = serde_json::Value::String(kind.to_string());
|
||||
let metadata = HashMap::from([(DEFINITION_META_KEY.to_string(), value.to_string())]);
|
||||
let err = materialized_view_kind(&metadata).unwrap_err();
|
||||
assert!(
|
||||
err.to_string()
|
||||
.contains("does not match its source namespace"),
|
||||
"kind '{kind}': {err}"
|
||||
);
|
||||
}
|
||||
.unwrap_err();
|
||||
assert!(err.to_string().contains("namespaced source"), "{err}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -170,7 +170,6 @@ pub(crate) async fn execute_refresh(
|
||||
let (replanned, mut planned_fields, _renames) = super::plan(
|
||||
source_schema,
|
||||
&definition.source_table,
|
||||
&definition.source_namespace,
|
||||
&projections,
|
||||
definition.filter.as_deref(),
|
||||
definition.limit,
|
||||
@@ -591,7 +590,7 @@ async fn open_source(view: &Table, definition: &MaterializedViewDefinition) -> R
|
||||
let source = database
|
||||
.open_table(OpenTableRequest {
|
||||
name: definition.source_table.clone(),
|
||||
namespace_path: definition.source_namespace.clone(),
|
||||
namespace_path: Vec::new(),
|
||||
index_cache_size: None,
|
||||
lance_read_params: None,
|
||||
location: None,
|
||||
@@ -2920,7 +2919,6 @@ mod tests {
|
||||
|
||||
let replacement = crate::materialized_view::MaterializedViewDefinition {
|
||||
source_table: "src".into(),
|
||||
source_namespace: Vec::new(),
|
||||
projections: vec![
|
||||
crate::materialized_view::ViewProjection {
|
||||
output: "x".into(),
|
||||
@@ -2960,7 +2958,6 @@ mod tests {
|
||||
|
||||
let narrower = crate::materialized_view::MaterializedViewDefinition {
|
||||
source_table: "src".into(),
|
||||
source_namespace: Vec::new(),
|
||||
projections: vec![crate::materialized_view::ViewProjection {
|
||||
output: "x".into(),
|
||||
expression: "x".into(),
|
||||
|
||||
+77
-165
@@ -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;
|
||||
@@ -26,7 +26,9 @@ use crate::database::{
|
||||
use crate::error::Result;
|
||||
use crate::function::{FunctionRegistrationRequest, FunctionVersion};
|
||||
use crate::job::Job;
|
||||
use crate::remote::job::{DescribeJobResponse, RemoteJob, job_state_to_client};
|
||||
use crate::remote::job::{
|
||||
DescribeJobResponse, PauseJobResponse, RemoteJob, ResumeJobResponse, job_state_to_client,
|
||||
};
|
||||
use crate::remote::util::stream_as_body;
|
||||
use crate::table::BaseTable;
|
||||
|
||||
@@ -533,19 +535,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 +590,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
|
||||
@@ -734,6 +686,40 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
|
||||
}
|
||||
}
|
||||
|
||||
async fn pause_job(&self, job_id: &str) -> Result<crate::database::PauseJobStatus> {
|
||||
let req = self
|
||||
.client
|
||||
.post("/v1/jobs/pause")
|
||||
.json(&serde_json::json!({ "job_id": job_id }));
|
||||
let (request_id, rsp) = self.client.send(req).await?;
|
||||
let rsp = self.client.check_response(&request_id, rsp).await?;
|
||||
let body: PauseJobResponse = rsp.json().await.err_to_http(request_id)?;
|
||||
Ok(if body.paused {
|
||||
crate::database::PauseJobStatus::Pausing
|
||||
} else if body.committing {
|
||||
crate::database::PauseJobStatus::Committing
|
||||
} else {
|
||||
crate::database::PauseJobStatus::AlreadyPaused
|
||||
})
|
||||
}
|
||||
|
||||
async fn resume_job(&self, job_id: &str) -> Result<crate::database::ResumeJobStatus> {
|
||||
let req = self
|
||||
.client
|
||||
.post("/v1/jobs/resume")
|
||||
.json(&serde_json::json!({ "job_id": job_id }));
|
||||
let (request_id, rsp) = self.client.send(req).await?;
|
||||
let rsp = self.client.check_response(&request_id, rsp).await?;
|
||||
let body: ResumeJobResponse = rsp.json().await.err_to_http(request_id)?;
|
||||
Ok(if body.resumed {
|
||||
crate::database::ResumeJobStatus::Resumed
|
||||
} else if body.still_pausing {
|
||||
crate::database::ResumeJobStatus::StillPausing
|
||||
} else {
|
||||
crate::database::ResumeJobStatus::NotPaused
|
||||
})
|
||||
}
|
||||
|
||||
async fn job_history(&self, job_id: Option<&str>) -> Result<Vec<arrow_array::RecordBatch>> {
|
||||
let mut body = serde_json::json!({});
|
||||
if let Some(job_id) = job_id {
|
||||
@@ -2669,6 +2655,45 @@ mod tests {
|
||||
assert!(!conn.cancel_job("nope").await.unwrap());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_pause_and_resume_job() {
|
||||
use crate::database::{PauseJobStatus, ResumeJobStatus};
|
||||
let conn = Connection::new_with_handler(|request| {
|
||||
assert_eq!(request.url().path(), "/v1/jobs/pause");
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"job_id": "job-1", "paused": true}"#)
|
||||
.unwrap()
|
||||
});
|
||||
assert_eq!(
|
||||
conn.pause_job("job-1").await.unwrap(),
|
||||
PauseJobStatus::Pausing
|
||||
);
|
||||
|
||||
let conn = Connection::new_with_handler(|_| {
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"job_id": "job-1", "paused": false, "committing": true}"#)
|
||||
.unwrap()
|
||||
});
|
||||
assert_eq!(
|
||||
conn.pause_job("job-1").await.unwrap(),
|
||||
PauseJobStatus::Committing
|
||||
);
|
||||
|
||||
let conn = Connection::new_with_handler(|request| {
|
||||
assert_eq!(request.url().path(), "/v1/jobs/resume");
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(r#"{"job_id": "job-1", "resumed": false, "still_pausing": true}"#)
|
||||
.unwrap()
|
||||
});
|
||||
assert_eq!(
|
||||
conn.resume_job("job-1").await.unwrap(),
|
||||
ResumeJobStatus::StillPausing
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_job_history_parses_arrow_stream() {
|
||||
let schema = Arc::new(Schema::new(vec![Field::new(
|
||||
@@ -2758,119 +2783,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| {
|
||||
|
||||
@@ -73,6 +73,28 @@ pub(super) struct ReportedFailure {
|
||||
retryable: Option<bool>,
|
||||
}
|
||||
|
||||
/// Forward-compatible `/v1/jobs/pause` wire envelope.
|
||||
#[derive(Deserialize)]
|
||||
pub(super) struct PauseJobResponse {
|
||||
/// False when the job was already paused, so a repeated pause changed
|
||||
/// nothing.
|
||||
#[serde(default)]
|
||||
pub(super) paused: bool,
|
||||
/// The job is finalizing its results and cannot be parked right now.
|
||||
#[serde(default)]
|
||||
pub(super) committing: bool,
|
||||
}
|
||||
|
||||
/// Forward-compatible `/v1/jobs/resume` wire envelope.
|
||||
#[derive(Deserialize)]
|
||||
pub(super) struct ResumeJobResponse {
|
||||
#[serde(default)]
|
||||
pub(super) resumed: bool,
|
||||
/// The pause's worker drain is not confirmed yet.
|
||||
#[serde(default)]
|
||||
pub(super) still_pausing: bool,
|
||||
}
|
||||
|
||||
/// Forward-compatible `/v1/jobs/describe` wire envelope.
|
||||
#[derive(Deserialize)]
|
||||
pub(super) struct DescribeJobResponse {
|
||||
|
||||
@@ -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());
|
||||
@@ -6237,40 +6233,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));
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user