mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-13 14:16:54 +00:00
chore: update lance dependency to v15.0.0-beta.6 (#4742)
Updates Rust and Java Lance dependencies to [v15.0.0-beta.6](https://github.com/lance-format/lance/releases/tag/v15.0.0-beta.6), pins Python tests to published pylance 15.0.0b6, and refreshes both lockfiles. Aligns Arrow 59, DataFusion 55, and PyO3 0.29; adapts execution plans and MemWAL directory handling, updates the ABI metadata test, and removes the obsolete Lance Arrow patch now fixed upstream. Validation: workspace clippy with all features/tests and warnings denied, cargo fmt, Python/CI Ruff checks, uv lock check, 16 CI-script tests, and targeted Python regressions (54 passed, 28 skipped); repository-wide Ruff retains existing documentation-example failures, and the upstream macOS wheel build failed while Linux and Windows wheels are published. --------- Co-authored-by: Daniel Rammer <hamersaw@protonmail.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
LanceDB Robot
co-authored by
Daniel Rammer
Claude Opus 5.5
Daniel Rammer
Claude Opus 5.5
parent
03c7d1f199
commit
d5013459eb
@@ -381,7 +381,7 @@ jobs:
|
||||
runs-on: ubuntu-24.04
|
||||
strategy:
|
||||
matrix:
|
||||
msrv: ["1.94.0"] # This should match up with rust-version in Cargo.toml
|
||||
msrv: ["1.95.0"] # This should match up with rust-version in Cargo.toml
|
||||
env:
|
||||
# Need up-to-date compilers for kernels
|
||||
CC: clang-18
|
||||
|
||||
Generated
+1421
-1245
File diff suppressed because it is too large
Load Diff
+35
-39
@@ -16,52 +16,52 @@ repository = "https://github.com/lancedb/lancedb"
|
||||
description = "Serverless, low-latency vector database for AI applications"
|
||||
keywords = ["lancedb", "lance", "database", "vector", "search"]
|
||||
categories = ["database-implementations"]
|
||||
rust-version = "1.94.0"
|
||||
rust-version = "1.95.0"
|
||||
|
||||
[workspace.dependencies]
|
||||
lance = { "version" = "=15.0.0-beta.3", default-features = false, "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=15.0.0-beta.3", default-features = false, "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-jsonb = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=15.0.0-beta.3", default-features = false, "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { version = "=15.0.0-beta.3", path = "python/vendor/lance-arrow" }
|
||||
lance = { "version" = "=15.0.0-beta.6", default-features = false, "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=15.0.0-beta.6", default-features = false, "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-jsonb = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=15.0.0-beta.6", default-features = false, "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=15.0.0-beta.6", "tag" = "v15.0.0-beta.6", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lancedb = { path = "rust/lancedb", default-features = false }
|
||||
ahash = "0.8"
|
||||
anyhow = "1"
|
||||
# Note that this one does not include pyarrow
|
||||
arrow = { version = "58.0.0", optional = false }
|
||||
arrow-array = "58.0.0"
|
||||
arrow-buffer = "58.0.0"
|
||||
arrow-data = "58.0.0"
|
||||
arrow-ipc = "58.0.0"
|
||||
arrow-ord = "58.0.0"
|
||||
arrow-schema = "58.0.0"
|
||||
arrow-select = "58.0.0"
|
||||
arrow-cast = "58.0.0"
|
||||
arrow-flight = { version = "58.0.0", features = ["flight-sql-experimental"] }
|
||||
arrow = { version = "59.2.0", optional = false }
|
||||
arrow-array = "59.2.0"
|
||||
arrow-buffer = "59.2.0"
|
||||
arrow-data = "59.2.0"
|
||||
arrow-ipc = "59.2.0"
|
||||
arrow-ord = "59.2.0"
|
||||
arrow-schema = "59.2.0"
|
||||
arrow-select = "59.2.0"
|
||||
arrow-cast = "59.2.0"
|
||||
arrow-flight = { version = "59.2.0", features = ["flight-sql-experimental"] }
|
||||
async-trait = "0"
|
||||
# Smithy JSON 0.63 requires the pre-1.7 Document representation; allow the MSRV pin.
|
||||
aws-smithy-types = ">=1.3.6, <1.7"
|
||||
base64 = "0.22"
|
||||
bytes = "1"
|
||||
datafusion = { version = "54.0.0", default-features = false }
|
||||
datafusion-catalog = "54.0.0"
|
||||
datafusion-common = { version = "54.0.0", default-features = false }
|
||||
datafusion-execution = "54.0.0"
|
||||
datafusion-expr = "54.0.0"
|
||||
datafusion-functions = "54.0.0"
|
||||
datafusion-physical-plan = "54.0.0"
|
||||
datafusion-physical-expr = "54.0.0"
|
||||
datafusion-sql = "54.0.0"
|
||||
datafusion = { version = "55.0.0", default-features = false }
|
||||
datafusion-catalog = "55.0.0"
|
||||
datafusion-common = { version = "55.0.0", default-features = false }
|
||||
datafusion-execution = "55.0.0"
|
||||
datafusion-expr = "55.0.0"
|
||||
datafusion-functions = "55.0.0"
|
||||
datafusion-physical-plan = "55.0.0"
|
||||
datafusion-physical-expr = "55.0.0"
|
||||
datafusion-sql = "55.0.0"
|
||||
env_logger = "0.11"
|
||||
half = { "version" = "2.7.1", default-features = false, features = [
|
||||
"num-traits",
|
||||
@@ -94,10 +94,6 @@ tracing = "0.1"
|
||||
uuid = { version = "1.7.0", features = ["v4", "v7"] }
|
||||
chrono = { version = "0.4", default-features = false, features = ["clock"] }
|
||||
|
||||
[patch."https://github.com/lance-format/lance.git"]
|
||||
# Carry the offset-safe validity fix until it is available in a Lance release.
|
||||
lance-arrow = { path = "python/vendor/lance-arrow" }
|
||||
|
||||
[patch.crates-io]
|
||||
# Lance v14.0.0-beta.18 requires multipart write modes from this upstream revision.
|
||||
object_store = { git = "https://github.com/apache/arrow-rs-object-store.git", rev = "c10758c207c0103355f940f0c736fb8ecb157b04" }
|
||||
|
||||
+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>15.0.0-beta.3</lance-core.version>
|
||||
<lance-core.version>15.0.0-beta.6</lance-core.version>
|
||||
<jackson.version>2.18.11</jackson.version>
|
||||
<spotless.skip>false</spotless.skip>
|
||||
<spotless.version>2.30.0</spotless.version>
|
||||
|
||||
@@ -1731,12 +1731,13 @@ describe("When creating an index", () => {
|
||||
config: Index.ivfFlat({ numPartitions: 2, targetPartitionSize: 100 }),
|
||||
});
|
||||
|
||||
// Lance records the target size the partition count implies: 300 rows / 2.
|
||||
const indices = await tbl.listIndices();
|
||||
expect(
|
||||
(indices[0].indexDetails as Record<string, unknown>)[
|
||||
"target_partition_size"
|
||||
],
|
||||
).toBeUndefined();
|
||||
).toBe(150);
|
||||
});
|
||||
|
||||
it("should allow me to replace (or not) an existing index", async () => {
|
||||
|
||||
+4
-4
@@ -8,7 +8,7 @@ license.workspace = true
|
||||
repository.workspace = true
|
||||
keywords.workspace = true
|
||||
categories.workspace = true
|
||||
rust-version = "1.94.0"
|
||||
rust-version = "1.95.0"
|
||||
|
||||
[lib]
|
||||
name = "_lancedb"
|
||||
@@ -30,9 +30,9 @@ env_logger.workspace = true
|
||||
log.workspace = true
|
||||
# Maturin enables extension-module mode for Python builds. Keeping it out of
|
||||
# Cargo features lets Rust unit tests link against libpython.
|
||||
pyo3 = { version = "0.28", features = ["abi3-py310", "chrono", "uuid"] }
|
||||
pyo3 = { version = "0.29", features = ["abi3-py310", "chrono", "uuid"] }
|
||||
chrono.workspace = true
|
||||
pyo3-async-runtimes = { version = "0.28", features = [
|
||||
pyo3-async-runtimes = { version = "0.29", features = [
|
||||
"attributes",
|
||||
"tokio-runtime",
|
||||
] }
|
||||
@@ -46,7 +46,7 @@ uuid.workspace = true
|
||||
libc = "0.2"
|
||||
|
||||
[build-dependencies]
|
||||
pyo3-build-config = { version = "0.28", features = ["abi3-py310"] }
|
||||
pyo3-build-config = "0.29"
|
||||
|
||||
[features]
|
||||
default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/cos", "lancedb/goosefs", "lancedb/metrics-otel"]
|
||||
|
||||
@@ -64,7 +64,7 @@ tests = [
|
||||
"polars>=0.19, <=1.32.3",
|
||||
"pyarrow<25",
|
||||
"pyarrow-stubs>=16.0",
|
||||
"pylance==15.0.0b3",
|
||||
"pylance==15.0.0b6",
|
||||
"requests>=2.31.0",
|
||||
"datafusion>=54,<55",
|
||||
"opentelemetry-sdk>=1.30.0",
|
||||
|
||||
@@ -5921,7 +5921,8 @@ class AsyncTable:
|
||||
"""Remove the RTEWriteSpec from this table.
|
||||
|
||||
Reverts to the standard `merge_insert` write path. Errors if no spec
|
||||
is currently set.
|
||||
is currently set, or if the RTE still holds rows that have not been
|
||||
merged into the base table, since removing it would drop them.
|
||||
"""
|
||||
await self._inner.disable_rte()
|
||||
|
||||
|
||||
@@ -386,7 +386,7 @@ def test_custom_stop_words_list(table):
|
||||
assert [token.text for token in tokens] == ["the", "data"]
|
||||
empty_tokens = ldb.tokenize("the lance data", stem=False, custom_stop_words=[])
|
||||
assert [token.text for token in empty_tokens] == ["the", "lance", "data"]
|
||||
with pytest.raises(TypeError, match=r"custom_stop_words.*int"):
|
||||
with pytest.raises(TypeError, match=r"'int' object is not an instance of 'str'"):
|
||||
ldb.tokenize(
|
||||
"the lance data",
|
||||
custom_stop_words=["lance", 42],
|
||||
|
||||
@@ -32,8 +32,10 @@ def test_pyo3_abi_matches_minimum_supported_python():
|
||||
expected_abi = f"abi3-py{major}{minor}"
|
||||
configured_abis = re.findall(r'"(abi3-py\d+)"', cargo_manifest)
|
||||
|
||||
assert configured_abis == [expected_abi, expected_abi], (
|
||||
"the pyo3 runtime and build ABI features must both match requires-python"
|
||||
# PyO3 0.29 configures the ABI through the runtime dependency; its
|
||||
# build-config crate no longer exposes ABI features.
|
||||
assert configured_abis == [expected_abi], (
|
||||
"the pyo3 runtime ABI feature must match requires-python"
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -1417,6 +1417,7 @@ def test_branches_preserve_namespace(tmp_path):
|
||||
"lance"
|
||||
) # namespace_path routes through lance's DirectoryNamespace
|
||||
db = lancedb.connect(tmp_path)
|
||||
db.create_namespace(["ns1"])
|
||||
table = db.create_table("t", [{"id": 1}], namespace_path=["ns1"])
|
||||
assert table.namespace == ["ns1"]
|
||||
|
||||
|
||||
Generated
+6
-8
@@ -2007,7 +2007,7 @@ requires-dist = [
|
||||
{ name = "pyarrow-stubs", marker = "extra == 'tests'", specifier = ">=16.0" },
|
||||
{ name = "pydantic", specifier = ">=2.7.4,<3" },
|
||||
{ name = "pylance", marker = "extra == 'pylance'", specifier = ">=15.0.0b3" },
|
||||
{ name = "pylance", marker = "extra == 'tests'", specifier = "==15.0.0b3" },
|
||||
{ name = "pylance", marker = "extra == 'tests'", specifier = "==15.0.0b6" },
|
||||
{ name = "pyright", marker = "extra == 'dev'", specifier = ">=1.1.350" },
|
||||
{ name = "pytest", marker = "extra == 'tests'", specifier = ">=7.0" },
|
||||
{ name = "pytest-asyncio", marker = "extra == 'tests'", specifier = ">=0.21" },
|
||||
@@ -3938,7 +3938,7 @@ crypto = [
|
||||
|
||||
[[package]]
|
||||
name = "pylance"
|
||||
version = "15.0.0b3"
|
||||
version = "15.0.0b6"
|
||||
source = { registry = "https://pypi.fury.io/lance-format/" }
|
||||
dependencies = [
|
||||
{ name = "lance-namespace" },
|
||||
@@ -3948,12 +3948,10 @@ dependencies = [
|
||||
{ name = "pyarrow" },
|
||||
]
|
||||
wheels = [
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_1ttHFr/pylance-15.0.0b3-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:aaca475121cef79ada6fc19e2e2b54860e411cc8291f9ffa5643500264513086" },
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_19DzNz/pylance-15.0.0b3-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:5632f269b9e138b9ed3e5c7b788dfe27e04debab9d0f4c72be468f56d4a0d45a" },
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_20XdL/pylance-15.0.0b3-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:1bdc92d7dc4d38323452543f71e61b493c3cfb5129dade33175f478e394b1959" },
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_YBklr/pylance-15.0.0b3-cp310-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:6fcaced6f675f027979e1928284e104ae45eb8282d7c1eb0e51e4a5b92b93392" },
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_1HLHhZ/pylance-15.0.0b3-cp310-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:f19240f3ccd621b746062fd2825e9340b99131c2cbda7ed5b226b6f7871b73f4" },
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_24P9Ax/pylance-15.0.0b3-cp310-abi3-win_amd64.whl", hash = "sha256:3f964e4ed093b6f170601d40772fcd3525fd73d15fd56eb30d884a446668d602" },
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_2jARvj/pylance-15.0.0b6-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:88bd0646118c57e1c0a3118d04162323a74ebea28578f362977c927cdcb0f0c9" },
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_28yD6d/pylance-15.0.0b6-cp310-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:66613ecb88dcb8e02f0cd21d241987018d924b9f587cef4015bf7ea1815ab44f" },
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_1Us94M/pylance-15.0.0b6-cp310-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:d0765efab1f21b4697207ab75b9d391f1962813afb08e78db794eeea280c0bcc" },
|
||||
{ url = "https://pypi.fury.io/lance-format/-/ver_ccoue/pylance-15.0.0b6-cp310-abi3-win_amd64.whl", hash = "sha256:f21eb48c6ec5d2269bb1817b2c0bb036956383d05033fb54278c4df64251ead4" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -47,7 +47,7 @@ lance-index = { workspace = true, features = ["tokenizer-jieba", "tokenizer-lind
|
||||
lance-table = { workspace = true }
|
||||
lance-linalg = { workspace = true }
|
||||
lance-encoding = { workspace = true }
|
||||
lance-arrow = { "version" = "=15.0.0-beta.3", "tag" = "v15.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { workspace = true }
|
||||
lance-namespace = { workspace = true }
|
||||
lance-namespace-impls = { workspace = true }
|
||||
metrics = { workspace = true, optional = true }
|
||||
|
||||
@@ -17,7 +17,7 @@ anyhow.workspace = true
|
||||
arrow-array.workspace = true
|
||||
arrow-buffer.workspace = true
|
||||
arrow-cast.workspace = true
|
||||
arrow-json = "58.0.0"
|
||||
arrow-json = "59.2.0"
|
||||
arrow-schema.workspace = true
|
||||
async-trait.workspace = true
|
||||
clap = { version = "4.5.13", features = ["derive", "env"] }
|
||||
|
||||
@@ -254,6 +254,7 @@ async fn replace_fragments(dataset: Dataset, stream: SendableRecordBatchStream)
|
||||
update_mode: None,
|
||||
inserted_rows_filter: Some(inserted_rows_filter),
|
||||
updated_fragment_offsets: None,
|
||||
mem_wal_dir_id: None,
|
||||
},
|
||||
None,
|
||||
))
|
||||
@@ -663,6 +664,7 @@ mod tests {
|
||||
update_mode: None,
|
||||
inserted_rows_filter: Some(filter),
|
||||
updated_fragment_offsets: None,
|
||||
mem_wal_dir_id: None,
|
||||
},
|
||||
None,
|
||||
))
|
||||
|
||||
@@ -1862,6 +1862,16 @@ impl DisplayAs for TakeRestoreExec {
|
||||
}
|
||||
|
||||
impl ExecutionPlan for TakeRestoreExec {
|
||||
fn apply_expressions(
|
||||
&self,
|
||||
_f: &mut dyn FnMut(
|
||||
&Arc<dyn datafusion_physical_expr::PhysicalExpr>,
|
||||
) -> datafusion_common::Result<
|
||||
datafusion_common::tree_node::TreeNodeRecursion,
|
||||
>,
|
||||
) -> datafusion_common::Result<datafusion_common::tree_node::TreeNodeRecursion> {
|
||||
Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
|
||||
}
|
||||
fn name(&self) -> &str {
|
||||
"TakeRestoreExec"
|
||||
}
|
||||
|
||||
@@ -589,6 +589,16 @@ impl<S: HttpSend + 'static> DisplayAs for RemoteWriteExec<S> {
|
||||
}
|
||||
|
||||
impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
|
||||
fn apply_expressions(
|
||||
&self,
|
||||
_f: &mut dyn FnMut(
|
||||
&Arc<dyn datafusion_physical_expr::PhysicalExpr>,
|
||||
) -> datafusion_common::Result<
|
||||
datafusion_common::tree_node::TreeNodeRecursion,
|
||||
>,
|
||||
) -> datafusion_common::Result<datafusion_common::tree_node::TreeNodeRecursion> {
|
||||
Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
|
||||
}
|
||||
fn name(&self) -> &str {
|
||||
Self::static_name()
|
||||
}
|
||||
@@ -1149,6 +1159,16 @@ mod tests {
|
||||
}
|
||||
|
||||
impl ExecutionPlan for ErroringExec {
|
||||
fn apply_expressions(
|
||||
&self,
|
||||
_f: &mut dyn FnMut(
|
||||
&Arc<dyn datafusion_physical_expr::PhysicalExpr>,
|
||||
) -> datafusion_common::Result<
|
||||
datafusion_common::tree_node::TreeNodeRecursion,
|
||||
>,
|
||||
) -> datafusion_common::Result<datafusion_common::tree_node::TreeNodeRecursion> {
|
||||
Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
|
||||
}
|
||||
fn name(&self) -> &str {
|
||||
"ErroringExec"
|
||||
}
|
||||
|
||||
@@ -2075,7 +2075,9 @@ impl Table {
|
||||
/// Remove the [`RTEWriteSpec`] from this table, reverting to the standard
|
||||
/// `merge_insert` write path.
|
||||
///
|
||||
/// Errors if no spec is currently set.
|
||||
/// Errors if no spec is currently set, or if the RTE still holds rows that
|
||||
/// have not been merged into the base table, since removing it would drop
|
||||
/// them.
|
||||
pub async fn disable_rte(&self) -> Result<()> {
|
||||
self.inner.disable_rte().await
|
||||
}
|
||||
|
||||
@@ -3576,9 +3576,8 @@ mod tests {
|
||||
assert!(matches!(err, Error::NotSupported { .. }), "declare won");
|
||||
}
|
||||
|
||||
/// The gate's reproducer: after catch-up activation, an LSM write, and
|
||||
/// unset, retained SSTable rows survive without a live spec. The catch-up
|
||||
/// flag is the durable marker; declaration refuses on it.
|
||||
/// Rows written through the RTE keep the spec installed, since
|
||||
/// `disable_rte` refuses while they are unmerged, so declaration refuses.
|
||||
#[tokio::test]
|
||||
#[cfg_attr(
|
||||
not(feature = "experimental-functions"),
|
||||
@@ -3622,7 +3621,8 @@ mod tests {
|
||||
.execute(Box::new(RecordBatchIterator::new(vec![Ok(batch)], schema)))
|
||||
.await
|
||||
.unwrap();
|
||||
table.disable_rte().await.unwrap();
|
||||
// The RTE holds the written rows, so unsetting it is refused.
|
||||
table.disable_rte().await.unwrap_err();
|
||||
|
||||
let err = add_computed(&table, &[("doubled".into(), "value * 2".into())])
|
||||
.await
|
||||
|
||||
@@ -104,6 +104,16 @@ impl DisplayAs for MetadataEraserExec {
|
||||
}
|
||||
|
||||
impl ExecutionPlan for MetadataEraserExec {
|
||||
fn apply_expressions(
|
||||
&self,
|
||||
_f: &mut dyn FnMut(
|
||||
&Arc<dyn datafusion_physical_expr::PhysicalExpr>,
|
||||
) -> datafusion_common::Result<
|
||||
datafusion_common::tree_node::TreeNodeRecursion,
|
||||
>,
|
||||
) -> datafusion_common::Result<datafusion_common::tree_node::TreeNodeRecursion> {
|
||||
Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
|
||||
}
|
||||
fn name(&self) -> &str {
|
||||
"MetadataEraserExec"
|
||||
}
|
||||
@@ -157,8 +167,21 @@ impl ExecutionPlan for MetadataEraserExec {
|
||||
)
|
||||
}
|
||||
|
||||
fn partition_statistics(&self, partition: Option<usize>) -> DataFusionResult<Arc<Statistics>> {
|
||||
self.input.partition_statistics(partition)
|
||||
fn child_stats_requests(
|
||||
&self,
|
||||
partition: Option<usize>,
|
||||
) -> Vec<datafusion_physical_plan::statistics::ChildStats> {
|
||||
vec![datafusion_physical_plan::statistics::ChildStats::At(
|
||||
partition,
|
||||
)]
|
||||
}
|
||||
|
||||
fn statistics_from_inputs(
|
||||
&self,
|
||||
input_stats: &[Arc<Statistics>],
|
||||
_args: &datafusion_physical_plan::statistics::StatisticsArgs,
|
||||
) -> DataFusionResult<Arc<Statistics>> {
|
||||
Ok(input_stats[0].clone())
|
||||
}
|
||||
|
||||
fn supports_limit_pushdown(&self) -> bool {
|
||||
@@ -1398,7 +1421,9 @@ pub mod tests {
|
||||
|
||||
assert_eq!(physical_plan.name(), "MetadataEraserExec");
|
||||
|
||||
let partition_stats = physical_plan.partition_statistics(None).unwrap();
|
||||
let partition_stats = datafusion_physical_plan::statistics::StatisticsContext::new()
|
||||
.compute(physical_plan.as_ref(), &Default::default())
|
||||
.unwrap();
|
||||
|
||||
assert!(matches!(partition_stats.num_rows, Precision::Exact(10)));
|
||||
}
|
||||
@@ -1492,7 +1517,9 @@ pub mod tests {
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let rewired = foo.with_new_children(vec![tbl2.clone()]).unwrap();
|
||||
let rewired =
|
||||
datafusion_physical_plan::replace_children_if_necessary(foo, vec![tbl2.clone()])
|
||||
.unwrap();
|
||||
assert_eq!(rewired.schema(), tbl2.schema());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -141,6 +141,16 @@ impl DisplayAs for InsertExec {
|
||||
}
|
||||
|
||||
impl ExecutionPlan for InsertExec {
|
||||
fn apply_expressions(
|
||||
&self,
|
||||
_f: &mut dyn FnMut(
|
||||
&Arc<dyn datafusion_physical_expr::PhysicalExpr>,
|
||||
) -> datafusion_common::Result<
|
||||
datafusion_common::tree_node::TreeNodeRecursion,
|
||||
>,
|
||||
) -> datafusion_common::Result<datafusion_common::tree_node::TreeNodeRecursion> {
|
||||
Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
|
||||
}
|
||||
fn name(&self) -> &str {
|
||||
Self::static_name()
|
||||
}
|
||||
|
||||
@@ -62,6 +62,16 @@ impl DisplayAs for ScannableExec {
|
||||
}
|
||||
|
||||
impl ExecutionPlan for ScannableExec {
|
||||
fn apply_expressions(
|
||||
&self,
|
||||
_f: &mut dyn FnMut(
|
||||
&Arc<dyn datafusion_physical_expr::PhysicalExpr>,
|
||||
) -> datafusion_common::Result<
|
||||
datafusion_common::tree_node::TreeNodeRecursion,
|
||||
>,
|
||||
) -> datafusion_common::Result<datafusion_common::tree_node::TreeNodeRecursion> {
|
||||
Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
|
||||
}
|
||||
fn name(&self) -> &str {
|
||||
"ScannableExec"
|
||||
}
|
||||
|
||||
@@ -507,6 +507,16 @@ impl DisplayAs for DuplicatePairsExec {
|
||||
}
|
||||
|
||||
impl ExecutionPlan for DuplicatePairsExec {
|
||||
fn apply_expressions(
|
||||
&self,
|
||||
_f: &mut dyn FnMut(
|
||||
&Arc<dyn datafusion_physical_expr::PhysicalExpr>,
|
||||
) -> datafusion_common::Result<
|
||||
datafusion_common::tree_node::TreeNodeRecursion,
|
||||
>,
|
||||
) -> datafusion_common::Result<datafusion_common::tree_node::TreeNodeRecursion> {
|
||||
Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
|
||||
}
|
||||
fn name(&self) -> &str {
|
||||
"DuplicatePairsExec"
|
||||
}
|
||||
|
||||
@@ -373,6 +373,16 @@ impl DisplayAs for VectorDedupExec {
|
||||
}
|
||||
|
||||
impl ExecutionPlan for VectorDedupExec {
|
||||
fn apply_expressions(
|
||||
&self,
|
||||
_f: &mut dyn FnMut(
|
||||
&Arc<dyn datafusion_physical_expr::PhysicalExpr>,
|
||||
) -> datafusion_common::Result<
|
||||
datafusion_common::tree_node::TreeNodeRecursion,
|
||||
>,
|
||||
) -> datafusion_common::Result<datafusion_common::tree_node::TreeNodeRecursion> {
|
||||
Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
|
||||
}
|
||||
fn name(&self) -> &str {
|
||||
"VectorDedupExec"
|
||||
}
|
||||
|
||||
+151
-10
@@ -1453,6 +1453,132 @@ mod lsm_tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Another process re-initializing the MemWAL retires this process's
|
||||
/// cached writer: its rows are not read, and the next write opens a
|
||||
/// writer for the new MemWAL.
|
||||
///
|
||||
/// `disable_rte` refuses while the writer holds rows, so the MemWAL index
|
||||
/// is dropped through Lance directly, standing in for a write that lands
|
||||
/// after that check. The writer's rows are lost to the base table either
|
||||
/// way; what this pins is that they are not served from a directory the
|
||||
/// index no longer names.
|
||||
#[tokio::test]
|
||||
async fn lsm_writer_of_a_retired_mem_wal_is_replaced() {
|
||||
use lance::index::DatasetIndexExt;
|
||||
|
||||
let dir = tempdir().unwrap();
|
||||
let uri = dir.path().to_str().unwrap();
|
||||
let conn = connect(uri)
|
||||
.read_consistency_interval(std::time::Duration::ZERO)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let table = conn
|
||||
.create_table("t", id_value_reader(vec![1, 2, 3]))
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
table.set_unenforced_primary_key(["id"]).await.unwrap();
|
||||
table.enable_rte(RTEWriteSpec::unsharded()).await.unwrap();
|
||||
lsm_upsert(&table, vec![4, 5]).await;
|
||||
let retired_dir_id = mem_wal_dir_id(&table).await;
|
||||
|
||||
let mut dataset = lance::Dataset::open(&table.uri().await.unwrap())
|
||||
.await
|
||||
.unwrap();
|
||||
dataset
|
||||
.drop_index(lance_index::mem_wal::MEM_WAL_INDEX_NAME)
|
||||
.await
|
||||
.unwrap();
|
||||
let other = connect(uri)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.open_table("t")
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
other.enable_rte(RTEWriteSpec::unsharded()).await.unwrap();
|
||||
let dir_id = mem_wal_dir_id(&table).await;
|
||||
assert_ne!(dir_id, retired_dir_id);
|
||||
|
||||
let ids = || async {
|
||||
let rows = collect_id_value(table.query().execute().await.unwrap()).await;
|
||||
rows.into_iter().map(|(id, _)| id).collect::<Vec<_>>()
|
||||
};
|
||||
assert_eq!(ids().await, vec![1, 2, 3]);
|
||||
|
||||
lsm_upsert(&table, vec![6]).await;
|
||||
assert_eq!(ids().await, vec![1, 2, 3, 6]);
|
||||
let native = table.as_native().unwrap();
|
||||
assert!(
|
||||
native
|
||||
.dataset
|
||||
.shard_writer()
|
||||
.read_snapshot(&dir_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.is_some(),
|
||||
"the write opened a writer for the new MemWAL"
|
||||
);
|
||||
}
|
||||
|
||||
/// `disable_rte` refuses while the MemWAL holds rows the base table does
|
||||
/// not, since a MemWAL installed later would not read them.
|
||||
#[tokio::test]
|
||||
async fn lsm_disable_is_refused_while_rows_are_unmerged() {
|
||||
let dir = tempdir().unwrap();
|
||||
let table = id_value_table(&dir).await;
|
||||
table.enable_rte(RTEWriteSpec::unsharded()).await.unwrap();
|
||||
lsm_upsert(&table, vec![4, 5]).await;
|
||||
|
||||
let err = table.disable_rte().await.unwrap_err();
|
||||
assert!(
|
||||
err.to_string()
|
||||
.contains("not been merged into the base table"),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
assert_eq!(
|
||||
table.get_rte_spec().await.unwrap(),
|
||||
Some(RTEWriteSpec::unsharded())
|
||||
);
|
||||
let rows = collect_id_value(table.query().execute().await.unwrap()).await;
|
||||
assert_eq!(
|
||||
rows.into_iter().map(|(id, _)| id).collect::<Vec<_>>(),
|
||||
vec![1, 2, 3, 4, 5]
|
||||
);
|
||||
}
|
||||
|
||||
/// Rows another process's writer has acknowledged but not flushed are in
|
||||
/// its WAL only, and still refuse `disable_rte`.
|
||||
#[tokio::test]
|
||||
async fn lsm_disable_is_refused_while_another_writer_holds_rows() {
|
||||
let dir = tempdir().unwrap();
|
||||
let table = id_value_table(&dir).await;
|
||||
table.enable_rte(RTEWriteSpec::unsharded()).await.unwrap();
|
||||
lsm_upsert(&table, vec![4, 5]).await;
|
||||
|
||||
let other = connect(dir.path().to_str().unwrap())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.open_table("t")
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let err = other.disable_rte().await.unwrap_err();
|
||||
assert!(
|
||||
err.to_string().contains("0 SSTable(s) not merged")
|
||||
&& !err.to_string().contains(" 0 WAL entry(ies) not flushed"),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
let rows = collect_id_value(table.query().execute().await.unwrap()).await;
|
||||
assert_eq!(
|
||||
rows.into_iter().map(|(id, _)| id).collect::<Vec<_>>(),
|
||||
vec![1, 2, 3, 4, 5]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn base_table_adapter_use_rte_false_scans_the_base_only() {
|
||||
use crate::table::datafusion::BaseTableAdapter;
|
||||
@@ -1728,9 +1854,10 @@ mod lsm_tests {
|
||||
|
||||
/// A spec is replaced by unsetting it and setting the new one.
|
||||
///
|
||||
/// `disable_rte` drains the open writer, so the rows it held
|
||||
/// survive the replacement, and the writer the next write opens builds its
|
||||
/// MemTables from the new set.
|
||||
/// The writer the next write opens builds its MemTables from the new set.
|
||||
/// Unsetting is refused while the MemWAL holds unmerged rows (see
|
||||
/// `lsm_disable_is_refused_while_rows_are_unmerged`), so nothing is
|
||||
/// written before the replacement.
|
||||
#[tokio::test]
|
||||
async fn lsm_spec_is_replaced_by_unsetting_it_first() {
|
||||
use crate::index::Index;
|
||||
@@ -1745,13 +1872,11 @@ mod lsm_tests {
|
||||
.unwrap();
|
||||
let fts_index = table.list_indices().await.unwrap()[0].name.clone();
|
||||
|
||||
// Installed maintaining nothing, then written to: the writer opens with
|
||||
// no FTS index.
|
||||
// Installed maintaining nothing.
|
||||
table
|
||||
.enable_rte(RTEWriteSpec::unsharded().with_maintained_indexes(Vec::new()))
|
||||
.await
|
||||
.unwrap();
|
||||
upsert_text(&table, vec![(99, "zebra")]).await;
|
||||
|
||||
// Setting over an installed spec is refused; unset, then set.
|
||||
let err = table
|
||||
@@ -2269,7 +2394,7 @@ mod lsm_tests {
|
||||
.unwrap()
|
||||
.dataset
|
||||
.shard_writer()
|
||||
.read_snapshot()
|
||||
.read_snapshot(&mem_wal_dir_id(&table).await)
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("the merge opened a writer");
|
||||
@@ -2288,6 +2413,20 @@ mod lsm_tests {
|
||||
assert_sstables_answer_through_region_bitmap(&table, vec![10, 11]).await;
|
||||
}
|
||||
|
||||
/// The `dir_id` of the table's current MemWAL, read from storage.
|
||||
async fn mem_wal_dir_id(table: &Table) -> String {
|
||||
use lance::dataset::mem_wal::DatasetMemWalExt;
|
||||
|
||||
lance::Dataset::open(&table.uri().await.unwrap())
|
||||
.await
|
||||
.unwrap()
|
||||
.mem_wal_index_details()
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("the table has a MemWAL")
|
||||
.dir_id
|
||||
}
|
||||
|
||||
/// The SSTables each shard's manifest publishes.
|
||||
async fn published_sstables(table: &Table) -> Vec<lance::Dataset> {
|
||||
use lance::dataset::mem_wal::ShardManifestStore;
|
||||
@@ -2296,13 +2435,15 @@ mod lsm_tests {
|
||||
let (store, base) = lance_io::object_store::ObjectStore::from_uri(&uri)
|
||||
.await
|
||||
.unwrap();
|
||||
let mem_wal = std::path::Path::new(&uri).join("_mem_wal");
|
||||
let dir_id = mem_wal_dir_id(table).await;
|
||||
let mem_wal = std::path::Path::new(&uri)
|
||||
.join(lance::dataset::mem_wal::util::mem_wal_dir_name(&dir_id));
|
||||
let mut sstables = Vec::new();
|
||||
for shard in std::fs::read_dir(&mem_wal).unwrap() {
|
||||
let Ok(shard_id) = shard.unwrap().file_name().to_string_lossy().parse() else {
|
||||
continue;
|
||||
};
|
||||
let manifest = ShardManifestStore::new(store.clone(), &base, shard_id, 2)
|
||||
let manifest = ShardManifestStore::new(store.clone(), &base, &dir_id, shard_id, 2)
|
||||
.latest()
|
||||
.await
|
||||
.unwrap()
|
||||
@@ -2417,7 +2558,7 @@ mod lsm_tests {
|
||||
.unwrap()
|
||||
.dataset
|
||||
.shard_writer()
|
||||
.read_snapshot()
|
||||
.read_snapshot(&mem_wal_dir_id(&table).await)
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("the merge opened a writer");
|
||||
|
||||
@@ -30,8 +30,8 @@ use datafusion_physical_plan::SendableRecordBatchStream;
|
||||
use futures::TryStreamExt;
|
||||
use lance::Dataset;
|
||||
use lance::dataset::mem_wal::{
|
||||
DatasetMemWalExt, ShardWriter, ShardWriterConfig, evaluate_sharding_spec,
|
||||
validate_maintained_indexes_with,
|
||||
DatasetMemWalExt, ShardManifestStore, ShardWriter, ShardWriterConfig, WalTailer,
|
||||
evaluate_sharding_spec, validate_maintained_indexes_with,
|
||||
};
|
||||
use lance::index::DatasetIndexExt;
|
||||
use lance_core::datatypes::Schema as LanceSchema;
|
||||
@@ -197,7 +197,9 @@ fn index_name_list(indices: &[IndexConfig]) -> String {
|
||||
/// Remove the [`RTEWriteSpec`] from the table by dropping the MemWAL index.
|
||||
///
|
||||
/// Any cached shard writers are drained and closed first. Errors if no spec is
|
||||
/// currently set.
|
||||
/// currently set, or if the MemWAL still holds rows that have not been merged
|
||||
/// into the base table: dropping the index would orphan them, since a MemWAL
|
||||
/// installed later gets a new directory.
|
||||
#[allow(clippy::redundant_pub_crate)]
|
||||
pub(crate) async fn disable_rte(table: &NativeTable) -> Result<()> {
|
||||
table.dataset.ensure_mutable()?;
|
||||
@@ -214,6 +216,9 @@ pub(crate) async fn disable_rte(table: &NativeTable) -> Result<()> {
|
||||
table.dataset.shard_writer().drain_and_close().await?;
|
||||
|
||||
let mut dataset = (*table.dataset.get().await?).clone();
|
||||
if let Some(details) = dataset.mem_wal_index_details().await? {
|
||||
ensure_mem_wal_merged(&dataset, &details).await?;
|
||||
}
|
||||
dataset
|
||||
.drop_index(lance_index::mem_wal::MEM_WAL_INDEX_NAME)
|
||||
.await?;
|
||||
@@ -221,6 +226,67 @@ pub(crate) async fn disable_rte(table: &NativeTable) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Error if any shard of the MemWAL holds rows the base table does not: an
|
||||
/// SSTable past the shard's compaction watermark, or a WAL entry past the
|
||||
/// position its flushed SSTables cover.
|
||||
///
|
||||
/// Reading the WAL tip catches rows a live writer in another process has
|
||||
/// acknowledged but not flushed. A write that lands after this check is not
|
||||
/// caught.
|
||||
async fn ensure_mem_wal_merged(dataset: &Dataset, details: &MemWalIndexDetails) -> Result<()> {
|
||||
let store = dataset.object_store(None).await?;
|
||||
let base_path = dataset.branch_location().path;
|
||||
let scan_batch_size = ShardWriterConfig::default().manifest_scan_batch_size;
|
||||
|
||||
let mut pending = Vec::new();
|
||||
for shard_id in dataset.list_mem_wal_latest_shard_ids().await? {
|
||||
// First occurrence per shard, as Lance's `compacted_generation_for_shard`.
|
||||
let compacted = details
|
||||
.compacted_sstables
|
||||
.iter()
|
||||
.find(|entry| entry.shard_id == shard_id)
|
||||
.map_or(0, |entry| entry.generation);
|
||||
let manifest = ShardManifestStore::new(
|
||||
store.clone(),
|
||||
&base_path,
|
||||
&details.dir_id,
|
||||
shard_id,
|
||||
scan_batch_size,
|
||||
)
|
||||
.latest()
|
||||
.await?;
|
||||
let (unmerged, flushed_through) = manifest.map_or((0, 0), |manifest| {
|
||||
let unmerged = manifest
|
||||
.sstables
|
||||
.iter()
|
||||
.filter(|sstable| sstable.generation > compacted)
|
||||
.count();
|
||||
(unmerged, manifest.replay_after_wal_entry_position)
|
||||
});
|
||||
let wal_tip = WalTailer::new(store.clone(), base_path.clone(), &details.dir_id, shard_id)
|
||||
.next_position()
|
||||
.await?
|
||||
.saturating_sub(1);
|
||||
let unflushed = wal_tip.saturating_sub(flushed_through);
|
||||
if unmerged > 0 || unflushed > 0 {
|
||||
pending.push(format!(
|
||||
"shard {shard_id}: {unmerged} SSTable(s) not merged, {unflushed} WAL entry(ies) not flushed"
|
||||
));
|
||||
}
|
||||
}
|
||||
if pending.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
Err(Error::InvalidInput {
|
||||
message: format!(
|
||||
"disable_rte: the RTE holds rows that have not been merged into the base \
|
||||
table and would be lost if it were removed ({}); disable_rte is only \
|
||||
allowed before any rows have been written through the RTE",
|
||||
pending.join("; ")
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// get_rte_spec
|
||||
// =============================================================================
|
||||
@@ -351,12 +417,17 @@ impl std::fmt::Debug for ShardWriterCache {
|
||||
}
|
||||
|
||||
struct ShardWriterEntry {
|
||||
/// The `dir_id` of the MemWAL the writer was opened for. Another process
|
||||
/// can drop and re-initialize the MemWAL under this one, leaving the cached
|
||||
/// writer bound to a directory the index no longer names.
|
||||
dir_id: String,
|
||||
inner: RwLock<Option<ShardWriter>>,
|
||||
}
|
||||
|
||||
impl ShardWriterEntry {
|
||||
fn new(writer: ShardWriter) -> Self {
|
||||
Self {
|
||||
dir_id: writer.config().dir_id.clone(),
|
||||
inner: RwLock::new(Some(writer)),
|
||||
}
|
||||
}
|
||||
@@ -420,37 +491,57 @@ impl ShardWriterCache {
|
||||
/// Return the cached writer, opening one for `shard_id` with `config` if
|
||||
/// the slot is empty. Errors if a writer is already open for a *different*
|
||||
/// shard — the caller must close it first.
|
||||
///
|
||||
/// A cached writer opened for a MemWAL other than `dir_id` is retired and
|
||||
/// replaced: the MemWAL it wrote to was dropped by another process, so its
|
||||
/// rows are no longer part of the table and its commits would be fenced.
|
||||
async fn writer_for_shard(
|
||||
&self,
|
||||
dataset: &Dataset,
|
||||
dir_id: &str,
|
||||
shard_id: Uuid,
|
||||
config: ShardWriterConfig,
|
||||
) -> Result<Arc<ShardWriterEntry>> {
|
||||
{
|
||||
let guard = self.slot.read().await;
|
||||
if let Some((cached, entry)) = guard.as_ref() {
|
||||
if let Some((cached, entry)) = guard.as_ref()
|
||||
&& entry.dir_id == dir_id
|
||||
{
|
||||
check_shard_match(*cached, shard_id)?;
|
||||
return Ok(entry.clone());
|
||||
}
|
||||
}
|
||||
let mut guard = self.slot.write().await;
|
||||
// Re-check: another caller may have opened the writer meanwhile.
|
||||
if let Some((cached, entry)) = guard.as_ref() {
|
||||
if let Some((cached, entry)) = guard.as_ref()
|
||||
&& entry.dir_id == dir_id
|
||||
{
|
||||
check_shard_match(*cached, shard_id)?;
|
||||
return Ok(entry.clone());
|
||||
}
|
||||
let writer = dataset
|
||||
let retired = guard.take();
|
||||
let opened = dataset
|
||||
.mem_wal_writer(shard_id, config)
|
||||
.await
|
||||
.map_err(|e| Error::Runtime {
|
||||
message: format!(
|
||||
"merge_insert: failed to open RTE shard writer for shard {}: {}",
|
||||
shard_id, e
|
||||
),
|
||||
})?;
|
||||
let entry = Arc::new(ShardWriterEntry::new(writer));
|
||||
*guard = Some((shard_id, entry.clone()));
|
||||
Ok(entry)
|
||||
.map(|writer| Arc::new(ShardWriterEntry::new(writer)));
|
||||
if let Ok(entry) = &opened {
|
||||
*guard = Some((shard_id, entry.clone()));
|
||||
}
|
||||
drop(guard);
|
||||
if let Some((retired_shard, retired)) = retired
|
||||
&& let Err(error) = retired.close().await
|
||||
{
|
||||
log::warn!(
|
||||
"the MemWAL was re-initialized, and the writer this process held for \
|
||||
shard {retired_shard} of the retired MemWAL could not be closed: {error}"
|
||||
);
|
||||
}
|
||||
opened.map_err(|e| Error::Runtime {
|
||||
message: format!(
|
||||
"merge_insert: failed to open RTE shard writer for shard {}: {}",
|
||||
shard_id, e
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
/// Install `dataset`'s maintained index set on the cached writer, if one is
|
||||
@@ -505,10 +596,12 @@ impl ShardWriterCache {
|
||||
/// Snapshot the cached writer's shard for the RTE read path: its shard id,
|
||||
/// authoritative in-memory manifest, and active + frozen memtable refs.
|
||||
/// Returns `None` when no writer is currently cached (e.g. nothing has been
|
||||
/// written this session, or the writer was closed).
|
||||
/// written this session, or the writer was closed), or when the cached
|
||||
/// writer belongs to a MemWAL other than `dir_id`.
|
||||
#[allow(clippy::redundant_pub_crate)]
|
||||
pub(crate) async fn read_snapshot(
|
||||
&self,
|
||||
dir_id: &str,
|
||||
) -> Result<
|
||||
Option<(
|
||||
Uuid,
|
||||
@@ -523,6 +616,10 @@ impl ShardWriterCache {
|
||||
let Some((shard_id, entry)) = cached else {
|
||||
return Ok(None);
|
||||
};
|
||||
// A writer for a retired MemWAL holds no rows of the current one.
|
||||
if entry.dir_id != dir_id {
|
||||
return Ok(None);
|
||||
}
|
||||
// Capture memtables before the manifest. If a flush interleaves, dedup
|
||||
// tolerates the same rows appearing in both a memtable and a freshly
|
||||
// flushed generation, but would drop rows present in neither. Manifest
|
||||
@@ -576,6 +673,8 @@ enum LsmMode {
|
||||
#[allow(clippy::redundant_pub_crate)]
|
||||
pub(crate) struct LsmPlan {
|
||||
mode: LsmMode,
|
||||
/// The `dir_id` of the MemWAL the write targets.
|
||||
dir_id: String,
|
||||
writer_config_defaults: HashMap<String, String>,
|
||||
}
|
||||
|
||||
@@ -644,6 +743,7 @@ pub(crate) async fn lsm_dispatch_decision(
|
||||
let mode = resolve_lsm_mode(&details)?;
|
||||
Ok(LsmDispatch::Lsm(LsmPlan {
|
||||
mode,
|
||||
dir_id: details.dir_id,
|
||||
writer_config_defaults: details.writer_config_defaults,
|
||||
}))
|
||||
}
|
||||
@@ -745,7 +845,7 @@ pub(crate) async fn execute_lsm_merge_insert(
|
||||
let writer = table
|
||||
.dataset
|
||||
.shard_writer()
|
||||
.writer_for_shard(dataset.as_ref(), shard_id, config)
|
||||
.writer_for_shard(dataset.as_ref(), &plan.dir_id, shard_id, config)
|
||||
.await?;
|
||||
writer.put(batches).await?;
|
||||
|
||||
|
||||
@@ -27,7 +27,7 @@ use datafusion_physical_plan::PhysicalExpr;
|
||||
use datafusion_physical_plan::projection::ProjectionExec;
|
||||
use datafusion_physical_plan::repartition::RepartitionExec;
|
||||
use datafusion_physical_plan::union::UnionExec;
|
||||
use datafusion_physical_plan::{ExecutionPlan, with_new_children_if_necessary};
|
||||
use datafusion_physical_plan::{ExecutionPlan, replace_children_if_necessary};
|
||||
use lance::dataset::mem_wal::DatasetMemWalExt;
|
||||
use lance::dataset::scanner::DatasetRecordBatchStream;
|
||||
use lance::dataset::scanner::Scanner;
|
||||
@@ -740,7 +740,7 @@ fn replace_ann_branches(
|
||||
.cloned()
|
||||
.map(|child| replace_ann_branches(child, internal_ann_plans, normalized_l2_indices))
|
||||
.collect::<Result<Vec<_>>>()?;
|
||||
Ok(with_new_children_if_necessary(public_plan, children)?)
|
||||
Ok(replace_children_if_necessary(public_plan, children)?)
|
||||
}
|
||||
|
||||
fn scale_distance_column(
|
||||
|
||||
@@ -120,12 +120,11 @@ pub(super) async fn create_lsm_plan(
|
||||
} else {
|
||||
plain_plan(
|
||||
&ds_ref,
|
||||
&details,
|
||||
&query,
|
||||
pk_columns.clone(),
|
||||
snapshots,
|
||||
in_memory,
|
||||
limit,
|
||||
offset,
|
||||
)
|
||||
.await?
|
||||
};
|
||||
@@ -304,8 +303,13 @@ async fn build_read_context(
|
||||
|
||||
let mut snapshots: Vec<ShardSnapshot> = Vec::new();
|
||||
for shard_id in shard_ids {
|
||||
let manifest_store =
|
||||
ShardManifestStore::new(store.clone(), &base_path, shard_id, scan_batch_size);
|
||||
let manifest_store = ShardManifestStore::new(
|
||||
store.clone(),
|
||||
&base_path,
|
||||
&details.dir_id,
|
||||
shard_id,
|
||||
scan_batch_size,
|
||||
);
|
||||
if let Some(manifest) = manifest_store.latest().await? {
|
||||
snapshots.push(snapshot_from_manifest(shard_id, &manifest, &exclude));
|
||||
}
|
||||
@@ -324,8 +328,11 @@ async fn build_read_context(
|
||||
// Override the active shard with the cached writer's in-memory view.
|
||||
let mut in_memory: HashMap<Uuid, InMemoryMemTables> = HashMap::new();
|
||||
if !wal_only
|
||||
&& let Some((shard_id, manifest, memtables)) =
|
||||
table.dataset.shard_writer().read_snapshot().await?
|
||||
&& let Some((shard_id, manifest, memtables)) = table
|
||||
.dataset
|
||||
.shard_writer()
|
||||
.read_snapshot(&details.dir_id)
|
||||
.await?
|
||||
{
|
||||
if let Some(manifest) = manifest {
|
||||
let snapshot = snapshot_from_manifest(shard_id, &manifest, &exclude);
|
||||
@@ -384,12 +391,14 @@ fn as_i64(value: Option<usize>) -> Option<i64> {
|
||||
/// scanner's `create_plan`.
|
||||
fn base_scanner(
|
||||
dataset: &Dataset,
|
||||
dir_id: &str,
|
||||
query: &VectorQueryRequest,
|
||||
pk_columns: Vec<String>,
|
||||
snapshots: Vec<ShardSnapshot>,
|
||||
in_memory: HashMap<Uuid, InMemoryMemTables>,
|
||||
) -> Result<LsmScanner> {
|
||||
let mut scanner = LsmScanner::new(Arc::new(dataset.clone()), snapshots, pk_columns);
|
||||
let mut scanner = LsmScanner::new(Arc::new(dataset.clone()), snapshots, pk_columns)
|
||||
.with_mem_wal_dir_id(dir_id);
|
||||
for (shard_id, memtables) in in_memory {
|
||||
scanner = scanner.with_in_memory_memtables(shard_id, memtables);
|
||||
}
|
||||
@@ -435,15 +444,21 @@ fn enrich_filter_error(error: DataFusionError, schema: &ArrowSchema, context: &s
|
||||
/// The plain scan applies limit and offset inside the planner.
|
||||
async fn plain_plan(
|
||||
dataset: &Dataset,
|
||||
details: &MemWalIndexDetails,
|
||||
query: &VectorQueryRequest,
|
||||
pk_columns: Vec<String>,
|
||||
snapshots: Vec<ShardSnapshot>,
|
||||
in_memory: HashMap<Uuid, InMemoryMemTables>,
|
||||
limit: Option<usize>,
|
||||
offset: Option<usize>,
|
||||
) -> Result<Arc<dyn ExecutionPlan>> {
|
||||
let scanner = base_scanner(dataset, query, pk_columns, snapshots, in_memory)?
|
||||
.limit(as_i64(limit), as_i64(offset))?;
|
||||
let scanner = base_scanner(
|
||||
dataset,
|
||||
&details.dir_id,
|
||||
query,
|
||||
pk_columns,
|
||||
snapshots,
|
||||
in_memory,
|
||||
)?
|
||||
.limit(as_i64(query.base.limit), as_i64(query.base.offset))?;
|
||||
Ok(scanner.create_plan().await?)
|
||||
}
|
||||
|
||||
@@ -491,10 +506,17 @@ async fn fts_plan(
|
||||
});
|
||||
}
|
||||
|
||||
let scanner = base_scanner(dataset, query, pk_columns, snapshots, in_memory)?
|
||||
.with_overfetch_factor(LSM_OVERFETCH_FACTOR)
|
||||
.full_text_search(fts)?
|
||||
.limit(as_i64(limit), as_i64(offset))?;
|
||||
let scanner = base_scanner(
|
||||
dataset,
|
||||
&details.dir_id,
|
||||
query,
|
||||
pk_columns,
|
||||
snapshots,
|
||||
in_memory,
|
||||
)?
|
||||
.with_overfetch_factor(LSM_OVERFETCH_FACTOR)
|
||||
.full_text_search(fts)?
|
||||
.limit(as_i64(limit), as_i64(offset))?;
|
||||
Ok(scanner.create_plan().await?)
|
||||
}
|
||||
|
||||
@@ -740,10 +762,17 @@ async fn vector_plan(
|
||||
// Guard `k` to at least 1 so a degenerate `limit(0)` is trimmed by `limit`
|
||||
// below rather than rejected by `nearest`.
|
||||
let k = limit.unwrap_or(DEFAULT_TOP_K).max(1);
|
||||
let mut scanner = base_scanner(dataset, query, pk_columns, snapshots, in_memory)?
|
||||
.with_overfetch_factor(LSM_OVERFETCH_FACTOR)
|
||||
.nearest(&column, query_vector.as_ref(), k)?
|
||||
.distance_metric(distance_type.into());
|
||||
let mut scanner = base_scanner(
|
||||
dataset,
|
||||
&details.dir_id,
|
||||
query,
|
||||
pk_columns,
|
||||
snapshots,
|
||||
in_memory,
|
||||
)?
|
||||
.with_overfetch_factor(LSM_OVERFETCH_FACTOR)
|
||||
.nearest(&column, query_vector.as_ref(), k)?
|
||||
.distance_metric(distance_type.into());
|
||||
if let Some(nprobes) = query.nprobes {
|
||||
scanner = if query.maximum_nprobes == Some(0) {
|
||||
scanner.minimum_nprobes(nprobes)
|
||||
|
||||
@@ -1805,8 +1805,8 @@ mod tests {
|
||||
assert!(matches!(err, Error::NotSupported { .. }));
|
||||
}
|
||||
|
||||
/// After catch-up activation and unset, no spec remains but the catch-up
|
||||
/// flag still marks retained SSTable rows; refresh refuses on the flag.
|
||||
/// Rows written through the RTE keep the spec installed, since
|
||||
/// `disable_rte` refuses while they are unmerged, so refresh refuses.
|
||||
#[tokio::test]
|
||||
#[cfg_attr(
|
||||
not(feature = "experimental-functions"),
|
||||
@@ -1849,7 +1849,8 @@ mod tests {
|
||||
)))
|
||||
.await
|
||||
.unwrap();
|
||||
table.disable_rte().await.unwrap();
|
||||
// The RTE holds the written rows, so unsetting it is refused.
|
||||
table.disable_rte().await.unwrap_err();
|
||||
super::super::computed_columns::add_foreign_kind(&table, "doubled", "sql").await;
|
||||
|
||||
let err = table.refresh_column("doubled").await.unwrap_err();
|
||||
|
||||
Reference in New Issue
Block a user