mirror of
https://github.com/lancedb/lancedb.git
synced 2026-08-29 17:38:31 +00:00
Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Gatefixer
|
1aa1c969d6 | ||
 Yang Cen
|
1d75638dea | ||
 LanceDB Robot
|
031c3585a8 | ||
|
Gatefixer
|
730334fe32 | ||
|
Gatefixer
|
6cb527dc80 | ||
|
Gatefixer
|
e619eb0942 | ||
|
Gatefixer
|
91ec4a695f | ||
|
Gatefixer
|
7829ead241 |
@@ -69,6 +69,16 @@ jobs:
|
||||
uses: actions/setup-python@v6
|
||||
with:
|
||||
python-version: "3.10"
|
||||
- name: Add swap for Arm fat LTO
|
||||
if: matrix.config.platform == 'aarch64'
|
||||
shell: bash
|
||||
run: |
|
||||
swap_file="$RUNNER_TEMP/lancedb-swap"
|
||||
sudo fallocate --length 16G "$swap_file"
|
||||
sudo chmod 600 "$swap_file"
|
||||
sudo mkswap "$swap_file"
|
||||
sudo swapon "$swap_file"
|
||||
free -h
|
||||
- uses: ./.github/workflows/build_linux_wheel
|
||||
with:
|
||||
python-minor-version: 10
|
||||
|
||||
Generated
+43
-42
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
|
||||
|
||||
[[package]]
|
||||
name = "fsst"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"rand 0.9.5",
|
||||
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
|
||||
|
||||
[[package]]
|
||||
name = "lance"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"arrow",
|
||||
@@ -4890,8 +4890,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-arrow"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
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.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
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.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -4936,8 +4936,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-bitpacking"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrayref",
|
||||
"crunchy",
|
||||
@@ -4947,8 +4947,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-core"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -4988,8 +4988,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-datafusion"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5019,8 +5019,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-datagen"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5037,8 +5037,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-derive"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -5047,8 +5047,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-encoding"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow-arith",
|
||||
"arrow-array",
|
||||
@@ -5082,8 +5082,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-file"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow-arith",
|
||||
"arrow-array",
|
||||
@@ -5114,8 +5114,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-index"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"arrow",
|
||||
@@ -5182,8 +5182,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-index-core"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -5205,8 +5205,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-io"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5242,8 +5242,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-linalg"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -5259,8 +5259,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"async-trait",
|
||||
@@ -5272,8 +5272,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-namespace-impls"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-ipc",
|
||||
@@ -5326,8 +5326,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-select"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
@@ -5342,8 +5342,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-table"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-array",
|
||||
@@ -5383,8 +5383,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-testing"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-schema",
|
||||
@@ -5397,8 +5397,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lance-tokenizer"
|
||||
version = "11.0.0-beta.6"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.6#5ab688688c411111d94d279bacd037d7c319dc10"
|
||||
version = "11.0.0-beta.7"
|
||||
source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
|
||||
dependencies = [
|
||||
"frostem",
|
||||
"icu_segmenter",
|
||||
@@ -5495,6 +5495,7 @@ dependencies = [
|
||||
"urlencoding",
|
||||
"uuid",
|
||||
"walkdir",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
+14
-14
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
|
||||
rust-version = "1.91.0"
|
||||
|
||||
[workspace.dependencies]
|
||||
lance = { "version" = "=11.0.0-beta.6", default-features = false, "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=11.0.0-beta.6", default-features = false, "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=11.0.0-beta.6", default-features = false, "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=11.0.0-beta.6", "tag" = "v11.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance = { "version" = "=11.0.0-beta.7", default-features = false, "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=11.0.0-beta.7", default-features = false, "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=11.0.0-beta.7", default-features = false, "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "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 }
|
||||
|
||||
+1
-1
@@ -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.6</lance-core.version>
|
||||
<lance-core.version>11.0.0-beta.7</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>
|
||||
|
||||
@@ -115,6 +115,11 @@ serial_test = "3"
|
||||
[target.'cfg(unix)'.dev-dependencies]
|
||||
pprof = { version = "0.14", features = ["flamegraph"] }
|
||||
|
||||
[target.'cfg(windows)'.dependencies]
|
||||
windows-sys = { version = "0.61", features = [
|
||||
"Win32_Foundation",
|
||||
"Win32_Storage_FileSystem",
|
||||
] }
|
||||
|
||||
[features]
|
||||
default = []
|
||||
@@ -188,6 +193,9 @@ required-features = ["bedrock"]
|
||||
[[example]]
|
||||
name = "bench_streaming_dataloader"
|
||||
|
||||
[[example]]
|
||||
name = "bench_open_missing_table"
|
||||
|
||||
[[example]]
|
||||
name = "simple"
|
||||
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
// Release benchmark for opening a missing table as sibling-table cardinality grows.
|
||||
//
|
||||
// The fixture uses real `.lance` directories and marker files. Fixture creation is
|
||||
// outside the timed section. Defaults intentionally cover 1k, 10k, and 100k siblings
|
||||
// with 10 warmups and 100 distinct missing-table opens per scale:
|
||||
//
|
||||
// ```text
|
||||
// cargo run --release -p lancedb --example bench_open_missing_table
|
||||
// ```
|
||||
//
|
||||
// `BENCH_SIBLINGS`, `BENCH_WARMUPS`, and `BENCH_TRIALS` override those defaults.
|
||||
// Reduced settings are useful only as a smoke test. Performance comparisons require
|
||||
// the same machine, filesystem, fixture sizes, settings, lockfile, and alternating
|
||||
// baseline/candidate execution order.
|
||||
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use anyhow::{Context, Result, bail};
|
||||
use lancedb::connection::Connection;
|
||||
use lancedb::{Error, connect};
|
||||
use object_store::ObjectStoreExt as _;
|
||||
use object_store::path::Path;
|
||||
|
||||
const MAX_SIBLINGS: usize = 1_000_000;
|
||||
const MAX_WARMUPS: usize = 10_000;
|
||||
const MAX_TRIALS: usize = 100_000;
|
||||
|
||||
fn env_usize(key: &str, default: usize, max: usize) -> Result<usize> {
|
||||
let value = match std::env::var(key) {
|
||||
Ok(value) => value
|
||||
.parse()
|
||||
.with_context(|| format!("invalid {key} value: {value}"))?,
|
||||
Err(std::env::VarError::NotPresent) => default,
|
||||
Err(error) => return Err(error).with_context(|| format!("reading {key}")),
|
||||
};
|
||||
if value == 0 || value > max {
|
||||
bail!("{key} must be between 1 and {max}");
|
||||
}
|
||||
Ok(value)
|
||||
}
|
||||
|
||||
fn sibling_counts() -> Result<Vec<usize>> {
|
||||
let raw = std::env::var("BENCH_SIBLINGS").unwrap_or_else(|_| "1000,10000,100000".into());
|
||||
let mut counts = raw
|
||||
.split(',')
|
||||
.map(|value| {
|
||||
value
|
||||
.trim()
|
||||
.parse::<usize>()
|
||||
.with_context(|| format!("invalid BENCH_SIBLINGS value: {value}"))
|
||||
})
|
||||
.collect::<Result<Vec<_>>>()?;
|
||||
counts.sort_unstable();
|
||||
counts.dedup();
|
||||
if counts.is_empty() || counts[0] == 0 || counts[counts.len() - 1] > MAX_SIBLINGS {
|
||||
bail!("BENCH_SIBLINGS values must be between 1 and {MAX_SIBLINGS}");
|
||||
}
|
||||
Ok(counts)
|
||||
}
|
||||
|
||||
async fn add_siblings(
|
||||
store: &object_store::local::LocalFileSystem,
|
||||
start: usize,
|
||||
end: usize,
|
||||
) -> Result<()> {
|
||||
for index in start..end {
|
||||
let marker = Path::from(format!("sibling_{index:06}.lance/_marker"));
|
||||
store
|
||||
.put(&marker, bytes::Bytes::new().into())
|
||||
.await
|
||||
.with_context(|| format!("creating benchmark marker {marker}"))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn time_missing_open(db: &Connection, name: &str) -> Result<Duration> {
|
||||
let started = Instant::now();
|
||||
let result = db.open_table(name).execute().await;
|
||||
let elapsed = started.elapsed();
|
||||
match result {
|
||||
Err(Error::TableNotFound { .. }) => Ok(elapsed),
|
||||
Err(error) => bail!("expected TableNotFound for {name}, got {error:?}"),
|
||||
Ok(_) => bail!("benchmark missing-table name unexpectedly exists: {name}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn percentile(sorted: &[Duration], percentile: usize) -> Duration {
|
||||
let rank = (sorted.len() * percentile).div_ceil(100).saturating_sub(1);
|
||||
sorted[rank]
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<()> {
|
||||
let counts = sibling_counts()?;
|
||||
let warmups = env_usize("BENCH_WARMUPS", 10, MAX_WARMUPS)?;
|
||||
let trials = env_usize("BENCH_TRIALS", 100, MAX_TRIALS)?;
|
||||
|
||||
let fixture = tempfile::tempdir().context("creating benchmark fixture")?;
|
||||
let database_path = fixture.path();
|
||||
let fixture_store = object_store::local::LocalFileSystem::new_with_prefix(database_path)
|
||||
.context("creating benchmark object store")?;
|
||||
let db = connect(database_path.to_str().context("non-UTF-8 fixture path")?)
|
||||
.execute()
|
||||
.await?;
|
||||
|
||||
println!(
|
||||
"config: siblings={counts:?} warmups={warmups} trials={trials} profile={} os={} arch={}",
|
||||
if cfg!(debug_assertions) {
|
||||
"debug"
|
||||
} else {
|
||||
"release"
|
||||
},
|
||||
std::env::consts::OS,
|
||||
std::env::consts::ARCH,
|
||||
);
|
||||
println!("lower is better; fixture setup and teardown are excluded");
|
||||
println!("| siblings | samples | p50 | p95 | max |");
|
||||
println!("| ---: | ---: | ---: | ---: | ---: |");
|
||||
|
||||
let mut created = 0;
|
||||
for sibling_count in counts {
|
||||
add_siblings(&fixture_store, created, sibling_count).await?;
|
||||
created = sibling_count;
|
||||
|
||||
for index in 0..warmups {
|
||||
let name = format!("__missing_warmup_{sibling_count}_{index}");
|
||||
let _ = time_missing_open(&db, &name).await?;
|
||||
}
|
||||
|
||||
let mut samples = Vec::with_capacity(trials);
|
||||
for index in 0..trials {
|
||||
let name = format!("__missing_trial_{sibling_count}_{index}");
|
||||
samples.push(time_missing_open(&db, &name).await?);
|
||||
}
|
||||
samples.sort_unstable();
|
||||
|
||||
println!(
|
||||
"| {sibling_count} | {} | {:?} | {:?} | {:?} |",
|
||||
samples.len(),
|
||||
percentile(&samples, 50),
|
||||
percentile(&samples, 95),
|
||||
samples[samples.len() - 1],
|
||||
);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -409,6 +409,11 @@ impl Connection {
|
||||
///
|
||||
/// The names will be returned in lexicographical order (ascending)
|
||||
///
|
||||
/// Listing databases discover physical `*.lance` entries without opening every
|
||||
/// dataset. The result is a point-in-time discovery snapshot: an entry may still be
|
||||
/// under creation, may contain only uncommitted storage, or may be concurrently
|
||||
/// dropped before it is opened.
|
||||
///
|
||||
/// The parameters `page_token` and `limit` can be used to paginate the results
|
||||
pub fn table_names(&self) -> TableNamesBuilder {
|
||||
TableNamesBuilder::new(self.internal.clone())
|
||||
@@ -456,10 +461,9 @@ impl Connection {
|
||||
///
|
||||
/// # Returns
|
||||
/// Created [`TableRef`], or [`Error::TableNotFound`] if the table does not exist.
|
||||
/// If the table's storage is present but holds no readable dataset (for example a
|
||||
/// `<name>.lance` directory left behind by an interrupted drop and re-create, which
|
||||
/// [`Self::table_names`] still lists) this returns [`Error::TableCorrupted`]
|
||||
/// instead.
|
||||
/// On listing databases, a committed Lance manifest is authoritative for table
|
||||
/// existence. Uncommitted files or a physical `<name>.lance` directory alone do not
|
||||
/// make a table openable.
|
||||
pub fn open_table(&self, name: impl Into<String>) -> OpenTableBuilder {
|
||||
OpenTableBuilder::new(
|
||||
self.internal.clone(),
|
||||
|
||||
@@ -25,7 +25,7 @@ use crate::database::namespace::LanceNamespaceDatabase;
|
||||
use crate::error::{CreateDirSnafu, Error, Result};
|
||||
use crate::io::object_store::MirroringObjectStoreWrapper;
|
||||
use crate::table::NativeTable;
|
||||
use crate::utils::validate_table_name;
|
||||
use crate::utils::{PatchStoreParam, validate_table_name};
|
||||
|
||||
use lance_namespace::models::{
|
||||
CreateNamespaceRequest, CreateNamespaceResponse, DescribeNamespaceRequest,
|
||||
@@ -355,6 +355,14 @@ impl ListingDatabase {
|
||||
url.to_string()
|
||||
}
|
||||
|
||||
#[cfg(any(windows, test))]
|
||||
fn uses_local_file_provider(object_store: &ObjectStore) -> bool {
|
||||
matches!(
|
||||
object_store.scheme(),
|
||||
"file" | "file-object-store" | "file+uring"
|
||||
)
|
||||
}
|
||||
|
||||
async fn prepare_namespace_root(
|
||||
uri: &str,
|
||||
storage_options: &HashMap<String, String>,
|
||||
@@ -581,6 +589,17 @@ impl ListingDatabase {
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
#[cfg(windows)]
|
||||
let write_store_wrapper = if Self::uses_local_file_provider(&object_store) {
|
||||
// Local manifest commits need create-only rename semantics,
|
||||
// including on filesystems that do not support hard links.
|
||||
Some(
|
||||
Arc::new(crate::io::object_store::windows::WindowsLocalFileSystemWrapper)
|
||||
as Arc<dyn WrappingObjectStore>,
|
||||
)
|
||||
} else {
|
||||
write_store_wrapper
|
||||
};
|
||||
|
||||
let namespace_database = Self::connect_namespace_database(
|
||||
&storage_base_uri,
|
||||
@@ -645,12 +664,22 @@ impl ListingDatabase {
|
||||
)
|
||||
.await?;
|
||||
|
||||
#[cfg(windows)]
|
||||
let write_store_wrapper = Self::uses_local_file_provider(&object_store).then(|| {
|
||||
// Local manifest commits need create-only rename semantics,
|
||||
// including on filesystems that do not support hard links.
|
||||
Arc::new(crate::io::object_store::windows::WindowsLocalFileSystemWrapper)
|
||||
as Arc<dyn WrappingObjectStore>
|
||||
});
|
||||
#[cfg(not(windows))]
|
||||
let write_store_wrapper = None;
|
||||
|
||||
Ok(Self {
|
||||
uri: path.to_string(),
|
||||
query_string: None,
|
||||
base_path,
|
||||
object_store,
|
||||
store_wrapper: None,
|
||||
store_wrapper: write_store_wrapper,
|
||||
read_consistency_interval,
|
||||
storage_options: HashMap::new(),
|
||||
storage_options_provider: None,
|
||||
@@ -1112,6 +1141,12 @@ impl Database for ListingDatabase {
|
||||
},
|
||||
..Default::default()
|
||||
};
|
||||
let storage_params = match self.store_wrapper.clone() {
|
||||
Some(wrapper) => Some(storage_params)
|
||||
.patch_with_store_wrapper(wrapper)?
|
||||
.expect("patching store params always returns parameters"),
|
||||
None => storage_params,
|
||||
};
|
||||
let read_params = ReadParams {
|
||||
store_options: Some(storage_params.clone()),
|
||||
session: Some(self.session.clone()),
|
||||
@@ -1291,16 +1326,37 @@ impl Database for ListingDatabase {
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::Table;
|
||||
use crate::arrow::{SendableRecordBatchStream, SimpleRecordBatchStream};
|
||||
use crate::connection::ConnectRequest;
|
||||
use crate::data::scannable::Scannable;
|
||||
use crate::database::{CreateTableMode, CreateTableRequest};
|
||||
use crate::io::object_store::io_tracking::IoStatsHolder;
|
||||
use crate::query::QueryRequest;
|
||||
use crate::table::{AnyQuery, WriteOptions};
|
||||
use arrow_array::{Int32Array, RecordBatch, StringArray};
|
||||
use arrow_schema::{DataType, Field, Schema};
|
||||
use futures::TryStreamExt;
|
||||
use arrow_schema::{DataType, Field, Schema, SchemaRef};
|
||||
use futures::{TryStreamExt, stream::once};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::Duration;
|
||||
use tempfile::tempdir;
|
||||
use tokio::sync::Barrier;
|
||||
use tokio::time::timeout;
|
||||
|
||||
#[derive(Debug)]
|
||||
struct PassthroughStoreWrapper(Arc<AtomicUsize>);
|
||||
|
||||
impl WrappingObjectStore for PassthroughStoreWrapper {
|
||||
fn wrap(
|
||||
&self,
|
||||
_store_prefix: &str,
|
||||
target: Arc<dyn object_store::ObjectStore>,
|
||||
) -> Arc<dyn object_store::ObjectStore> {
|
||||
self.0.fetch_add(1, Ordering::Relaxed);
|
||||
target
|
||||
}
|
||||
}
|
||||
|
||||
async fn setup_database() -> (tempfile::TempDir, ListingDatabase) {
|
||||
let tempdir = tempdir().unwrap();
|
||||
@@ -1324,6 +1380,114 @@ mod tests {
|
||||
(tempdir, db)
|
||||
}
|
||||
|
||||
struct BarrierScannable {
|
||||
batch: RecordBatch,
|
||||
barrier: Arc<Barrier>,
|
||||
}
|
||||
|
||||
impl Scannable for BarrierScannable {
|
||||
fn schema(&self) -> SchemaRef {
|
||||
self.batch.schema()
|
||||
}
|
||||
|
||||
fn scan_as_stream(&mut self) -> SendableRecordBatchStream {
|
||||
let batch = self.batch.clone();
|
||||
let schema = batch.schema();
|
||||
let barrier = self.barrier.clone();
|
||||
Box::pin(SimpleRecordBatchStream {
|
||||
schema,
|
||||
stream: once(async move {
|
||||
barrier.wait().await;
|
||||
Ok(batch)
|
||||
}),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn create_request(name: &str, data: Box<dyn Scannable>) -> CreateTableRequest {
|
||||
CreateTableRequest {
|
||||
name: name.to_string(),
|
||||
namespace_path: vec![],
|
||||
data,
|
||||
mode: CreateTableMode::Create,
|
||||
write_options: Default::default(),
|
||||
location: None,
|
||||
namespace_client: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_create_ignores_uncommitted_storage_without_manifest() {
|
||||
let (tmp_dir, db) = setup_database().await;
|
||||
let data_dir = tmp_dir.path().join("test.lance/data");
|
||||
std::fs::create_dir_all(&data_dir).unwrap();
|
||||
std::fs::write(data_dir.join("orphan.lance"), b"uncommitted").unwrap();
|
||||
|
||||
let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
|
||||
let batch =
|
||||
RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1]))]).unwrap();
|
||||
|
||||
let table = db
|
||||
.create_table(create_request("test", Box::new(batch)))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(table.count_rows(None).await.unwrap(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_concurrent_create_is_arbitrated_by_manifest_commit() {
|
||||
let uri = format!("memory:///concurrent-create-{}", uuid::Uuid::new_v4());
|
||||
let db = crate::connect(&uri).execute().await.unwrap();
|
||||
let store: Arc<dyn object_store::ObjectStore> =
|
||||
Arc::new(object_store::memory::InMemory::new());
|
||||
let table_url = url::Url::parse("memory:///database/test.lance").unwrap();
|
||||
let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
|
||||
let batch =
|
||||
RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1]))]).unwrap();
|
||||
let barrier = Arc::new(Barrier::new(2));
|
||||
|
||||
#[allow(deprecated)]
|
||||
let request = |batch, barrier| {
|
||||
let mut request = create_request("test", Box::new(BarrierScannable { batch, barrier }));
|
||||
request.write_options = WriteOptions {
|
||||
lance_write_params: Some(lance::dataset::WriteParams {
|
||||
store_params: Some(ObjectStoreParams {
|
||||
object_store: Some((store.clone(), table_url.clone())),
|
||||
..Default::default()
|
||||
}),
|
||||
commit_handler: Some(Arc::new(
|
||||
lance_table::io::commit::ConditionalPutCommitHandler,
|
||||
)),
|
||||
..Default::default()
|
||||
}),
|
||||
};
|
||||
request
|
||||
};
|
||||
|
||||
let left = db
|
||||
.database()
|
||||
.create_table(request(batch.clone(), barrier.clone()));
|
||||
let right = db.database().create_table(request(batch, barrier));
|
||||
let (left, right) = timeout(Duration::from_secs(30), async { tokio::join!(left, right) })
|
||||
.await
|
||||
.expect("concurrent creates deadlocked");
|
||||
|
||||
let results = [left, right];
|
||||
assert_eq!(
|
||||
results.iter().filter(|result| result.is_ok()).count(),
|
||||
1,
|
||||
"expected one successful create, got {results:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
results
|
||||
.iter()
|
||||
.filter(|result| matches!(result, Err(Error::TableAlreadyExists { .. })))
|
||||
.count(),
|
||||
1,
|
||||
"expected one manifest conflict, got {results:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_listing_database_root_ops_do_not_create_manifest() {
|
||||
let tempdir = tempdir().unwrap();
|
||||
@@ -1378,6 +1542,25 @@ mod tests {
|
||||
assert!(!tempdir.path().join("__manifest").exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_file_object_store_uses_local_file_provider() {
|
||||
let tempdir = tempdir().unwrap();
|
||||
let path = tempdir.path().to_string_lossy().replace('\\', "/");
|
||||
let uri = if path.starts_with('/') {
|
||||
format!("file-object-store://{path}")
|
||||
} else {
|
||||
format!("file-object-store:///{path}")
|
||||
};
|
||||
let registry = Arc::new(lance_io::object_store::ObjectStoreRegistry::default());
|
||||
let (store, _) =
|
||||
ObjectStore::from_uri_and_params(registry, &uri, &ObjectStoreParams::default())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(store.scheme(), "file-object-store");
|
||||
assert!(ListingDatabase::uses_local_file_provider(&store));
|
||||
}
|
||||
|
||||
/// Regression test for https://github.com/lancedb/lancedb/issues/1600.
|
||||
///
|
||||
/// Opening a table used to create a separate object-store client instead of
|
||||
@@ -1401,9 +1584,13 @@ mod tests {
|
||||
read_consistency_interval: None,
|
||||
session: Some(session),
|
||||
};
|
||||
let db = ListingDatabase::connect_with_options(&request)
|
||||
let mut db = ListingDatabase::connect_with_options(&request)
|
||||
.await
|
||||
.unwrap();
|
||||
// A connection-level write wrapper must not prevent table opens from
|
||||
// reusing the connection's registered object store.
|
||||
let wrapper_calls = Arc::new(AtomicUsize::new(0));
|
||||
db.store_wrapper = Some(Arc::new(PassthroughStoreWrapper(wrapper_calls.clone())));
|
||||
|
||||
let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
|
||||
db.create_table(CreateTableRequest {
|
||||
@@ -1419,6 +1606,7 @@ mod tests {
|
||||
.unwrap();
|
||||
|
||||
let before_open = registry.stats();
|
||||
let wrapper_calls_before_open = wrapper_calls.load(Ordering::Relaxed);
|
||||
for _ in 0..3 {
|
||||
let table = db
|
||||
.open_table(OpenTableRequest {
|
||||
@@ -1438,6 +1626,7 @@ mod tests {
|
||||
let after_open = registry.stats();
|
||||
assert_eq!(after_open.misses, before_open.misses);
|
||||
assert!(after_open.hits >= before_open.hits + 3);
|
||||
assert!(wrapper_calls.load(Ordering::Relaxed) >= wrapper_calls_before_open + 3);
|
||||
}
|
||||
|
||||
/// Regression test for https://github.com/lancedb/lancedb/issues/3197.
|
||||
@@ -1581,6 +1770,46 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_clone_table_uses_connection_store_wrapper() {
|
||||
let (_tempdir, mut db) = setup_database().await;
|
||||
let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
|
||||
db.create_table(CreateTableRequest {
|
||||
name: "source_table".to_string(),
|
||||
namespace_path: vec![],
|
||||
data: Box::new(RecordBatch::new_empty(schema)) as Box<dyn Scannable>,
|
||||
mode: CreateTableMode::Create,
|
||||
write_options: Default::default(),
|
||||
location: None,
|
||||
namespace_client: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let source_uri = db.table_uri("source_table").unwrap();
|
||||
let tracker = IoStatsHolder::default();
|
||||
db.store_wrapper = Some(Arc::new(tracker.clone()));
|
||||
let _ = tracker.incremental_stats();
|
||||
|
||||
db.clone_table(CloneTableRequest {
|
||||
target_table_name: "cloned_table".to_string(),
|
||||
target_namespace_path: vec![],
|
||||
source_uri,
|
||||
source_version: None,
|
||||
source_tag: None,
|
||||
is_shallow: true,
|
||||
namespace_client: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let stats = tracker.incremental_stats();
|
||||
assert!(
|
||||
stats.write_iops > 0,
|
||||
"clone bypassed the wrapper: {stats:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_clone_table_with_data() {
|
||||
let (_tempdir, db) = setup_database().await;
|
||||
|
||||
@@ -18,6 +18,9 @@ use async_trait::async_trait;
|
||||
#[cfg(test)]
|
||||
pub mod io_tracking;
|
||||
|
||||
#[cfg(windows)]
|
||||
pub(crate) mod windows;
|
||||
|
||||
#[derive(Debug)]
|
||||
struct MirroringObjectStore {
|
||||
primary: Arc<dyn ObjectStore>,
|
||||
|
||||
@@ -0,0 +1,223 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
//! Windows local filesystem compatibility for atomic manifest commits.
|
||||
|
||||
use std::ffi::OsStr;
|
||||
use std::fmt::{Display, Formatter};
|
||||
use std::os::windows::ffi::OsStrExt;
|
||||
use std::path::{Path as StdPath, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use bytes::Bytes;
|
||||
use futures::stream::BoxStream;
|
||||
use lance::io::WrappingObjectStore;
|
||||
use object_store::{
|
||||
CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
|
||||
ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult, RenameOptions,
|
||||
RenameTargetMode, Result, UploadPart, path::Path,
|
||||
};
|
||||
use windows_sys::Win32::Foundation::{ERROR_ALREADY_EXISTS, ERROR_FILE_EXISTS};
|
||||
use windows_sys::Win32::Storage::FileSystem::MoveFileExW;
|
||||
|
||||
const STORE_NAME: &str = "WindowsLocalFileSystem";
|
||||
|
||||
/// Uses the Windows move primitive for create-only renames on local stores.
|
||||
///
|
||||
/// `object_store` implements create-only local renames with a hard link followed
|
||||
/// by a delete. Some Windows filesystems do not support hard links, but
|
||||
/// `MoveFileExW` without `MOVEFILE_REPLACE_EXISTING` provides the same atomic
|
||||
/// create-only rename semantics without requiring them.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct WindowsLocalFileSystemWrapper;
|
||||
|
||||
impl WrappingObjectStore for WindowsLocalFileSystemWrapper {
|
||||
fn wrap(&self, _store_prefix: &str, target: Arc<dyn ObjectStore>) -> Arc<dyn ObjectStore> {
|
||||
Arc::new(WindowsLocalFileSystem { target })
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct WindowsLocalFileSystem {
|
||||
target: Arc<dyn ObjectStore>,
|
||||
}
|
||||
|
||||
impl Display for WindowsLocalFileSystem {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "{STORE_NAME}({})", self.target)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
#[deny(clippy::missing_trait_methods)]
|
||||
impl ObjectStore for WindowsLocalFileSystem {
|
||||
async fn put_opts(
|
||||
&self,
|
||||
location: &Path,
|
||||
bytes: PutPayload,
|
||||
opts: PutOptions,
|
||||
) -> Result<PutResult> {
|
||||
self.target.put_opts(location, bytes, opts).await
|
||||
}
|
||||
|
||||
async fn put_multipart_opts(
|
||||
&self,
|
||||
location: &Path,
|
||||
opts: PutMultipartOptions,
|
||||
) -> Result<Box<dyn MultipartUpload>> {
|
||||
self.target.put_multipart_opts(location, opts).await
|
||||
}
|
||||
|
||||
async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
|
||||
self.target.get_opts(location, options).await
|
||||
}
|
||||
|
||||
async fn get_ranges(
|
||||
&self,
|
||||
location: &Path,
|
||||
ranges: &[std::ops::Range<u64>],
|
||||
) -> Result<Vec<Bytes>> {
|
||||
self.target.get_ranges(location, ranges).await
|
||||
}
|
||||
|
||||
fn delete_stream(
|
||||
&self,
|
||||
locations: BoxStream<'static, Result<Path>>,
|
||||
) -> BoxStream<'static, Result<Path>> {
|
||||
self.target.delete_stream(locations)
|
||||
}
|
||||
|
||||
fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
|
||||
self.target.list(prefix)
|
||||
}
|
||||
|
||||
fn list_with_offset(
|
||||
&self,
|
||||
prefix: Option<&Path>,
|
||||
offset: &Path,
|
||||
) -> BoxStream<'static, Result<ObjectMeta>> {
|
||||
self.target.list_with_offset(prefix, offset)
|
||||
}
|
||||
|
||||
async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
|
||||
self.target.list_with_delimiter(prefix).await
|
||||
}
|
||||
|
||||
async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
|
||||
self.target.copy_opts(from, to, options).await
|
||||
}
|
||||
|
||||
async fn rename_opts(&self, from: &Path, to: &Path, options: RenameOptions) -> Result<()> {
|
||||
if options.target_mode != RenameTargetMode::Create {
|
||||
return self.target.rename_opts(from, to, options).await;
|
||||
}
|
||||
|
||||
let from = PathBuf::from(from.as_ref());
|
||||
let to = PathBuf::from(to.as_ref());
|
||||
tokio::task::spawn_blocking(move || move_file_if_not_exists(&from, &to))
|
||||
.await
|
||||
.map_err(|source| Error::Generic {
|
||||
store: STORE_NAME,
|
||||
source: Box::new(source),
|
||||
})?
|
||||
}
|
||||
}
|
||||
|
||||
fn move_file_if_not_exists(from: &StdPath, to: &StdPath) -> Result<()> {
|
||||
let from_wide = null_terminated_wide(from.as_os_str());
|
||||
let to_wide = null_terminated_wide(to.as_os_str());
|
||||
|
||||
// SAFETY: both pointers reference null-terminated UTF-16 buffers that remain
|
||||
// alive for the duration of this call. A zero flag value deliberately omits
|
||||
// MOVEFILE_REPLACE_EXISTING, giving this operation create-only semantics.
|
||||
if unsafe { MoveFileExW(from_wide.as_ptr(), to_wide.as_ptr(), 0) } != 0 {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let source = std::io::Error::last_os_error();
|
||||
let path = to.to_string_lossy().into_owned();
|
||||
match source.raw_os_error().map(|code| code as u32) {
|
||||
Some(ERROR_ALREADY_EXISTS | ERROR_FILE_EXISTS) => Err(Error::AlreadyExists {
|
||||
path,
|
||||
source: Box::new(source),
|
||||
}),
|
||||
_ if source.kind() == std::io::ErrorKind::NotFound => Err(Error::NotFound {
|
||||
path,
|
||||
source: Box::new(source),
|
||||
}),
|
||||
_ => Err(Error::Generic {
|
||||
store: STORE_NAME,
|
||||
source: Box::new(source),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
fn null_terminated_wide(value: &OsStr) -> Vec<u16> {
|
||||
value.encode_wide().chain(Some(0)).collect()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use object_store::memory::InMemory;
|
||||
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_only_rename_does_not_use_hard_links() {
|
||||
let tempdir = tempfile::tempdir().unwrap();
|
||||
let source_path = tempdir.path().join("staged.manifest");
|
||||
let destination_path = tempdir.path().join("1.manifest");
|
||||
std::fs::write(&source_path, b"manifest").unwrap();
|
||||
|
||||
let source = Path::from_absolute_path(&source_path).unwrap();
|
||||
let destination = Path::from_absolute_path(&destination_path).unwrap();
|
||||
let store = WindowsLocalFileSystem {
|
||||
// The source does not exist in this inner store. Delegating the
|
||||
// rename would fail, proving the wrapper uses the native move path.
|
||||
target: Arc::new(InMemory::new()),
|
||||
};
|
||||
|
||||
store
|
||||
.rename_opts(
|
||||
&source,
|
||||
&destination,
|
||||
RenameOptions::new().with_target_mode(RenameTargetMode::Create),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(!source_path.exists());
|
||||
assert_eq!(std::fs::read(destination_path).unwrap(), b"manifest");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_only_rename_preserves_existing_destination() {
|
||||
let tempdir = tempfile::tempdir().unwrap();
|
||||
let source_path = tempdir.path().join("staged.manifest");
|
||||
let destination_path = tempdir.path().join("1.manifest");
|
||||
std::fs::write(&source_path, b"new manifest").unwrap();
|
||||
std::fs::write(&destination_path, b"existing manifest").unwrap();
|
||||
|
||||
let source = Path::from_absolute_path(&source_path).unwrap();
|
||||
let destination = Path::from_absolute_path(&destination_path).unwrap();
|
||||
let store = WindowsLocalFileSystem {
|
||||
target: Arc::new(InMemory::new()),
|
||||
};
|
||||
|
||||
let error = store
|
||||
.rename_opts(
|
||||
&source,
|
||||
&destination,
|
||||
RenameOptions::new().with_target_mode(RenameTargetMode::Create),
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(matches!(error, Error::AlreadyExists { .. }));
|
||||
assert_eq!(std::fs::read(source_path).unwrap(), b"new manifest");
|
||||
assert_eq!(
|
||||
std::fs::read(destination_path).unwrap(),
|
||||
b"existing manifest"
|
||||
);
|
||||
}
|
||||
}
|
||||
+296
-112
@@ -50,7 +50,6 @@ use crate::DistanceType;
|
||||
use crate::blob::BlobRangeRequest;
|
||||
use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions};
|
||||
use crate::database::Database;
|
||||
use crate::database::listing::LANCE_FILE_EXTENSION;
|
||||
use crate::database::read_freshness::TableFreshness;
|
||||
use crate::embeddings::{EmbeddingDefinition, EmbeddingRegistry, MemoryRegistry};
|
||||
use crate::error::{Error, Result};
|
||||
@@ -152,55 +151,6 @@ pub(crate) fn map_namespace_lance_error(err: lance::Error, table_name: &str) ->
|
||||
}
|
||||
}
|
||||
|
||||
/// Map a `lance::Error::DatasetNotFound` for the table at `uri` into a `lancedb::Error`.
|
||||
///
|
||||
/// Lance reports "there is nothing at this location" and "there is a table directory
|
||||
/// here but nothing loadable inside it" with the same error. Only the first is a
|
||||
/// `TableNotFound`: a `<name>.lance` directory left behind by an interrupted drop and
|
||||
/// re-create is still reported by `Connection::table_names`, so callers need to be able
|
||||
/// to tell "never existed" from "exists but is broken".
|
||||
///
|
||||
/// See <https://github.com/lancedb/lancedb/issues/3127>.
|
||||
async fn map_dataset_not_found(
|
||||
uri: &str,
|
||||
name: &str,
|
||||
params: ReadParams,
|
||||
err: lance::Error,
|
||||
) -> Error {
|
||||
let name = name.to_string();
|
||||
let source = Box::new(err);
|
||||
if table_dir_exists(uri, params).await.unwrap_or(false) {
|
||||
Error::TableCorrupted { name, source }
|
||||
} else {
|
||||
Error::TableNotFound { name, source }
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether a table directory is present at `uri`, even though no dataset could be
|
||||
/// loaded from it.
|
||||
///
|
||||
/// This looks for a `<name>.lance` entry in the parent directory, which is exactly what
|
||||
/// `ListingDatabase::table_names` lists, so the two APIs agree on whether a table is
|
||||
/// present. Probing `uri` itself would not work: object stores have no empty
|
||||
/// directories to probe, and on a local filesystem the interesting case is precisely an
|
||||
/// empty directory.
|
||||
async fn table_dir_exists(uri: &str, params: ReadParams) -> Result<bool> {
|
||||
let (object_store, path, _) = DatasetBuilder::from_uri(uri)
|
||||
.with_read_params(params)
|
||||
.build_object_store()
|
||||
.await?;
|
||||
// Only `*.lance` entries are ever reported as tables, so nothing else can produce
|
||||
// the list-then-open mismatch this guards against.
|
||||
if path.extension() != Some(LANCE_FILE_EXTENSION) {
|
||||
return Ok(false);
|
||||
}
|
||||
let (Some(parent), Some(dir_name)) = (path.parent(), path.filename()) else {
|
||||
return Ok(false);
|
||||
};
|
||||
let entries = object_store.read_dir(parent).await?;
|
||||
Ok(entries.iter().any(|entry| entry.as_str() == dir_name))
|
||||
}
|
||||
|
||||
/// Defines the type of column
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum ColumnKind {
|
||||
@@ -2385,10 +2335,19 @@ impl NativeTable {
|
||||
managed_versioning: Option<bool>,
|
||||
) -> Result<Self> {
|
||||
let params = params.unwrap_or_default();
|
||||
// patch the params if we have a write store wrapper
|
||||
let params = match write_store_wrapper.clone() {
|
||||
Some(wrapper) => params.patch_with_store_wrapper(wrapper)?,
|
||||
None => params,
|
||||
let has_caller_store_wrapper = params
|
||||
.store_options
|
||||
.as_ref()
|
||||
.and_then(|options| options.object_store_wrapper.as_ref())
|
||||
.is_some();
|
||||
// A caller wrapper must remain outside connection-level compatibility
|
||||
// behavior. When there is no caller wrapper, apply the compatibility
|
||||
// layer after loading so the session's registered store can be reused.
|
||||
let (params, wrapper_after_load) = match write_store_wrapper {
|
||||
Some(wrapper) if has_caller_store_wrapper => {
|
||||
(params.patch_with_store_wrapper(wrapper)?, None)
|
||||
}
|
||||
wrapper => (params, wrapper),
|
||||
};
|
||||
|
||||
// Build table_id from namespace + name
|
||||
@@ -2420,8 +2379,6 @@ impl NativeTable {
|
||||
None => false,
|
||||
};
|
||||
|
||||
// Kept so that a `DatasetNotFound` can be re-checked against storage below.
|
||||
let recovery_params = params.clone();
|
||||
let mut builder = DatasetBuilder::from_uri(uri).with_read_params(params);
|
||||
|
||||
// Set up commit handler when managed_versioning is enabled
|
||||
@@ -2440,10 +2397,23 @@ impl NativeTable {
|
||||
let dataset = match builder.load().await {
|
||||
Ok(dataset) => dataset,
|
||||
Err(e @ lance::Error::DatasetNotFound { .. }) => {
|
||||
return Err(map_dataset_not_found(uri, name, recovery_params, e).await);
|
||||
// The manifest load is the existence check. A physical prefix may be
|
||||
// from a concurrent or abandoned create, so it cannot refine this error.
|
||||
return Err(Error::TableNotFound {
|
||||
name: name.to_string(),
|
||||
source: Box::new(e),
|
||||
});
|
||||
}
|
||||
Err(e) => return Err(e.into()),
|
||||
};
|
||||
// Resolve the store from the session registry before applying a
|
||||
// connection-level write wrapper. Wrapper identity is part of the
|
||||
// registry key, so including it in ReadParams prevents reuse when the
|
||||
// opened table (and its wrapped store) is short-lived.
|
||||
let dataset = match wrapper_after_load {
|
||||
Some(wrapper) => dataset.with_object_store_wrappers([wrapper]),
|
||||
None => dataset,
|
||||
};
|
||||
|
||||
let dataset = DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval);
|
||||
let id = Self::build_id(&namespace, name);
|
||||
@@ -2545,11 +2515,16 @@ impl NativeTable {
|
||||
if let Some(sess) = session {
|
||||
params.session(sess);
|
||||
}
|
||||
|
||||
// patch the params if we have a write store wrapper
|
||||
let params = match write_store_wrapper.clone() {
|
||||
Some(wrapper) => params.patch_with_store_wrapper(wrapper)?,
|
||||
None => params,
|
||||
let has_caller_store_wrapper = params
|
||||
.store_options
|
||||
.as_ref()
|
||||
.and_then(|options| options.object_store_wrapper.as_ref())
|
||||
.is_some();
|
||||
let (params, wrapper_after_load) = match write_store_wrapper {
|
||||
Some(wrapper) if has_caller_store_wrapper => {
|
||||
(params.patch_with_store_wrapper(wrapper)?, None)
|
||||
}
|
||||
wrapper => (params, wrapper),
|
||||
};
|
||||
|
||||
// Build table_id from namespace + name
|
||||
@@ -2573,6 +2548,13 @@ impl NativeTable {
|
||||
},
|
||||
e => e.into(),
|
||||
})?;
|
||||
// Apply the write wrapper after the session registry has resolved the
|
||||
// shared store. The cloned dataset retains the wrapper for subsequent
|
||||
// reads, manifest commits, and any additional base stores.
|
||||
let dataset = match wrapper_after_load {
|
||||
Some(wrapper) => dataset.with_object_store_wrappers([wrapper]),
|
||||
None => dataset,
|
||||
};
|
||||
|
||||
let uri = dataset.uri().to_string();
|
||||
let dataset = DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval);
|
||||
@@ -3707,8 +3689,8 @@ pub struct FragmentSummaryStats {
|
||||
#[cfg(test)]
|
||||
#[allow(deprecated)]
|
||||
mod tests {
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use arrow_array::{
|
||||
@@ -3790,73 +3772,50 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Write a table and then break it, leaving the `<name>.lance` directory in place.
|
||||
///
|
||||
/// `remove_all` reproduces an interrupted drop + re-create (the directory is left
|
||||
/// empty); otherwise only the manifests are removed, leaving the data files behind.
|
||||
async fn write_then_corrupt_table(dir: &std::path::Path, remove_all: bool) -> String {
|
||||
let dataset_path = dir.join("test.lance");
|
||||
let uri = dataset_path.to_str().unwrap().to_string();
|
||||
|
||||
let batch = make_test_batches();
|
||||
let reader = RecordBatchIterator::new(vec![Ok(batch.clone())], batch.schema());
|
||||
Dataset::write(reader, &uri, None).await.unwrap();
|
||||
|
||||
if remove_all {
|
||||
for entry in std::fs::read_dir(&dataset_path).unwrap() {
|
||||
let entry = entry.unwrap();
|
||||
if entry.file_type().unwrap().is_dir() {
|
||||
std::fs::remove_dir_all(entry.path()).unwrap();
|
||||
} else {
|
||||
std::fs::remove_file(entry.path()).unwrap();
|
||||
}
|
||||
}
|
||||
assert_eq!(std::fs::read_dir(&dataset_path).unwrap().count(), 0);
|
||||
} else {
|
||||
let versions = dataset_path.join("_versions");
|
||||
assert!(versions.is_dir(), "expected manifests under {versions:?}");
|
||||
std::fs::remove_dir_all(&versions).unwrap();
|
||||
assert!(std::fs::read_dir(&dataset_path).unwrap().count() > 0);
|
||||
}
|
||||
|
||||
uri
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_open_corrupt_empty_dir() {
|
||||
async fn test_open_not_found_when_empty_directory_exists() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let uri = write_then_corrupt_table(tmp_dir.path(), true).await;
|
||||
let dataset_path = tmp_dir.path().join("test.lance");
|
||||
std::fs::create_dir(&dataset_path).unwrap();
|
||||
|
||||
let err = NativeTable::open(&uri).await.unwrap_err();
|
||||
let err = NativeTable::open(dataset_path.to_str().unwrap())
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
matches!(&err, Error::TableCorrupted { name, .. } if name == "test"),
|
||||
matches!(&err, Error::TableNotFound { name, .. } if name == "test"),
|
||||
"got {err:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_open_corrupt_missing_manifest() {
|
||||
async fn test_open_not_found_when_only_uncommitted_storage_exists() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let uri = write_then_corrupt_table(tmp_dir.path(), false).await;
|
||||
let dataset_path = tmp_dir.path().join("test.lance");
|
||||
let data_dir = dataset_path.join("data");
|
||||
std::fs::create_dir_all(&data_dir).unwrap();
|
||||
std::fs::write(data_dir.join("orphan.lance"), b"uncommitted").unwrap();
|
||||
|
||||
let err = NativeTable::open(&uri).await.unwrap_err();
|
||||
let err = NativeTable::open(dataset_path.to_str().unwrap())
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
matches!(&err, Error::TableCorrupted { name, .. } if name == "test"),
|
||||
matches!(&err, Error::TableNotFound { name, .. } if name == "test"),
|
||||
"got {err:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// A table listed by `table_names()` must not be reported as missing by
|
||||
/// `open_table()`. See <https://github.com/lancedb/lancedb/issues/3127>.
|
||||
/// Listing databases discover physical `*.lance` entries. That snapshot is not an
|
||||
/// authoritative table-existence check: only a committed manifest makes a table
|
||||
/// openable, and the entry could also be concurrently created or dropped.
|
||||
#[tokio::test]
|
||||
async fn test_open_table_corrupt_is_still_listed() {
|
||||
async fn test_table_names_may_include_uncommitted_storage() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let db = connect(tmp_dir.path().to_str().unwrap())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
write_then_corrupt_table(tmp_dir.path(), true).await;
|
||||
std::fs::create_dir(tmp_dir.path().join("test.lance")).unwrap();
|
||||
|
||||
assert_eq!(
|
||||
db.table_names().execute().await.unwrap(),
|
||||
@@ -3864,12 +3823,177 @@ mod tests {
|
||||
);
|
||||
let err = db.open_table("test").execute().await.unwrap_err();
|
||||
assert!(
|
||||
matches!(&err, Error::TableCorrupted { name, .. } if name == "test"),
|
||||
matches!(&err, Error::TableNotFound { name, .. } if name == "test"),
|
||||
"physical storage without a committed manifest is not a table: {err:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct ParentListGuardStore {
|
||||
inner: Arc<dyn object_store::ObjectStore>,
|
||||
parent: object_store::path::Path,
|
||||
parent_list_calls: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for ParentListGuardStore {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.write_str("ParentListGuardStore")
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
#[deny(clippy::missing_trait_methods)]
|
||||
impl object_store::ObjectStore for ParentListGuardStore {
|
||||
async fn put_opts(
|
||||
&self,
|
||||
location: &object_store::path::Path,
|
||||
payload: object_store::PutPayload,
|
||||
opts: object_store::PutOptions,
|
||||
) -> object_store::Result<object_store::PutResult> {
|
||||
self.inner.put_opts(location, payload, opts).await
|
||||
}
|
||||
|
||||
async fn put_multipart_opts(
|
||||
&self,
|
||||
location: &object_store::path::Path,
|
||||
opts: object_store::PutMultipartOptions,
|
||||
) -> object_store::Result<Box<dyn object_store::MultipartUpload>> {
|
||||
self.inner.put_multipart_opts(location, opts).await
|
||||
}
|
||||
|
||||
async fn get_opts(
|
||||
&self,
|
||||
location: &object_store::path::Path,
|
||||
options: object_store::GetOptions,
|
||||
) -> object_store::Result<object_store::GetResult> {
|
||||
self.inner.get_opts(location, options).await
|
||||
}
|
||||
|
||||
async fn get_ranges(
|
||||
&self,
|
||||
location: &object_store::path::Path,
|
||||
ranges: &[std::ops::Range<u64>],
|
||||
) -> object_store::Result<Vec<bytes::Bytes>> {
|
||||
self.inner.get_ranges(location, ranges).await
|
||||
}
|
||||
|
||||
fn delete_stream(
|
||||
&self,
|
||||
locations: futures::stream::BoxStream<
|
||||
'static,
|
||||
object_store::Result<object_store::path::Path>,
|
||||
>,
|
||||
) -> futures::stream::BoxStream<'static, object_store::Result<object_store::path::Path>>
|
||||
{
|
||||
self.inner.delete_stream(locations)
|
||||
}
|
||||
|
||||
fn list(
|
||||
&self,
|
||||
prefix: Option<&object_store::path::Path>,
|
||||
) -> futures::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>>
|
||||
{
|
||||
if prefix == Some(&self.parent) {
|
||||
self.parent_list_calls.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
self.inner.list(prefix)
|
||||
}
|
||||
|
||||
fn list_with_offset(
|
||||
&self,
|
||||
prefix: Option<&object_store::path::Path>,
|
||||
offset: &object_store::path::Path,
|
||||
) -> futures::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>>
|
||||
{
|
||||
if prefix == Some(&self.parent) {
|
||||
self.parent_list_calls.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
self.inner.list_with_offset(prefix, offset)
|
||||
}
|
||||
|
||||
async fn list_with_delimiter(
|
||||
&self,
|
||||
prefix: Option<&object_store::path::Path>,
|
||||
) -> object_store::Result<object_store::ListResult> {
|
||||
if prefix == Some(&self.parent) {
|
||||
self.parent_list_calls.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
self.inner.list_with_delimiter(prefix).await
|
||||
}
|
||||
|
||||
async fn copy_opts(
|
||||
&self,
|
||||
from: &object_store::path::Path,
|
||||
to: &object_store::path::Path,
|
||||
options: object_store::CopyOptions,
|
||||
) -> object_store::Result<()> {
|
||||
self.inner.copy_opts(from, to, options).await
|
||||
}
|
||||
|
||||
async fn rename_opts(
|
||||
&self,
|
||||
from: &object_store::path::Path,
|
||||
to: &object_store::path::Path,
|
||||
options: object_store::RenameOptions,
|
||||
) -> object_store::Result<()> {
|
||||
self.inner.rename_opts(from, to, options).await
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct ParentListGuardWrapper {
|
||||
parent_list_calls: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
impl WrappingObjectStore for ParentListGuardWrapper {
|
||||
fn wrap(
|
||||
&self,
|
||||
_store_prefix: &str,
|
||||
inner: Arc<dyn object_store::ObjectStore>,
|
||||
) -> Arc<dyn object_store::ObjectStore> {
|
||||
Arc::new(ParentListGuardStore {
|
||||
inner,
|
||||
parent: object_store::path::Path::from("database"),
|
||||
parent_list_calls: self.parent_list_calls.clone(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_open_missing_never_lists_database_parent() {
|
||||
let parent_list_calls = Arc::new(AtomicUsize::new(0));
|
||||
let params = ReadParams {
|
||||
store_options: Some(ObjectStoreParams {
|
||||
object_store_wrapper: Some(Arc::new(ParentListGuardWrapper {
|
||||
parent_list_calls: parent_list_calls.clone(),
|
||||
})),
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let err = NativeTable::open_with_params(
|
||||
"memory:///database/missing.lance",
|
||||
"missing",
|
||||
Vec::new(),
|
||||
None,
|
||||
Some(params),
|
||||
None,
|
||||
None,
|
||||
HashSet::new(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(
|
||||
matches!(&err, Error::TableNotFound { name, .. } if name == "missing"),
|
||||
"got {err:?}"
|
||||
);
|
||||
assert!(
|
||||
err.to_string().contains("exists but could not be loaded"),
|
||||
"got {err}"
|
||||
assert_eq!(
|
||||
parent_list_calls.load(Ordering::Relaxed),
|
||||
0,
|
||||
"opening one missing table must not enumerate sibling tables"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -3938,6 +4062,66 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct OrderedStoreWrapper {
|
||||
name: &'static str,
|
||||
order: Arc<Mutex<Vec<&'static str>>>,
|
||||
}
|
||||
|
||||
impl WrappingObjectStore for OrderedStoreWrapper {
|
||||
fn wrap(
|
||||
&self,
|
||||
_store_prefix: &str,
|
||||
original: Arc<dyn object_store::ObjectStore>,
|
||||
) -> Arc<dyn object_store::ObjectStore> {
|
||||
self.order.lock().unwrap().push(self.name);
|
||||
original
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_open_with_params_keeps_caller_store_wrapper_outermost() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let dataset_path = tmp_dir.path().join("test.lance");
|
||||
let uri = dataset_path.to_str().unwrap();
|
||||
let batch = make_test_batches();
|
||||
let reader = RecordBatchIterator::new(vec![Ok(batch.clone())], batch.schema());
|
||||
Dataset::write(reader, uri, None).await.unwrap();
|
||||
|
||||
let order = Arc::new(Mutex::new(Vec::new()));
|
||||
let caller_wrapper = Arc::new(OrderedStoreWrapper {
|
||||
name: "caller",
|
||||
order: order.clone(),
|
||||
});
|
||||
let compatibility_wrapper = Arc::new(OrderedStoreWrapper {
|
||||
name: "compatibility",
|
||||
order: order.clone(),
|
||||
});
|
||||
let params = ReadParams {
|
||||
store_options: Some(ObjectStoreParams {
|
||||
object_store_wrapper: Some(caller_wrapper),
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
NativeTable::open_with_params(
|
||||
uri,
|
||||
"test",
|
||||
vec![],
|
||||
Some(compatibility_wrapper),
|
||||
Some(params),
|
||||
None,
|
||||
None,
|
||||
HashSet::new(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(*order.lock().unwrap(), vec!["compatibility", "caller"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_open_table_options() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
|
||||
@@ -14,6 +14,7 @@ use lance::arrow::json::JsonDataType;
|
||||
use lance::dataset::{ReadParams, WriteParams};
|
||||
use lance::index::vector::utils::infer_vector_dim;
|
||||
use lance::io::{ObjectStoreParams, WrappingObjectStore};
|
||||
use lance_io::object_store::ChainedWrappingObjectStore;
|
||||
use std::pin::Pin;
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
@@ -37,13 +38,13 @@ impl PatchStoreParam for Option<ObjectStoreParams> {
|
||||
wrapper: Arc<dyn WrappingObjectStore>,
|
||||
) -> Result<Option<ObjectStoreParams>> {
|
||||
let mut params = self.unwrap_or_default();
|
||||
if params.object_store_wrapper.is_some() {
|
||||
return Err(Error::Other {
|
||||
message: "can not patch param because object store is already set".into(),
|
||||
source: None,
|
||||
});
|
||||
}
|
||||
params.object_store_wrapper = Some(wrapper);
|
||||
params.object_store_wrapper = Some(match params.object_store_wrapper.take() {
|
||||
// The wrapper being patched in is connection-level compatibility
|
||||
// behavior. Keep it closest to the target store so an existing
|
||||
// caller wrapper remains outermost and can observe every operation.
|
||||
Some(existing) => Arc::new(ChainedWrappingObjectStore::new(vec![wrapper, existing])),
|
||||
None => wrapper,
|
||||
});
|
||||
|
||||
Ok(Some(params))
|
||||
}
|
||||
@@ -472,14 +473,60 @@ impl Stream for MaxBatchLengthStream {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::Mutex;
|
||||
|
||||
use arrow_array::Int32Array;
|
||||
use arrow_schema::Field;
|
||||
use datafusion_physical_plan::stream::RecordBatchStreamAdapter;
|
||||
use futures::{StreamExt, stream};
|
||||
use object_store::{ObjectStore, memory::InMemory};
|
||||
use tokio::time::sleep;
|
||||
|
||||
use super::*;
|
||||
|
||||
#[derive(Debug)]
|
||||
struct OrderedStoreWrapper {
|
||||
name: &'static str,
|
||||
order: Arc<Mutex<Vec<&'static str>>>,
|
||||
}
|
||||
|
||||
impl WrappingObjectStore for OrderedStoreWrapper {
|
||||
fn wrap(
|
||||
&self,
|
||||
_store_prefix: &str,
|
||||
original: Arc<dyn ObjectStore>,
|
||||
) -> Arc<dyn ObjectStore> {
|
||||
self.order.lock().unwrap().push(self.name);
|
||||
original
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_patch_store_param_keeps_caller_wrapper_outermost() {
|
||||
let order = Arc::new(Mutex::new(Vec::new()));
|
||||
let params = Some(ObjectStoreParams {
|
||||
object_store_wrapper: Some(Arc::new(OrderedStoreWrapper {
|
||||
name: "caller",
|
||||
order: order.clone(),
|
||||
})),
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let params = params
|
||||
.patch_with_store_wrapper(Arc::new(OrderedStoreWrapper {
|
||||
name: "compatibility",
|
||||
order: order.clone(),
|
||||
}))
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
params
|
||||
.object_store_wrapper
|
||||
.unwrap()
|
||||
.wrap("memory", Arc::new(InMemory::new()) as Arc<dyn ObjectStore>);
|
||||
|
||||
assert_eq!(*order.lock().unwrap(), vec!["compatibility", "caller"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_guess_default_column() {
|
||||
let schema_no_vector = Schema::new(vec![
|
||||
|
||||
Reference in New Issue
Block a user