mirror of
https://github.com/lancedb/lancedb.git
synced 2026-08-28 08:58:41 +00:00
Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| da86d804ba | |||
| 77a93fee76 | |||
| 7bb501839a | |||
| 5b347afd99 | |||
| 706a9c327f | |||
| be290447d9 | |||
| 79ba076429 |
+1
-1
@@ -1,5 +1,5 @@
|
||||
[tool.bumpversion]
|
||||
current_version = "0.37.1-beta.0"
|
||||
current_version = "0.37.1-beta.1"
|
||||
parse = """(?x)
|
||||
(?P<major>0|[1-9]\\d*)\\.
|
||||
(?P<minor>0|[1-9]\\d*)\\.
|
||||
|
||||
Generated
+47
-46
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
|
||||
|
||||
[[package]]
|
||||
name = "fsst"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"rand 0.9.5",
|
||||
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
|
||||
|
||||
[[package]]
|
||||
name = "lance"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"arrow",
|
||||
@@ -4890,8 +4890,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-arrow"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4913,7 +4913,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-arrow-scalar"
|
||||
version = "58.0.0"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4927,7 +4927,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "lance-arrow-stats"
|
||||
version = "58.0.0"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -4936,8 +4936,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-bitpacking"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrayref",
|
||||
"crunchy",
|
||||
@@ -4947,8 +4947,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-core"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4988,8 +4988,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-datafusion"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5019,8 +5019,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-datagen"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5037,8 +5037,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-derive"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -5047,8 +5047,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-encoding"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-arith",
|
||||
"arrow-array",
|
||||
@@ -5082,8 +5082,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-file"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-arith",
|
||||
"arrow-array",
|
||||
@@ -5114,8 +5114,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-index"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"arrow",
|
||||
@@ -5182,8 +5182,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-index-core"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -5205,8 +5205,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-io"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5242,8 +5242,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-linalg"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -5259,8 +5259,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"async-trait",
|
||||
@@ -5272,8 +5272,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace-impls"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-ipc",
|
||||
@@ -5303,7 +5303,6 @@ dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sha2 0.10.9",
|
||||
"time",
|
||||
"tokio",
|
||||
"tower",
|
||||
"tower-http 0.5.2",
|
||||
@@ -5327,8 +5326,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-select"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -5343,8 +5342,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-table"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5354,6 +5353,7 @@ dependencies = [
|
||||
"async-trait",
|
||||
"aws-credential-types",
|
||||
"aws-sdk-dynamodb",
|
||||
"blake3",
|
||||
"byteorder",
|
||||
"bytes",
|
||||
"chrono",
|
||||
@@ -5383,8 +5383,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-testing"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -5397,8 +5397,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-tokenizer"
|
||||
version = "11.0.0-beta.2"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.2#35da5d920159b49d1b53032652f7615ab699c160"
|
||||
version = "11.0.0-beta.3"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
|
||||
dependencies = [
|
||||
"frostem",
|
||||
"icu_segmenter",
|
||||
@@ -5411,7 +5411,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lancedb"
|
||||
version = "0.37.1-beta.0"
|
||||
version = "0.37.1-beta.1"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"anyhow",
|
||||
@@ -5480,6 +5480,7 @@ dependencies = [
|
||||
"random_word",
|
||||
"regex",
|
||||
"reqwest 0.12.28",
|
||||
"roaring",
|
||||
"rstest",
|
||||
"semver",
|
||||
"serde",
|
||||
@@ -5499,7 +5500,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lancedb-nodejs"
|
||||
version = "0.37.1-beta.0"
|
||||
version = "0.37.1-beta.1"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -5524,7 +5525,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lancedb-python"
|
||||
version = "0.37.1-beta.0"
|
||||
version = "0.37.1-beta.1"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"async-trait",
|
||||
|
||||
+14
-14
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
|
||||
rust-version = "1.91.0"
|
||||
|
||||
[workspace.dependencies]
|
||||
lance = { "version" = "=11.0.0-beta.2", default-features = false, "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=11.0.0-beta.2", default-features = false, "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=11.0.0-beta.2", default-features = false, "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
ahash = "0.8"
|
||||
# Note that this one does not include pyarrow
|
||||
arrow = { version = "58.0.0", optional = false }
|
||||
|
||||
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
|
||||
<dependency>
|
||||
<groupId>com.lancedb</groupId>
|
||||
<artifactId>lancedb-core</artifactId>
|
||||
<version>0.37.1-beta.0</version>
|
||||
<version>0.37.1-beta.1</version>
|
||||
</dependency>
|
||||
```
|
||||
|
||||
|
||||
@@ -431,9 +431,10 @@ Read the [LsmWriteSpec](../interfaces/LsmWriteSpec.md) currently installed on th
|
||||
|
||||
Resolves to `undefined` when the MemWAL LSM write path is not enabled (no
|
||||
spec has been set, or it was removed with [Table#unsetLsmWriteSpec](Table.md#unsetlsmwritespec)).
|
||||
The returned spec — including its `maintainedIndexes` and
|
||||
`writerConfigDefaults` — mirrors what was passed to
|
||||
[Table#setLsmWriteSpec](Table.md#setlsmwritespec).
|
||||
The returned spec mirrors what was passed to
|
||||
[Table#setLsmWriteSpec](Table.md#setlsmwritespec), except that `maintainedIndexes` always
|
||||
reports the concrete list resolved when the spec was set — `undefined`
|
||||
never round-trips.
|
||||
|
||||
#### Returns
|
||||
|
||||
@@ -806,6 +807,11 @@ All variants require the table to have an unenforced primary key
|
||||
([Table#setUnenforcedPrimaryKey](Table.md#setunenforcedprimarykey)); bucket sharding additionally
|
||||
requires it to be the single column being bucketed.
|
||||
|
||||
Omitting `maintainedIndexes` maintains every index on the table, resolved
|
||||
here, failing if one cannot be maintained — name them to install anyway.
|
||||
Naming them pins an exact set, and a still-building index is rejected
|
||||
rather than quietly omitted.
|
||||
|
||||
#### Parameters
|
||||
|
||||
* **spec**: [`LsmWriteSpec`](../interfaces/LsmWriteSpec.md)
|
||||
|
||||
@@ -34,7 +34,9 @@ Bucket and identity variants: the sharding column.
|
||||
optional maintainedIndexes: string[];
|
||||
```
|
||||
|
||||
Names of indexes the MemWAL should keep up to date during writes.
|
||||
Indexes the MemWAL keeps up to date. Omit to maintain every supported
|
||||
index, resolved on install — a snapshot, so indexes created later are not
|
||||
maintained. Pass `[]` for none.
|
||||
|
||||
***
|
||||
|
||||
|
||||
@@ -44,4 +44,7 @@ The number of rows in the table
|
||||
totalBytes: number;
|
||||
```
|
||||
|
||||
The total number of bytes in the table
|
||||
The total size, in bytes, of the table's data files, index files, and
|
||||
overlay files
|
||||
|
||||
Read from the manifest, so this excludes deletion files and manifests.
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
<parent>
|
||||
<groupId>com.lancedb</groupId>
|
||||
<artifactId>lancedb-parent</artifactId>
|
||||
<version>0.37.1-beta.0</version>
|
||||
<version>0.37.1-beta.1</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
|
||||
|
||||
+2
-2
@@ -6,7 +6,7 @@
|
||||
|
||||
<groupId>com.lancedb</groupId>
|
||||
<artifactId>lancedb-parent</artifactId>
|
||||
<version>0.37.1-beta.0</version>
|
||||
<version>0.37.1-beta.1</version>
|
||||
<packaging>pom</packaging>
|
||||
<name>${project.artifactId}</name>
|
||||
<description>LanceDB Java SDK Parent POM</description>
|
||||
@@ -28,7 +28,7 @@
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
<arrow.version>15.0.0</arrow.version>
|
||||
<lance-core.version>11.0.0-beta.2</lance-core.version>
|
||||
<lance-core.version>11.0.0-beta.3</lance-core.version>
|
||||
<spotless.skip>false</spotless.skip>
|
||||
<spotless.version>2.30.0</spotless.version>
|
||||
<spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
[package]
|
||||
name = "lancedb-nodejs"
|
||||
edition.workspace = true
|
||||
version = "0.37.1-beta.0"
|
||||
version = "0.37.1-beta.1"
|
||||
publish = false
|
||||
license.workspace = true
|
||||
description.workspace = true
|
||||
|
||||
@@ -69,6 +69,33 @@ describe("given a connection", () => {
|
||||
await expect(tbl.countRows()).resolves.toBe(1);
|
||||
});
|
||||
|
||||
it("should isolate object-form table creation across databases", async () => {
|
||||
const otherTmpDir = tmp.dirSync({ unsafeCleanup: true });
|
||||
const otherDb = await connect(otherTmpDir.name);
|
||||
|
||||
try {
|
||||
const firstTable = await db.createTable({
|
||||
name: "defaultTable",
|
||||
data: [{ rowId: "id1", vector: Array(384).fill(0) }],
|
||||
});
|
||||
const secondTable = await otherDb.createTable({
|
||||
name: "defaultTable",
|
||||
data: [{ rowId: "id2", vector: Array(384).fill(0) }],
|
||||
});
|
||||
|
||||
await expect(db.tableNames()).resolves.toEqual(["defaultTable"]);
|
||||
await expect(otherDb.tableNames()).resolves.toEqual(["defaultTable"]);
|
||||
|
||||
const firstRows = await firstTable.query().select(["rowId"]).toArray();
|
||||
const secondRows = await secondTable.query().select(["rowId"]).toArray();
|
||||
expect(firstRows.map((row) => row.rowId)).toEqual(["id1"]);
|
||||
expect(secondRows.map((row) => row.rowId)).toEqual(["id2"]);
|
||||
} finally {
|
||||
otherDb.close();
|
||||
otherTmpDir.removeCallback();
|
||||
}
|
||||
});
|
||||
|
||||
it("should be able to drop tables`", async () => {
|
||||
await db.createTable("test", [{ id: 1 }, { id: 2 }]);
|
||||
await db.createTable("test2", [{ id: 1 }, { id: 2 }]);
|
||||
|
||||
@@ -277,8 +277,16 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
|
||||
},
|
||||
numIndices: 0,
|
||||
numRows: 3,
|
||||
totalBytes: 44,
|
||||
// Full on-disk size of the two data files, footers and metadata included.
|
||||
totalBytes: 684,
|
||||
});
|
||||
|
||||
// Index files count toward totalBytes too (only deletion files and
|
||||
// manifests are excluded).
|
||||
await table.createIndex("id", { config: Index.btree() });
|
||||
const statsWithIndex = await table.stats();
|
||||
expect(statsWithIndex.numIndices).toBe(1);
|
||||
expect(statsWithIndex.totalBytes).toBeGreaterThan(684);
|
||||
});
|
||||
|
||||
it("should overwrite data if asked", async () => {
|
||||
|
||||
+14
-4
@@ -197,7 +197,11 @@ export interface LsmWriteSpec {
|
||||
column?: string;
|
||||
/** Bucket variant: the number of buckets, in `[1, 1024]`. */
|
||||
numBuckets?: number;
|
||||
/** Names of indexes the MemWAL should keep up to date during writes. */
|
||||
/**
|
||||
* Indexes the MemWAL keeps up to date. Omit to maintain every supported
|
||||
* index, resolved on install — a snapshot, so indexes created later are not
|
||||
* maintained. Pass `[]` for none.
|
||||
*/
|
||||
maintainedIndexes?: string[];
|
||||
/** Default `ShardWriter` configuration recorded in the MemWAL index. */
|
||||
writerConfigDefaults?: Record<string, string>;
|
||||
@@ -595,6 +599,11 @@ export abstract class Table {
|
||||
* All variants require the table to have an unenforced primary key
|
||||
* ({@link Table#setUnenforcedPrimaryKey}); bucket sharding additionally
|
||||
* requires it to be the single column being bucketed.
|
||||
*
|
||||
* Omitting `maintainedIndexes` maintains every index on the table, resolved
|
||||
* here, failing if one cannot be maintained — name them to install anyway.
|
||||
* Naming them pins an exact set, and a still-building index is rejected
|
||||
* rather than quietly omitted.
|
||||
* @param {LsmWriteSpec} spec The sharding spec to install.
|
||||
* @returns {Promise<void>}
|
||||
* @example
|
||||
@@ -622,9 +631,10 @@ export abstract class Table {
|
||||
*
|
||||
* Resolves to `undefined` when the MemWAL LSM write path is not enabled (no
|
||||
* spec has been set, or it was removed with {@link Table#unsetLsmWriteSpec}).
|
||||
* The returned spec — including its `maintainedIndexes` and
|
||||
* `writerConfigDefaults` — mirrors what was passed to
|
||||
* {@link Table#setLsmWriteSpec}.
|
||||
* The returned spec mirrors what was passed to
|
||||
* {@link Table#setLsmWriteSpec}, except that `maintainedIndexes` always
|
||||
* reports the concrete list resolved when the spec was set — `undefined`
|
||||
* never round-trips.
|
||||
* @returns {Promise<LsmWriteSpec | undefined>}
|
||||
*/
|
||||
abstract getLsmWriteSpec(): Promise<LsmWriteSpec | undefined>;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-darwin-arm64",
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"os": ["darwin"],
|
||||
"cpu": ["arm64"],
|
||||
"main": "lancedb.darwin-arm64.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-arm64-gnu",
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"os": ["linux"],
|
||||
"cpu": ["arm64"],
|
||||
"main": "lancedb.linux-arm64-gnu.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-arm64-musl",
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"os": ["linux"],
|
||||
"cpu": ["arm64"],
|
||||
"main": "lancedb.linux-arm64-musl.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-x64-gnu",
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"os": ["linux"],
|
||||
"cpu": ["x64"],
|
||||
"main": "lancedb.linux-x64-gnu.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-linux-x64-musl",
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"os": ["linux"],
|
||||
"cpu": ["x64"],
|
||||
"main": "lancedb.linux-x64-musl.node",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-win32-arm64-msvc",
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"os": [
|
||||
"win32"
|
||||
],
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb-win32-x64-msvc",
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"os": ["win32"],
|
||||
"cpu": ["x64"],
|
||||
"main": "lancedb.win32-x64-msvc.node",
|
||||
|
||||
Generated
+2
-2
@@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "@lancedb/lancedb",
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "@lancedb/lancedb",
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"cpu": [
|
||||
"x64",
|
||||
"arm64"
|
||||
|
||||
+1
-1
@@ -11,7 +11,7 @@
|
||||
"ann"
|
||||
],
|
||||
"private": false,
|
||||
"version": "0.37.1-beta.0",
|
||||
"version": "0.37.1-beta.1",
|
||||
"main": "dist/index.js",
|
||||
"exports": {
|
||||
".": "./dist/index.js",
|
||||
|
||||
+10
-7
@@ -772,7 +772,8 @@ pub struct LsmWriteSpec {
|
||||
pub column: Option<String>,
|
||||
/// Bucket variant: the number of buckets, in `[1, 1024]`.
|
||||
pub num_buckets: Option<u32>,
|
||||
/// Names of indexes the MemWAL should keep up to date during writes.
|
||||
/// Indexes the MemWAL keeps up to date. Omitted resolves every
|
||||
/// maintainable index on install; an empty array means none.
|
||||
pub maintained_indexes: Option<Vec<String>>,
|
||||
/// Default `ShardWriter` configuration recorded in the MemWAL index.
|
||||
pub writer_config_defaults: Option<HashMap<String, String>>,
|
||||
@@ -782,7 +783,6 @@ impl TryFrom<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
|
||||
type Error = napi::Error;
|
||||
|
||||
fn try_from(value: LsmWriteSpec) -> napi::Result<Self> {
|
||||
let maintained = value.maintained_indexes.unwrap_or_default();
|
||||
let writer_config_defaults = value.writer_config_defaults.unwrap_or_default();
|
||||
let spec = match value.spec_type.as_str() {
|
||||
"bucket" => {
|
||||
@@ -809,7 +809,7 @@ impl TryFrom<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
|
||||
}
|
||||
};
|
||||
Ok(spec
|
||||
.with_maintained_indexes(maintained)
|
||||
.with_maintained_indexes(value.maintained_indexes)
|
||||
.with_writer_config_defaults(writer_config_defaults))
|
||||
}
|
||||
}
|
||||
@@ -827,7 +827,7 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
|
||||
spec_type: "bucket".to_string(),
|
||||
column: Some(column),
|
||||
num_buckets: Some(num_buckets),
|
||||
maintained_indexes: Some(maintained_indexes),
|
||||
maintained_indexes,
|
||||
writer_config_defaults: Some(writer_config_defaults),
|
||||
},
|
||||
Native::Identity {
|
||||
@@ -838,7 +838,7 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
|
||||
spec_type: "identity".to_string(),
|
||||
column: Some(column),
|
||||
num_buckets: None,
|
||||
maintained_indexes: Some(maintained_indexes),
|
||||
maintained_indexes,
|
||||
writer_config_defaults: Some(writer_config_defaults),
|
||||
},
|
||||
Native::Unsharded {
|
||||
@@ -848,7 +848,7 @@ impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
|
||||
spec_type: "unsharded".to_string(),
|
||||
column: None,
|
||||
num_buckets: None,
|
||||
maintained_indexes: Some(maintained_indexes),
|
||||
maintained_indexes,
|
||||
writer_config_defaults: Some(writer_config_defaults),
|
||||
},
|
||||
}
|
||||
@@ -1043,7 +1043,10 @@ impl From<lancedb::index::IndexStatistics> for IndexStatistics {
|
||||
|
||||
#[napi(object)]
|
||||
pub struct TableStatistics {
|
||||
/// The total number of bytes in the table
|
||||
/// The total size, in bytes, of the table's data files, index files, and
|
||||
/// overlay files
|
||||
///
|
||||
/// Read from the manifest, so this excludes deletion files and manifests.
|
||||
pub total_bytes: i64,
|
||||
|
||||
/// The number of rows in the table
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "lancedb-python"
|
||||
version = "0.37.1-beta.0"
|
||||
version = "0.37.1-beta.1"
|
||||
publish = false
|
||||
edition.workspace = true
|
||||
description = "Python bindings for LanceDB"
|
||||
|
||||
@@ -355,6 +355,10 @@ class Table:
|
||||
async def set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ...
|
||||
async def unset_lsm_write_spec(self) -> None: ...
|
||||
async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ...
|
||||
async def checkpoint_lsm(self) -> None: ...
|
||||
async def flush_lsm(self) -> None: ...
|
||||
async def compact_lsm(self) -> None: ...
|
||||
async def get_lsm_stats(self, include_generation_rows: bool) -> Optional[dict]: ...
|
||||
async def close_lsm_writers(self) -> None: ...
|
||||
@property
|
||||
def tags(self) -> Tags: ...
|
||||
@@ -649,9 +653,10 @@ class LsmWriteSpec:
|
||||
def identity(column: str) -> "LsmWriteSpec": ...
|
||||
@staticmethod
|
||||
def unsharded() -> "LsmWriteSpec": ...
|
||||
def with_maintained_indexes(self, indexes: List[str]) -> "LsmWriteSpec":
|
||||
"""Return a copy of this spec asking the MemWAL to keep the named
|
||||
indexes up to date as rows are appended."""
|
||||
def with_maintained_indexes(self, indexes: Optional[List[str]]) -> "LsmWriteSpec":
|
||||
"""Set which indexes the MemWAL keeps up to date. None resolves every
|
||||
index on the table at install, failing if one cannot be maintained;
|
||||
a list is verbatim, empty means none."""
|
||||
...
|
||||
def with_writer_config_defaults(self, defaults: Dict[str, str]) -> "LsmWriteSpec":
|
||||
"""Return a copy of this spec recording the given default
|
||||
@@ -666,7 +671,9 @@ class LsmWriteSpec:
|
||||
@property
|
||||
def num_buckets(self) -> Optional[int]: ...
|
||||
@property
|
||||
def maintained_indexes(self) -> List[str]: ...
|
||||
def maintained_indexes(self) -> Optional[List[str]]:
|
||||
"""Indexes the MemWAL keeps up to date, or None for every supported one."""
|
||||
...
|
||||
@property
|
||||
def writer_config_defaults(self) -> Dict[str, str]: ...
|
||||
|
||||
|
||||
@@ -87,12 +87,13 @@ class JinaEmbeddings(EmbeddingFunction):
|
||||
if isinstance(image, bytes):
|
||||
image_dict = {"image": base64.b64encode(image).decode("utf-8")}
|
||||
elif isinstance(image, (str, Path)):
|
||||
parsed = urlparse.urlparse(image)
|
||||
# TODO handle drive letter on windows.
|
||||
parsed = urlparse(str(image))
|
||||
PIL_Image = attempt_import_or_raise("PIL.Image", "pillow")
|
||||
if parsed.scheme == "file":
|
||||
pil_image = PIL_Image.open(parsed.path)
|
||||
elif parsed.scheme == "":
|
||||
elif parsed.scheme == "" or (os.name == "nt" and len(parsed.scheme) == 1):
|
||||
# A Windows drive letter parses as a one-character scheme
|
||||
# ("C:\\img.png" -> scheme="c"), so treat it as a local path.
|
||||
pil_image = PIL_Image.open(image if os.name == "nt" else parsed.path)
|
||||
elif parsed.scheme.startswith("http"):
|
||||
pil_image = PIL_Image.open(io.BytesIO(url_retrieve(image)))
|
||||
|
||||
@@ -3976,6 +3976,28 @@ class LanceTable(Table):
|
||||
[`AsyncTable.get_lsm_write_spec`][lancedb.AsyncTable.get_lsm_write_spec]."""
|
||||
return LOOP.run(self._table.get_lsm_write_spec())
|
||||
|
||||
def checkpoint_lsm(self) -> None:
|
||||
"""Synchronous version of
|
||||
[`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
|
||||
return LOOP.run(self._table.checkpoint_lsm())
|
||||
|
||||
def flush_lsm(self) -> None:
|
||||
"""Synchronous version of
|
||||
[`AsyncTable.flush_lsm`][lancedb.AsyncTable.flush_lsm]."""
|
||||
return LOOP.run(self._table.flush_lsm())
|
||||
|
||||
def compact_lsm(self) -> None:
|
||||
"""Synchronous version of
|
||||
[`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
|
||||
return LOOP.run(self._table.compact_lsm())
|
||||
|
||||
def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
|
||||
"""Synchronous version of
|
||||
[`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
|
||||
return LOOP.run(
|
||||
self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
|
||||
)
|
||||
|
||||
def close_lsm_writers(self) -> None:
|
||||
"""Close cached MemWAL shard writers. See
|
||||
[`AsyncTable.close_lsm_writers`][lancedb.AsyncTable.close_lsm_writers]."""
|
||||
@@ -4654,6 +4676,13 @@ class AsyncTable:
|
||||
via [`set_unenforced_primary_key`]; bucket sharding additionally
|
||||
requires it to be the single column being bucketed.
|
||||
|
||||
By default the MemWAL maintains every index on the table, resolved
|
||||
here — a snapshot, so an index created afterwards needs the spec unset
|
||||
and set again. This fails if one cannot be maintained; name the set
|
||||
with ``with_maintained_indexes`` to install anyway. That pins an exact
|
||||
set (a still-building index is rejected, not omitted); ``[]`` maintains
|
||||
none.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
spec : LsmWriteSpec
|
||||
@@ -4680,12 +4709,73 @@ class AsyncTable:
|
||||
|
||||
Returns ``None`` when the MemWAL LSM write path is not enabled (no
|
||||
spec has been set, or it was removed with `unset_lsm_write_spec`).
|
||||
The returned spec — including its ``maintained_indexes`` and
|
||||
``writer_config_defaults`` — mirrors what was passed to
|
||||
`set_lsm_write_spec`.
|
||||
The returned spec mirrors what was passed to `set_lsm_write_spec`,
|
||||
except that ``maintained_indexes`` always reports the concrete list
|
||||
resolved when the spec was set — ``None`` never round-trips.
|
||||
"""
|
||||
return await self._inner.get_lsm_write_spec()
|
||||
|
||||
async def checkpoint_lsm(self) -> None:
|
||||
"""Converge this table's LSM write path into its base table.
|
||||
|
||||
One flush, sealing every memtable into L0, then compaction triggers
|
||||
until every generation that existed at that moment has reached base.
|
||||
The loop runs client-side, reading progress from ``get_lsm_stats``.
|
||||
|
||||
Best-effort: generations created *while* it runs are deliberately not
|
||||
waited on, which is what lets it terminate on a table taking writes.
|
||||
Idempotent and safe on a cadence.
|
||||
|
||||
There is no deadline, and the caller owns that. It returns when the
|
||||
target generations are gone, raises on a terminal server fault, and
|
||||
otherwise waits however long the server takes. A slow table and a
|
||||
stuck one are the same picture from the client: the compactor pool is
|
||||
shared across every table on the node, so a checkpoint queued behind
|
||||
unrelated work looks exactly like one that is merging. Wrap this in
|
||||
``asyncio.wait_for`` for a wall-clock bound; abandoning it partway
|
||||
costs nothing.
|
||||
"""
|
||||
return await self._inner.checkpoint_lsm()
|
||||
|
||||
async def flush_lsm(self) -> None:
|
||||
"""Seal every bucket's active memtable into L0.
|
||||
|
||||
Does not touch the base table — moving L0 into base is
|
||||
`compact_lsm`. On a node that has not claimed this table, this claims
|
||||
it and replays its WAL log first.
|
||||
"""
|
||||
return await self._inner.flush_lsm()
|
||||
|
||||
async def compact_lsm(self) -> None:
|
||||
"""Trigger a background L0 to base compaction pass per bucket.
|
||||
|
||||
Returns once the passes are dispatched, not once they finish: watch
|
||||
``get_lsm_stats`` for progress, or use ``checkpoint_lsm`` to loop
|
||||
until the current L0 has reached base.
|
||||
"""
|
||||
return await self._inner.compact_lsm()
|
||||
|
||||
async def get_lsm_stats(
|
||||
self, *, include_generation_rows: bool = False
|
||||
) -> Optional[dict]:
|
||||
"""Read live per-bucket LSM state.
|
||||
|
||||
Answers "how far behind is my fresh tier", "which bucket is hot", and
|
||||
"why is my fresh-tier vector search brute-force". Mutates no table
|
||||
state, though on a node that has not claimed this table it claims it,
|
||||
exactly as a read would.
|
||||
|
||||
Returns ``None`` only when the LSM write path is not enabled.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
include_generation_rows
|
||||
Report a row count per L0 generation. Off by default: each count
|
||||
opens an uncached Lance dataset, and ``checkpoint_lsm`` polls this
|
||||
needing only generation numbers.
|
||||
"""
|
||||
return await self._inner.get_lsm_stats(include_generation_rows)
|
||||
|
||||
async def close_lsm_writers(self) -> None:
|
||||
"""Drain and close any cached MemWAL shard writers for this table.
|
||||
|
||||
@@ -6251,7 +6341,9 @@ class TableStatistics:
|
||||
Attributes
|
||||
----------
|
||||
total_bytes: int
|
||||
The total number of bytes in the table.
|
||||
The total size, in bytes, of the table's data files, index files, and
|
||||
overlay files. Read from the manifest, so this excludes deletion files
|
||||
and manifests.
|
||||
num_rows: int
|
||||
The total number of rows in the table.
|
||||
num_indices: int
|
||||
|
||||
@@ -631,3 +631,23 @@ def test_url_retrieve_downloads_image():
|
||||
image_bytes = url_retrieve(image_url)
|
||||
img = Image.open(io.BytesIO(image_bytes))
|
||||
assert img.size[0] > 0 and img.size[1] > 0
|
||||
|
||||
|
||||
def test_jina_generate_image_input_dict_local_path(tmp_path):
|
||||
"""
|
||||
JinaEmbeddings._generate_image_input_dict must accept a local image path
|
||||
(str or Path), not just bytes. Previously it crashed with
|
||||
`AttributeError: 'function' object has no attribute 'urlparse'` on any
|
||||
str/Path input because it called `urlparse.urlparse(image)` instead of
|
||||
`urlparse(image)` (urlparse was imported as a function, not a module).
|
||||
"""
|
||||
Image = pytest.importorskip("PIL.Image")
|
||||
from lancedb.embeddings.jinaai import JinaEmbeddings
|
||||
|
||||
image_path = tmp_path / "test.png"
|
||||
Image.new("RGB", (4, 4), color="red").save(image_path, format="PNG")
|
||||
|
||||
for image in (str(image_path), image_path):
|
||||
image_dict = JinaEmbeddings._generate_image_input_dict(image)
|
||||
assert "image" in image_dict
|
||||
assert isinstance(image_dict["image"], str) and len(image_dict["image"]) > 0
|
||||
|
||||
@@ -83,7 +83,9 @@ def test_lsm_write_spec_repr():
|
||||
assert s.spec_type == "bucket"
|
||||
assert s.column == "id"
|
||||
assert s.num_buckets == 4
|
||||
assert s.maintained_indexes == []
|
||||
# A fresh spec defers its maintained set to install time.
|
||||
assert s.maintained_indexes is None
|
||||
assert s.with_maintained_indexes([]).maintained_indexes == []
|
||||
assert "bucket" in repr(s)
|
||||
assert "id" in repr(s)
|
||||
assert "4" in repr(s)
|
||||
@@ -169,18 +171,23 @@ def test_get_lsm_write_spec(tmp_path):
|
||||
table.unset_lsm_write_spec()
|
||||
assert table.get_lsm_write_spec() is None
|
||||
|
||||
# Identity round-trips (column recovered from the schema).
|
||||
# Identity round-trips (column recovered from the schema). Leaving the
|
||||
# maintained set to be inferred picks up the index on the table, so the
|
||||
# spec reads back naming it rather than as "infer".
|
||||
table.set_lsm_write_spec(LsmWriteSpec.identity("id"))
|
||||
spec = table.get_lsm_write_spec()
|
||||
assert spec.spec_type == "identity"
|
||||
assert spec.column == "id"
|
||||
assert spec.maintained_indexes == [idx_name]
|
||||
table.unset_lsm_write_spec()
|
||||
|
||||
# Unsharded round-trips (no routing column).
|
||||
table.set_lsm_write_spec(LsmWriteSpec.unsharded())
|
||||
# Unsharded round-trips (no routing column). Opting out is distinct from
|
||||
# the inferred default.
|
||||
table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
|
||||
spec = table.get_lsm_write_spec()
|
||||
assert spec.spec_type == "unsharded"
|
||||
assert spec.column is None
|
||||
assert spec.maintained_indexes == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
@@ -544,7 +544,7 @@ def test_lsm_read_fts_unmaintained_index_errors(tmp_path):
|
||||
table.create_index("text", config=FTS())
|
||||
# No maintained indexes: the active memtable FTS arm cannot serve un-compacted
|
||||
# docs, so the search would silently omit them — reject instead.
|
||||
table.set_lsm_write_spec(LsmWriteSpec.unsharded())
|
||||
table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
|
||||
with pytest.raises(Exception, match="maintained"):
|
||||
table.search("fox", query_type="fts", fts_columns="text").to_arrow()
|
||||
|
||||
@@ -631,7 +631,7 @@ def test_lsm_read_vector_unmaintained_index_errors(tmp_path):
|
||||
)
|
||||
# Spec with NO maintained indexes: the base vector index's catch-up is untracked,
|
||||
# so the scanner rejects rather than risk dropping compacted-but-unindexed rows.
|
||||
table.set_lsm_write_spec(LsmWriteSpec.unsharded())
|
||||
table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
|
||||
with pytest.raises(Exception, match="maintained"):
|
||||
table.search([1.0] * VECTOR_DIM).to_arrow()
|
||||
|
||||
|
||||
@@ -3713,7 +3713,8 @@ def test_stats(mem_db: DBConnection):
|
||||
stats = table.stats()
|
||||
print(f"{stats=}")
|
||||
assert stats == {
|
||||
"total_bytes": 60,
|
||||
# Full on-disk size of the data file, footer and metadata included.
|
||||
"total_bytes": 633,
|
||||
"num_rows": 2,
|
||||
"num_indices": 0,
|
||||
"fragment_stats": {
|
||||
@@ -3731,6 +3732,13 @@ def test_stats(mem_db: DBConnection):
|
||||
},
|
||||
}
|
||||
|
||||
# Index files count toward total_bytes too (only deletion files and
|
||||
# manifests are excluded).
|
||||
table.create_index("id", config=BTree())
|
||||
stats_with_index = table.stats()
|
||||
assert stats_with_index["num_indices"] == 1
|
||||
assert stats_with_index["total_bytes"] > stats["total_bytes"]
|
||||
|
||||
|
||||
def test_create_table_empty_list_with_schema(mem_db: DBConnection):
|
||||
"""Test creating table with empty list data and schema
|
||||
|
||||
+138
-16
@@ -28,11 +28,72 @@ use pyo3::{
|
||||
Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python,
|
||||
exceptions::{PyRuntimeError, PyValueError},
|
||||
pyclass, pyfunction, pymethods,
|
||||
types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods},
|
||||
types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods, PyList, PyListMethods},
|
||||
};
|
||||
|
||||
mod scannable;
|
||||
|
||||
/// Convert `LsmStats` to a Python dict, preserving the per-bucket list.
|
||||
///
|
||||
/// Deliberately not flattened to a table-level summary: a table is N
|
||||
/// buckets on one node, and the per-bucket detail is the reason the
|
||||
/// endpoint exists — flattening hides the single hot bucket someone opened
|
||||
/// it to find.
|
||||
fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult<Py<PyDict>> {
|
||||
let out = PyDict::new(py);
|
||||
let buckets = PyList::empty(py);
|
||||
for b in &stats.buckets {
|
||||
let e = PyDict::new(py);
|
||||
e.set_item("shard_id", &b.shard_id)?;
|
||||
e.set_item("status", &b.status)?;
|
||||
e.set_item("writer_epoch", b.writer_epoch)?;
|
||||
e.set_item("manifest_version", b.manifest_version)?;
|
||||
e.set_item("current_generation", b.current_generation)?;
|
||||
e.set_item(
|
||||
"replay_after_wal_entry_position",
|
||||
b.replay_after_wal_entry_position,
|
||||
)?;
|
||||
e.set_item(
|
||||
"wal_entry_position_last_seen",
|
||||
b.wal_entry_position_last_seen,
|
||||
)?;
|
||||
|
||||
let generations = PyList::empty(py);
|
||||
for g in &b.generations {
|
||||
let ge = PyDict::new(py);
|
||||
ge.set_item("generation", g.generation)?;
|
||||
ge.set_item("bytes", g.bytes)?;
|
||||
ge.set_item("rows", g.rows)?;
|
||||
generations.append(ge)?;
|
||||
}
|
||||
e.set_item("generations", generations)?;
|
||||
e.set_item("compacting", b.compacting)?;
|
||||
|
||||
e.set_item(
|
||||
"memtables",
|
||||
b.memtables
|
||||
.as_ref()
|
||||
.map(|ms| {
|
||||
let l = PyList::empty(py);
|
||||
for m in ms {
|
||||
let d = PyDict::new(py);
|
||||
d.set_item("generation", m.generation)?;
|
||||
d.set_item("rows", m.rows)?;
|
||||
d.set_item("bytes", m.bytes)?;
|
||||
d.set_item("batches", m.batches)?;
|
||||
d.set_item("indexes", m.indexes.clone())?;
|
||||
l.append(d)?;
|
||||
}
|
||||
PyResult::Ok(l.unbind())
|
||||
})
|
||||
.transpose()?,
|
||||
)?;
|
||||
buckets.append(e)?;
|
||||
}
|
||||
out.set_item("buckets", buckets)?;
|
||||
Ok(out.unbind())
|
||||
}
|
||||
|
||||
#[derive(FromPyObject)]
|
||||
enum PredicateArg {
|
||||
Expr(PyExpr),
|
||||
@@ -185,12 +246,22 @@ impl From<lancedb::table::MergeResult> for MergeResult {
|
||||
}
|
||||
}
|
||||
|
||||
/// Render for `__repr__`, so the default reads as Python's `None` rather than
|
||||
/// Rust's `Some([..])`.
|
||||
fn fmt_maintained(maintained: &Option<Vec<String>>) -> String {
|
||||
match maintained {
|
||||
Some(names) => format!("{:?}", names),
|
||||
None => "None".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Specification selecting Lance's MemWAL LSM-style write path for
|
||||
/// `merge_insert`.
|
||||
///
|
||||
/// Constructed via the `bucket(...)`, `identity(...)`, or `unsharded()`
|
||||
/// classmethods, then optionally chain `with_maintained_indexes(...)` and
|
||||
/// `with_writer_config_defaults(...)`.
|
||||
/// `with_writer_config_defaults(...)`. A fresh spec maintains every index the
|
||||
/// MemWAL supports, resolved on install.
|
||||
#[pyclass(from_py_object)]
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct LsmWriteSpec {
|
||||
@@ -230,11 +301,11 @@ impl LsmWriteSpec {
|
||||
}
|
||||
}
|
||||
|
||||
/// Replace the list of indexes the MemWAL should keep up to date as
|
||||
/// rows are appended. Each name must reference an index that
|
||||
/// already exists on the table at the time `set_lsm_write_spec`
|
||||
/// is called.
|
||||
pub fn with_maintained_indexes(&self, indexes: Vec<String>) -> Self {
|
||||
/// Set which indexes the MemWAL maintains. `None` (the default)
|
||||
/// resolves every supported index on install; a list is verbatim,
|
||||
/// and an empty list maintains nothing.
|
||||
#[pyo3(signature = (indexes))]
|
||||
pub fn with_maintained_indexes(&self, indexes: Option<Vec<String>>) -> Self {
|
||||
Self {
|
||||
inner: self.inner.clone().with_maintained_indexes(indexes),
|
||||
}
|
||||
@@ -256,23 +327,29 @@ impl LsmWriteSpec {
|
||||
maintained_indexes,
|
||||
writer_config_defaults,
|
||||
} => format!(
|
||||
"LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={:?}, writer_config_defaults={:?})",
|
||||
column, num_buckets, maintained_indexes, writer_config_defaults,
|
||||
"LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
|
||||
column,
|
||||
num_buckets,
|
||||
fmt_maintained(maintained_indexes),
|
||||
writer_config_defaults,
|
||||
),
|
||||
lancedb::table::LsmWriteSpec::Identity {
|
||||
column,
|
||||
maintained_indexes,
|
||||
writer_config_defaults,
|
||||
} => format!(
|
||||
"LsmWriteSpec.identity(column={:?}, maintained_indexes={:?}, writer_config_defaults={:?})",
|
||||
column, maintained_indexes, writer_config_defaults,
|
||||
"LsmWriteSpec.identity(column={:?}, maintained_indexes={}, writer_config_defaults={:?})",
|
||||
column,
|
||||
fmt_maintained(maintained_indexes),
|
||||
writer_config_defaults,
|
||||
),
|
||||
lancedb::table::LsmWriteSpec::Unsharded {
|
||||
maintained_indexes,
|
||||
writer_config_defaults,
|
||||
} => format!(
|
||||
"LsmWriteSpec.unsharded(maintained_indexes={:?}, writer_config_defaults={:?})",
|
||||
maintained_indexes, writer_config_defaults,
|
||||
"LsmWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
|
||||
fmt_maintained(maintained_indexes),
|
||||
writer_config_defaults,
|
||||
),
|
||||
}
|
||||
}
|
||||
@@ -307,10 +384,10 @@ impl LsmWriteSpec {
|
||||
}
|
||||
}
|
||||
|
||||
/// Names of indexes the MemWAL should keep up to date during writes.
|
||||
/// Indexes the MemWAL keeps up to date, or `None` for every supported one.
|
||||
#[getter]
|
||||
pub fn maintained_indexes(&self) -> Vec<String> {
|
||||
self.inner.maintained_indexes().to_vec()
|
||||
pub fn maintained_indexes(&self) -> Option<Vec<String>> {
|
||||
self.inner.maintained_indexes().map(<[String]>::to_vec)
|
||||
}
|
||||
|
||||
/// Default `ShardWriter` configuration recorded by this spec.
|
||||
@@ -1339,6 +1416,51 @@ impl Table {
|
||||
})
|
||||
}
|
||||
|
||||
/// Converge the table's LSM write path into its base table.
|
||||
///
|
||||
/// Best-effort: with writes flowing, new rows may land after the last
|
||||
/// pass. Errors if the table stops making progress.
|
||||
pub fn checkpoint_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.inner_ref()?.clone();
|
||||
future_into_py(self_.py(), async move {
|
||||
inner.checkpoint_lsm().await.infer_error()
|
||||
})
|
||||
}
|
||||
|
||||
/// Seal every bucket's active memtable into L0.
|
||||
pub fn flush_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.inner_ref()?.clone();
|
||||
future_into_py(
|
||||
self_.py(),
|
||||
async move { inner.flush_lsm().await.infer_error() },
|
||||
)
|
||||
}
|
||||
|
||||
/// Trigger a background L0 → base pass per bucket. Returns once the
|
||||
/// passes are dispatched, not once they finish — watch `get_lsm_stats`.
|
||||
pub fn compact_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.inner_ref()?.clone();
|
||||
future_into_py(self_.py(), async move {
|
||||
inner.compact_lsm().await.infer_error()
|
||||
})
|
||||
}
|
||||
|
||||
/// Live LSM state, or `None` when the LSM write path is not enabled.
|
||||
#[pyo3(signature = (include_generation_rows=false))]
|
||||
pub fn get_lsm_stats(
|
||||
self_: PyRef<'_, Self>,
|
||||
include_generation_rows: bool,
|
||||
) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.inner_ref()?.clone();
|
||||
future_into_py(self_.py(), async move {
|
||||
let stats = inner
|
||||
.get_lsm_stats(include_generation_rows)
|
||||
.await
|
||||
.infer_error()?;
|
||||
Python::attach(|py| stats.map(|s| lsm_stats_to_py(py, &s)).transpose())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn close_lsm_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.inner_ref()?.clone();
|
||||
future_into_py(self_.py(), async move {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "lancedb"
|
||||
version = "0.37.1-beta.0"
|
||||
version = "0.37.1-beta.1"
|
||||
edition.workspace = true
|
||||
description = "LanceDB: A serverless, low-latency vector database for AI applications"
|
||||
license.workspace = true
|
||||
@@ -100,7 +100,8 @@ anyhow = "1"
|
||||
lance-testing = { workspace = true }
|
||||
tempfile = "3.5.0"
|
||||
random_word = { version = "0.4.3", features = ["en"] }
|
||||
tokio = { version = "1.23", features = ["io-util", "macros", "net", "rt-multi-thread", "sync"] }
|
||||
roaring = "0.11.4"
|
||||
tokio = { version = "1.23", features = ["io-util", "macros", "net", "rt-multi-thread", "sync", "test-util"] }
|
||||
uuid = { version = "1.7.0", features = ["v4"] }
|
||||
walkdir = "2"
|
||||
aws-sdk-dynamodb = { version = "1.55.0" }
|
||||
|
||||
@@ -23,11 +23,13 @@ use crate::table::AddResult;
|
||||
use crate::table::BranchDiff;
|
||||
use crate::table::DeleteResult;
|
||||
use crate::table::DropColumnsResult;
|
||||
use crate::table::LsmStats;
|
||||
use crate::table::LsmWriteSpec;
|
||||
use crate::table::MergeBranchResult;
|
||||
use crate::table::MergeResult;
|
||||
use crate::table::Tags;
|
||||
use crate::table::UpdateResult;
|
||||
use crate::table::lsm_stats::GetLsmStatsResponse;
|
||||
use crate::table::merge::MergeFilter;
|
||||
use crate::table::query::create_multi_vector_plan;
|
||||
use crate::table::write_progress::FinishOnDrop;
|
||||
@@ -991,6 +993,18 @@ impl<S: HttpSend> RemoteTable<S> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Send an LSM operator request with the transport retry layer **off**.
|
||||
///
|
||||
/// Retry policy on these routes belongs to the checkpoint loop, which
|
||||
/// reads the status and can tell contention from a lost claim. Leaving the
|
||||
/// transport layer on would re-ask on its own schedule first, and surface
|
||||
/// an `Error::Retry` whose status the loop would then have to unwrap.
|
||||
async fn send_lsm_route(&self, request: RequestBuilder) -> Result<(String, reqwest::Response)> {
|
||||
let (request_id, response) = self.send(request, false).await?;
|
||||
let response = self.check_table_response(&request_id, response).await?;
|
||||
Ok((request_id, response))
|
||||
}
|
||||
|
||||
/// Build a POST request and attach the read-freshness headers
|
||||
/// (`x-lancedb-min-version`, `x-lancedb-min-timestamp`).
|
||||
fn post_read(&self, uri: &str) -> RequestBuilder {
|
||||
@@ -2468,13 +2482,47 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
|
||||
})
|
||||
}
|
||||
|
||||
async fn flush_lsm(&self) -> Result<()> {
|
||||
let request = self
|
||||
.client
|
||||
.post(&format!("/v1/table/{}/flush_lsm/", self.identifier));
|
||||
self.send_lsm_route(request).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn compact_lsm(&self) -> Result<()> {
|
||||
let request = self
|
||||
.client
|
||||
.post(&format!("/v1/table/{}/compact_lsm/", self.identifier));
|
||||
self.send_lsm_route(request).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_lsm_stats(&self, include_generation_rows: bool) -> Result<Option<LsmStats>> {
|
||||
// Read-semantics POST, like `get_lsm_write_spec`.
|
||||
let request = self
|
||||
.post_read(&format!("/v1/table/{}/get_lsm_stats/", self.identifier))
|
||||
.json(&serde_json::json!({
|
||||
"include_generation_rows": include_generation_rows,
|
||||
}));
|
||||
let (request_id, response) = self.send_lsm_route(request).await?;
|
||||
let body = response.text().await.err_to_http(request_id.clone())?;
|
||||
let parsed: GetLsmStatsResponse = serde_json::from_str(&body).map_err(|e| Error::Http {
|
||||
source: format!("Failed to parse get_lsm_stats response: {e}").into(),
|
||||
request_id,
|
||||
status_code: None,
|
||||
})?;
|
||||
// `null` — and only — when the table has no LSM write path.
|
||||
Ok(parsed.lsm_stats)
|
||||
}
|
||||
|
||||
async fn set_lsm_write_spec(&self, spec: LsmWriteSpec) -> Result<()> {
|
||||
self.check_mutable().await?;
|
||||
|
||||
// Map the spec onto the server's request DTO. `sharding` is internally
|
||||
// tagged on `mode` to mirror sophon's `Sharding` enum; `maintained_indexes`
|
||||
// and `writer_config_defaults` are sent verbatim (an empty list means "no
|
||||
// maintained indexes", not "default to all").
|
||||
// tagged on `mode` to mirror sophon's `Sharding` enum. A null
|
||||
// `maintained_indexes` asks the server to resolve every maintainable
|
||||
// index at HEAD; a list is verbatim, an empty one meaning none.
|
||||
let sharding = match &spec {
|
||||
LsmWriteSpec::Bucket {
|
||||
column,
|
||||
@@ -6551,7 +6599,7 @@ mod tests {
|
||||
.unwrap()
|
||||
});
|
||||
let spec = crate::table::LsmWriteSpec::unsharded()
|
||||
.with_maintained_indexes(["id_idx"])
|
||||
.with_maintained_indexes(vec!["id_idx".to_string()])
|
||||
.with_writer_config_defaults([("max_memtable_rows", "1000")]);
|
||||
table.set_lsm_write_spec(spec).await.unwrap();
|
||||
}
|
||||
@@ -6570,7 +6618,8 @@ mod tests {
|
||||
body["sharding"],
|
||||
serde_json::json!({ "mode": "bucket", "column": "id", "num_buckets": 16 })
|
||||
);
|
||||
assert_eq!(body["maintained_indexes"], serde_json::json!([]));
|
||||
// An unpinned maintained set sends null: resolve server-side.
|
||||
assert_eq!(body["maintained_indexes"], serde_json::Value::Null);
|
||||
http::Response::builder().status(200).body("{}").unwrap()
|
||||
});
|
||||
table
|
||||
@@ -6579,6 +6628,23 @@ mod tests {
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
/// `[]` (none) must stay distinguishable on the wire from null (all).
|
||||
#[tokio::test]
|
||||
async fn test_set_lsm_write_spec_no_maintained_indexes() {
|
||||
let table = Table::new_with_handler("my_table", |request| {
|
||||
let body = request.body().unwrap().as_bytes().unwrap();
|
||||
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
|
||||
assert_eq!(body["maintained_indexes"], serde_json::json!([]));
|
||||
http::Response::builder().status(200).body("{}").unwrap()
|
||||
});
|
||||
table
|
||||
.set_lsm_write_spec(
|
||||
crate::table::LsmWriteSpec::bucket("id", 16).with_maintained_indexes(Vec::new()),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_set_lsm_write_spec_identity() {
|
||||
let table = Table::new_with_handler("my_table", |request| {
|
||||
@@ -6653,7 +6719,7 @@ mod tests {
|
||||
} => {
|
||||
assert_eq!(column, "id");
|
||||
assert_eq!(num_buckets, 4);
|
||||
assert_eq!(maintained_indexes, vec!["id_idx".to_string()]);
|
||||
assert_eq!(maintained_indexes, Some(vec!["id_idx".to_string()]));
|
||||
assert_eq!(
|
||||
writer_config_defaults
|
||||
.get("durable_write")
|
||||
@@ -6682,6 +6748,499 @@ mod tests {
|
||||
assert!(table.get_lsm_write_spec().await.unwrap().is_none());
|
||||
}
|
||||
|
||||
/// Build a `get_lsm_stats` body for one bucket holding `generations`.
|
||||
fn stats_body(generations: &[u64], compacting: bool) -> String {
|
||||
serde_json::json!({
|
||||
"lsm_stats": {
|
||||
"buckets": [{
|
||||
"shard_id": "b0",
|
||||
"status": "Active",
|
||||
"writer_epoch": 1,
|
||||
"manifest_version": 1,
|
||||
"current_generation": generations.iter().max().copied().unwrap_or(0) + 1,
|
||||
"replay_after_wal_entry_position": 0,
|
||||
"wal_entry_position_last_seen": 0,
|
||||
"generations": generations.iter()
|
||||
.map(|g| serde_json::json!({ "generation": g, "bytes": 1 }))
|
||||
.collect::<Vec<_>>(),
|
||||
"compacting": compacting,
|
||||
"memtables": [],
|
||||
}],
|
||||
}
|
||||
})
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// `flush_lsm` / `compact_lsm` answer 202 with no body at all.
|
||||
fn accepted() -> http::Response<String> {
|
||||
http::Response::builder()
|
||||
.status(202)
|
||||
.body(String::new())
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn ok_json(body: String) -> http::Response<String> {
|
||||
http::Response::builder().status(200).body(body).unwrap()
|
||||
}
|
||||
|
||||
/// A flush landing in an empty L0 finishes on the opening stats read
|
||||
/// alone. Asserting zero compacts is the point: "it returned Ok" is also
|
||||
/// true of a loop that ran a pointless pass.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_checkpoint_short_circuits_on_empty_l0() {
|
||||
let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = compacts.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
if path.contains("compact_lsm") {
|
||||
seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
panic!("an already-converged table must issue no compact calls");
|
||||
}
|
||||
if path.contains("flush_lsm") {
|
||||
return accepted();
|
||||
}
|
||||
assert_eq!(path, "/v1/table/my_table/get_lsm_stats/");
|
||||
ok_json(stats_body(&[], false))
|
||||
});
|
||||
|
||||
table.checkpoint_lsm().await.unwrap();
|
||||
assert_eq!(compacts.load(std::sync::atomic::Ordering::SeqCst), 0);
|
||||
}
|
||||
|
||||
/// The loop triggers compaction until every generation that existed at
|
||||
/// the start is gone, one bounded prefix per pass.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_checkpoint_triggers_until_targets_are_drained() {
|
||||
let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = compacts.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
if path.contains("flush_lsm") {
|
||||
return accepted();
|
||||
}
|
||||
if path.contains("compact_lsm") {
|
||||
seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
return accepted();
|
||||
}
|
||||
// Each pass drains the oldest generation.
|
||||
let drained = seen.load(std::sync::atomic::Ordering::SeqCst);
|
||||
let left: Vec<u64> = [1u64, 2, 3].into_iter().skip(drained).collect();
|
||||
ok_json(stats_body(&left, false))
|
||||
});
|
||||
|
||||
table.checkpoint_lsm().await.unwrap();
|
||||
assert_eq!(
|
||||
compacts.load(std::sync::atomic::Ordering::SeqCst),
|
||||
3,
|
||||
"one trigger per generation prefix, then stop"
|
||||
);
|
||||
}
|
||||
|
||||
/// Generations created *during* the checkpoint are not waited on, which
|
||||
/// is what lets the loop terminate on a table taking writes where "L0 is
|
||||
/// empty" never becomes true.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_checkpoint_ignores_generations_created_while_it_runs() {
|
||||
let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = compacts.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
if path.contains("flush_lsm") {
|
||||
return accepted();
|
||||
}
|
||||
if path.contains("compact_lsm") {
|
||||
seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
return accepted();
|
||||
}
|
||||
// Target is 5. One pass drains it; a writer keeps adding above.
|
||||
let n = seen.load(std::sync::atomic::Ordering::SeqCst);
|
||||
let body = if n == 0 {
|
||||
stats_body(&[5], false)
|
||||
} else {
|
||||
stats_body(&[6, 7], false)
|
||||
};
|
||||
ok_json(body)
|
||||
});
|
||||
|
||||
table.checkpoint_lsm().await.unwrap();
|
||||
assert_eq!(
|
||||
compacts.load(std::sync::atomic::Ordering::SeqCst),
|
||||
1,
|
||||
"the loop must not chase generations written after it started"
|
||||
);
|
||||
}
|
||||
|
||||
/// Contention is a 429 and must be retried. The server keeps it off 503
|
||||
/// precisely so the client can act on the status alone — reading it as
|
||||
/// terminal stops the checkpoint early on a healthy node.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_checkpoint_retries_contention() {
|
||||
let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = compacts.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
if path.contains("flush_lsm") {
|
||||
return accepted();
|
||||
}
|
||||
if path.contains("compact_lsm") {
|
||||
// First two triggers: every bucket already latched.
|
||||
if seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < 2 {
|
||||
return http::Response::builder()
|
||||
.status(429)
|
||||
.body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
return accepted();
|
||||
}
|
||||
let accepted_triggers = seen
|
||||
.load(std::sync::atomic::Ordering::SeqCst)
|
||||
.saturating_sub(2);
|
||||
let left: Vec<u64> = if accepted_triggers == 0 {
|
||||
vec![1]
|
||||
} else {
|
||||
vec![]
|
||||
};
|
||||
ok_json(stats_body(&left, false))
|
||||
});
|
||||
|
||||
table
|
||||
.checkpoint_lsm()
|
||||
.await
|
||||
.expect("contention must not abort the checkpoint");
|
||||
assert_eq!(
|
||||
compacts.load(std::sync::atomic::Ordering::SeqCst),
|
||||
3,
|
||||
"assert the retry count, not just the outcome"
|
||||
);
|
||||
}
|
||||
|
||||
/// A transient fault on the poll must not abort the checkpoint. This route
|
||||
/// meets the most contention — it runs every `POLL_INTERVAL` for the
|
||||
/// checkpoint's whole life, with the transport retry layer disabled — yet
|
||||
/// was the one call reached with a bare `?`.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_checkpoint_retries_a_contended_stats_poll() {
|
||||
let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = polls.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
if path.contains("flush_lsm") || path.contains("compact_lsm") {
|
||||
return accepted();
|
||||
}
|
||||
// The opening read lands; the next two polls are latched out.
|
||||
let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
if (1..3).contains(&n) {
|
||||
return http::Response::builder()
|
||||
.status(429)
|
||||
.body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
ok_json(stats_body(if n < 4 { &[1] } else { &[] }, false))
|
||||
});
|
||||
|
||||
table
|
||||
.checkpoint_lsm()
|
||||
.await
|
||||
.expect("a contended poll must be retried, not surfaced");
|
||||
assert_eq!(
|
||||
polls.load(std::sync::atomic::Ordering::SeqCst),
|
||||
5,
|
||||
"the two rejected polls must be re-issued, not skipped"
|
||||
);
|
||||
}
|
||||
|
||||
/// Contention and a lost claim draw on separate budgets: five straight
|
||||
/// 429s on `flush`, more than `MAX_REISSUES`, must still converge. On one
|
||||
/// shared counter this spent the re-issue cap and then reported a lost
|
||||
/// claim nothing had ever reported.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_contention_does_not_exhaust_the_reissue_budget() {
|
||||
let flushes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = flushes.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
if path.contains("flush_lsm") {
|
||||
if seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < 5 {
|
||||
return http::Response::builder()
|
||||
.status(429)
|
||||
.body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
return accepted();
|
||||
}
|
||||
if path.contains("compact_lsm") {
|
||||
return accepted();
|
||||
}
|
||||
ok_json(stats_body(&[], false))
|
||||
});
|
||||
|
||||
table
|
||||
.checkpoint_lsm()
|
||||
.await
|
||||
.expect("contention must not be reported as a lost claim");
|
||||
assert_eq!(
|
||||
flushes.load(std::sync::atomic::Ordering::SeqCst),
|
||||
6,
|
||||
"five retries against one seal, then it lands"
|
||||
);
|
||||
}
|
||||
|
||||
/// An exhausted retry budget surfaces the fault that consumed it, not a
|
||||
/// message the loop invented: "429, nine times" points an operator at a
|
||||
/// saturated pool, a generic runtime error points them nowhere.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_exhausted_retries_surface_the_underlying_fault() {
|
||||
let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = calls.clone();
|
||||
let table = Table::new_with_handler("my_table", move |_request| {
|
||||
seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
http::Response::builder()
|
||||
.status(429)
|
||||
.body(r#"{"code":21,"error":"Too many concurrent writes"}"#.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
let err = table.checkpoint_lsm().await.unwrap_err();
|
||||
assert!(
|
||||
matches!(&err, Error::Http { status_code: Some(s), .. } if s.as_u16() == 429),
|
||||
"the fault that spent the budget must be the one reported: {err:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
calls.load(std::sync::atomic::Ordering::SeqCst),
|
||||
9,
|
||||
"one call plus MAX_RETRIES — the re-issue budget is not spent on top"
|
||||
);
|
||||
}
|
||||
|
||||
/// A draining node is terminal, but the client does not know that from the
|
||||
/// status: draining and a proxy blip are both 503, and telling them apart
|
||||
/// takes parsing the body for a namespace code. So it spends the retry
|
||||
/// budget and then reports what the server said — the drain gate never
|
||||
/// releases, so the answer does not change, and the operator still reads
|
||||
/// "WAL node draining" in the error.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_draining_surfaces_after_the_retry_budget() {
|
||||
let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = calls.clone();
|
||||
let table = Table::new_with_handler("my_table", move |_request| {
|
||||
seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
http::Response::builder()
|
||||
.status(503)
|
||||
.body(r#"{"code":19,"error":"WAL node draining"}"#.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
let err = table.checkpoint_lsm().await.unwrap_err();
|
||||
let message = err.to_string();
|
||||
assert!(
|
||||
matches!(&err, Error::Http { status_code: Some(s), .. } if s.as_u16() == 503),
|
||||
"the 503 must surface as itself: {err:?}"
|
||||
);
|
||||
assert!(
|
||||
message.contains("WAL node draining"),
|
||||
"the server's own diagnosis must survive to the caller: {message}"
|
||||
);
|
||||
assert_eq!(
|
||||
calls.load(std::sync::atomic::Ordering::SeqCst),
|
||||
9,
|
||||
"one call plus MAX_RETRIES, then it reports rather than spinning"
|
||||
);
|
||||
}
|
||||
|
||||
/// A long stall with nothing compacting must keep waiting, not fail. The
|
||||
/// client cannot judge this: a checkpoint queued behind unrelated tables
|
||||
/// on the pod-wide compactor pool reports exactly these numbers — flat
|
||||
/// generations, an idle latch — as one whose merges are failing. The
|
||||
/// deadline is the caller's.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_checkpoint_waits_out_a_long_stall_rather_than_failing() {
|
||||
let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = polls.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
if path.contains("flush_lsm") || path.contains("compact_lsm") {
|
||||
return accepted();
|
||||
}
|
||||
// Flat for far longer than any bound this loop ever had, with
|
||||
// `compacting: false` throughout — then it drains.
|
||||
let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
ok_json(stats_body(if n < 40 { &[1, 2] } else { &[] }, false))
|
||||
});
|
||||
|
||||
table
|
||||
.checkpoint_lsm()
|
||||
.await
|
||||
.expect("a stall is the server being slow, not the client's call to make");
|
||||
assert!(
|
||||
polls.load(std::sync::atomic::Ordering::SeqCst) > 40,
|
||||
"the loop must have kept polling well past the old ten-poll bound"
|
||||
);
|
||||
}
|
||||
|
||||
/// A pass already owns the latch on every outstanding bucket, so the loop
|
||||
/// waits rather than piling on triggers it would only refuse. This is the
|
||||
/// sole thing `compacting` is read for.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_checkpoint_waits_while_a_pass_is_running() {
|
||||
let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let compacts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen_polls = polls.clone();
|
||||
let seen_compacts = compacts.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
if path.contains("flush_lsm") {
|
||||
return accepted();
|
||||
}
|
||||
if path.contains("compact_lsm") {
|
||||
seen_compacts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
return accepted();
|
||||
}
|
||||
// Latched for many polls, then done.
|
||||
let n = seen_polls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
ok_json(if n > 15 {
|
||||
stats_body(&[], false)
|
||||
} else {
|
||||
stats_body(&[1], true)
|
||||
})
|
||||
});
|
||||
|
||||
table
|
||||
.checkpoint_lsm()
|
||||
.await
|
||||
.expect("a running pass is progress, not a stall");
|
||||
assert_eq!(
|
||||
compacts.load(std::sync::atomic::Ordering::SeqCst),
|
||||
0,
|
||||
"never trigger against a bucket already compacting"
|
||||
);
|
||||
}
|
||||
|
||||
/// WAL off ⇒ `None`; WAL on ⇒ a fully populated `Some` with no field
|
||||
/// defaulting to a zero it did not measure. `include_generation_rows`
|
||||
/// rides in the body and is off unless asked for.
|
||||
#[tokio::test]
|
||||
async fn test_get_lsm_stats_round_trip() {
|
||||
let table = Table::new_with_handler("my_table", |request| {
|
||||
assert_eq!(request.url().path(), "/v1/table/my_table/get_lsm_stats/");
|
||||
let body = request.body().unwrap().as_bytes().unwrap();
|
||||
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
|
||||
assert_eq!(
|
||||
body["include_generation_rows"], true,
|
||||
"the flag must reach the server, not be silently dropped"
|
||||
);
|
||||
let response = serde_json::json!({
|
||||
"lsm_stats": {
|
||||
"buckets": [{
|
||||
"shard_id": "b0",
|
||||
"status": "Active",
|
||||
"writer_epoch": 3,
|
||||
"manifest_version": 11,
|
||||
"current_generation": 9,
|
||||
"replay_after_wal_entry_position": 100,
|
||||
"wal_entry_position_last_seen": 140,
|
||||
"generations": [{ "generation": 8, "bytes": 4096, "rows": 30 }],
|
||||
"compacting": false,
|
||||
"memtables": [
|
||||
{ "generation": 9, "rows": 12, "bytes": 900, "batches": 2,
|
||||
"indexes": ["vec_idx"] }
|
||||
],
|
||||
}],
|
||||
}
|
||||
});
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(response.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
let stats = table
|
||||
.get_lsm_stats(true)
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("a WAL-backed table reports Some");
|
||||
let bucket = &stats.buckets[0];
|
||||
assert_eq!(bucket.replay_after_wal_entry_position, 100);
|
||||
assert_eq!(bucket.wal_entry_position_last_seen, 140);
|
||||
assert!(!bucket.compacting);
|
||||
assert_eq!(bucket.generations[0].generation, 8);
|
||||
assert_eq!(bucket.generations[0].rows, Some(30));
|
||||
// The line that answers "why is my fresh-tier vector search
|
||||
// brute-force" — an absent index name is the whole explanation.
|
||||
let memtables = bucket.memtables.as_ref().unwrap();
|
||||
assert_eq!(memtables[0].indexes, vec!["vec_idx".to_string()]);
|
||||
}
|
||||
|
||||
/// A 404 arrives as `TableNotFound`, not as a lost claim the loop
|
||||
/// re-issues from flush until its cap. The two are distinguished by
|
||||
/// status: 404 is "no such table", 421 is "this node holds no claim".
|
||||
/// They shared 404 once, and the loop chased a name that never existed.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_missing_table_is_not_read_as_a_lost_claim() {
|
||||
let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = calls.clone();
|
||||
let table = Table::new_with_handler("my_table", move |_request| {
|
||||
seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
http::Response::builder()
|
||||
.status(404)
|
||||
.body(r#"{"code":4,"error":"Not found: Table not found: my_table"}"#.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
let err = table.checkpoint_lsm().await.unwrap_err();
|
||||
assert!(
|
||||
matches!(err, Error::TableNotFound { .. }),
|
||||
"a missing table must say so: {err:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
calls.load(std::sync::atomic::Ordering::SeqCst),
|
||||
1,
|
||||
"no point re-claiming a table that does not exist"
|
||||
);
|
||||
}
|
||||
|
||||
/// A lost claim — 421, not 404 — does re-issue from flush, the call that
|
||||
/// re-claims and replays.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_registry_miss_reissues_from_flush() {
|
||||
let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = calls.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
if path.contains("flush_lsm") {
|
||||
// First flush lands; the claim is then lost, and the
|
||||
// re-issued flush succeeds.
|
||||
return accepted();
|
||||
}
|
||||
if path.contains("compact_lsm") {
|
||||
if n < 4 {
|
||||
return http::Response::builder()
|
||||
.status(421)
|
||||
.body(r#"{"code":19,"error":"table not claimed"}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
return accepted();
|
||||
}
|
||||
ok_json(stats_body(if n < 6 { &[1] } else { &[] }, false))
|
||||
});
|
||||
|
||||
table
|
||||
.checkpoint_lsm()
|
||||
.await
|
||||
.expect("a lost claim must be recovered by re-flushing, not surfaced");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_lsm_stats_absent_when_wal_off() {
|
||||
let table = Table::new_with_handler("my_table", |_request| {
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(serde_json::json!({ "lsm_stats": null }).to_string())
|
||||
.unwrap()
|
||||
});
|
||||
assert!(table.get_lsm_stats(false).await.unwrap().is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_wait_for_index() {
|
||||
let table = _make_table_with_indices(0);
|
||||
|
||||
+496
-41
@@ -68,10 +68,12 @@ use self::merge::MergeInsertBuilder;
|
||||
pub mod add_columns;
|
||||
mod add_data;
|
||||
pub mod branch_merge;
|
||||
pub mod checkpoint;
|
||||
mod create_index;
|
||||
pub mod datafusion;
|
||||
pub(crate) mod dataset;
|
||||
pub mod delete;
|
||||
pub mod lsm_stats;
|
||||
pub mod merge;
|
||||
pub mod optimize;
|
||||
mod primary_key;
|
||||
@@ -93,8 +95,8 @@ pub use delete::DeleteResult;
|
||||
use futures::future::join_all;
|
||||
pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTags};
|
||||
pub use lance::dataset::scanner::DatasetRecordBatchStream;
|
||||
use lance::dataset::statistics::DatasetStatisticsExt;
|
||||
pub use lance_index::optimize::OptimizeOptions;
|
||||
pub use lsm_stats::{BucketStats, GenerationStats, LsmStats, MemtableStats};
|
||||
pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats};
|
||||
pub use schema_evolution::{
|
||||
AddColumnsResult, AlterColumnsResult, DropColumnsResult, FieldMetadataUpdate,
|
||||
@@ -368,6 +370,8 @@ pub use self::merge::MergeResult;
|
||||
/// date) and [`LsmWriteSpec::with_writer_config_defaults`] (default
|
||||
/// `ShardWriter` configuration recorded in the MemWAL index).
|
||||
///
|
||||
/// A fresh spec maintains every index on the table, resolved on install.
|
||||
///
|
||||
/// Install a spec with [`Table::set_lsm_write_spec`] and remove it with
|
||||
/// [`Table::unset_lsm_write_spec`]. The actual `merge_insert` dispatch
|
||||
/// onto the MemWAL writer is a follow-up.
|
||||
@@ -382,9 +386,12 @@ pub enum LsmWriteSpec {
|
||||
Bucket {
|
||||
column: String,
|
||||
num_buckets: u32,
|
||||
/// Names of indexes (already created on the table) that the
|
||||
/// MemWAL should maintain in-memory as rows are appended.
|
||||
maintained_indexes: Vec<String>,
|
||||
/// Indexes the MemWAL maintains in-memory as rows are appended.
|
||||
///
|
||||
/// `None` means every index it can maintain, resolved on install — a
|
||||
/// snapshot, so indexes created later need the spec unset and re-set.
|
||||
/// `Some([])` maintains nothing.
|
||||
maintained_indexes: Option<Vec<String>>,
|
||||
/// Default `ShardWriter` configuration recorded in the MemWAL index.
|
||||
writer_config_defaults: HashMap<String, String>,
|
||||
},
|
||||
@@ -394,35 +401,41 @@ pub enum LsmWriteSpec {
|
||||
/// distinct value of `column` becomes its own shard.
|
||||
Identity {
|
||||
column: String,
|
||||
/// Names of indexes (already created on the table) that the
|
||||
/// MemWAL should maintain in-memory as rows are appended.
|
||||
maintained_indexes: Vec<String>,
|
||||
/// Indexes the MemWAL maintains in-memory as rows are appended.
|
||||
///
|
||||
/// `None` means every index it can maintain, resolved on install — a
|
||||
/// snapshot, so indexes created later need the spec unset and re-set.
|
||||
/// `Some([])` maintains nothing.
|
||||
maintained_indexes: Option<Vec<String>>,
|
||||
/// Default `ShardWriter` configuration recorded in the MemWAL index.
|
||||
writer_config_defaults: HashMap<String, String>,
|
||||
},
|
||||
/// No sharding — every `merge_insert` call writes to a single MemWAL shard.
|
||||
Unsharded {
|
||||
/// Names of indexes (already created on the table) that the
|
||||
/// MemWAL should maintain in-memory as rows are appended.
|
||||
maintained_indexes: Vec<String>,
|
||||
/// Indexes the MemWAL maintains in-memory as rows are appended.
|
||||
///
|
||||
/// `None` means every index it can maintain, resolved on install — a
|
||||
/// snapshot, so indexes created later need the spec unset and re-set.
|
||||
/// `Some([])` maintains nothing.
|
||||
maintained_indexes: Option<Vec<String>>,
|
||||
/// Default `ShardWriter` configuration recorded in the MemWAL index.
|
||||
writer_config_defaults: HashMap<String, String>,
|
||||
},
|
||||
}
|
||||
|
||||
impl LsmWriteSpec {
|
||||
/// Construct a hash-bucket sharding spec with no maintained indexes.
|
||||
/// Construct a hash-bucket sharding spec maintaining every index on the table.
|
||||
pub fn bucket(column: impl Into<String>, num_buckets: u32) -> Self {
|
||||
Self::Bucket {
|
||||
column: column.into(),
|
||||
num_buckets,
|
||||
maintained_indexes: Vec::new(),
|
||||
maintained_indexes: None,
|
||||
writer_config_defaults: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Construct an identity-sharding spec (shard by the raw value of
|
||||
/// `column`) with no maintained indexes.
|
||||
/// `column`) maintaining every index on the table.
|
||||
///
|
||||
/// `column` must be a deterministic function of the unenforced primary
|
||||
/// key: every row with a given primary key must always produce the same
|
||||
@@ -434,28 +447,37 @@ impl LsmWriteSpec {
|
||||
pub fn identity(column: impl Into<String>) -> Self {
|
||||
Self::Identity {
|
||||
column: column.into(),
|
||||
maintained_indexes: Vec::new(),
|
||||
maintained_indexes: None,
|
||||
writer_config_defaults: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Construct an unsharded spec with no maintained indexes.
|
||||
/// Construct an unsharded spec maintaining every index on the table.
|
||||
pub fn unsharded() -> Self {
|
||||
Self::Unsharded {
|
||||
maintained_indexes: Vec::new(),
|
||||
maintained_indexes: None,
|
||||
writer_config_defaults: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Replace the list of indexes the MemWAL should keep up to date as
|
||||
/// rows are appended. Each name must reference an index that already
|
||||
/// exists on the table at the time `set_lsm_write_spec` is called.
|
||||
pub fn with_maintained_indexes<I, S>(mut self, indexes: I) -> Self
|
||||
where
|
||||
I: IntoIterator<Item = S>,
|
||||
S: Into<String>,
|
||||
{
|
||||
let v: Vec<String> = indexes.into_iter().map(Into::into).collect();
|
||||
/// Set which indexes the MemWAL maintains.
|
||||
///
|
||||
/// `None` (the default) resolves to every index on the table at install,
|
||||
/// failing if one cannot be maintained — name the set to install anyway. A
|
||||
/// list is verbatim: each name must already exist and be maintainable, and
|
||||
/// an empty list maintains nothing.
|
||||
///
|
||||
/// ```
|
||||
/// # use lancedb::table::LsmWriteSpec;
|
||||
/// // Every index the table has when the spec is installed:
|
||||
/// LsmWriteSpec::unsharded().with_maintained_indexes(None);
|
||||
/// // Exactly these:
|
||||
/// LsmWriteSpec::unsharded().with_maintained_indexes(vec!["id_idx".to_string()]);
|
||||
/// // None at all:
|
||||
/// LsmWriteSpec::unsharded().with_maintained_indexes(Vec::new());
|
||||
/// ```
|
||||
pub fn with_maintained_indexes(mut self, indexes: impl Into<Option<Vec<String>>>) -> Self {
|
||||
let indexes = indexes.into();
|
||||
match &mut self {
|
||||
Self::Bucket {
|
||||
maintained_indexes, ..
|
||||
@@ -465,7 +487,7 @@ impl LsmWriteSpec {
|
||||
}
|
||||
| Self::Unsharded {
|
||||
maintained_indexes, ..
|
||||
} => *maintained_indexes = v,
|
||||
} => *maintained_indexes = indexes,
|
||||
}
|
||||
self
|
||||
}
|
||||
@@ -501,8 +523,9 @@ impl LsmWriteSpec {
|
||||
self
|
||||
}
|
||||
|
||||
/// Borrow the list of index names this spec asks MemWAL to maintain.
|
||||
pub fn maintained_indexes(&self) -> &[String] {
|
||||
/// Borrow the list of index names this spec asks MemWAL to maintain, or
|
||||
/// `None` when it asks for every index on the table.
|
||||
pub fn maintained_indexes(&self) -> Option<&[String]> {
|
||||
match self {
|
||||
Self::Bucket {
|
||||
maintained_indexes, ..
|
||||
@@ -512,7 +535,7 @@ impl LsmWriteSpec {
|
||||
}
|
||||
| Self::Unsharded {
|
||||
maintained_indexes, ..
|
||||
} => maintained_indexes,
|
||||
} => maintained_indexes.as_deref(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -685,6 +708,31 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
|
||||
message: "get_lsm_write_spec is not supported on this table type".into(),
|
||||
})
|
||||
}
|
||||
/// Seal every bucket's active memtable into L0.
|
||||
///
|
||||
/// The default implementation returns `NotSupported`.
|
||||
async fn flush_lsm(&self) -> Result<()> {
|
||||
Err(Error::NotSupported {
|
||||
message: "flush_lsm is not supported on this table type".into(),
|
||||
})
|
||||
}
|
||||
/// Trigger a background L0 → base compaction pass per bucket.
|
||||
///
|
||||
/// The default implementation returns `NotSupported`.
|
||||
async fn compact_lsm(&self) -> Result<()> {
|
||||
Err(Error::NotSupported {
|
||||
message: "compact_lsm is not supported on this table type".into(),
|
||||
})
|
||||
}
|
||||
/// Read live LSM state, or `None` when the LSM write path is not
|
||||
/// enabled for this table.
|
||||
///
|
||||
/// The default implementation returns `NotSupported`.
|
||||
async fn get_lsm_stats(&self, _include_generation_rows: bool) -> Result<Option<LsmStats>> {
|
||||
Err(Error::NotSupported {
|
||||
message: "get_lsm_stats is not supported on this table type".into(),
|
||||
})
|
||||
}
|
||||
/// Drain and close any cached MemWAL shard writers for this table.
|
||||
///
|
||||
/// The default implementation is a no-op; table types that maintain
|
||||
@@ -1685,7 +1733,7 @@ impl Table {
|
||||
/// # async fn example(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
|
||||
/// table
|
||||
/// .set_lsm_write_spec(
|
||||
/// LsmWriteSpec::bucket("id", 16).with_maintained_indexes(["id_idx"]),
|
||||
/// LsmWriteSpec::bucket("id", 16).with_maintained_indexes(vec!["id_idx".to_string()]),
|
||||
/// )
|
||||
/// .await?;
|
||||
/// # Ok(())
|
||||
@@ -1707,9 +1755,10 @@ impl Table {
|
||||
///
|
||||
/// Returns `Ok(None)` when the MemWAL LSM write path is not enabled (no
|
||||
/// spec has been set, or it was removed with [`Table::unset_lsm_write_spec`]).
|
||||
/// The returned spec — including its [`LsmWriteSpec::maintained_indexes`] and
|
||||
/// [`LsmWriteSpec::writer_config_defaults`] — mirrors what was passed to
|
||||
/// [`Table::set_lsm_write_spec`].
|
||||
/// The returned spec mirrors what was passed to
|
||||
/// [`Table::set_lsm_write_spec`], except that
|
||||
/// [`LsmWriteSpec::maintained_indexes`] always reports the concrete list
|
||||
/// resolved when the spec was set — `None` never round-trips.
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
@@ -1726,6 +1775,85 @@ impl Table {
|
||||
self.inner.get_lsm_write_spec().await
|
||||
}
|
||||
|
||||
/// Converge this table's LSM write path into its base table.
|
||||
///
|
||||
/// One `flush` to seal every memtable into L0, then compaction triggers
|
||||
/// until every generation that existed at that moment has reached base.
|
||||
/// The loop runs client-side, reading progress from `get_lsm_stats`, so
|
||||
/// there is no held socket and nothing to reconcile if you drop this
|
||||
/// future partway through.
|
||||
///
|
||||
/// **Best-effort.** Generations created *after* the opening flush are
|
||||
/// deliberately not waited on — that is what lets this terminate on a
|
||||
/// table taking writes. Idempotent and safe on a cadence: an
|
||||
/// already-converged table costs two round trips and triggers nothing.
|
||||
///
|
||||
/// **No deadline, and the caller owns that.** It returns when the target
|
||||
/// generations are gone, propagates a terminal server fault, and
|
||||
/// otherwise waits however long the server takes. A slow table and a
|
||||
/// stuck one are the same picture from here: the compactor pool is shared
|
||||
/// across every table on the node, so a checkpoint queued behind
|
||||
/// unrelated work is indistinguishable from one that is merging. Wrap
|
||||
/// this in `tokio::time::timeout` for a wall-clock bound; abandoning it
|
||||
/// partway costs nothing.
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```no_run
|
||||
/// # use lancedb::Table;
|
||||
/// # async fn example(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
|
||||
/// let before = table.get_lsm_stats(false).await?;
|
||||
/// table.checkpoint_lsm().await?;
|
||||
/// let after = table.get_lsm_stats(false).await?;
|
||||
/// # Ok(())
|
||||
/// # }
|
||||
/// ```
|
||||
pub async fn checkpoint_lsm(&self) -> Result<()> {
|
||||
checkpoint::checkpoint_lsm(self).await
|
||||
}
|
||||
|
||||
/// Seal every bucket's active memtable into L0 without touching the
|
||||
/// base table.
|
||||
///
|
||||
/// Independently useful: flushing makes memtable rows readable from L0 at
|
||||
/// a lower per-query cost. On a node that has not claimed this table it
|
||||
/// claims it and replays the WAL log first — reporting "nothing to flush"
|
||||
/// without replaying would lie about durable data.
|
||||
pub async fn flush_lsm(&self) -> Result<()> {
|
||||
self.inner.flush_lsm().await
|
||||
}
|
||||
|
||||
/// Run one bounded L0 → base compaction pass per bucket, reporting what
|
||||
/// it merged and what is left.
|
||||
///
|
||||
/// One pass, not convergence: that bounds each request's cost and gives a
|
||||
/// caller driving its own cadence a progress signal per round trip.
|
||||
pub async fn compact_lsm(&self) -> Result<()> {
|
||||
self.inner.compact_lsm().await
|
||||
}
|
||||
|
||||
/// Read live per-bucket LSM state.
|
||||
///
|
||||
/// Answers "how far behind is my fresh tier", "which bucket is hot", and
|
||||
/// "why is my fresh-tier vector search brute-force". Mutates no table
|
||||
/// state, though on a node that has not claimed this table it claims it,
|
||||
/// exactly as a read would.
|
||||
///
|
||||
/// `include_generation_rows` reports a row count per L0 generation. Off by
|
||||
/// default: each count opens an uncached Lance dataset, and
|
||||
/// `checkpoint_lsm` polls this needing only generation numbers.
|
||||
///
|
||||
/// `Ok(None)` only when the LSM write path is not enabled, matching
|
||||
/// [`Table::get_lsm_write_spec`]. Stats is fresh-tier only, so with the
|
||||
/// WAL off there is no manifest to report and a struct of zeros would
|
||||
/// read as measurements.
|
||||
///
|
||||
/// Do not build a checkpoint's termination on this: the completion
|
||||
/// predicate lives in the `flush` and `compact` responses.
|
||||
pub async fn get_lsm_stats(&self, include_generation_rows: bool) -> Result<Option<LsmStats>> {
|
||||
self.inner.get_lsm_stats(include_generation_rows).await
|
||||
}
|
||||
|
||||
/// Drain and close any cached MemWAL shard writers held for this table.
|
||||
///
|
||||
/// When an [`LsmWriteSpec`] is installed, `merge_insert` opens MemWAL shard
|
||||
@@ -3441,9 +3569,24 @@ impl BaseTable for NativeTable {
|
||||
let num_rows = self.count_rows(None).await?;
|
||||
let num_indices = self.list_indices().await?.len();
|
||||
let ds = self.dataset.get().await?;
|
||||
let ds_clone = (*ds).clone();
|
||||
let ds_stats = Arc::new(ds_clone).calculate_data_stats().await?;
|
||||
let total_bytes = ds_stats.fields.iter().map(|f| f.bytes_on_disk).sum::<u64>() as usize;
|
||||
// Sizes come from the manifest. Summing per-field `bytes_on_disk` instead
|
||||
// would open every data file to read its column metadata, which costs one
|
||||
// IO per fragment and reports 0 for legacy v1 storage.
|
||||
//
|
||||
// The manifest summary covers only the fragments' base data files, so
|
||||
// overlay files (recorded on each fragment) and index files (recorded in
|
||||
// the manifest's index section) are added separately.
|
||||
let mut total_bytes = ds.manifest().summary().total_files_size as usize;
|
||||
for frag in ds.manifest().fragments.iter() {
|
||||
for overlay in &frag.overlays {
|
||||
if let Some(size) = overlay.data_file.file_size_bytes.get() {
|
||||
total_bytes += size.get() as usize;
|
||||
}
|
||||
}
|
||||
}
|
||||
for index in ds.load_indices().await?.iter() {
|
||||
total_bytes += index.total_size_bytes().unwrap_or(0) as usize;
|
||||
}
|
||||
|
||||
let frags = ds.get_fragments();
|
||||
let mut sorted_sizes = join_all(
|
||||
@@ -3515,7 +3658,12 @@ impl BaseTable for NativeTable {
|
||||
#[skip_serializing_none]
|
||||
#[derive(Debug, Deserialize, PartialEq)]
|
||||
pub struct TableStatistics {
|
||||
/// The total number of bytes in the table
|
||||
/// The total size, in bytes, of the table's data files, index files, and
|
||||
/// overlay files
|
||||
///
|
||||
/// Read from the manifest, so this excludes deletion files and manifests,
|
||||
/// and it excludes any file whose size the manifest does not record
|
||||
/// (tables and indices written before writers persisted file sizes).
|
||||
pub total_bytes: usize,
|
||||
|
||||
/// The number of rows in the table
|
||||
@@ -3576,6 +3724,7 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::connect;
|
||||
use crate::connection::ConnectBuilder;
|
||||
use crate::io::object_store::io_tracking::IoTrackingStore;
|
||||
use crate::query::Select;
|
||||
use crate::query::{ExecutableQuery, QueryBase};
|
||||
use crate::test_utils::connection::new_test_connection;
|
||||
@@ -4958,7 +5107,7 @@ mod tests {
|
||||
// Bucket spec round-trips exactly, including the routing column (recovered
|
||||
// from its field id), maintained indexes, and writer config defaults.
|
||||
let spec = LsmWriteSpec::bucket("id", 4)
|
||||
.with_maintained_indexes([idx_name])
|
||||
.with_maintained_indexes(vec![idx_name.clone()])
|
||||
.with_writer_config_defaults([("durable_write", "false")]);
|
||||
table.set_lsm_write_spec(spec.clone()).await.unwrap();
|
||||
assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
|
||||
@@ -4968,15 +5117,125 @@ mod tests {
|
||||
assert_eq!(table.get_lsm_write_spec().await.unwrap(), None);
|
||||
|
||||
// Identity sharding round-trips (column recovered from the schema).
|
||||
// A spec left at its default maintains every index on the table, so it
|
||||
// reads back naming the one on the table rather than as "infer".
|
||||
let spec = LsmWriteSpec::identity("region");
|
||||
table.set_lsm_write_spec(spec.clone()).await.unwrap();
|
||||
assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
|
||||
assert_eq!(
|
||||
table.get_lsm_write_spec().await.unwrap(),
|
||||
Some(spec.with_maintained_indexes(vec![idx_name.clone()]))
|
||||
);
|
||||
table.unset_lsm_write_spec().await.unwrap();
|
||||
|
||||
// Unsharded round-trips (no routing column).
|
||||
let spec = LsmWriteSpec::unsharded();
|
||||
table.set_lsm_write_spec(spec.clone()).await.unwrap();
|
||||
assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec));
|
||||
assert_eq!(
|
||||
table.get_lsm_write_spec().await.unwrap(),
|
||||
Some(spec.with_maintained_indexes(vec![idx_name]))
|
||||
);
|
||||
}
|
||||
|
||||
/// The maintained set defaults to every index on the table, resolved at
|
||||
/// install. An index the memtable cannot build fails the install rather
|
||||
/// than being dropped: maintaining it would take the table offline for
|
||||
/// writes, dropping it would hide that from the caller.
|
||||
#[tokio::test]
|
||||
async fn test_set_lsm_write_spec_infers_maintained_indexes() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let uri = tmp_dir.path().to_str().unwrap();
|
||||
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
Field::new("id", DataType::Int64, false),
|
||||
Field::new("tag", DataType::Utf8, true),
|
||||
]));
|
||||
let batch = RecordBatch::try_new(
|
||||
schema.clone(),
|
||||
vec![
|
||||
Arc::new(arrow_array::Int64Array::from(vec![1, 2, 3])),
|
||||
Arc::new(StringArray::from(vec!["a", "b", "c"])),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let reader: Box<dyn arrow_array::RecordBatchReader + Send> =
|
||||
Box::new(RecordBatchIterator::new(vec![Ok(batch)], schema.clone()));
|
||||
let conn = ConnectBuilder::new(uri)
|
||||
.read_consistency_interval(Duration::from_secs(0))
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let table = conn.create_table("t", reader).execute().await.unwrap();
|
||||
|
||||
table
|
||||
.create_index(&["id"], Index::BTree(Default::default()))
|
||||
.name("id_btree".to_string())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
table
|
||||
.create_index(&["tag"], Index::Bitmap(Default::default()))
|
||||
.name("tag_bitmap".to_string())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Explicitly naming the bitmap index fails before anything commits.
|
||||
let err = table
|
||||
.set_lsm_write_spec(
|
||||
LsmWriteSpec::unsharded().with_maintained_indexes(vec!["tag_bitmap".to_string()]),
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
matches!(err, Error::InvalidInput { ref message } if message.contains("tag_bitmap")),
|
||||
"expected the bitmap index to be rejected, got {err:?}"
|
||||
);
|
||||
assert_eq!(table.get_lsm_write_spec().await.unwrap(), None);
|
||||
|
||||
// The default covers every index, so the bitmap fails it too.
|
||||
let err = table
|
||||
.set_lsm_write_spec(LsmWriteSpec::unsharded())
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
matches!(err, Error::InvalidInput { ref message }
|
||||
if message.contains("tag_bitmap") && message.contains("maintained_indexes")),
|
||||
"expected the inferred set to be rejected, got {err:?}"
|
||||
);
|
||||
assert_eq!(table.get_lsm_write_spec().await.unwrap(), None);
|
||||
|
||||
// Naming the maintainable subset installs.
|
||||
table
|
||||
.set_lsm_write_spec(
|
||||
LsmWriteSpec::unsharded().with_maintained_indexes(vec!["id_btree".to_string()]),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
table
|
||||
.get_lsm_write_spec()
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.maintained_indexes(),
|
||||
Some(["id_btree".to_string()].as_slice())
|
||||
);
|
||||
|
||||
// Opting out entirely is distinct from the default.
|
||||
table.unset_lsm_write_spec().await.unwrap();
|
||||
table
|
||||
.set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(Vec::new()))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
table
|
||||
.get_lsm_write_spec()
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.maintained_indexes(),
|
||||
Some([].as_slice())
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -5024,12 +5283,16 @@ mod tests {
|
||||
|
||||
let res = table.stats().await.unwrap();
|
||||
println!("{:#?}", res);
|
||||
// `total_bytes` is the full on-disk size of the 11 data files (this table
|
||||
// has no index or overlay files), so it is well above the 2000 bytes of
|
||||
// column data these 250 int32 pairs hold: each file carries its own footer
|
||||
// and metadata.
|
||||
assert_eq!(
|
||||
res,
|
||||
TableStatistics {
|
||||
num_rows: 250,
|
||||
num_indices: 0,
|
||||
total_bytes: 2300,
|
||||
total_bytes: 8925,
|
||||
fragment_stats: FragmentStatistics {
|
||||
num_fragments: 11,
|
||||
num_small_fragments: 11,
|
||||
@@ -5069,4 +5332,196 @@ mod tests {
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
/// `total_bytes` counts more than the base data files: index files and
|
||||
/// overlay files recorded in the manifest are included too.
|
||||
#[tokio::test]
|
||||
pub async fn test_stats_includes_index_and_overlay_files() {
|
||||
use lance::dataset::WriteDestination;
|
||||
use lance::dataset::transaction::{DataOverlayGroup, Operation};
|
||||
use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
|
||||
use lance_file::writer::FileWriterOptions;
|
||||
use lance_io::utils::CachedFileSize;
|
||||
use lance_table::format::DataFile;
|
||||
use lance_table::format::overlay::{DataOverlayFile, OverlayCoverage};
|
||||
use roaring::RoaringBitmap;
|
||||
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let uri = tmp_dir.path().to_str().unwrap();
|
||||
let conn = ConnectBuilder::new(uri)
|
||||
.read_consistency_interval(Duration::from_secs(0))
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
Field::new("id", DataType::Int32, false),
|
||||
Field::new("foo", DataType::Int32, true),
|
||||
]));
|
||||
let batch = RecordBatch::try_new(
|
||||
schema.clone(),
|
||||
vec![
|
||||
Arc::new(Int32Array::from_iter_values(0..100)),
|
||||
Arc::new(Int32Array::from_iter_values(0..100)),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let table = conn
|
||||
.create_table("test_stats_extra_files", batch)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let data_only = table.stats().await.unwrap().total_bytes;
|
||||
assert!(data_only > 0);
|
||||
|
||||
// A scalar index adds index files whose sizes are recorded in the
|
||||
// manifest's index section.
|
||||
table
|
||||
.create_index(&["id"], Index::Auto)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let with_index = table.stats().await.unwrap().total_bytes;
|
||||
let dataset = {
|
||||
let native = table.as_native().unwrap();
|
||||
(*native.dataset.get().await.unwrap()).clone()
|
||||
};
|
||||
let index_bytes: usize = dataset
|
||||
.load_indices()
|
||||
.await
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|idx| idx.total_size_bytes().unwrap_or(0) as usize)
|
||||
.sum();
|
||||
assert!(index_bytes > 0);
|
||||
assert_eq!(with_index, data_only + index_bytes);
|
||||
|
||||
// Commit an overlay file supplying new `foo` values for the first three
|
||||
// rows of fragment 0. There is no high-level API that writes overlays
|
||||
// yet, so write the overlay's data file and commit the `DataOverlay`
|
||||
// operation by hand.
|
||||
let read_version = dataset.version().version;
|
||||
let fragment_id = dataset.get_fragments()[0].id() as u64;
|
||||
let foo_field_id = dataset.schema().field("foo").unwrap().id;
|
||||
let overlay_schema = dataset.schema().project_by_ids(&[foo_field_id], true);
|
||||
let file_version = ConcreteFileVersion::from(LanceFileVersion::Stable);
|
||||
|
||||
let filename = "overlay.lance".to_string();
|
||||
let store = dataset.object_store(None).await.unwrap();
|
||||
let path = dataset.data_dir().child(filename.clone());
|
||||
let obj_writer = store.create(&path).await.unwrap();
|
||||
let mut writer = lance_file::versions::create_writer(
|
||||
file_version,
|
||||
obj_writer,
|
||||
overlay_schema,
|
||||
FileWriterOptions::default(),
|
||||
)
|
||||
.unwrap();
|
||||
writer
|
||||
.write_column(0, Arc::new(Int32Array::from(vec![1000, 1001, 1002])) as _)
|
||||
.await
|
||||
.unwrap();
|
||||
let summary = writer.finish().await.unwrap();
|
||||
let overlay_bytes = summary.size_bytes as usize;
|
||||
assert!(overlay_bytes > 0);
|
||||
|
||||
let mut data_file = DataFile::new_unstarted(filename, file_version);
|
||||
data_file.fields = writer
|
||||
.field_id_to_column_indices()
|
||||
.iter()
|
||||
.map(|(field_id, _)| *field_id as i32)
|
||||
.collect::<Vec<_>>()
|
||||
.into();
|
||||
data_file.column_indices = writer
|
||||
.field_id_to_column_indices()
|
||||
.iter()
|
||||
.map(|(_, column_index)| *column_index as i32)
|
||||
.collect::<Vec<_>>()
|
||||
.into();
|
||||
data_file.file_size_bytes = CachedFileSize::new(summary.size_bytes);
|
||||
|
||||
let overlay = DataOverlayFile {
|
||||
data_file,
|
||||
coverage: OverlayCoverage::dense(RoaringBitmap::from_iter(0..3)),
|
||||
committed_version: 0,
|
||||
};
|
||||
Dataset::commit(
|
||||
WriteDestination::Dataset(Arc::new(dataset)),
|
||||
Operation::DataOverlay {
|
||||
groups: vec![DataOverlayGroup {
|
||||
fragment_id,
|
||||
overlays: vec![overlay],
|
||||
}],
|
||||
},
|
||||
Some(read_version),
|
||||
None,
|
||||
None,
|
||||
Arc::new(Default::default()),
|
||||
false,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
table.checkout_latest().await.unwrap();
|
||||
let with_overlay = table.stats().await.unwrap().total_bytes;
|
||||
assert_eq!(with_overlay, with_index + overlay_bytes);
|
||||
}
|
||||
|
||||
/// `stats()` must stay manifest-only. Summing per-field `bytes_on_disk`
|
||||
/// instead opens every data file, so cost would grow with fragment count.
|
||||
#[tokio::test]
|
||||
pub async fn test_stats_does_not_read_data_files() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let uri = tmp_dir.path().to_str().unwrap();
|
||||
|
||||
let conn = ConnectBuilder::new(uri).execute().await.unwrap();
|
||||
|
||||
let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
|
||||
let batch = RecordBatch::try_new(
|
||||
schema.clone(),
|
||||
vec![Arc::new(Int32Array::from_iter_values(0..10))],
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
conn.create_table("test_stats_io", batch.clone())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let table = conn.open_table("test_stats_io").execute().await.unwrap();
|
||||
const NUM_APPENDS: usize = 20;
|
||||
for _ in 0..NUM_APPENDS {
|
||||
table.add(batch.clone()).execute().await.unwrap();
|
||||
}
|
||||
|
||||
// Reopen through a tracking store so the counters cover `stats()` alone and
|
||||
// not the writes above.
|
||||
let (wrapper, io_stats) = IoTrackingStore::new_wrapper();
|
||||
let table = conn
|
||||
.open_table("test_stats_io")
|
||||
.lance_read_params(ReadParams {
|
||||
store_options: Some(ObjectStoreParams {
|
||||
object_store_wrapper: Some(wrapper),
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
})
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
io_stats.lock().unwrap().read_iops = 0;
|
||||
|
||||
let stats = table.stats().await.unwrap();
|
||||
let read_iops = io_stats.lock().unwrap().read_iops;
|
||||
|
||||
assert_eq!(stats.fragment_stats.num_fragments, NUM_APPENDS + 1);
|
||||
assert!(stats.total_bytes > 0);
|
||||
// Reading the fragments' data files would take at least one IOP each.
|
||||
assert!(
|
||||
read_iops < stats.fragment_stats.num_fragments as u64,
|
||||
"stats() issued {} read IOPs across {} fragments",
|
||||
read_iops,
|
||||
stats.fragment_stats.num_fragments
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,315 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
//! Converging a table's LSM write path into its base table.
|
||||
//!
|
||||
//! `checkpoint_lsm` seals once, then triggers compaction and watches
|
||||
//! generation numbers until the L0 that existed at the start is gone.
|
||||
//!
|
||||
//! The loop runs in the client, not the server: `compact_lsm` dispatches a
|
||||
//! pass and returns, so nothing holds a socket and a client can vanish
|
||||
//! mid-operation with nothing to reconcile. Completion is read from
|
||||
//! generation numbers in the shard manifest — durable state, unlike a count
|
||||
//! in a compact response, which a concurrent write invalidates.
|
||||
//!
|
||||
//! The target set is fixed at the start, so generations created *during* the
|
||||
//! checkpoint are ignored. That is what lets it terminate under write load,
|
||||
//! and what makes it best-effort: it converges the fresh tier as of some
|
||||
//! instant. Idempotent, abandonable at any point, safe on a cadence.
|
||||
//!
|
||||
//! No liveness bound — the caller owns the deadline. The compactor pool is
|
||||
//! shared pod-wide, so a checkpoint queued behind unrelated tables looks
|
||||
//! exactly like one that is merging.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::future::Future;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::{Error, Result, Table};
|
||||
|
||||
/// The HTTP status a failed request carried, if it carried one.
|
||||
///
|
||||
/// `None` for anything with no retry story: a `TableNotFound` that
|
||||
/// `check_table_response` already translated, or a connection failure that
|
||||
/// never reached the server. Both are terminal.
|
||||
fn status_of(e: &Error) -> Option<u16> {
|
||||
#[cfg(feature = "remote")]
|
||||
{
|
||||
match e {
|
||||
Error::Http {
|
||||
status_code: Some(status),
|
||||
..
|
||||
} => Some(status.as_u16()),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
#[cfg(not(feature = "remote"))]
|
||||
{
|
||||
let _ = e;
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// 429 (latch held, pool saturated, or the pod replaying its WAL) and 503 (a
|
||||
/// draining node, or a proxy between here and it).
|
||||
///
|
||||
/// The status is the whole signal: the server deliberately keeps contention
|
||||
/// off 503, so a latch collision is a 429. A draining node *is* terminal, but
|
||||
/// it is also a 503 that stays a 503, so retrying spends one budget and then
|
||||
/// reports the server's own message — cheaper than parsing the body for the
|
||||
/// namespace code it would take to tell the two apart.
|
||||
fn is_retryable(e: &Error) -> bool {
|
||||
matches!(status_of(e), Some(429 | 503))
|
||||
}
|
||||
|
||||
/// 421: the owning node holds no claim. Only `flush` re-claims and replays,
|
||||
/// so this cannot be retried in place — the caller has to start over.
|
||||
fn is_lost_claim(e: &Error) -> bool {
|
||||
status_of(e) == Some(421)
|
||||
}
|
||||
|
||||
/// Interval between `get_lsm_stats` polls. One interval is roughly one
|
||||
/// compaction pass, the granularity at which the answer can change.
|
||||
///
|
||||
/// Fixed rather than configurable, matching `wait_for_index`. It costs
|
||||
/// nothing on an already-converged table and at most one interval of tail
|
||||
/// latency after the final pass lands.
|
||||
const POLL_INTERVAL: Duration = Duration::from_secs(5);
|
||||
|
||||
/// Cap on re-issues from `flush` after a 421, so a crash-looping node cannot
|
||||
/// turn flush → compact → 421 → flush into a spin.
|
||||
///
|
||||
/// Deliberately not shared with [`MAX_RETRIES`]: a claim that keeps
|
||||
/// evaporating is a broken node, while contention is routine and wants a real
|
||||
/// budget. One shared counter let a merely contended table exhaust this cap
|
||||
/// and then blame a claim it never lost.
|
||||
const MAX_REISSUES: usize = 3;
|
||||
|
||||
/// Retryable faults tolerated on a *single* request, reset on every success —
|
||||
/// scattered contention across a long checkpoint must not accumulate toward a
|
||||
/// cap. Roughly 16s of retrying against the backoff below.
|
||||
const MAX_RETRIES: usize = 8;
|
||||
|
||||
/// Backoff between retries, doubling up to [`RETRY_BACKOFF_MAX`]. Latch
|
||||
/// contention clears in about the time one pass takes, so start small; a
|
||||
/// saturated pool wants the ceiling.
|
||||
const RETRY_BACKOFF_BASE: Duration = Duration::from_millis(100);
|
||||
const RETRY_BACKOFF_MAX: Duration = Duration::from_secs(5);
|
||||
|
||||
/// Sleep before re-issuing a retryable request.
|
||||
async fn backoff(attempt: usize) {
|
||||
let delay = RETRY_BACKOFF_BASE
|
||||
.saturating_mul(1u32 << attempt.min(8) as u32)
|
||||
.min(RETRY_BACKOFF_MAX);
|
||||
tokio::time::sleep(delay).await;
|
||||
}
|
||||
|
||||
/// Whether the drain loop finished or needs the table re-claimed first.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum CheckpointOutcome {
|
||||
Done,
|
||||
ReissueFromFlush,
|
||||
}
|
||||
|
||||
/// What one LSM request produced: its value, or word that the owning node
|
||||
/// holds no claim and only `flush` can get it back.
|
||||
enum Attempt<T> {
|
||||
Ok(T),
|
||||
ReissueFromFlush,
|
||||
}
|
||||
|
||||
/// Issue one LSM request, retrying in place while the fault is retryable.
|
||||
///
|
||||
/// The two recoverable faults have separate budgets: contention clears on its
|
||||
/// own and retries here against [`MAX_RETRIES`], while a 421 needs `flush` to
|
||||
/// re-claim, which only the caller can drive.
|
||||
///
|
||||
/// An exhausted budget propagates the last error *as itself* rather than a
|
||||
/// synthesized one — "429 after nine tries" beats "checkpoint failed", and a
|
||||
/// draining node arrives carrying the server's own message.
|
||||
async fn issue<T, F, Fut>(mut call: F) -> Result<Attempt<T>>
|
||||
where
|
||||
F: FnMut() -> Fut,
|
||||
Fut: Future<Output = Result<T>>,
|
||||
{
|
||||
let mut retries = 0;
|
||||
loop {
|
||||
let e = match call().await {
|
||||
Ok(value) => return Ok(Attempt::Ok(value)),
|
||||
Err(e) => e,
|
||||
};
|
||||
if is_lost_claim(&e) {
|
||||
return Ok(Attempt::ReissueFromFlush);
|
||||
}
|
||||
if !is_retryable(&e) || retries >= MAX_RETRIES {
|
||||
return Err(e);
|
||||
}
|
||||
backoff(retries).await;
|
||||
retries += 1;
|
||||
}
|
||||
}
|
||||
|
||||
/// Drive [`Table::checkpoint_lsm`]: seal once, fix the target watermark
|
||||
/// from the resulting L0, then trigger and poll until it drains.
|
||||
pub(crate) async fn checkpoint_lsm(table: &Table) -> Result<()> {
|
||||
for reissue in 0..=MAX_REISSUES {
|
||||
// The seal turns everything written before this call into a
|
||||
// generation, so the watermark has to be read after it. Idempotent:
|
||||
// sealing an empty memtable is a no-op, so a re-issue does not churn
|
||||
// empty generations.
|
||||
match issue(|| table.flush_lsm()).await? {
|
||||
Attempt::Ok(()) => {}
|
||||
Attempt::ReissueFromFlush => {
|
||||
backoff(reissue).await;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
let stats = match issue(|| table.get_lsm_stats(false)).await? {
|
||||
Attempt::Ok(stats) => stats,
|
||||
Attempt::ReissueFromFlush => {
|
||||
backoff(reissue).await;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let Some(stats) = stats else {
|
||||
// Not WAL-backed; `flush_lsm` would have errored first but for a race.
|
||||
return Ok(());
|
||||
};
|
||||
let targets: HashMap<String, u64> = stats
|
||||
.buckets
|
||||
.iter()
|
||||
.filter_map(|b| Some((b.shard_id.clone(), b.newest_generation()?)))
|
||||
.collect();
|
||||
if targets.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
match drain_to_targets(table, &targets).await? {
|
||||
CheckpointOutcome::Done => return Ok(()),
|
||||
CheckpointOutcome::ReissueFromFlush => {
|
||||
backoff(reissue).await;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(Error::Runtime {
|
||||
message: "checkpoint_lsm: the owning node kept losing its claim; \
|
||||
re-issued from flush the maximum number of times"
|
||||
.into(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Trigger and poll until no bucket holds a generation at or below its
|
||||
/// target.
|
||||
///
|
||||
/// No liveness bound, deliberately. The pod-wide compactor pool (a semaphore
|
||||
/// of 2 by default, shared across every table on the node) is taken *inside*
|
||||
/// the pass, after the bucket latch, so a checkpoint queued behind unrelated
|
||||
/// tables is indistinguishable from one that is merging. An idle-poll counter
|
||||
/// here could only ever have fired on a table that would have finished.
|
||||
async fn drain_to_targets(
|
||||
table: &Table,
|
||||
targets: &HashMap<String, u64>,
|
||||
) -> Result<CheckpointOutcome> {
|
||||
loop {
|
||||
let stats = match issue(|| table.get_lsm_stats(false)).await? {
|
||||
Attempt::Ok(stats) => stats,
|
||||
Attempt::ReissueFromFlush => return Ok(CheckpointOutcome::ReissueFromFlush),
|
||||
};
|
||||
let Some(stats) = stats else {
|
||||
return Ok(CheckpointOutcome::Done);
|
||||
};
|
||||
// `compacting` is the bucket's compaction latch, held from dispatch
|
||||
// until the pass ends — including while it waits on the pod-wide
|
||||
// permit. So it answers one question only: do not pile on. Buckets
|
||||
// with nothing outstanding are skipped, not counted as idle.
|
||||
let mut outstanding = 0;
|
||||
let mut all_compacting = true;
|
||||
for b in &stats.buckets {
|
||||
let Some(target) = targets.get(&b.shard_id) else {
|
||||
continue;
|
||||
};
|
||||
let n = b.outstanding_generations(*target);
|
||||
if n > 0 {
|
||||
outstanding += n;
|
||||
all_compacting &= b.compacting;
|
||||
}
|
||||
}
|
||||
if outstanding == 0 {
|
||||
return Ok(CheckpointOutcome::Done);
|
||||
}
|
||||
|
||||
if !all_compacting {
|
||||
match table.compact_lsm().await {
|
||||
Ok(()) => {}
|
||||
Err(e) if is_lost_claim(&e) => return Ok(CheckpointOutcome::ReissueFromFlush),
|
||||
Err(e) if !is_retryable(&e) => return Err(e),
|
||||
// A 429 here means the server could latch no bucket at all,
|
||||
// which the poll above already handles. Not retried in place:
|
||||
// the latch it would contend for is the one doing the work, so
|
||||
// fall through and re-read — `POLL_INTERVAL` is the backoff.
|
||||
Err(_) => {}
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(POLL_INTERVAL).await;
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(test, feature = "remote"))]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn http(status: u16) -> Error {
|
||||
Error::Http {
|
||||
source: "server said no".into(),
|
||||
request_id: "rid".into(),
|
||||
status_code: reqwest::StatusCode::from_u16(status).ok(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Every status the loop acts on. The two predicates are checked together
|
||||
/// because their overlap is what would be wrong: a status must never be
|
||||
/// both, and 421 in particular must not read as retryable — retrying it in
|
||||
/// place re-issues the call that just said the node holds no claim.
|
||||
#[test]
|
||||
fn taxonomy_round_trips() {
|
||||
for status in [429, 503] {
|
||||
assert!(is_retryable(&http(status)), "{status} must retry");
|
||||
assert!(
|
||||
!is_lost_claim(&http(status)),
|
||||
"{status} is not a lost claim"
|
||||
);
|
||||
}
|
||||
assert!(is_lost_claim(&http(421)), "a lost claim must re-claim");
|
||||
assert!(
|
||||
!is_retryable(&http(421)),
|
||||
"retrying a lost claim in place only asks the same node again"
|
||||
);
|
||||
for status in [400, 404, 409, 500] {
|
||||
assert!(!is_retryable(&http(status)), "{status} is terminal");
|
||||
assert!(!is_lost_claim(&http(status)), "{status} is terminal");
|
||||
}
|
||||
}
|
||||
|
||||
/// An error carrying no status has no retry story and must be terminal —
|
||||
/// a connection that never reached the server, or a `TableNotFound` that
|
||||
/// `check_table_response` translated before the loop saw it.
|
||||
#[test]
|
||||
fn errors_without_a_status_are_terminal() {
|
||||
let no_status = Error::Http {
|
||||
source: "connection reset".into(),
|
||||
request_id: "rid".into(),
|
||||
status_code: None,
|
||||
};
|
||||
assert!(!is_retryable(&no_status));
|
||||
assert!(!is_lost_claim(&no_status));
|
||||
|
||||
let translated = Error::TableNotFound {
|
||||
name: "t".into(),
|
||||
source: "gone".into(),
|
||||
};
|
||||
assert!(!is_retryable(&translated));
|
||||
assert!(!is_lost_claim(&translated));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
//! Live per-bucket LSM state — the shape [`crate::Table::get_lsm_stats`]
|
||||
//! returns and [`super::checkpoint`] polls.
|
||||
//!
|
||||
//! Nothing here is derived: sums and differences (total L0 bytes, WAL lag)
|
||||
//! are the caller's to compute. There is no "WAL is off" shape — that case is
|
||||
//! `None`, because a struct of zeros would read as measurements.
|
||||
|
||||
use serde::Deserialize;
|
||||
|
||||
/// One flushed L0 generation.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct GenerationStats {
|
||||
pub generation: u64,
|
||||
pub bytes: u64,
|
||||
/// Present only when `include_generation_rows` was requested. Off by
|
||||
/// default because each count opens an uncached Lance dataset, and the
|
||||
/// checkpoint loop polls this route needing only generation numbers.
|
||||
#[serde(default)]
|
||||
pub rows: Option<u64>,
|
||||
}
|
||||
|
||||
/// One in-memory memtable.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct MemtableStats {
|
||||
pub generation: u64,
|
||||
pub rows: u64,
|
||||
pub bytes: u64,
|
||||
pub batches: u64,
|
||||
/// Names of the indexes this memtable carries. An absent name is the whole
|
||||
/// answer to "why is my fresh-tier search on that column brute-force".
|
||||
pub indexes: Vec<String>,
|
||||
}
|
||||
|
||||
/// Live state of one bucket. A table is N buckets on one node; flattening to
|
||||
/// a single number hides the one hot bucket that is usually why someone
|
||||
/// opened this endpoint.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct BucketStats {
|
||||
pub shard_id: String,
|
||||
/// `Active` | `Sealed` (drop-table 2PC in flight).
|
||||
pub status: String,
|
||||
pub writer_epoch: u64,
|
||||
pub manifest_version: u64,
|
||||
pub current_generation: u64,
|
||||
pub replay_after_wal_entry_position: u64,
|
||||
pub wal_entry_position_last_seen: u64,
|
||||
pub generations: Vec<GenerationStats>,
|
||||
/// Whether a pass owns this bucket's compaction latch right now. Says *a*
|
||||
/// driver is running, not *whose*, and the latch is held from dispatch —
|
||||
/// including while the pass queues for a pod-wide compactor permit. Read
|
||||
/// it as "do not pile on", never as "mine is progressing".
|
||||
pub compacting: bool,
|
||||
/// Oldest first, active last. Absent for a `Sealed` bucket, whose
|
||||
/// in-memory state is torn down.
|
||||
#[serde(default)]
|
||||
pub memtables: Option<Vec<MemtableStats>>,
|
||||
}
|
||||
|
||||
impl BucketStats {
|
||||
/// The newest flushed generation, or `None` when L0 is empty.
|
||||
pub(crate) fn newest_generation(&self) -> Option<u64> {
|
||||
self.generations.iter().map(|g| g.generation).max()
|
||||
}
|
||||
|
||||
/// How many generations at or below `target` are still in L0.
|
||||
///
|
||||
/// A count, not a boolean: one pass drains a bounded prefix rather than
|
||||
/// the whole target set, so a boolean would read as "no progress" for
|
||||
/// every pass but the last. Compaction drains oldest-first, so this
|
||||
/// decreases monotonically.
|
||||
pub(crate) fn outstanding_generations(&self, target: u64) -> usize {
|
||||
self.generations
|
||||
.iter()
|
||||
.filter(|g| g.generation <= target)
|
||||
.count()
|
||||
}
|
||||
}
|
||||
|
||||
/// Live LSM state, one entry per bucket.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct LsmStats {
|
||||
pub buckets: Vec<BucketStats>,
|
||||
}
|
||||
|
||||
/// Server-side JSON envelope for `get_lsm_stats`. `lsm_stats` is null when
|
||||
/// the table has no LSM write path.
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct GetLsmStatsResponse {
|
||||
#[serde(default)]
|
||||
pub lsm_stats: Option<LsmStats>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn bucket(shard: &str, generations: &[u64], compacting: bool) -> BucketStats {
|
||||
BucketStats {
|
||||
shard_id: shard.into(),
|
||||
status: "Active".into(),
|
||||
writer_epoch: 1,
|
||||
manifest_version: 1,
|
||||
current_generation: generations.iter().max().copied().unwrap_or(0) + 1,
|
||||
replay_after_wal_entry_position: 0,
|
||||
wal_entry_position_last_seen: 0,
|
||||
generations: generations
|
||||
.iter()
|
||||
.map(|g| GenerationStats {
|
||||
generation: *g,
|
||||
bytes: 1,
|
||||
rows: None,
|
||||
})
|
||||
.collect(),
|
||||
compacting,
|
||||
memtables: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// The target watermark is the newest generation at the start, and a
|
||||
/// generation created after it must not hold the loop open — that is why
|
||||
/// the predicate terminates under write load.
|
||||
#[test]
|
||||
fn newer_generations_do_not_extend_the_target() {
|
||||
let start = bucket("b0", &[7, 8], false);
|
||||
let target = start.newest_generation().expect("L0 is non-empty");
|
||||
assert_eq!(target, 8);
|
||||
|
||||
// Compaction drained 7 and 8; 9 and 10 arrived while it ran.
|
||||
let later = bucket("b0", &[9, 10], false);
|
||||
assert_eq!(
|
||||
later.outstanding_generations(target),
|
||||
0,
|
||||
"generations above the target are somebody else's problem"
|
||||
);
|
||||
|
||||
// Still holding 8 means still outstanding.
|
||||
assert_eq!(
|
||||
bucket("b0", &[8, 9], false).outstanding_generations(target),
|
||||
1
|
||||
);
|
||||
}
|
||||
|
||||
/// The metric counts generations, not buckets: a pass drains a bounded
|
||||
/// prefix, so one bucket going 3 → 2 → 1 → 0 is three steps.
|
||||
#[test]
|
||||
fn progress_is_measured_in_generations() {
|
||||
let target = 3;
|
||||
let counts: Vec<usize> = [&[1u64, 2, 3][..], &[2, 3][..], &[3][..], &[][..]]
|
||||
.iter()
|
||||
.map(|gens| bucket("b0", gens, false).outstanding_generations(target))
|
||||
.collect();
|
||||
assert_eq!(counts, vec![3, 2, 1, 0]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_l0_has_no_target() {
|
||||
assert!(bucket("b0", &[], false).newest_generation().is_none());
|
||||
}
|
||||
}
|
||||
@@ -1161,7 +1161,7 @@ mod lsm_tests {
|
||||
.unwrap();
|
||||
let fts_index = table.list_indices().await.unwrap()[0].name.clone();
|
||||
table
|
||||
.set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes([fts_index]))
|
||||
.set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(vec![fts_index]))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
@@ -1254,7 +1254,7 @@ mod lsm_tests {
|
||||
.unwrap();
|
||||
let vec_index = table.list_indices().await.unwrap()[0].name.clone();
|
||||
table
|
||||
.set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes([vec_index]))
|
||||
.set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(vec![vec_index]))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
||||
@@ -29,6 +29,7 @@ use arrow_schema::{DataType, Schema as ArrowSchema, SchemaRef};
|
||||
use lance::Dataset;
|
||||
use lance::dataset::mem_wal::{
|
||||
DatasetMemWalExt, ShardWriter, ShardWriterConfig, evaluate_sharding_spec,
|
||||
validate_maintained_indexes,
|
||||
};
|
||||
use lance::index::DatasetIndexExt;
|
||||
use lance_core::datatypes::Schema as LanceSchema;
|
||||
@@ -37,8 +38,9 @@ use tokio::sync::RwLock;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
use crate::index::IndexConfig;
|
||||
use crate::table::merge::{MergeInsertBuilder, MergeResult};
|
||||
use crate::table::{LsmWriteSpec, NativeTable};
|
||||
use crate::table::{BaseTable, LsmWriteSpec, NativeTable};
|
||||
|
||||
/// Spec id of the sole sharding spec installed by [`set_lsm_write_spec`].
|
||||
/// Must match Lance's `InitializeMemWalBuilder` (`SHARDING_SPEC_ID`).
|
||||
@@ -80,32 +82,44 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec)
|
||||
}
|
||||
}
|
||||
|
||||
// Before the builder borrows the dataset clone. `list_indices` merges an
|
||||
// index's segments into one entry, so the result needs no dedup.
|
||||
let maintained_indexes = {
|
||||
let dataset = table.dataset.get().await?;
|
||||
resolve_maintained_indexes(
|
||||
&dataset,
|
||||
&table.list_indices().await?,
|
||||
spec.maintained_indexes(),
|
||||
)
|
||||
.await?
|
||||
};
|
||||
|
||||
let mut dataset = (*table.dataset.get().await?).clone();
|
||||
let mut builder = dataset.initialize_mem_wal();
|
||||
let (maintained_indexes, writer_config_defaults) = match spec {
|
||||
let writer_config_defaults = match spec {
|
||||
LsmWriteSpec::Bucket {
|
||||
column,
|
||||
num_buckets,
|
||||
maintained_indexes,
|
||||
writer_config_defaults,
|
||||
..
|
||||
} => {
|
||||
builder = builder.bucket_sharding(column, num_buckets);
|
||||
(maintained_indexes, writer_config_defaults)
|
||||
writer_config_defaults
|
||||
}
|
||||
LsmWriteSpec::Identity {
|
||||
column,
|
||||
maintained_indexes,
|
||||
writer_config_defaults,
|
||||
..
|
||||
} => {
|
||||
builder = builder.identity_sharding(column);
|
||||
(maintained_indexes, writer_config_defaults)
|
||||
writer_config_defaults
|
||||
}
|
||||
LsmWriteSpec::Unsharded {
|
||||
maintained_indexes,
|
||||
writer_config_defaults,
|
||||
..
|
||||
} => {
|
||||
builder = builder.unsharded();
|
||||
(maintained_indexes, writer_config_defaults)
|
||||
writer_config_defaults
|
||||
}
|
||||
};
|
||||
builder = builder.maintained_indexes(maintained_indexes);
|
||||
@@ -117,6 +131,58 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec)
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Resolve a spec's maintained-index selection against `indices`, as reported
|
||||
/// by [`Table::list_indices`](crate::Table::list_indices).
|
||||
///
|
||||
/// `None` means every index on the table, snapshotted now. Lance validates
|
||||
/// either selection against its shard-writer rules, so a spec that installs is
|
||||
/// one the MemWAL can open.
|
||||
///
|
||||
/// An unmaintainable index fails an inferred set rather than being dropped from
|
||||
/// it — dropping would leave the caller believing it is maintained.
|
||||
async fn resolve_maintained_indexes(
|
||||
dataset: &Dataset,
|
||||
indices: &[IndexConfig],
|
||||
requested: Option<&[String]>,
|
||||
) -> Result<Vec<String>> {
|
||||
let Some(requested) = requested else {
|
||||
let all: Vec<String> = indices.iter().map(|index| index.name.clone()).collect();
|
||||
validate_maintained_indexes(dataset, &all)
|
||||
.await
|
||||
.map_err(|source| Error::InvalidInput {
|
||||
message: format!(
|
||||
"cannot maintain every index on this table: {source}. Set \
|
||||
maintained_indexes explicitly to choose from {}",
|
||||
index_name_list(indices),
|
||||
),
|
||||
})?;
|
||||
return Ok(all);
|
||||
};
|
||||
for name in requested {
|
||||
if !indices.iter().any(|index| &index.name == name) {
|
||||
return Err(Error::InvalidInput {
|
||||
message: format!(
|
||||
"maintained index '{}' does not exist on this table; it has {}",
|
||||
name,
|
||||
index_name_list(indices),
|
||||
),
|
||||
});
|
||||
}
|
||||
}
|
||||
validate_maintained_indexes(dataset, requested).await?;
|
||||
Ok(requested.to_vec())
|
||||
}
|
||||
|
||||
/// Index names for an error message.
|
||||
fn index_name_list(indices: &[IndexConfig]) -> String {
|
||||
if indices.is_empty() {
|
||||
return "no indexes".to_string();
|
||||
}
|
||||
let mut names: Vec<&str> = indices.iter().map(|index| index.name.as_str()).collect();
|
||||
names.sort_unstable();
|
||||
format!("[{}]", names.join(", "))
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// unset_lsm_write_spec
|
||||
// =============================================================================
|
||||
|
||||
Reference in New Issue
Block a user