Compare commits

..

1 Commits

Author SHA1 Message Date
Jack Ye b81e5c2971 chore: use streamed-copy lance branch 2026-08-25 22:46:47 -07:00
58 changed files with 240 additions and 1799 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
[tool.bumpversion]
current_version = "0.38.0-beta.11"
current_version = "0.38.0-beta.10"
parse = """(?x)
(?P<major>0|[1-9]\\d*)\\.
(?P<minor>0|[1-9]\\d*)\\.
Generated
+37 -37
View File
@@ -1756,7 +1756,7 @@ version = "3.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -3034,7 +3034,7 @@ dependencies = [
"libc",
"option-ext",
"redox_users",
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -3257,7 +3257,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
dependencies = [
"libc",
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -3456,7 +3456,7 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]]
name = "fsst"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow-array",
"rand 0.9.5",
@@ -4561,7 +4561,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
dependencies = [
"hermit-abi",
"libc",
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -4816,7 +4816,7 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
[[package]]
name = "lance"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arc-swap",
"arrow",
@@ -4889,7 +4889,7 @@ dependencies = [
[[package]]
name = "lance-arrow"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
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.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
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.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -4935,7 +4935,7 @@ dependencies = [
[[package]]
name = "lance-bitpacking"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrayref",
"crunchy",
@@ -4946,7 +4946,7 @@ dependencies = [
[[package]]
name = "lance-core"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4984,7 +4984,7 @@ dependencies = [
[[package]]
name = "lance-datafusion"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow",
"arrow-array",
@@ -5014,7 +5014,7 @@ dependencies = [
[[package]]
name = "lance-datagen"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow",
"arrow-array",
@@ -5032,7 +5032,7 @@ dependencies = [
[[package]]
name = "lance-derive"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"proc-macro2",
"quote",
@@ -5042,7 +5042,7 @@ dependencies = [
[[package]]
name = "lance-encoding"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -5076,7 +5076,7 @@ dependencies = [
[[package]]
name = "lance-file"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -5108,7 +5108,7 @@ dependencies = [
[[package]]
name = "lance-index"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arc-swap",
"arrow",
@@ -5173,7 +5173,7 @@ dependencies = [
[[package]]
name = "lance-index-core"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5196,7 +5196,7 @@ dependencies = [
[[package]]
name = "lance-io"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow",
"arrow-array",
@@ -5237,7 +5237,7 @@ dependencies = [
[[package]]
name = "lance-linalg"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5252,7 +5252,7 @@ dependencies = [
[[package]]
name = "lance-namespace"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow",
"async-trait",
@@ -5265,7 +5265,7 @@ dependencies = [
[[package]]
name = "lance-namespace-impls"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow",
"arrow-ipc",
@@ -5319,7 +5319,7 @@ dependencies = [
[[package]]
name = "lance-select"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5334,7 +5334,7 @@ dependencies = [
[[package]]
name = "lance-table"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow",
"arrow-array",
@@ -5375,7 +5375,7 @@ dependencies = [
[[package]]
name = "lance-testing"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5389,7 +5389,7 @@ dependencies = [
[[package]]
name = "lance-tokenizer"
version = "12.0.0-beta.2"
source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
dependencies = [
"frostem",
"icu_segmenter",
@@ -6240,7 +6240,7 @@ version = "0.50.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -7632,8 +7632,8 @@ version = "0.14.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7"
dependencies = [
"heck 0.5.0",
"itertools 0.14.0",
"heck 0.4.1",
"itertools 0.11.0",
"log",
"multimap",
"petgraph",
@@ -7652,7 +7652,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b"
dependencies = [
"anyhow",
"itertools 0.14.0",
"itertools 0.11.0",
"proc-macro2",
"quote",
"syn 2.0.117",
@@ -7923,7 +7923,7 @@ dependencies = [
"once_cell",
"socket2 0.6.3",
"tracing",
"windows-sys 0.60.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -8701,7 +8701,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys",
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -8772,7 +8772,7 @@ dependencies = [
"security-framework",
"security-framework-sys",
"webpki-root-certs",
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -9328,7 +9328,7 @@ version = "0.8.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c1c97747dbf44bb1ca44a561ece23508e99cb592e862f22222dcf42f51d1e451"
dependencies = [
"heck 0.5.0",
"heck 0.4.1",
"proc-macro2",
"quote",
"syn 2.0.117",
@@ -9340,7 +9340,7 @@ version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "54254b8531cafa275c5e096f62d48c81435d1015405a91198ddb11e967301d40"
dependencies = [
"heck 0.5.0",
"heck 0.4.1",
"proc-macro2",
"quote",
"syn 2.0.117",
@@ -9762,7 +9762,7 @@ dependencies = [
"getrandom 0.4.2",
"once_cell",
"rustix",
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -10739,7 +10739,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
+14 -14
View File
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
rust-version = "1.91.0"
[workspace.dependencies]
lance = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance = { "version" = "=12.0.0-beta.2", default-features = false, "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-core = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-datagen = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-file = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-io = { "version" = "=12.0.0-beta.2", default-features = false, "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-index = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-linalg = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-namespace = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-namespace-impls = { "version" = "=12.0.0-beta.2", default-features = false, "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-table = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-testing = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-datafusion = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-encoding = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lance-arrow = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
lancedb = { path = "rust/lancedb", default-features = false }
ahash = "0.8"
# Note that this one does not include pyarrow
+1 -1
View File
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
<dependency>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-core</artifactId>
<version>0.38.0-beta.11</version>
<version>0.38.0-beta.10</version>
</dependency>
```
-12
View File
@@ -1292,18 +1292,6 @@ abstract updateFieldMetadata(updates): Promise<UpdateFieldMetadataResult>
Update per-field (column) metadata.
The following keys are treated specially, by convention, and should be
used when appropriate:
- `lancedb:description`: for a human-readable description of a field.
- `lancedb:tag:<name>`: for a user-defined key-value tag, where the suffix
names the tag category; e.g. `lancedb:tag:model: "clip"`.
- `lancedb:logical-column`: for a column grouping; e.g. `feature_v1` and
`feature_v2` might be in the same logical column.
- `lancedb:status`: for status options (`production`, `candidate`,
`deprecated`, `archived`) to designate the current life cycle state of
this column.
#### Parameters
* **updates**: [`FieldMetadataUpdate`](../interfaces/FieldMetadataUpdate.md)[]
@@ -17,8 +17,7 @@ metadata: Record<string, null | string>;
```
Metadata key/value pairs. Merged into the field's existing metadata by
default; a value of `null` deletes that key. See
[Table.updateFieldMetadata](../classes/Table.md#updatefieldmetadata) for the conventional `lancedb:*` keys.
default; a value of `null` deletes that key.
***
-2
View File
@@ -159,8 +159,6 @@ and combined with [BooleanQuery][lancedb.query.BooleanQuery].
::: lancedb.query.FullTextOperator
::: lancedb.query.DocumentGranularity
::: lancedb.query.Occur
## Embeddings
+1 -1
View File
@@ -8,7 +8,7 @@
<parent>
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.38.0-beta.11</version>
<version>0.38.0-beta.10</version>
<relativePath>../pom.xml</relativePath>
</parent>
+1 -1
View File
@@ -6,7 +6,7 @@
<groupId>com.lancedb</groupId>
<artifactId>lancedb-parent</artifactId>
<version>0.38.0-beta.11</version>
<version>0.38.0-beta.10</version>
<packaging>pom</packaging>
<name>${project.artifactId}</name>
<description>LanceDB Java SDK Parent POM</description>
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
name = "lancedb-nodejs"
edition.workspace = true
version = "0.38.0-beta.11"
version = "0.38.0-beta.10"
publish = false
license.workspace = true
description.workspace = true
-21
View File
@@ -6,9 +6,6 @@ import * as arrow17 from "apache-arrow-17";
import * as arrow18 from "apache-arrow-18";
import {
Field as CurrentField,
LargeBinary as CurrentLargeBinary,
Schema as CurrentSchema,
Vector as CurrentVector,
convertToTable,
tableFromIPC as currentTableFromIPC,
@@ -39,24 +36,6 @@ function sampleRecords(): Array<Record<string, any>> {
},
];
}
it("preserves field metadata from a provided schema", async function () {
const jsonMetadata = new Map([["ARROW:extension:name", "lance.json"]]);
const schema = new CurrentSchema([
new CurrentField("meta", new CurrentLargeBinary(), true, jsonMetadata),
]);
const table = makeArrowTable(
[{ meta: Buffer.from(JSON.stringify({ source: "test" })) }],
{ schema },
);
expect(table.schema.fields[0].metadata).toEqual(jsonMetadata);
const roundTripped = currentTableFromIPC(await fromTableToBuffer(table));
expect(roundTripped.schema.fields[0].metadata).toEqual(jsonMetadata);
});
describe.each([arrow15, arrow16, arrow17, arrow18])(
"Arrow",
(
-21
View File
@@ -3561,27 +3561,6 @@ describe("when creating an empty table", () => {
expect((actualSchema.fields[1].type as Float64).precision).toBe(2);
});
it("can add and query JSON data", async () => {
const schema = new Schema([
new Field("id", new Int32(), true),
new Field(
"meta",
new Utf8(),
true,
new Map([["ARROW:extension:name", "arrow.json"]]),
),
]);
const table = await con.createEmptyTable("json", schema);
const meta = JSON.stringify({ x: 1 });
await table.add([{ id: 1, meta }]);
const rows = await table.query().toArray();
expect(rows).toHaveLength(1);
expect(rows[0].id).toBe(1);
expect(rows[0].meta).toBe(meta);
});
it("can create an empty table from schema that specifies field types by name", async () => {
const schemaLike = {
fields: [
+1 -1
View File
@@ -170,7 +170,7 @@ test("basic table examples", async () => {
// --8<-- [end:create_index]
// --8<-- [start:delete_rows]
await tbl.delete("item = 'fizz'");
await tbl.delete('item = "fizz"');
// --8<-- [end:delete_rows]
// --8<-- [start:drop_table]
+5 -5
View File
@@ -727,11 +727,11 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
* Add a query vector to the search
*
* This method can be called multiple times to add multiple query vectors
* to the search. A column called `query_index` will be added to indicate the index
* of the query vector that produced the result. Flat searches share one table scan
* across the query vectors, avoiding the scan and memory amplification of running
* multiple queries concurrently. Indexed searches may still perform per-vector
* index work.
* to the search. If multiple query vectors are added, then they will be searched
* in parallel, and the results will be concatenated. A column called `query_index`
* will be added to indicate the index of the query vector that produced the result.
*
* Performance wise, this is equivalent to running multiple queries concurrently.
*/
addQueryVector(vector: IntoVector): VectorQuery {
if (vector instanceof Promise) {
+1 -2
View File
@@ -406,11 +406,10 @@ function matchingFields(fields: Field[], tree: FieldTree): Field[] {
field.name,
new Struct(matchingFields(struct.children, value)),
field.nullable,
field.metadata,
),
);
} else {
matches.push(field);
matches.push(new Field(field.name, value as DataType, field.nullable));
}
}
return matches;
+1 -14
View File
@@ -630,18 +630,6 @@ export abstract class Table {
/**
* Update per-field (column) metadata.
*
* The following keys are treated specially, by convention, and should be
* used when appropriate:
*
* - `lancedb:description`: for a human-readable description of a field.
* - `lancedb:tag:<name>`: for a user-defined key-value tag, where the suffix
* names the tag category; e.g. `lancedb:tag:model: "clip"`.
* - `lancedb:logical-column`: for a column grouping; e.g. `feature_v1` and
* `feature_v2` might be in the same logical column.
* - `lancedb:status`: for status options (`production`, `candidate`,
* `deprecated`, `archived`) to designate the current life cycle state of
* this column.
* @param {FieldMetadataUpdate[]} updates One or more per-field updates. Each
* update's metadata is merged into the field's existing metadata by default;
* a value of `null` deletes that key, and `replace: true` swaps the whole map.
@@ -1567,8 +1555,7 @@ export interface FieldMetadataUpdate {
path: string;
/**
* Metadata key/value pairs. Merged into the field's existing metadata by
* default; a value of `null` deletes that key. See
* {@link Table.updateFieldMetadata} for the conventional `lancedb:*` keys.
* default; a value of `null` deletes that key.
*/
metadata: Record<string, string | null>;
/** If true, replace the field's entire metadata map instead of merging. */
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-darwin-arm64",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.10",
"os": ["darwin"],
"cpu": ["arm64"],
"main": "lancedb.darwin-arm64.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-gnu",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.10",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-arm64-musl",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.10",
"os": ["linux"],
"cpu": ["arm64"],
"main": "lancedb.linux-arm64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-gnu",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.10",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-gnu.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-linux-x64-musl",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.10",
"os": ["linux"],
"cpu": ["x64"],
"main": "lancedb.linux-x64-musl.node",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-arm64-msvc",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.10",
"os": [
"win32"
],
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@lancedb/lancedb-win32-x64-msvc",
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.10",
"os": ["win32"],
"cpu": ["x64"],
"main": "lancedb.win32-x64-msvc.node",
+1 -1
View File
@@ -11,7 +11,7 @@
"ann"
],
"private": false,
"version": "0.38.0-beta.11",
"version": "0.38.0-beta.10",
"main": "dist/index.js",
"exports": {
".": "./dist/index.js",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.38.0-beta.11"
version = "0.38.0-beta.10"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
+3 -26
View File
@@ -222,7 +222,6 @@ class PythonEnvironmentSpec(_RemoteValue):
kind: str
packages: tuple[str, ...] = ()
channels: tuple[str, ...] = ()
path: Optional[str] = None
modules: tuple[str, ...] = ()
image: Optional[str] = None
@@ -910,25 +909,13 @@ class UdfDefinition:
pip: tuple[str, ...],
env: Mapping[str, str],
python_version: Optional[str],
conda: tuple[str, ...] = (),
conda_channels: tuple[str, ...] = (),
):
function_name = name or function.__name__
if not _FUNCTION_NAME.fullmatch(function_name):
raise ValueError(f"invalid Function name: {function_name!r}")
if pip and conda:
raise ValueError("a Function environment is pip or conda, not both")
if conda_channels and not conda:
raise ValueError("conda_channels requires conda packages")
packages = tuple(sorted(set(conda if conda else pip)))
packages = tuple(sorted(set(pip)))
if any(not package or package != package.strip() for package in packages):
raise ValueError("package requirements must be non-empty and trimmed")
if conda:
environment_spec = PythonEnvironmentSpec(
kind="conda", packages=packages, channels=tuple(conda_channels)
)
else:
environment_spec = PythonEnvironmentSpec(kind="pip", packages=packages)
raise ValueError("pip requirements must be non-empty and trimmed")
environment = dict(env)
if any(
not isinstance(key, str) or not isinstance(value, str)
@@ -942,7 +929,7 @@ class UdfDefinition:
kind="python",
python_version=python_version
or f"{sys.version_info.major}.{sys.version_info.minor}",
environment=environment_spec,
environment=PythonEnvironmentSpec(kind="pip", packages=packages),
env=environment,
)
self._function = function
@@ -989,8 +976,6 @@ def udf(
pip: tuple[str, ...] | list[str] = (),
env: Optional[Mapping[str, str]] = None,
python_version: Optional[str] = None,
conda: tuple[str, ...] | list[str] = (),
conda_channels: tuple[str, ...] | list[str] = (),
) -> Callable[[Callable[..., Any]], UdfDefinition]: ...
@@ -1003,8 +988,6 @@ def udf(
pip: tuple[str, ...] | list[str] = (),
env: Optional[Mapping[str, str]] = None,
python_version: Optional[str] = None,
conda: tuple[str, ...] | list[str] = (),
conda_channels: tuple[str, ...] | list[str] = (),
):
"""Prepare a scalar Python callable for remote Function registration.
@@ -1027,10 +1010,6 @@ def udf(
provided together with ``input_schema``.
pip : sequence of str, optional
Pip requirements for the remote environment.
conda : sequence of str, optional
Conda packages for the remote environment, instead of ``pip``.
conda_channels : sequence of str, optional
Conda channels in priority order; requires ``conda``.
env : mapping of str to str, optional
Environment variables included in the Function definition.
python_version : str, optional
@@ -1070,8 +1049,6 @@ def udf(
pip=tuple(pip),
env={} if env is None else env,
python_version=python_version,
conda=tuple(conda),
conda_channels=tuple(conda_channels),
)
if function is None:
-12
View File
@@ -7,7 +7,6 @@ from typing import List, Literal, Optional
from ._lancedb import (
IndexConfig,
)
from .query import DocumentGranularity
from .types import BaseTokenizerType
lang_mapping = {
@@ -122,11 +121,6 @@ class FTS:
>>> config = FTS(block_size=256)
Create an index that treats each deepest-list element as one document:
>>> from lancedb.query import DocumentGranularity
>>> config = FTS(document_granularity=DocumentGranularity.LIST_ELEMENT)
Attributes
----------
with_position : bool, default False
@@ -178,11 +172,6 @@ class FTS:
roughly half of the available CPU cores. The effective value is
limited by the available compute capacity. This build-only setting is
not persisted with the index and does not apply to remote tables.
document_granularity : DocumentGranularity, default ROW
``ROW`` treats the selected text in one table row as one document.
``LIST_ELEMENT`` treats each element of the deepest list on the indexed
field path as one document and returns its physical coordinates in
``_doc_index`` for matching queries.
Notes
-----
@@ -207,7 +196,6 @@ class FTS:
custom_stop_words: Optional[List[str]] = None
memory_limit: Optional[int] = None
num_workers: Optional[int] = None
document_granularity: DocumentGranularity = DocumentGranularity.ROW
@dataclass
+5 -29
View File
@@ -375,13 +375,6 @@ class FullTextOperator(str, Enum):
OR = "OR"
class DocumentGranularity(str, Enum):
"""The unit treated as one full-text-search document."""
ROW = "row"
LIST_ELEMENT = "list_element"
class Occur(str, Enum):
SHOULD = "SHOULD"
MUST = "MUST"
@@ -485,10 +478,6 @@ class MatchQuery(FullTextQuery):
prefix_length : int, optional
The number of beginning characters being unchanged for fuzzy matching.
This is useful to achieve prefix matching.
document_granularity : DocumentGranularity, optional
Explicitly select row or deepest-list-element documents. If omitted,
the indexed granularity is inferred. When both granularities are indexed
for the field, this must be specified. With no index, row granularity is used.
"""
query: str
@@ -498,9 +487,6 @@ class MatchQuery(FullTextQuery):
max_expansions: int = pydantic.Field(50, kw_only=True)
operator: FullTextOperator = pydantic.Field(FullTextOperator.OR, kw_only=True)
prefix_length: int = pydantic.Field(0, kw_only=True)
document_granularity: Optional[DocumentGranularity] = pydantic.Field(
None, kw_only=True
)
def query_type(self) -> FullTextQueryType:
return FullTextQueryType.MATCH
@@ -517,20 +503,11 @@ class PhraseQuery(FullTextQuery):
The query string to match against.
column : str
The name of the column to match against.
slop : int, default 0
The maximum number of intervening positions permitted in the phrase.
document_granularity : DocumentGranularity, optional
Explicitly select row or deepest-list-element documents. If omitted,
the indexed granularity is inferred. When both granularities are indexed
for the field, this must be specified. With no index, row granularity is used.
"""
query: str
column: str
slop: int = pydantic.Field(0, kw_only=True)
document_granularity: Optional[DocumentGranularity] = pydantic.Field(
None, kw_only=True
)
def query_type(self) -> FullTextQueryType:
return FullTextQueryType.MATCH_PHRASE
@@ -3401,10 +3378,9 @@ class AsyncQuery(AsyncStandardQuery):
pass in multiple vectors. When multiple vectors are passed in, if the vector
column is with multivector type, then the vectors will be treated as a single
query. Or the vectors will be treated as multiple queries, this can be useful
if you want to find the nearest vectors to multiple query vectors. Flat
searches share one table scan across the query vectors, avoiding the scan
and memory amplification of making multiple queries concurrently. If
multiple vectors are passed in then
if you want to find the nearest vectors to multiple query vectors.
This is not expected to be faster than making multiple queries concurrently;
it is just a convenience method. If multiple vectors are passed in then
an additional column `query_index` will be added to the results. This column
will contain the index of the query vector that the result is nearest to.
"""
@@ -3533,8 +3509,8 @@ class AsyncFTSQuery(AsyncStandardQuery):
Typically, a single vector is passed in as the query. However, you can also
pass in multiple vectors. This can be useful if you want to find the nearest
vectors to multiple query vectors. Flat searches share one table scan across
the query vectors instead of issuing concurrent full scans.
vectors to multiple query vectors. This is not expected to be faster than
making multiple queries concurrently; it is just a convenience method.
If multiple vectors are passed in then an additional column `query_index`
will be added to the results. This column will contain the index of the
query vector that the result is nearest to.
-3
View File
@@ -61,7 +61,6 @@ from lancedb.table import _normalize_progress
from ..query import (
AnalyzePlanDistributedMetrics,
DocumentGranularity,
LanceQueryBuilder,
LanceTakeQueryBuilder,
LanceVectorQueryBuilder,
@@ -350,7 +349,6 @@ class RemoteTable(Table):
ngram_max_length: int = 3,
prefix_only: bool = False,
block_size: int = 128,
document_granularity: DocumentGranularity = DocumentGranularity.ROW,
name: Optional[str] = None,
):
"""Create a full-text search index on a column.
@@ -373,7 +371,6 @@ class RemoteTable(Table):
ngram_max_length=ngram_max_length,
prefix_only=prefix_only,
block_size=block_size,
document_granularity=document_granularity,
)
LOOP.run(
self._table.create_index(
+1 -26
View File
@@ -85,7 +85,6 @@ from .query import (
AsyncQuery,
AsyncTakeQuery,
AsyncVectorQuery,
DocumentGranularity,
FullTextQuery,
LanceEmptyQueryBuilder,
LanceFtsQueryBuilder,
@@ -1169,7 +1168,6 @@ class Table(ABC):
ngram_max_length: int = 3,
prefix_only: bool = False,
block_size: int = 128,
document_granularity: DocumentGranularity = DocumentGranularity.ROW,
wait_timeout: Optional[timedelta] = None,
name: Optional[str] = None,
):
@@ -1248,11 +1246,6 @@ class Table(ABC):
The number of documents per compressed posting block. Must be 128
or 256. A value of 256 uses the experimental FTS V3 format and
may introduce breaking changes.
document_granularity: DocumentGranularity, default ROW
``ROW`` treats the selected text in one table row as one document.
``LIST_ELEMENT`` treats each element of the deepest list on the field
path as one document and returns its physical coordinates in
``_doc_index`` for matching queries.
wait_timeout: timedelta, optional
The timeout to wait if indexing is asynchronous.
name: str, optional
@@ -2127,25 +2120,12 @@ class Table(ABC):
----------
updates : dict
One or more dicts, each with:
- "path": str dot-path to the field (e.g. "embedding" or "a.b.c").
- "metadata": dict[str, str | None] keys to set; a value of ``None``
deletes that key.
- "replace": bool, optional replace the field's whole metadata map
instead of merging (default False).
The following keys are treated specially, by convention, and should
be used when appropriate:
- "lancedb:description": for a human-readable description of a field.
- ``"lancedb:tag:<name>"`` for a user-defined key-value tag, where the
suffix names the tag category; e.g. "lancedb:tag:model": "clip".
- "lancedb:logical-column" for a column grouping; e.g. "feature_v1"
and "feature_v2" might be in the same logical column.
- "lancedb:status" for status options ("production", "candidate",
"deprecated", "archived") to designate the current life cycle
state of this column.
Returns
-------
UpdateFieldMetadataResult
@@ -3293,7 +3273,6 @@ class LanceTable(Table):
ngram_max_length: int = 3,
prefix_only: bool = False,
block_size: int = 128,
document_granularity: DocumentGranularity = DocumentGranularity.ROW,
name: Optional[str] = None,
):
"""Create a full-text search index on a column.
@@ -3345,11 +3324,7 @@ class LanceTable(Table):
tokenizer_configs = self.infer_tokenizer_configs(tokenizer_name)
tokenizer_configs["custom_stop_words"] = custom_stop_words
config = FTS(
block_size=block_size,
document_granularity=document_granularity,
**tokenizer_configs,
)
config = FTS(block_size=block_size, **tokenizer_configs)
try:
LOOP.run(
+2 -2
View File
@@ -105,7 +105,7 @@ def test_quickstart(tmp_path):
tbl.create_index(num_sub_vectors=1)
# --8<-- [end:create_index]
# --8<-- [start:delete_rows]
tbl.delete("item = 'fizz'")
tbl.delete('item = "fizz"')
# --8<-- [end:delete_rows]
# --8<-- [start:drop_table]
db.drop_table("my_table")
@@ -201,7 +201,7 @@ async def test_quickstart_async(tmp_path):
await tbl.create_index("vector")
# --8<-- [end:create_index_async]
# --8<-- [start:delete_rows_async]
await tbl.delete("item = 'fizz'")
await tbl.delete('item = "fizz"')
# --8<-- [end:delete_rows_async]
# --8<-- [start:drop_table_async]
await db.drop_table("my_table_async")
@@ -266,7 +266,7 @@ def test_table():
tbl.add(pydantic_model_items)
# --8<-- [end:add_table_from_pydantic]
# --8<-- [start:delete_row]
tbl.delete("item = 'fizz'")
tbl.delete('item = "fizz"')
# --8<-- [end:delete_row]
# --8<-- [start:delete_specific_row]
data = [
@@ -538,7 +538,7 @@ async def test_table_async():
await async_tbl.add(pydantic_model_items)
# --8<-- [end:add_table_async_from_pydantic]
# --8<-- [start:delete_row_async]
await async_tbl.delete("item = 'fizz'")
await async_tbl.delete('item = "fizz"')
# --8<-- [end:delete_row_async]
# --8<-- [start:delete_specific_row_async]
data = [
@@ -69,26 +69,6 @@ def _run_packaged(definition, *args):
return namespace[definition.registration_request.artifact.entrypoint](*args)
def test_udf_conda_environment():
@udf(conda=["scipy", "numpy"], conda_channels=["conda-forge", "defaults"])
def halve(value: float) -> float:
return value / 2
request = json.loads(halve.registration_request.to_canonical_json())
assert request["runtime"]["environment"] == {
"kind": "conda",
"packages": ["numpy", "scipy"],
"channels": ["conda-forge", "defaults"],
}
pip_request = json.loads(normalize_score.registration_request.to_canonical_json())
assert "channels" not in pip_request["runtime"]["environment"]
with pytest.raises(ValueError, match="not both"):
udf(name="both", pip=["numpy"], conda=["numpy"])(lambda value: value)
with pytest.raises(ValueError, match="requires conda"):
udf(name="channels", conda_channels=["conda-forge"])(lambda value: value)
def test_udf_packages_attribute_access_and_body_imports():
@udf
def word_norm(body: str) -> float:
-77
View File
@@ -25,7 +25,6 @@ from lancedb.db import DBConnection
from lancedb.index import FTS
from lancedb.query import (
BoostQuery,
DocumentGranularity,
MatchQuery,
MultiMatchQuery,
PhraseQuery,
@@ -246,55 +245,6 @@ def test_create_inverted_index_rejects_invalid_block_size(table):
table.create_index("text", config=FTS(block_size=129))
def test_list_element_document_granularity(tmp_path):
docs_type = pa.list_(pa.struct([pa.field("content", pa.string())]))
docs = pa.array(
[
[
{"content": "alpha beta"},
None,
{"content": ""},
{"content": "the and"},
{"content": "alpha beta"},
]
],
type=docs_type,
)
table = ldb.connect(tmp_path).create_table(
"list_element_docs", pa.table({"id": [0], "docs": docs})
)
row_table = ldb.connect(tmp_path).create_table(
"row_docs", pa.table({"id": [0], "docs": docs})
)
row_table.create_index("docs.content", config=FTS())
row_result = row_table.search(MatchQuery("alpha", "docs.content")).to_arrow()
assert row_result.num_rows == 1
assert "_doc_index" not in row_result.column_names
granularity = DocumentGranularity.LIST_ELEMENT
table.create_index(
"docs.content",
config=FTS(with_position=True, document_granularity=granularity),
)
assert table.list_indices()[0].columns == ["docs.content"]
def coordinates(query):
result = table.search(query).limit(10).to_arrow()
doc_index_type = result.schema.field("_doc_index").type
assert pa.types.is_list(doc_index_type)
assert doc_index_type.value_type == pa.uint32()
return sorted(result["_doc_index"].to_pylist())
assert coordinates(
MatchQuery("alpha", "docs.content", document_granularity=granularity)
) == [[0], [4]]
assert coordinates(
PhraseQuery("alpha beta", "docs.content", document_granularity=granularity)
) == [[0], [4]]
assert coordinates(MatchQuery("alpha", "docs.content")) == [[0], [4]]
assert FTS().document_granularity is DocumentGranularity.ROW
def test_create_inverted_index_respects_build_memory_limit(table):
with pytest.raises(ValueError, match="exceeds worker memory limit"):
table.create_index(
@@ -1139,20 +1089,6 @@ def test_fts_query_to_json():
)
assert json_str == expected
# Test MatchQuery with list-element document granularity
match_query = MatchQuery(
"hello world",
"text",
document_granularity=DocumentGranularity.LIST_ELEMENT,
)
json_str = match_query.to_json()
expected = (
'{"match":{"column":"text","terms":"hello world","boost":1.0,'
'"fuzziness":0,"max_expansions":50,"operator":"Or","prefix_length":0,'
'"document_granularity":"list_element"}}'
)
assert json_str == expected
# Test MatchQuery with options
match_query = MatchQuery("puppy", "text", fuzziness=2, boost=1.5, prefix_length=3)
json_str = match_query.to_json()
@@ -1162,19 +1098,6 @@ def test_fts_query_to_json():
)
assert json_str == expected
# Test PhraseQuery with list-element document granularity
phrase_query = PhraseQuery(
"quick brown fox",
"title",
document_granularity=DocumentGranularity.LIST_ELEMENT,
)
json_str = phrase_query.to_json()
expected = (
'{"phrase":{"column":"title","terms":"quick brown fox","slop":0,'
'"document_granularity":"list_element"}}'
)
assert json_str == expected
# Test PhraseQuery
phrase_query = PhraseQuery("quick brown fox", "title")
json_str = phrase_query.to_json()
-17
View File
@@ -897,23 +897,6 @@ def test_query_builder_batches(table):
assert rs_list["id"][1] == 2
def test_batch_vector_query_shares_filtered_flat_scan(table):
query = (
table.search([[1.0, 2.0], [3.0, 4.0]])
.where("id > 0", prefilter=True)
.limit(1)
.select(["id"])
)
plan = query.explain_plan(verbose=True)
assert "KNNVectorDistance: queries=2" in plan
assert "UnionExec" not in plan
results = query.to_arrow()
assert len(results) == 2
assert results["query_index"].to_pylist() == [0, 1]
def test_dynamic_projection(table):
rs = (
LanceVectorQueryBuilder(table, [0, 0], "vector")
-43
View File
@@ -1618,49 +1618,6 @@ def test_query_sync_fts():
)
def test_query_sync_fts_document_granularity():
from lancedb.query import DocumentGranularity, MatchQuery
def handler(body):
assert body == {
"full_text_query": {
"query": {
"match": {
"column": "docs.content",
"terms": "alpha",
"boost": 1.0,
"fuzziness": 0,
"max_expansions": 50,
"operator": "Or",
"prefix_length": 0,
"document_granularity": "list_element",
}
}
},
"k": 10,
"prefilter": True,
"vector": [],
"version": None,
}
return pa.table(
{
"id": [1, 1],
"_doc_index": pa.array([[0], [4]], type=pa.list_(pa.uint32())),
}
)
with query_test_table(handler, server_version=Version("0.6.0")) as table:
result = table.search(
MatchQuery(
"alpha",
"docs.content",
document_granularity=DocumentGranularity.LIST_ELEMENT,
)
).to_arrow()
assert result["_doc_index"].to_pylist() == [[0], [4]]
def test_query_sync_hybrid():
def handler(body):
if "full_text_query" in body:
-20
View File
@@ -4,7 +4,6 @@
import asyncio
import copy
from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta
import threading
@@ -87,25 +86,6 @@ def test_s3_lifecycle(s3_bucket: str):
asyncio.run(test())
@pytest.mark.s3_test
def test_concurrent_open_table(s3_bucket: str):
uri = f"s3://{s3_bucket}/test_concurrent_open_table"
db = lancedb.connect(uri, storage_options=copy.copy(CONFIG))
db.create_table("test", pa.table({"x": [1, 2, 3]}))
num_workers = 32
barrier = threading.Barrier(num_workers)
def open_and_count(_):
barrier.wait()
return db.open_table("test").count_rows()
with ThreadPoolExecutor(max_workers=num_workers) as pool:
row_counts = list(pool.map(open_and_count, range(num_workers)))
assert row_counts == [3] * num_workers
@pytest.fixture()
def kms_key():
kms = get_boto3_client("kms", endpoint_url=CONFIG["aws_endpoint"])
+2 -8
View File
@@ -8,7 +8,7 @@ use lancedb::index::vector::{
};
use lancedb::index::{
Index as LanceDbIndex,
scalar::{BTreeIndexBuilder, DocumentGranularity, FmIndexBuilder, FtsIndexBuilder},
scalar::{BTreeIndexBuilder, FmIndexBuilder, FtsIndexBuilder},
};
use pyo3::IntoPyObject;
use pyo3::types::PyStringMethods;
@@ -60,11 +60,7 @@ pub fn extract_index_params(source: &Option<Bound<'_, PyAny>>) -> PyResult<Lance
.ngram_min_length(params.ngram_min_length)
.ngram_max_length(params.ngram_max_length)
.ngram_prefix_only(params.prefix_only)
.custom_stop_words(params.custom_stop_words)
.document_granularity(
DocumentGranularity::try_from(params.document_granularity.as_str())
.map_err(|err| PyValueError::new_err(err.to_string()))?,
);
.custom_stop_words(params.custom_stop_words);
if let Some(memory_limit) = params.memory_limit {
inner_opts = inner_opts.memory_limit_mb(memory_limit);
}
@@ -225,7 +221,6 @@ struct FtsParams {
block_size: usize,
memory_limit: Option<u64>,
num_workers: Option<usize>,
document_granularity: String,
}
#[derive(FromPyObject)]
@@ -486,7 +481,6 @@ mod tests {
block_size = 128
memory_limit = 2048
num_workers = 7
document_granularity = 'row'
config = FTS()",
None,
+12 -45
View File
@@ -16,8 +16,8 @@ use arrow::pyarrow::FromPyArrow;
use arrow::pyarrow::IntoPyArrow;
use arrow::pyarrow::ToPyArrow;
use lancedb::index::scalar::{
BooleanQuery, BoostQuery, DocumentGranularity, FtsQuery, FullTextSearchQuery, MatchQuery,
MultiMatchQuery, Occur, Operator, PhraseQuery,
BooleanQuery, BoostQuery, FtsQuery, FullTextSearchQuery, MatchQuery, MultiMatchQuery, Occur,
Operator, PhraseQuery,
};
use lancedb::query::AnalyzePlanDistributedMetrics;
use lancedb::query::QueryBase;
@@ -76,16 +76,8 @@ impl<'a, 'py> FromPyObject<'a, 'py> for PyLanceDB<FtsQuery> {
let max_expansions = ob.getattr("max_expansions")?.extract()?;
let operator = ob.getattr("operator")?.extract::<String>()?;
let prefix_length = ob.getattr("prefix_length")?.extract()?;
let document_granularity = ob
.getattr("document_granularity")?
.extract::<Option<String>>()?
.map(|value| {
DocumentGranularity::try_from(value.as_str())
.map_err(|err| PyValueError::new_err(err.to_string()))
})
.transpose()?;
let mut query =
Ok(Self(
MatchQuery::new(query)
.with_column(Some(column))
.with_boost(boost)
@@ -94,32 +86,21 @@ impl<'a, 'py> FromPyObject<'a, 'py> for PyLanceDB<FtsQuery> {
.with_operator(Operator::try_from(operator.as_str()).map_err(|e| {
PyValueError::new_err(format!("Invalid operator: {}", e))
})?)
.with_prefix_length(prefix_length);
if let Some(document_granularity) = document_granularity {
query = query.with_document_granularity(document_granularity);
}
Ok(Self(query.into()))
.with_prefix_length(prefix_length)
.into(),
))
}
"PhraseQuery" => {
let query = ob.getattr("query")?.extract()?;
let column = ob.getattr("column")?.extract()?;
let slop = ob.getattr("slop")?.extract()?;
let document_granularity = ob
.getattr("document_granularity")?
.extract::<Option<String>>()?
.map(|value| {
DocumentGranularity::try_from(value.as_str())
.map_err(|err| PyValueError::new_err(err.to_string()))
})
.transpose()?;
let mut query = PhraseQuery::new(query)
.with_column(Some(column))
.with_slop(slop);
if let Some(document_granularity) = document_granularity {
query = query.with_document_granularity(document_granularity);
}
Ok(Self(query.into()))
Ok(Self(
PhraseQuery::new(query)
.with_column(Some(column))
.with_slop(slop)
.into(),
))
}
"BoostQuery" => {
let positive: Self = ob.getattr("positive")?.extract()?;
@@ -186,13 +167,6 @@ impl<'py> IntoPyObject<'py> for PyLanceDB<FtsQuery> {
kwargs.set_item("max_expansions", query.max_expansions)?;
kwargs.set_item::<_, &str>("operator", query.operator.into())?;
kwargs.set_item("prefix_length", query.prefix_length)?;
if let Some(document_granularity) = query.document_granularity {
let value = match document_granularity {
DocumentGranularity::Row => "row",
DocumentGranularity::ListElement => "list_element",
};
kwargs.set_item("document_granularity", value)?;
}
namespace
.getattr(intern!(py, "MatchQuery"))?
.call((query.terms, query.column.unwrap()), Some(&kwargs))
@@ -200,13 +174,6 @@ impl<'py> IntoPyObject<'py> for PyLanceDB<FtsQuery> {
FtsQuery::Phrase(query) => {
let kwargs = PyDict::new(py);
kwargs.set_item("slop", query.slop)?;
if let Some(document_granularity) = query.document_granularity {
let value = match document_granularity {
DocumentGranularity::Row => "row",
DocumentGranularity::ListElement => "list_element",
};
kwargs.set_item("document_granularity", value)?;
}
namespace
.getattr(intern!(py, "PhraseQuery"))?
.call((query.terms, query.column.unwrap()), Some(&kwargs))
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb"
version = "0.38.0-beta.11"
version = "0.38.0-beta.10"
edition.workspace = true
description = "LanceDB: A serverless, low-latency vector database for AI applications"
license.workspace = true
+1 -54
View File
@@ -1476,7 +1476,7 @@ mod tests {
use crate::table::{AnyQuery, WriteOptions};
use arrow_array::{Int32Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use futures::{TryStreamExt, future::try_join_all, stream::once};
use futures::{TryStreamExt, stream::once};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
@@ -1614,59 +1614,6 @@ mod tests {
);
}
#[tokio::test]
async fn test_concurrent_open_table_reuses_connection_object_store() {
let tempdir = tempdir().unwrap();
let uri = tempdir.path().to_str().unwrap();
let session = Arc::new(lance::session::Session::default());
let request = ConnectRequest {
uri: uri.to_string(),
#[cfg(feature = "remote")]
client_config: Default::default(),
options: Default::default(),
namespace_client_properties: Default::default(),
manifest_enabled: false,
read_consistency_interval: None,
session: Some(session.clone()),
};
let db = ListingDatabase::connect_with_options(&request)
.await
.unwrap();
let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
db.create_table(CreateTableRequest {
name: "test".to_string(),
namespace_path: vec![],
data: Box::new(RecordBatch::new_empty(schema)) as Box<dyn Scannable>,
mode: CreateTableMode::Create,
write_options: Default::default(),
location: None,
namespace_client: None,
})
.await
.unwrap();
let before = session.store_registry().stats();
let opened_tables = try_join_all((0..32).map(|_| {
db.open_table(OpenTableRequest {
name: "test".to_string(),
namespace_path: vec![],
index_cache_size: None,
lance_read_params: None,
location: None,
namespace_client: None,
managed_versioning: None,
})
}))
.await
.unwrap();
let after = session.store_registry().stats();
assert_eq!(opened_tables.len(), 32);
assert_eq!(after.misses, before.misses);
assert_eq!(after.active_stores, before.active_stores);
assert!(after.hits >= before.hits + 32);
}
#[tokio::test]
async fn test_listing_database_root_ops_do_not_create_manifest() {
let tempdir = tempdir().unwrap();
-1
View File
@@ -19,7 +19,6 @@
mod sql;
pub(crate) use sql::canonicalize_sql_predicate;
pub use sql::expr_to_sql_string;
use std::sync::Arc;
+2 -111
View File
@@ -1,16 +1,10 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
use std::any::TypeId;
use datafusion_common::ScalarValue;
use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
use datafusion_expr::Expr;
use datafusion_sql::sqlparser::{
dialect::{Dialect as SqlParserDialect, GenericDialect},
tokenizer::{Token, Tokenizer},
};
use datafusion_sql::unparser::{self, dialect::Dialect as UnparserDialect};
use datafusion_sql::unparser::{self, dialect::Dialect};
/// Unparser dialect that matches the quoting style expected by the Lance SQL
/// parser. Lance uses backtick (`` ` ``) as the only delimited-identifier
@@ -25,7 +19,7 @@ use datafusion_sql::unparser::{self, dialect::Dialect as UnparserDialect};
/// lower-case by the SQL parser, which would break case-sensitive schemas).
struct LanceSqlDialect;
impl UnparserDialect for LanceSqlDialect {
impl Dialect for LanceSqlDialect {
fn identifier_quote_style(&self, identifier: &str) -> Option<char> {
let needs_quote = identifier.chars().any(|c| c.is_ascii_uppercase())
|| !identifier
@@ -36,61 +30,6 @@ impl UnparserDialect for LanceSqlDialect {
}
}
/// Lance's tokenizer dialect with SQL-standard double-quoted identifiers added.
///
/// Keep this deliberately small: Lance's parser wraps `GenericDialect` and
/// delegates only identifier recognition, leaving every other dialect option at
/// its default. In particular, `/*! ... */` remains an ordinary block comment.
#[derive(Debug, Default)]
struct PredicateDialect(GenericDialect);
impl SqlParserDialect for PredicateDialect {
fn dialect(&self) -> TypeId {
self.0.dialect()
}
fn is_identifier_start(&self, ch: char) -> bool {
self.0.is_identifier_start(ch)
}
fn is_identifier_part(&self, ch: char) -> bool {
self.0.is_identifier_part(ch)
}
fn is_delimited_identifier_start(&self, ch: char) -> bool {
ch == '"' || ch == '`'
}
}
/// Canonicalize a raw SQL predicate for Lance's parser.
///
/// Lance wraps [`GenericDialect`] for identifier recognition while retaining the
/// default dialect behavior for every other lexical option. [`PredicateDialect`]
/// mirrors that contract and additionally recognizes `"` as an identifier
/// delimiter, allowing this function to rewrite only those identifier tokens.
pub fn canonicalize_sql_predicate(predicate: &str) -> crate::Result<String> {
let dialect = PredicateDialect::default();
let tokens = Tokenizer::new(&dialect, predicate)
.with_unescape(false)
.tokenize()
.map_err(|err| crate::Error::InvalidInput {
message: format!("invalid SQL predicate: {err}"),
})?;
Ok(tokens
.into_iter()
.map(|token| match token {
Token::Word(word) if word.quote_style == Some('"') => {
// with_unescape(false) retains doubled double quotes. Decode
// those before escaping any backticks for Lance's delimiter.
let identifier = word.value.replace("\"\"", "\"").replace('`', "``");
format!("`{identifier}`")
}
other => other.to_string(),
})
.collect())
}
/// Prefix for placeholder strings inserted in place of binary literals. Chosen
/// to be extremely unlikely to occur in user data.
const BINARY_PLACEHOLDER_PREFIX: &str = "__lancedb_binary_placeholder_";
@@ -174,51 +113,3 @@ pub fn expr_to_sql_string(expr: &Expr) -> crate::Result<String> {
}
Ok(sql)
}
#[cfg(test)]
mod tests {
use super::canonicalize_sql_predicate;
#[test]
fn normalizes_double_quoted_identifiers() {
assert_eq!(
canonicalize_sql_predicate(r#""PartyAbbrev" = 'D'"#).unwrap(),
"`PartyAbbrev` = 'D'"
);
assert_eq!(
canonicalize_sql_predicate(r#""MetaData"."userId" = 5"#).unwrap(),
"`MetaData`.`userId` = 5"
);
assert_eq!(
canonicalize_sql_predicate(r#""a""b" = 1"#).unwrap(),
"`a\"b` = 1"
);
}
#[test]
fn preserves_quotes_inside_literals_and_backticks() {
let filter = r#"name = 'Alice "Ace"' AND `quoted"field` = 1"#;
assert_eq!(canonicalize_sql_predicate(filter).unwrap(), filter);
}
#[test]
fn preserves_literals_and_comments_using_lance_dialect_rules() {
let predicate = r#"path = '\' AND "PartyAbbrev" = 'D' -- unmatched " in comment"#;
assert_eq!(
canonicalize_sql_predicate(predicate).unwrap(),
r#"path = '\' AND `PartyAbbrev` = 'D' -- unmatched " in comment"#
);
let predicate = r#"id = 1 /* unmatched " in block comment */"#;
assert_eq!(canonicalize_sql_predicate(predicate).unwrap(), predicate);
let predicate = r#"id = 1 /*! OR "PartyAbbrev" = 'D' */"#;
assert_eq!(canonicalize_sql_predicate(predicate).unwrap(), predicate);
}
#[test]
fn rejects_unterminated_double_quoted_identifier() {
let error = canonicalize_sql_predicate(r#""PartyAbbrev = 'D'"#).unwrap_err();
assert!(matches!(error, crate::Error::InvalidInput { .. }));
}
}
-26
View File
@@ -186,9 +186,6 @@ pub struct PythonEnvironmentSpec {
pub kind: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub packages: Vec<String>,
/// Conda channels in priority order; conda environments only.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub channels: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub path: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
@@ -586,26 +583,3 @@ impl RefreshColumnResult {
}
impl_json!(RefreshColumnResult);
#[cfg(test)]
mod conda_environment_tests {
use super::PythonEnvironmentSpec;
#[test]
fn conda_channels_round_trip_and_pip_stays_bare() {
let conda: PythonEnvironmentSpec = serde_json::from_str(
r#"{"kind":"conda","packages":["numpy"],"channels":["conda-forge"]}"#,
)
.unwrap();
assert_eq!(conda.channels, ["conda-forge"]);
assert!(
serde_json::to_string(&conda)
.unwrap()
.contains(r#""channels":["conda-forge"]"#)
);
let pip: PythonEnvironmentSpec =
serde_json::from_str(r#"{"kind":"pip","packages":["numpy"]}"#).unwrap();
assert!(!serde_json::to_string(&pip).unwrap().contains("channels"));
}
}
-1
View File
@@ -63,5 +63,4 @@ pub struct FmIndexBuilder {}
pub use lance_index::scalar::FullTextSearchQuery;
pub use lance_index::scalar::InvertedIndexParams as FtsIndexBuilder;
pub use lance_index::scalar::InvertedIndexParams;
pub use lance_index::scalar::inverted::DocumentGranularity;
pub use lance_index::scalar::inverted::query::*;
+2 -11
View File
@@ -170,15 +170,6 @@ pub(crate) fn plan(
filter: Option<&str>,
limit: Option<u64>,
) -> Result<(MaterializedViewDefinition, Vec<ArrowField>, Lineage)> {
let filter = filter
.map(crate::expr::canonicalize_sql_predicate)
.transpose()
.map_err(|err| match err {
Error::InvalidInput { message } => Error::InvalidInput {
message: format!("invalid view filter: {message}"),
},
err => err,
})?;
let projections: Vec<(String, String)> = if projections.is_empty() {
source_schema
.fields()
@@ -283,7 +274,7 @@ pub(crate) fn plan(
declared.push(output);
}
if let Some(filter) = filter.as_deref() {
if let Some(filter) = filter {
let expr = planner
.parse_filter(filter)
.map_err(|e| Error::InvalidInput {
@@ -323,7 +314,7 @@ pub(crate) fn plan(
.into_iter()
.map(|(output, expression)| ViewProjection { output, expression })
.collect(),
filter,
filter: filter.map(String::from),
limit,
inputs,
};
+24 -160
View File
@@ -46,9 +46,8 @@ use lance_table::format::Fragment;
use serde::{Deserialize, Serialize};
use super::{
DEFINITION_META_KEY, INCARNATION_META_KEY, MaterializedViewDefinition,
REFRESHED_AT_MS_META_KEY, SOURCE_ROW_ID_COLUMN, SOURCE_VERSION_META_KEY,
definition_to_metadata,
INCARNATION_META_KEY, MaterializedViewDefinition, REFRESHED_AT_MS_META_KEY,
SOURCE_ROW_ID_COLUMN, SOURCE_VERSION_META_KEY,
};
use crate::database::OpenTableRequest;
use crate::table::{NativeTable, NativeTableExt, Table};
@@ -198,28 +197,8 @@ pub(crate) async fn execute_refresh(
),
});
}
let definition_changed =
definition.filter != replanned.filter || definition.inputs != replanned.inputs;
let definition = &replanned;
// A watermark written for a legacy raw filter certifies the rows that
// filter produced, not the canonical predicate above. Rebuild instead of
// accepting or advancing it, and persist the migrated definition in the
// same metadata commit that certifies the replacement rows.
if definition_changed {
return rebuild(
view_native,
&view_ds,
&source_ds,
source_version,
source_ts,
definition,
true,
expected_incarnation,
)
.await;
}
let metadata = &view_ds.schema().metadata;
let watermark: Option<u64> = metadata
.get(SOURCE_VERSION_META_KEY)
@@ -278,7 +257,6 @@ pub(crate) async fn execute_refresh(
source_version,
source_ts,
definition,
false,
expected_incarnation,
)
.await
@@ -293,7 +271,6 @@ pub(crate) async fn execute_refresh(
source_version,
source_ts,
definition,
false,
expected_incarnation,
)
.await
@@ -706,7 +683,6 @@ async fn incremental(
view_ds.clone(),
source_version,
source_ts,
None,
expected_incarnation,
)
.await?;
@@ -728,7 +704,6 @@ async fn incremental(
published,
source_version,
source_ts,
None,
expected_incarnation,
)
.await?;
@@ -800,7 +775,6 @@ async fn incremental(
published,
source_version,
source_ts,
None,
expected_incarnation,
)
.await?;
@@ -850,14 +824,12 @@ async fn incremental(
appended,
source_version,
source_ts,
None,
expected_incarnation,
)
.await?;
Ok(Some(result))
}
#[allow(clippy::too_many_arguments)]
async fn rebuild(
view_native: &NativeTable,
view_ds: &Dataset,
@@ -865,7 +837,6 @@ async fn rebuild(
source_version: u64,
source_ts: u128,
definition: &MaterializedViewDefinition,
persist_definition: bool,
expected_incarnation: Option<&str>,
) -> Result<RefreshMaterializedViewResult> {
let rows_written = Arc::new(AtomicU64::new(0));
@@ -896,7 +867,6 @@ async fn rebuild(
replaced,
source_version,
source_ts,
persist_definition.then_some(definition),
expected_incarnation,
)
.await?;
@@ -1011,7 +981,6 @@ async fn stamp_watermark(
mut dataset: Dataset,
source_version: u64,
source_ts: u128,
definition: Option<&MaterializedViewDefinition>,
expected_incarnation: Option<&str>,
) -> Result<u64> {
ensure_incarnation(&dataset, expected_incarnation, dataset.uri()).await?;
@@ -1024,32 +993,27 @@ async fn stamp_watermark(
.get(INCARNATION_META_KEY)
.cloned()
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
let mut metadata = vec![(INCARNATION_META_KEY.to_string(), Some(incarnation))];
if let Some(definition) = definition {
metadata.push((
DEFINITION_META_KEY.to_string(),
Some(definition_to_metadata(definition)?),
));
}
metadata.extend([
(
SOURCE_VERSION_META_KEY.to_string(),
Some(source_version.to_string()),
),
(
SOURCE_VERSION_TS_META_KEY.to_string(),
Some(source_ts.to_string()),
),
(
REFRESHED_AT_MS_META_KEY.to_string(),
Some(now_ms().to_string()),
),
(
VIEW_VERSION_META_KEY.to_string(),
Some(predicted.to_string()),
),
]);
dataset.update_schema_metadata(metadata).await?;
dataset
.update_schema_metadata([
(INCARNATION_META_KEY.to_string(), Some(incarnation)),
(
SOURCE_VERSION_META_KEY.to_string(),
Some(source_version.to_string()),
),
(
SOURCE_VERSION_TS_META_KEY.to_string(),
Some(source_ts.to_string()),
),
(
REFRESHED_AT_MS_META_KEY.to_string(),
Some(now_ms().to_string()),
),
(
VIEW_VERSION_META_KEY.to_string(),
Some(predicted.to_string()),
),
])
.await?;
let actual = dataset.version().version;
if actual != predicted {
return Err(Error::Runtime {
@@ -1621,106 +1585,6 @@ mod tests {
assert_eq!(read(view.table(), "x").await, vec![20, 40]);
}
#[tokio::test]
async fn test_mixed_case_filter_is_canonicalized_for_lineage_and_refresh() {
let conn = connect("memory://").execute().await.unwrap();
let batch = record_batch!(
("id", Int32, [1, 2, 3]),
("PartyAbbrev", Utf8, ["D", "R", "D"])
)
.unwrap();
conn.create_table("src", batch)
.write_options(crate::materialized_view::tests::stable_row_ids())
.execute()
.await
.unwrap();
conn.create_materialized_view("democrats", "src")
.select([("id", "id")])
.only_if(r#""PartyAbbrev" = 'D'"#)
.execute()
.await
.unwrap();
// Reopen from schema metadata so these assertions cover the stored
// predicate and lineage, not only the declaration-time handle.
let view = conn.open_materialized_view("democrats").await.unwrap();
assert_eq!(
view.definition().filter.as_deref(),
Some("`PartyAbbrev` = 'D'")
);
assert_eq!(view.definition().inputs, ["PartyAbbrev", "id"]);
let result = view.refresh().execute().await.unwrap();
assert_eq!(result.rows_written, 2);
assert_eq!(read(view.table(), "id").await, vec![1, 3]);
}
#[tokio::test]
async fn test_legacy_raw_filter_rebuilds_and_persists_canonical_definition() {
let conn = connect("memory://").execute().await.unwrap();
let batch = record_batch!(
("id", Int32, [1, 2, 3]),
("PartyAbbrev", Utf8, ["D", "R", "D"])
)
.unwrap();
conn.create_table("legacy_src", batch)
.write_options(crate::materialized_view::tests::stable_row_ids())
.execute()
.await
.unwrap();
let view = conn
.create_materialized_view("legacy_view", "legacy_src")
.select([("id", "id")])
.only_if(r#""PartyAbbrev" = 'X'"#)
.execute()
.await
.unwrap();
assert_eq!(view.refresh().execute().await.unwrap().rows_written, 0);
// Model a definition and up-to-date watermark written before filter
// canonicalization was applied to materialized views.
let mut legacy = view.definition().clone();
legacy.filter = Some(r#""PartyAbbrev" = 'D'"#.into());
legacy.inputs = vec!["id".into()];
let native = view.table().as_native().unwrap();
let mut dataset = native.dataset.get().await.unwrap().as_ref().clone();
let predicted = dataset.version().version + 1;
dataset
.update_schema_metadata([
(
DEFINITION_META_KEY.to_string(),
Some(definition_to_metadata(&legacy).unwrap()),
),
(
VIEW_VERSION_META_KEY.to_string(),
Some(predicted.to_string()),
),
])
.await
.unwrap();
native.dataset.update(dataset);
let reopened = conn.open_materialized_view("legacy_view").await.unwrap();
let result = reopened.refresh().execute().await.unwrap();
assert_eq!(result.mode, RefreshMode::Rebuild);
assert_eq!(result.rows_written, 2);
assert_eq!(read(reopened.table(), "id").await, vec![1, 3]);
// A fresh handle proves the migration was stored alongside the new
// watermark and therefore happens only once.
let migrated = conn.open_materialized_view("legacy_view").await.unwrap();
assert_eq!(
migrated.definition().filter.as_deref(),
Some("`PartyAbbrev` = 'D'")
);
assert_eq!(migrated.definition().inputs, ["PartyAbbrev", "id"]);
assert_eq!(
migrated.refresh().execute().await.unwrap().mode,
RefreshMode::NoOp
);
assert_eq!(read(migrated.table(), "id").await, vec![1, 3]);
}
#[tokio::test]
async fn test_append_refreshes_incrementally() {
let (_conn, source, view) = refreshed_doubled(vec![1, 2]).await;
@@ -2903,7 +2767,7 @@ mod tests {
let stale = view_native.dataset.get().await.unwrap().as_ref().clone();
view.table().delete("x = 1").await.unwrap();
let err = stamp_watermark(view_native, stale, 99, 99, None, None).await;
let err = stamp_watermark(view_native, stale, 99, 99, None).await;
assert!(err.is_err());
let result = view.refresh().execute().await.unwrap();
+9 -273
View File
@@ -399,9 +399,6 @@ pub trait QueryBase {
/// x > 5 OR y = 'test'
/// ```
///
/// Identifiers may be delimited with SQL-standard double quotes or
/// backticks. String literals must use single quotes.
///
/// Filtering performance can often be improved by creating a scalar index
/// on the filter column(s).
///
@@ -916,17 +913,6 @@ impl QueryRequest {
/// use different representations) the error is recorded and surfaced later
/// by [`Self::check_filter`].
pub(crate) fn add_filter(&mut self, new: QueryFilter) {
let new = match new {
QueryFilter::Sql(filter) => match crate::expr::canonicalize_sql_predicate(&filter) {
Ok(filter) => QueryFilter::Sql(filter),
Err(err) => {
self.filter_error = Some(err.to_string());
return;
}
},
other => other,
};
self.filter = Some(match self.filter.take() {
None => new,
Some(existing) => match and_filters(existing, new) {
@@ -1188,12 +1174,12 @@ impl VectorQuery {
/// Add another query vector to the search.
///
/// Multiple searches will be dispatched as a batch. Flat searches share
/// one table scan across the query vectors, avoiding the scan and memory
/// amplification of issuing the searches concurrently. Indexed searches
/// may still perform per-vector index work.
/// Multiple searches will be dispatched as part of the query.
/// This is a convenience method for adding multiple query vectors
/// to the search. It is not expected to be faster than issuing
/// multiple queries concurrently.
///
/// The output data will contain an additional column `query_index` which
/// The output data will contain an additional columns `query_index` which
/// will contain the index of the query vector that was used to generate the
/// result.
pub fn add_query_vector(mut self, vector: impl IntoQueryVector) -> Result<Self> {
@@ -1660,14 +1646,10 @@ mod tests {
use std::{collections::HashSet, sync::Arc};
use super::*;
use arrow::{
array::downcast_array,
compute::concat_batches,
datatypes::{Int32Type, UInt8Type},
};
use arrow::{array::downcast_array, compute::concat_batches, datatypes::Int32Type};
use arrow_array::{
FixedSizeListArray, Float32Array, Int32Array, RecordBatch, RecordBatchIterator,
StringArray, cast::AsArray, types::Float32Type,
FixedSizeListArray, Float32Array, Int32Array, RecordBatch, StringArray, cast::AsArray,
types::Float32Type,
};
use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
use futures::{StreamExt, TryStreamExt};
@@ -1896,157 +1878,6 @@ mod tests {
query.execute().await.unwrap();
}
#[tokio::test]
async fn test_double_quoted_predicates_across_table_operations() {
let tmp_dir = tempdir().unwrap();
let dataset_path = tmp_dir.path().join("test.lance");
let uri = dataset_path.to_str().unwrap();
let schema = Arc::new(ArrowSchema::new(vec![
ArrowField::new("id", DataType::Int32, false),
ArrowField::new("PartyAbbrev", DataType::Utf8, false),
ArrowField::new("path", DataType::Utf8, false),
]));
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int32Array::from(vec![1, 2, 3, 4])),
Arc::new(StringArray::from(vec!["D", "R", "R", "D"])),
Arc::new(StringArray::from(vec!["\\", "\\", "x", "x"])),
],
)
.unwrap();
let conn = connect(uri).execute().await.unwrap();
let table = conn.create_table("parties", batch).execute().await.unwrap();
let batches = table
.query()
.only_if(r#""PartyAbbrev" = 'D'"#)
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
assert_eq!(
table
.count_rows(Some(r#""PartyAbbrev" = 'D'"#.to_string()))
.await
.unwrap(),
2
);
// Public BaseTable dispatch cannot bypass canonicalization.
let query = AnyQuery::Query(QueryRequest {
filter: Some(QueryFilter::Sql(r#""PartyAbbrev" = 'D'"#.to_string())),
..Default::default()
});
let batches = table
.base_table()
.query(&query, Default::default())
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
assert_eq!(
table
.base_table()
.count_rows(Some(crate::table::Filter::Sql(
r#""PartyAbbrev" = 'D'"#.to_string(),
)))
.await
.unwrap(),
2
);
for predicate in [
r#"id = 1 -- unmatched " in a valid SQL comment"#,
r#"id = 1 /* unmatched " in a valid SQL comment */"#,
r#"id = 1 /*! OR "PartyAbbrev" = 'D' */"#,
r#"path = '\' AND "PartyAbbrev" = 'D'"#,
] {
let batches = table
.query()
.only_if(predicate)
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
}
// The same canonical predicate contract applies to both merge filters.
let source = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int32Array::from(vec![1, 2, 3])),
Arc::new(StringArray::from(vec!["D", "R", "R"])),
Arc::new(StringArray::from(vec!["\\", "\\", "x"])),
],
)
.unwrap();
let mut merge = table.merge_insert(&["id"]);
merge.when_not_matched_by_source_delete(Some(r#""PartyAbbrev" = 'D'"#.to_string()));
let result = table
.base_table()
.merge_insert(
merge,
Box::new(RecordBatchIterator::new(vec![Ok(source)], schema.clone())),
)
.await
.unwrap();
assert_eq!(result.num_deleted_rows, 1);
let source = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int32Array::from(vec![1, 2, 3])),
Arc::new(StringArray::from(vec!["U", "U", "U"])),
Arc::new(StringArray::from(vec!["\\", "\\", "x"])),
],
)
.unwrap();
let mut merge = table.merge_insert(&["id"]);
merge.when_matched_update_all(Some(r#"target."PartyAbbrev" = 'D'"#.to_string()));
merge
.execute(Box::new(RecordBatchIterator::new(vec![Ok(source)], schema)))
.await
.unwrap();
assert_eq!(
table
.count_rows(Some(r#""PartyAbbrev" = 'U'"#.to_string()))
.await
.unwrap(),
1
);
let update = table
.update()
.only_if(r#""PartyAbbrev" = 'R'"#)
.column("PartyAbbrev", "'X'");
table.base_table().update(update).await.unwrap();
assert_eq!(
table
.count_rows(Some(r#""PartyAbbrev" = 'X'"#.to_string()))
.await
.unwrap(),
2
);
let result = table
.base_table()
.delete(crate::table::Predicate::String(r#""PartyAbbrev" = 'X'"#))
.await
.unwrap();
assert_eq!(result.num_deleted_rows, 2);
assert_eq!(table.count_rows(None).await.unwrap(), 1);
}
#[tokio::test]
async fn test_select_with_transform() {
let batches = make_non_empty_batches();
@@ -2503,8 +2334,7 @@ mod tests {
.limit(1);
let plan = query.explain_plan(true).await.unwrap();
assert!(plan.contains("KNNVectorDistance: queries=2"));
assert!(!plan.contains("UnionExec"));
assert!(plan.contains("UnionExec"));
let results = query
.execute()
@@ -2519,100 +2349,6 @@ mod tests {
// We don't guarantee order.
assert!(query_index.values().contains(&0));
assert!(query_index.values().contains(&1));
// Batch KNN does not support a per-query offset, so offset queries keep
// the legacy per-vector plan to preserve their result semantics.
let offset_query = table
.query()
.nearest_to(&[0.1, 0.2, 0.3, 0.4])
.unwrap()
.add_query_vector(&[0.5, 0.6, 0.7, 0.8])
.unwrap()
.limit(1)
.offset(1);
assert!(
offset_query
.explain_plan(true)
.await
.unwrap()
.contains("UnionExec")
);
let offset_results = offset_query
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(
offset_results
.iter()
.map(RecordBatch::num_rows)
.sum::<usize>(),
2
);
}
#[tokio::test]
async fn test_multiple_binary_query_vectors() {
let vectors = FixedSizeListArray::from_iter_primitive::<UInt8Type, _, _>(
vec![
Some(vec![Some(0), Some(0)]),
Some(vec![Some(255), Some(255)]),
],
2,
);
let schema = Arc::new(ArrowSchema::new(vec![
ArrowField::new("id", DataType::Int32, false),
ArrowField::new("vector", vectors.data_type().clone(), false),
]));
let batch = RecordBatch::try_new(
schema,
vec![Arc::new(Int32Array::from(vec![0, 1])), Arc::new(vectors)],
)
.unwrap();
let conn = connect("memory://").execute().await.unwrap();
let table = conn
.create_table("binary_batch", batch)
.execute()
.await
.unwrap();
let query = table
.query()
.nearest_to(&[0.0, 0.0])
.unwrap()
.add_query_vector(&[255.0, 255.0])
.unwrap()
.distance_type(DistanceType::Hamming)
.limit(1);
// Binary queries retain the per-vector plan because Lance's binary
// nearest path requires primitive UInt8 query arrays.
assert!(
query
.explain_plan(true)
.await
.unwrap()
.contains("UnionExec")
);
let results = query
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
let results = concat_batches(&results[0].schema(), &results).unwrap();
assert_eq!(results.num_rows(), 2);
let ids = results["id"].as_primitive::<Int32Type>();
assert!(ids.values().contains(&0));
assert!(ids.values().contains(&1));
let query_index = results["query_index"].as_primitive::<Int32Type>();
assert!(query_index.values().contains(&0));
assert!(query_index.values().contains(&1));
}
#[tokio::test]
-4
View File
@@ -87,10 +87,6 @@ impl ServerVersion {
pub fn support_blobs(&self) -> bool {
self.0 >= semver::Version::new(0, 5, 0)
}
pub fn support_fts_document_granularity(&self) -> bool {
self.0 >= semver::Version::new(0, 6, 0)
}
}
pub const OPT_REMOTE_PREFIX: &str = "remote_database_";
+60 -286
View File
@@ -14,7 +14,6 @@ use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitio
use crate::expr::expr_to_sql_string;
use crate::index::Index;
use crate::index::IndexStatistics;
use crate::index::scalar::FtsQuery;
use crate::index::waiter::wait_for_index;
use crate::job::Job;
use crate::query::{QueryFilter, QueryRequest, Select, VectorQueryRequest};
@@ -40,8 +39,7 @@ use crate::table::{
use crate::table::{AnyQuery, Filter, Predicate, PreprocessingOutput, TableStatistics};
use crate::utils::background_cache::BackgroundCache;
use crate::utils::{
resolve_arrow_field_path, resolve_arrow_fts_field_path, supported_btree_data_type,
supported_vector_data_type,
resolve_arrow_field_path, supported_btree_data_type, supported_vector_data_type,
};
use crate::{DistanceType, Error};
use crate::{
@@ -90,32 +88,6 @@ const SCHEMA_CACHE_TTL: Duration = Duration::from_secs(30);
const SCHEMA_CACHE_REFRESH_WINDOW: Duration = Duration::from_secs(5);
const SCHEMA_SELECTOR_CHANGED: &str = "table selector changed while fetching schema";
fn fts_query_requires_document_granularity_support(query: &FtsQuery) -> bool {
match query {
FtsQuery::Match(query) => query
.document_granularity
.is_some_and(|granularity| granularity.is_list_element()),
FtsQuery::Phrase(query) => query
.document_granularity
.is_some_and(|granularity| granularity.is_list_element()),
FtsQuery::Boost(query) => {
fts_query_requires_document_granularity_support(&query.positive)
|| fts_query_requires_document_granularity_support(&query.negative)
}
FtsQuery::MultiMatch(query) => query.match_queries.iter().any(|query| {
query
.document_granularity
.is_some_and(|granularity| granularity.is_list_element())
}),
FtsQuery::Boolean(query) => query
.must
.iter()
.chain(&query.should)
.chain(&query.must_not)
.any(fts_query_requires_document_granularity_support),
}
}
/// Per-table state driving the freshness headers (`x-lancedb-min-version`,
/// `x-lancedb-min-timestamp`, and `x-lancedb-min-read-version`) sent on table
/// requests.
@@ -508,21 +480,8 @@ impl<S: HttpSend> RemoteTable<S> {
});
}
};
if matches!(
&index.index,
Index::FTS(params) if params.get_document_granularity().is_list_element()
) && !self.server_version.support_fts_document_granularity()
{
return Err(Error::NotSupported {
message: "FTS document granularity requires remote server version 0.6.0 or later"
.into(),
});
}
let schema = self.schema().await?;
let (canonical_column, field) = match &index.index {
Index::FTS(_) => resolve_arrow_fts_field_path(&schema, &column)?,
_ => resolve_arrow_field_path(&schema, &column)?,
};
let (canonical_column, field) = resolve_arrow_field_path(&schema, &column)?;
let mut body = serde_json::json!({
"column": canonical_column
});
@@ -558,13 +517,7 @@ impl<S: HttpSend> RemoteTable<S> {
Index::Bitmap(p) => ("BITMAP", Some(to_json(p)?)),
Index::LabelList(p) => ("LABEL_LIST", Some(to_json(p)?)),
Index::Fm(p) => ("FM", Some(to_json(p)?)),
Index::FTS(p) => {
let mut params = to_json(p)?;
if p.get_document_granularity().is_list_element() {
params["document_granularity"] = "list_element".into();
}
("FTS", Some(params))
}
Index::FTS(p) => ("FTS", Some(to_json(p)?)),
Index::Auto => {
if supported_vector_data_type(field.data_type()) {
body[METRIC_TYPE_KEY] =
@@ -1046,18 +999,6 @@ impl<S: HttpSend> RemoteTable<S> {
});
}
let requires_document_granularity_support =
fts_query_requires_document_granularity_support(&full_text_search.query);
if requires_document_granularity_support
&& !self.server_version.support_fts_document_granularity()
{
return Err(Error::NotSupported {
message:
"FTS document granularity requires remote server version 0.6.0 or later"
.into(),
});
}
if self.server_version.support_structural_fts() {
body["full_text_query"] = serde_json::json!({
"query": full_text_search.query.clone(),
@@ -1379,11 +1320,10 @@ impl<S: HttpSend> RemoteTable<S> {
query: &AnyQuery,
version: Option<u64>,
) -> Result<Vec<serde_json::Value>> {
let query = query.canonicalized()?;
let mut base_body = serde_json::json!({ "version": version });
self.apply_branch_body(&mut base_body);
match &query {
match query {
AnyQuery::Query(query) => {
let mut body = base_body.clone();
self.apply_query_params(&mut body, query)?;
@@ -2492,7 +2432,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
let mut body = if let Some(filter) = filter {
let filter_sql = match filter {
Filter::Sql(sql) => crate::expr::canonicalize_sql_predicate(&sql)?,
Filter::Sql(sql) => sql.clone(),
Filter::Datafusion(expr) => expr_to_sql_string(&expr)?,
};
serde_json::json!({ "predicate": filter_sql, "version": read_snapshot.version })
@@ -2748,8 +2688,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
Ok(final_analyze)
}
async fn update(&self, mut update: UpdateBuilder) -> Result<UpdateResult> {
update.canonicalize_filter()?;
async fn update(&self, update: UpdateBuilder) -> Result<UpdateResult> {
self.check_mutable().await?;
let request = self
.client
@@ -2796,7 +2735,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
async fn delete(&self, predicate: Predicate<'_>) -> Result<DeleteResult> {
self.check_mutable().await?;
let predicate_sql = match predicate {
Predicate::String(s) => crate::expr::canonicalize_sql_predicate(s)?,
Predicate::String(s) => s.to_string(),
Predicate::Expr(expr) => expr_to_sql_string(expr)?,
};
let mut body = serde_json::json!({ "predicate": predicate_sql });
@@ -2853,10 +2792,9 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
async fn merge_insert(
&self,
mut params: MergeInsertBuilder,
params: MergeInsertBuilder,
new_data: Box<dyn RecordBatchReader + Send>,
) -> Result<MergeResult> {
params.canonicalize_filters()?;
self.check_mutable().await?;
let timeout = params.timeout;
@@ -3691,7 +3629,7 @@ mod tests {
use arrow_schema::{DataType, Field, Schema};
use chrono::{DateTime, Utc};
use futures::{StreamExt, TryFutureExt, future::BoxFuture};
use lance_index::scalar::inverted::{DocumentGranularity, query::MatchQuery};
use lance_index::scalar::inverted::query::MatchQuery;
use lance_index::scalar::{FullTextSearchQuery, InvertedIndexParams};
use reqwest::Body;
use rstest::rstest;
@@ -3867,17 +3805,13 @@ mod tests {
);
assert_eq!(
request.body().unwrap().as_bytes().unwrap(),
br#"{"predicate":"`A` > 10","version":null}"#
br#"{"predicate":"a > 10","version":null}"#
);
http::Response::builder().status(200).body("42").unwrap()
});
let count = table
.base_table()
.count_rows(Some(Filter::Sql(r#""A" > 10"#.into())))
.await
.unwrap();
let count = table.count_rows(Some("a > 10".into())).await.unwrap();
assert_eq!(count, 42);
}
@@ -3972,15 +3906,6 @@ mod tests {
DataType::Struct(vec![Field::new("text", DataType::Utf8, false)].into()),
false,
),
Field::new(
"docs",
DataType::List(Arc::new(Field::new(
"item",
DataType::Struct(vec![Field::new("content", DataType::Utf8, true)].into()),
true,
))),
true,
),
Field::new(
"meta-data",
DataType::Struct(vec![Field::new("user-id", DataType::Int32, false)].into()),
@@ -4360,7 +4285,7 @@ mod tests {
assert_eq!(expression, "b - 1");
let only_if = value.get("predicate").unwrap().as_str().unwrap();
assert_eq!(only_if, "`B` > 10");
assert_eq!(only_if, "b > 10");
}
if old_server {
@@ -4376,12 +4301,14 @@ mod tests {
}
});
let update = table
let result = table
.update()
.column("a", "a + 1")
.column("b", "b - 1")
.only_if(r#""B" > 10"#);
let result = table.base_table().update(update).await.unwrap();
.only_if("b > 10")
.execute()
.await
.unwrap();
assert_eq!(result.version, if old_server { 0 } else { 43 });
assert_eq!(result.rows_updated, if old_server { 0 } else { 5 });
@@ -4468,10 +4395,10 @@ mod tests {
let params = request.url().query_pairs().collect::<HashMap<_, _>>();
assert_eq!(params["on"], "some_col");
assert_eq!(params["when_matched_update_all"], "true");
assert_eq!(params["when_matched_update_all"], "false");
assert_eq!(params["when_not_matched_insert_all"], "false");
assert_eq!(params["when_not_matched_by_source_delete"], "false");
assert_eq!(params["when_matched_update_all_filt"], "target.`A` > 0");
assert!(!params.contains_key("when_matched_update_all_filt"));
assert!(!params.contains_key("when_not_matched_by_source_delete_filt"));
assert!(!params.contains_key("use_index"));
@@ -4488,9 +4415,11 @@ mod tests {
}
});
let mut merge = table.merge_insert(&["some_col"]);
merge.when_matched_update_all(Some(r#"target."A" > 0"#.into()));
let result = table.base_table().merge_insert(merge, data).await.unwrap();
let result = table
.merge_insert(&["some_col"])
.execute(data)
.await
.unwrap();
assert_eq!(result.version, if old_server { 0 } else { 43 });
if !old_server {
@@ -4552,7 +4481,7 @@ mod tests {
let body = request.body().unwrap().as_bytes().unwrap();
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
let predicate = body.get("predicate").unwrap().as_str().unwrap();
assert_eq!(predicate, "`ID` in (1, 2, 3)");
assert_eq!(predicate, "id in (1, 2, 3)");
if old_server {
http::Response::builder()
@@ -4570,11 +4499,7 @@ mod tests {
}
});
let result = table
.base_table()
.delete(Predicate::String(r#""ID" in (1, 2, 3)"#))
.await
.unwrap();
let result = table.delete("id in (1, 2, 3)").await.unwrap();
assert_eq!(result.version, if old_server { 0 } else { 43 });
}
@@ -4666,7 +4591,6 @@ mod tests {
let body = request.body().unwrap().as_bytes().unwrap();
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
let expected_body = serde_json::json!({
"filter": "`A` > 0",
"k": isize::MAX as usize,
"prefilter": true,
"vector": [], // Empty vector means no vector query.
@@ -4682,13 +4606,9 @@ mod tests {
.unwrap()
});
let query = AnyQuery::Query(QueryRequest {
filter: Some(QueryFilter::Sql(r#""A" > 0"#.into())),
..Default::default()
});
let data = table
.base_table()
.query(&query, Default::default())
.query()
.execute()
.await
.unwrap()
.collect::<Vec<_>>()
@@ -5614,7 +5534,7 @@ mod tests {
#[tokio::test]
async fn test_query_structured_fts() {
let table =
Table::new_with_handler_version("my_table", semver::Version::new(0, 6, 0), |request| {
Table::new_with_handler_version("my_table", semver::Version::new(0, 3, 0), |request| {
assert_eq!(request.method(), "POST");
assert_eq!(request.url().path(), "/v1/table/my_table/query/");
assert_eq!(
@@ -5635,7 +5555,6 @@ mod tests {
"max_expansions": 50,
"operator": "Or",
"prefix_length": 0,
"document_granularity": "list_element",
},
}
},
@@ -5665,7 +5584,6 @@ mod tests {
.full_text_search(FullTextSearchQuery::new_query(
MatchQuery::new("hello world".to_owned())
.with_column(Some("payload.text".to_owned()))
.with_document_granularity(DocumentGranularity::ListElement)
.into(),
))
.with_row_id()
@@ -5675,76 +5593,6 @@ mod tests {
.unwrap();
}
#[tokio::test]
async fn test_query_row_document_granularity_uses_structured_fts() {
let table =
Table::new_with_handler_version("my_table", semver::Version::new(0, 3, 0), |request| {
let body = request.body().unwrap().as_bytes().unwrap();
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
assert_eq!(
body["full_text_query"]["query"]["match"]["document_granularity"],
"row"
);
let data = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])),
vec![Arc::new(Int32Array::from(vec![1]))],
)
.unwrap();
http::Response::builder()
.status(200)
.header(CONTENT_TYPE, ARROW_FILE_CONTENT_TYPE)
.body(write_ipc_file(&data))
.unwrap()
});
table
.query()
.full_text_search(FullTextSearchQuery::new_query(
MatchQuery::new("hello world".to_owned())
.with_column(Some("payload.text".to_owned()))
.with_document_granularity(DocumentGranularity::Row)
.into(),
))
.execute()
.await
.unwrap();
}
#[rstest]
#[case(DEFAULT_SERVER_VERSION.clone())]
#[case(semver::Version::new(0, 3, 0))]
#[case(semver::Version::new(0, 5, 0))]
#[tokio::test]
async fn test_query_document_granularity_requires_server_support(
#[case] version: semver::Version,
) {
let table =
Table::new_with_handler_version("my_table", version, |_| -> http::Response<String> {
panic!("unsupported remote query must fail before sending a request")
});
let result = table
.query()
.full_text_search(FullTextSearchQuery::new_query(
MatchQuery::new("hello world".to_owned())
.with_column(Some("payload.text".to_owned()))
.with_document_granularity(DocumentGranularity::ListElement)
.into(),
))
.execute()
.await;
let err = match result {
Ok(_) => panic!("legacy remote query unexpectedly succeeded"),
Err(err) => err,
};
assert!(
err.to_string()
.contains("document granularity requires remote server version 0.6.0 or later")
);
}
#[rstest]
#[case(DEFAULT_SERVER_VERSION.clone())]
#[case(semver::Version::new(0, 2, 0))]
@@ -6023,56 +5871,40 @@ mod tests {
"CAT".to_string(),
]))),
),
(
"FTS",
{
let mut body = serde_json::to_value(InvertedIndexParams::default()).unwrap();
body["document_granularity"] = "list_element".into();
body
},
Index::FTS(
InvertedIndexParams::default()
.document_granularity(DocumentGranularity::ListElement),
),
),
];
for (index_type, expected_body, index) in cases {
let table = Table::new_with_handler_version(
"my_table",
semver::Version::new(0, 6, 0),
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/" => {
assert_eq!(
request.headers().get("Content-Type").unwrap(),
JSON_CONTENT_TYPE
);
let body = request.body().unwrap().as_bytes().unwrap();
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
let mut expected_body = expected_body.clone();
expected_body["column"] = "a".into();
expected_body[INDEX_TYPE_KEY] = index_type.into();
assert_eq!(body, expected_body);
http::Response::builder()
.status(200)
.body("{}".to_string())
.unwrap()
}
path => panic!("Unexpected path: {}", path),
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/" => {
assert_eq!(
request.headers().get("Content-Type").unwrap(),
JSON_CONTENT_TYPE
);
let body = request.body().unwrap().as_bytes().unwrap();
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
let mut expected_body = expected_body.clone();
expected_body["column"] = "a".into();
expected_body[INDEX_TYPE_KEY] = index_type.into();
assert_eq!(body, expected_body);
http::Response::builder()
.status(200)
.body("{}".to_string())
.unwrap()
}
path => panic!("Unexpected path: {}", path),
}
});
table.create_index(&["a"], index).execute().await.unwrap();
}
@@ -6331,19 +6163,6 @@ mod tests {
body["index_type"] = "FTS".into();
body
},
{
let mut body = serde_json::to_value(InvertedIndexParams::default()).unwrap();
body["column"] = "docs.content".into();
body["index_type"] = "FTS".into();
body
},
{
let mut body = serde_json::to_value(InvertedIndexParams::default()).unwrap();
body["column"] = "docs.content".into();
body["index_type"] = "FTS".into();
body["document_granularity"] = "list_element".into();
body
},
json!({
"column": "`meta-data`.`user-id`",
"index_type": "BTREE",
@@ -6354,7 +6173,7 @@ mod tests {
}),
]);
let request_idx = Arc::new(AtomicUsize::new(0));
let table = Table::new_with_handler_version("my_table", semver::Version::new(0, 6, 0), {
let table = Table::new_with_handler("my_table", {
let schema = schema.clone();
let expected_requests = expected_requests.clone();
let request_idx = request_idx.clone();
@@ -6419,22 +6238,6 @@ mod tests {
.execute()
.await
.unwrap();
table
.create_index(&["Docs.Content"], Index::FTS(Default::default()))
.execute()
.await
.unwrap();
table
.create_index(
&["Docs.Content"],
Index::FTS(
InvertedIndexParams::default()
.document_granularity(DocumentGranularity::ListElement),
),
)
.execute()
.await
.unwrap();
table
.create_index(&["`META-DATA`.`USER-ID`"], Index::BTree(Default::default()))
.execute()
@@ -6449,35 +6252,6 @@ mod tests {
assert_eq!(request_idx.load(Ordering::SeqCst), expected_requests.len());
}
#[tokio::test]
async fn test_create_list_element_fts_requires_server_support() {
let table = Table::new_with_handler_version(
"my_table",
semver::Version::new(0, 5, 0),
|_| -> http::Response<String> {
panic!("unsupported index creation must fail before sending a request")
},
);
let result = table
.create_index(
&["docs.content"],
Index::FTS(
InvertedIndexParams::default()
.document_granularity(DocumentGranularity::ListElement),
),
)
.execute()
.await;
assert!(
result
.unwrap_err()
.to_string()
.contains("document granularity requires remote server version 0.6.0 or later")
);
}
#[tokio::test]
async fn test_list_indices() {
let schema = Schema::new(vec![
+6 -43
View File
@@ -59,9 +59,7 @@ use crate::index::{IndexConfig, IndexStatisticsImpl, IndexType};
use crate::job::Job;
use crate::query::{IntoQueryVector, Query, QueryExecutionOptions, TakeQuery, VectorQuery};
use crate::table::datafusion::insert::InsertExec;
use crate::utils::{
PatchReadParam, PatchWriteParam, public_fts_field_path_by_id, resolve_arrow_field_path,
};
use crate::utils::{PatchReadParam, PatchWriteParam, resolve_arrow_field_path};
use self::dataset::DatasetConsistencyWrapper;
use self::merge::MergeInsertBuilder;
@@ -1164,10 +1162,7 @@ impl Table {
///
/// * `filter` if present, only count rows matching the filter
pub async fn count_rows(&self, filter: Option<String>) -> Result<usize> {
let filter = filter
.map(|predicate| crate::expr::canonicalize_sql_predicate(&predicate).map(Filter::Sql))
.transpose()?;
self.inner.count_rows(filter).await
self.inner.count_rows(filter.map(Filter::Sql)).await
}
/// Names of the blob v2 columns in this table, in declaration order.
@@ -1367,13 +1362,7 @@ impl Table {
/// # });
/// ```
pub async fn delete(&self, predicate: impl Into<Predicate<'_>>) -> Result<DeleteResult> {
match predicate.into() {
Predicate::String(predicate) => {
let predicate = crate::expr::canonicalize_sql_predicate(predicate)?;
self.inner.delete(Predicate::String(&predicate)).await
}
predicate @ Predicate::Expr(_) => self.inner.delete(predicate).await,
}
self.inner.delete(predicate.into()).await
}
/// Create an index on the provided column(s).
@@ -1786,23 +1775,7 @@ impl Table {
self.inner.alter_columns(alterations).await
}
/// Update per-field (column) metadata.
///
/// Each [`FieldMetadataUpdate`] is merged into the field's existing metadata
/// by default; use [`FieldMetadataUpdate::remove`] to delete a key, or
/// [`FieldMetadataUpdate::replace`] to swap the field's entire metadata map.
///
/// The following keys are treated specially, by convention, and should be
/// used when appropriate:
///
/// - `lancedb:description`: for a human-readable description of a field.
/// - `lancedb:tag:<name>`: for a user-defined key-value tag, where the suffix
/// names the tag category; e.g. `lancedb:tag:model: "clip"`.
/// - `lancedb:logical-column`: for a column grouping; e.g. `feature_v1` and
/// `feature_v2` might be in the same logical column.
/// - `lancedb:status`: for status options (`production`, `candidate`,
/// `deprecated`, `archived`) to designate the current life cycle state of
/// this column.
/// Update per-field metadata (merges by default).
pub async fn update_field_metadata(
&self,
updates: &[FieldMetadataUpdate],
@@ -3248,10 +3221,7 @@ impl BaseTable for NativeTable {
let dataset = self.dataset.get().await?;
match filter {
None => Ok(dataset.count_rows(None).await?),
Some(Filter::Sql(sql)) => {
let sql = crate::expr::canonicalize_sql_predicate(&sql)?;
Ok(dataset.count_rows(Some(sql)).await?)
}
Some(Filter::Sql(sql)) => Ok(dataset.count_rows(Some(sql)).await?),
Some(Filter::Datafusion(_)) => Err(Error::NotSupported {
message: "Datafusion filters are not yet supported".to_string(),
}),
@@ -3589,14 +3559,7 @@ impl BaseTable for NativeTable {
let field_ids = idx_desc.field_ids();
let mut columns = Vec::with_capacity(field_ids.len());
for field_id in field_ids {
let field_path = match if index_type == crate::index::IndexType::FTS {
public_fts_field_path_by_id(dataset.schema(), *field_id as i32)
} else {
dataset
.schema()
.field_path(*field_id as i32)
.map_err(Into::into)
} {
let field_path = match dataset.schema().field_path(*field_id as i32) {
Ok(field_path) => field_path,
Err(e) => {
log::warn!(
+4 -50
View File
@@ -28,9 +28,8 @@ pub(super) type PreparedIndex = (String, Box<dyn lance::index::IndexParams>, Ind
use crate::index::Index;
use crate::index::vector::{VectorIndex, suggested_num_sub_vectors};
use crate::utils::{
resolve_lance_fts_field_path, supported_bitmap_data_type, supported_btree_data_type,
supported_fm_data_type, supported_fts_data_type, supported_label_list_data_type,
supported_vector_data_type,
supported_bitmap_data_type, supported_btree_data_type, supported_fm_data_type,
supported_fts_data_type, supported_label_list_data_type, supported_vector_data_type,
};
use super::NativeTable;
@@ -123,20 +122,7 @@ impl NativeTable {
}
self.dataset.ensure_mutable()?;
let dataset = self.dataset.get().await?;
let (column, field) = if let Index::FTS(params) = &opts.index {
let resolved = resolve_lance_fts_field_path(dataset.schema(), &opts.columns[0])?;
if params.get_document_granularity().is_list_element() && resolved.list_depth == 0 {
return Err(Error::InvalidInput {
message: format!(
"FTS field path '{}' has no List layer and cannot use ListElement document granularity",
resolved.canonical_path
),
});
}
(resolved.canonical_path, resolved.field)
} else {
Self::resolve_index_field(dataset.schema(), &opts.columns[0])?
};
let (column, field) = Self::resolve_index_field(dataset.schema(), &opts.columns[0])?;
let params = self.make_index_params(&field, opts.index.clone()).await?;
let index_type = self.get_index_type_for_field(&field, &opts.index);
Ok((column, params, index_type))
@@ -450,7 +436,7 @@ mod tests {
use crate::connection::ConnectBuilder;
use crate::index::Index;
use crate::index::scalar::{
BTreeIndexBuilder, BitmapIndexBuilder, DocumentGranularity, FmIndexBuilder, FtsIndexBuilder,
BTreeIndexBuilder, BitmapIndexBuilder, FmIndexBuilder, FtsIndexBuilder,
};
use crate::index::vector::{
IvfHnswFlatIndexBuilder, IvfHnswPqIndexBuilder, IvfHnswSqIndexBuilder,
@@ -567,38 +553,6 @@ mod tests {
job.cancel().await.unwrap();
}
#[tokio::test]
async fn test_execute_async_validates_fts_input_before_starting_job() {
let conn = connect("memory://").execute().await.unwrap();
let batch =
record_batch!(("id", Int32, [1, 2]), ("text", Utf8, ["alpha", "beta"])).unwrap();
let table = conn.create_table("t", batch).execute().await.unwrap();
let missing = table
.create_index(&["missing"], Index::FTS(FtsIndexBuilder::default()))
.execute_async()
.await;
assert!(missing.is_err());
let invalid_type = table
.create_index(&["id"], Index::FTS(FtsIndexBuilder::default()))
.execute_async()
.await;
assert!(invalid_type.is_err());
let invalid_granularity = table
.create_index(
&["text"],
Index::FTS(
FtsIndexBuilder::default()
.document_granularity(DocumentGranularity::ListElement),
),
)
.execute_async()
.await;
assert!(invalid_granularity.is_err());
}
/// Concurrent waiters, and a wait issued after the job settled, all
/// succeed once the build does.
#[tokio::test]
+1 -2
View File
@@ -31,9 +31,8 @@ pub(crate) async fn execute_delete(
table.dataset.ensure_mutable()?;
match predicate {
Predicate::String(s) => {
let predicate = crate::expr::canonicalize_sql_predicate(s)?;
let mut dataset = (*table.dataset.get().await?).clone();
let delete_result = dataset.delete(&predicate).boxed().await?;
let delete_result = dataset.delete(s).boxed().await?;
let num_deleted_rows = delete_result.num_deleted_rows;
let version = dataset.version().version;
table.dataset.update(dataset);
+2 -26
View File
@@ -220,32 +220,9 @@ impl MergeInsertBuilder {
///
/// Returns version and statistics about the merge operation including the number of rows
/// inserted, updated, and deleted.
pub async fn execute(
mut self,
new_data: Box<dyn RecordBatchReader + Send>,
) -> Result<MergeResult> {
self.canonicalize_filters()?;
pub async fn execute(self, new_data: Box<dyn RecordBatchReader + Send>) -> Result<MergeResult> {
self.table.clone().merge_insert(self, new_data).await
}
pub(crate) fn canonicalize_filters(&mut self) -> Result<()> {
self.when_matched_update_all_filt =
canonicalize_merge_filter(self.when_matched_update_all_filt.take())?;
self.when_not_matched_by_source_delete_filt =
canonicalize_merge_filter(self.when_not_matched_by_source_delete_filt.take())?;
Ok(())
}
}
fn canonicalize_merge_filter(filter: Option<MergeFilter>) -> Result<Option<MergeFilter>> {
filter
.map(|filter| match filter {
MergeFilter::Sql(predicate) => {
crate::expr::canonicalize_sql_predicate(&predicate).map(MergeFilter::Sql)
}
filter @ MergeFilter::Expr(_) => Ok(filter),
})
.transpose()
}
/// Internal implementation of the merge insert logic
@@ -253,10 +230,9 @@ fn canonicalize_merge_filter(filter: Option<MergeFilter>) -> Result<Option<Merge
/// This logic was moved from NativeTable::merge_insert to keep table.rs clean.
pub(crate) async fn execute_merge_insert(
table: &NativeTable,
mut params: MergeInsertBuilder,
params: MergeInsertBuilder,
new_data: Box<dyn RecordBatchReader + Send>,
) -> Result<MergeResult> {
params.canonicalize_filters()?;
super::computed_columns::ensure_no_function_bindings_for_mutation(
table.schema().await?.as_ref(),
"merge_insert",
+24 -76
View File
@@ -21,6 +21,7 @@ use datafusion_physical_plan::ExecutionPlan;
use datafusion_physical_plan::projection::ProjectionExec;
use datafusion_physical_plan::repartition::RepartitionExec;
use datafusion_physical_plan::union::UnionExec;
use futures::future::try_join_all;
use lance::dataset::mem_wal::DatasetMemWalExt;
use lance::dataset::scanner::DatasetRecordBatchStream;
use lance::dataset::scanner::Scanner;
@@ -44,22 +45,6 @@ impl AnyQuery {
Self::VectorQuery(query) => &query.base,
}
}
fn base_mut(&mut self) -> &mut QueryRequest {
match self {
Self::Query(query) => query,
Self::VectorQuery(query) => &mut query.base,
}
}
/// Canonicalize any raw SQL filter immediately before backend dispatch.
pub(crate) fn canonicalized(&self) -> Result<Self> {
let mut query = self.clone();
if let Some(QueryFilter::Sql(predicate)) = &mut query.base_mut().filter {
*predicate = crate::expr::canonicalize_sql_predicate(predicate)?;
}
Ok(query)
}
}
//Decide between namespace or local
@@ -68,16 +53,15 @@ pub async fn execute_query(
query: &AnyQuery,
options: QueryExecutionOptions,
) -> Result<DatasetRecordBatchStream> {
let query = query.canonicalized()?;
// QueryTable pushdown runs the query server-side, but only on the main
// branch: the namespace request carries no branch yet, so a branch handle
// must fall through to local execution.
if can_execute_namespace_query(table, &query).await?
if can_execute_namespace_query(table, query).await?
&& let Some(ref namespace_client) = table.namespace_client
{
return execute_namespace_query(table, namespace_client.clone(), &query, options).await;
return execute_namespace_query(table, namespace_client.clone(), query, options).await;
}
execute_generic_query(table, &query, options).await
execute_generic_query(table, query, options).await
}
async fn can_execute_namespace_query(table: &NativeTable, query: &AnyQuery) -> Result<bool> {
@@ -152,10 +136,9 @@ pub async fn create_plan(
query: &AnyQuery,
options: QueryExecutionOptions,
) -> Result<Arc<dyn ExecutionPlan>> {
let query = query.canonicalized()?;
let query = match query {
AnyQuery::VectorQuery(query) => query,
AnyQuery::Query(query) => VectorQueryRequest::from_plain_query(query),
AnyQuery::VectorQuery(query) => query.clone(),
AnyQuery::Query(query) => VectorQueryRequest::from_plain_query(query.clone()),
};
query.base.check_filter()?;
@@ -187,7 +170,6 @@ pub async fn create_plan(
let mut column = query.column.clone();
let mut query_vector = query.query_vector.first().cloned();
let mut is_batch_query = false;
if query.query_vector.len() > 1 {
if column.is_none() {
// Infer a vector column with the same dimension of the query vector.
@@ -198,37 +180,16 @@ pub async fn create_plan(
)?);
}
let vector_field = schema.field(column.as_ref().unwrap()).unwrap();
let (_, element_type) =
lance::index::vector::utils::get_vector_type(schema, column.as_ref().unwrap())?;
let is_binary = matches!(element_type, DataType::UInt8);
if matches!(vector_field.data_type(), DataType::List(_))
|| (query.base.offset.unwrap_or(0) == 0 && !is_binary)
{
// Lance distinguishes these cases from the vector column type: a
// list-like query against a List column is one multivector query,
// while the same query against a FixedSizeList column is a batch of
// independent queries. The batch path shares a single flat scan and
// bounds retained candidate data instead of running one scan per
// query vector.
if let DataType::List(_) = vector_field.data_type() {
// Multivector handling: concatenate into FixedSizeList<FixedSizeList<_>>
let vectors = query
.query_vector
.iter()
.map(|arr| arr.as_ref())
.collect::<Vec<_>>();
let dim = vectors[0].len();
if let Some((query_index, actual_dim)) = vectors
.iter()
.enumerate()
.find_map(|(index, vector)| (vector.len() != dim).then_some((index, vector.len())))
{
return Err(Error::InvalidInput {
message: format!(
"query vector at index {query_index} has dimension {actual_dim}, expected {dim}"
),
});
}
let mut fsl_builder = FixedSizeListBuilder::with_capacity(
Float32Builder::with_capacity(dim * vectors.len()),
Float32Builder::with_capacity(dim),
dim as i32,
vectors.len(),
);
@@ -239,12 +200,8 @@ pub async fn create_plan(
fsl_builder.append(true);
}
query_vector = Some(Arc::new(fsl_builder.finish()));
is_batch_query = !matches!(vector_field.data_type(), DataType::List(_));
} else {
// Lance's batch path has no per-query offset, and its binary path
// requires primitive UInt8 queries rather than a fixed-size list.
// Keep the prior plan shape for these cases so offsets are applied
// per query and binary query vectors retain their primitive shape.
// Multiple query vectors: create a plan for each and union them
let query_vecs = query.query_vector.clone();
let plan_futures = query_vecs
.into_iter()
@@ -257,7 +214,7 @@ pub async fn create_plan(
}
})
.collect::<Vec<_>>();
let plans = futures::future::try_join_all(plan_futures).await?;
let plans = try_join_all(plan_futures).await?;
return create_multi_vector_plan(plans);
}
}
@@ -294,14 +251,10 @@ pub async fn create_plan(
}
}
// For a batch query, `nearest` already applies k to each query vector.
// Adding Scanner's global limit would truncate the combined result to k rows.
if !is_batch_query {
scanner.limit(
query.base.limit.map(|limit| limit as i64),
query.base.offset.map(|offset| offset as i64),
)?;
}
scanner.limit(
query.base.limit.map(|limit| limit as i64),
query.base.offset.map(|offset| offset as i64),
)?;
if let Some(ef) = query.ef {
scanner.ef(ef);
@@ -1135,7 +1088,7 @@ mod tests {
}
#[tokio::test]
async fn test_create_plan_batch_vector_uses_shared_scan() {
async fn test_create_plan_multivector_structure() {
use arrow_array::{Float32Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema};
use datafusion_physical_plan::display::DisplayableExecutionPlan;
@@ -1162,18 +1115,11 @@ mod tests {
.unwrap();
let native_table = table.as_native().unwrap();
// A batch of vectors against a fixed-size vector column should use
// Lance's native batch KNN path instead of independent scan plans.
// This triggers the "create_multi_vector_plan" logic branch
let q1 = Arc::new(Float32Array::from(vec![1.0, 2.0]));
let q2 = Arc::new(Float32Array::from(vec![3.0, 4.0]));
let req = VectorQueryRequest {
base: QueryRequest {
filter: Some(QueryFilter::Sql("id >= 0".to_string())),
limit: Some(1),
select: Select::Columns(vec!["id".to_string()]),
..Default::default()
},
column: Some("vector".to_string()),
query_vector: vec![q1, q2],
..Default::default()
@@ -1190,17 +1136,19 @@ mod tests {
.indent(true)
.to_string();
// We expect a RepartitionExec wrapping a UnionExec
assert!(
display.contains("KNNVectorDistance: queries=2"),
"plan should use native batch KNN, got:\n{display}"
display.contains("RepartitionExec"),
"Plan should include Repartitioning"
);
assert!(
!display.contains("UnionExec"),
"flat batch KNN should share one scan, got:\n{display}"
display.contains("UnionExec"),
"Plan should include a Union of multiple searches"
);
// We expect the projection to add the 'query_index' column (logic inside multi_vector_plan)
assert!(
display.contains("query_index"),
"plan should add query_index column, got:\n{display}"
"Plan should add query_index column"
);
}
+1 -3
View File
@@ -55,9 +55,7 @@ pub struct DropColumnsResult {
pub struct FieldMetadataUpdate {
/// Dot-separated path to the field (e.g. `"embedding"` or `"address.zip"`).
pub path: String,
/// Keys to set (`Some`) or delete (`None`). See
/// [`Table::update_field_metadata`](crate::Table::update_field_metadata) for
/// the conventional `lancedb:*` keys.
/// Keys to set (`Some`) or delete (`None`).
pub metadata: HashMap<String, Option<String>>,
/// If `true`, replace the field's entire metadata map instead of merging.
pub replace: bool,
+2 -13
View File
@@ -62,33 +62,22 @@ impl UpdateBuilder {
}
/// Executes the update operation.
pub async fn execute(mut self) -> Result<UpdateResult> {
pub async fn execute(self) -> Result<UpdateResult> {
if self.columns.is_empty() {
Err(Error::InvalidInput {
message: "at least one column must be specified in an update operation".to_string(),
})
} else {
self.canonicalize_filter()?;
self.parent.clone().update(self).await
}
}
pub(crate) fn canonicalize_filter(&mut self) -> Result<()> {
self.filter = self
.filter
.take()
.map(|predicate| crate::expr::canonicalize_sql_predicate(&predicate))
.transpose()?;
Ok(())
}
}
/// Internal implementation of the update logic
pub(crate) async fn execute_update(
table: &NativeTable,
mut update: UpdateBuilder,
update: UpdateBuilder,
) -> Result<UpdateResult> {
update.canonicalize_filter()?;
table.dataset.ensure_mutable()?;
// 1. Snapshot the current dataset
-183
View File
@@ -225,159 +225,6 @@ pub(crate) fn resolve_arrow_field_path(schema: &Schema, column: &str) -> Result<
Ok((canonical_path, Field::from(*field)))
}
pub(crate) struct ResolvedFtsField {
pub canonical_path: String,
pub field: Field,
pub list_depth: usize,
}
/// Canonicalize a public FTS field path while keeping Arrow list item names hidden.
pub(crate) fn resolve_lance_fts_field_path(
schema: &lance_core::datatypes::Schema,
column: &str,
) -> Result<ResolvedFtsField> {
let names =
lance_core::datatypes::parse_field_path(column).map_err(|e| Error::InvalidInput {
message: format!("Invalid field path `{}`: {}", column, e),
})?;
let (root_name, remaining_names) = names.split_first().ok_or_else(|| Error::InvalidInput {
message: "FTS field path cannot be empty".to_string(),
})?;
let mut field = schema
.fields
.iter()
.find(|field| field.name == *root_name)
.or_else(|| {
schema
.fields
.iter()
.find(|field| field.name.eq_ignore_ascii_case(root_name))
})
.ok_or_else(|| fts_field_not_found(schema, column))?;
let mut canonical_names = vec![field.name.clone()];
let mut list_depth = 0;
for name in remaining_names {
while matches!(
field.data_type(),
DataType::List(_) | DataType::LargeList(_)
) {
list_depth += 1;
field = field.children.first().ok_or_else(|| Error::Schema {
message: format!(
"FTS field path `{}` has a list without an item field",
column
),
})?;
}
if !matches!(field.data_type(), DataType::Struct(_)) {
return Err(fts_field_not_found(schema, column));
}
field = field
.children
.iter()
.find(|field| field.name == *name)
.or_else(|| {
field
.children
.iter()
.find(|field| field.name.eq_ignore_ascii_case(name))
})
.ok_or_else(|| fts_field_not_found(schema, column))?;
canonical_names.push(field.name.clone());
}
let mut terminal = field;
while matches!(
terminal.data_type(),
DataType::List(_) | DataType::LargeList(_)
) {
list_depth += 1;
terminal = terminal.children.first().ok_or_else(|| Error::Schema {
message: format!(
"FTS field path `{}` has a list without an item field",
column
),
})?;
}
let canonical_path = lance_core::datatypes::format_field_path(
&canonical_names
.iter()
.map(String::as_str)
.collect::<Vec<_>>(),
);
Ok(ResolvedFtsField {
canonical_path,
field: Field::from(field),
list_depth,
})
}
fn fts_field_not_found(schema: &lance_core::datatypes::Schema, column: &str) -> Error {
Error::Schema {
message: format!(
"Field path `{}` not found in schema. Available field paths: {}",
column,
schema.field_paths().join(", ")
),
}
}
fn find_public_fts_field_path_by_id(
field: &lance_core::datatypes::Field,
field_id: i32,
path: &mut Vec<String>,
) -> bool {
if field.id == field_id {
return true;
}
match field.data_type() {
DataType::List(_) | DataType::LargeList(_) => field
.children
.first()
.is_some_and(|child| find_public_fts_field_path_by_id(child, field_id, path)),
DataType::Struct(_) => field.children.iter().any(|child| {
path.push(child.name.clone());
let found = find_public_fts_field_path_by_id(child, field_id, path);
if !found {
path.pop();
}
found
}),
_ => false,
}
}
pub(crate) fn public_fts_field_path_by_id(
schema: &lance_core::datatypes::Schema,
field_id: i32,
) -> Result<String> {
for root in &schema.fields {
let mut path = vec![root.name.clone()];
if find_public_fts_field_path_by_id(root, field_id, &mut path) {
return Ok(lance_core::datatypes::format_field_path(
&path.iter().map(String::as_str).collect::<Vec<_>>(),
));
}
}
Err(Error::Schema {
message: format!("Field id `{}` not found in schema", field_id),
})
}
pub(crate) fn resolve_arrow_fts_field_path(
schema: &Schema,
column: &str,
) -> Result<(String, Field)> {
let lance_schema =
lance_core::datatypes::Schema::try_from(schema).map_err(|e| Error::Schema {
message: format!("Invalid schema: {}", e),
})?;
let resolved = resolve_lance_fts_field_path(&lance_schema, column)?;
Ok((resolved.canonical_path, resolved.field))
}
pub fn supported_btree_data_type(dtype: &DataType) -> bool {
dtype.is_integer()
|| dtype.is_floating()
@@ -633,36 +480,6 @@ mod tests {
use super::*;
#[test]
fn test_public_fts_field_path_prefers_exact_case() {
let text_list = || {
DataType::List(Arc::new(Field::new(
"item",
DataType::Struct(vec![Field::new("content", DataType::Utf8, true)].into()),
true,
)))
};
let schema = Schema::new(vec![
Field::new("Docs", text_list(), true),
Field::new("docs", text_list(), true),
]);
let (path, _) = resolve_arrow_fts_field_path(&schema, "docs.content").unwrap();
assert_eq!(path, "docs.content");
let lance_schema = lance_core::datatypes::Schema::try_from(&schema).unwrap();
let field_id = lance_schema
.resolve_case_insensitive("docs.item.content")
.unwrap()
.last()
.unwrap()
.id;
assert_eq!(
public_fts_field_path_by_id(&lance_schema, field_id).unwrap(),
"docs.content"
);
}
#[test]
fn test_guess_default_column() {
let schema_no_vector = Schema::new(vec![