Compare commits

..
Author SHA1 Message Date
lancedb automation 3e4be3c4b5 chore: update lance dependency to v4.0.0-beta.7 2026-03-04 02:16:34 +00:00
26 changed files with 217 additions and 1358 deletions
Generated
+67 -49
View File
@@ -3088,8 +3088,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]] [[package]]
name = "fsst" name = "fsst"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow-array", "arrow-array",
"rand 0.9.2", "rand 0.9.2",
@@ -4260,8 +4260,8 @@ dependencies = [
[[package]] [[package]]
name = "lance" name = "lance"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow", "arrow",
"arrow-arith", "arrow-arith",
@@ -4315,7 +4315,7 @@ dependencies = [
"semver", "semver",
"serde", "serde",
"serde_json", "serde_json",
"snafu", "snafu 0.9.0",
"tantivy", "tantivy",
"tokio", "tokio",
"tokio-stream", "tokio-stream",
@@ -4327,8 +4327,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-arrow" name = "lance-arrow"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow-array", "arrow-array",
"arrow-buffer", "arrow-buffer",
@@ -4338,6 +4338,7 @@ dependencies = [
"arrow-schema", "arrow-schema",
"arrow-select", "arrow-select",
"bytes", "bytes",
"futures",
"getrandom 0.2.16", "getrandom 0.2.16",
"half", "half",
"jsonb", "jsonb",
@@ -4347,8 +4348,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-bitpacking" name = "lance-bitpacking"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrayref", "arrayref",
"paste", "paste",
@@ -4357,8 +4358,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-core" name = "lance-core"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow-array", "arrow-array",
"arrow-buffer", "arrow-buffer",
@@ -4384,7 +4385,7 @@ dependencies = [
"rand 0.9.2", "rand 0.9.2",
"roaring", "roaring",
"serde_json", "serde_json",
"snafu", "snafu 0.9.0",
"tempfile", "tempfile",
"tokio", "tokio",
"tokio-stream", "tokio-stream",
@@ -4395,8 +4396,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-datafusion" name = "lance-datafusion"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow", "arrow",
"arrow-array", "arrow-array",
@@ -4419,15 +4420,15 @@ dependencies = [
"pin-project", "pin-project",
"prost", "prost",
"prost-build", "prost-build",
"snafu", "snafu 0.9.0",
"tokio", "tokio",
"tracing", "tracing",
] ]
[[package]] [[package]]
name = "lance-datagen" name = "lance-datagen"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow", "arrow",
"arrow-array", "arrow-array",
@@ -4445,8 +4446,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-encoding" name = "lance-encoding"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow-arith", "arrow-arith",
"arrow-array", "arrow-array",
@@ -4473,7 +4474,7 @@ dependencies = [
"prost-build", "prost-build",
"prost-types", "prost-types",
"rand 0.9.2", "rand 0.9.2",
"snafu", "snafu 0.9.0",
"strum", "strum",
"tokio", "tokio",
"tracing", "tracing",
@@ -4483,8 +4484,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-file" name = "lance-file"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow-arith", "arrow-arith",
"arrow-array", "arrow-array",
@@ -4509,15 +4510,15 @@ dependencies = [
"prost", "prost",
"prost-build", "prost-build",
"prost-types", "prost-types",
"snafu", "snafu 0.9.0",
"tokio", "tokio",
"tracing", "tracing",
] ]
[[package]] [[package]]
name = "lance-index" name = "lance-index"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow", "arrow",
"arrow-arith", "arrow-arith",
@@ -4569,7 +4570,7 @@ dependencies = [
"serde", "serde",
"serde_json", "serde_json",
"smallvec", "smallvec",
"snafu", "snafu 0.9.0",
"tantivy", "tantivy",
"tempfile", "tempfile",
"tokio", "tokio",
@@ -4580,8 +4581,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-io" name = "lance-io"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow", "arrow",
"arrow-arith", "arrow-arith",
@@ -4613,7 +4614,7 @@ dependencies = [
"prost", "prost",
"rand 0.9.2", "rand 0.9.2",
"serde", "serde",
"snafu", "snafu 0.9.0",
"tempfile", "tempfile",
"tokio", "tokio",
"tracing", "tracing",
@@ -4622,8 +4623,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-linalg" name = "lance-linalg"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow-array", "arrow-array",
"arrow-buffer", "arrow-buffer",
@@ -4639,21 +4640,21 @@ dependencies = [
[[package]] [[package]]
name = "lance-namespace" name = "lance-namespace"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow", "arrow",
"async-trait", "async-trait",
"bytes", "bytes",
"lance-core", "lance-core",
"lance-namespace-reqwest-client", "lance-namespace-reqwest-client",
"snafu", "snafu 0.9.0",
] ]
[[package]] [[package]]
name = "lance-namespace-impls" name = "lance-namespace-impls"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow", "arrow",
"arrow-ipc", "arrow-ipc",
@@ -4675,7 +4676,7 @@ dependencies = [
"reqwest", "reqwest",
"serde", "serde",
"serde_json", "serde_json",
"snafu", "snafu 0.9.0",
"tokio", "tokio",
"tower", "tower",
"tower-http 0.5.2", "tower-http 0.5.2",
@@ -4697,8 +4698,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-table" name = "lance-table"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow", "arrow",
"arrow-array", "arrow-array",
@@ -4728,7 +4729,7 @@ dependencies = [
"semver", "semver",
"serde", "serde",
"serde_json", "serde_json",
"snafu", "snafu 0.9.0",
"tokio", "tokio",
"tracing", "tracing",
"url", "url",
@@ -4737,8 +4738,8 @@ dependencies = [
[[package]] [[package]]
name = "lance-testing" name = "lance-testing"
version = "3.0.0-rc.2" version = "4.0.0-beta.7"
source = "git+https://github.com/lance-format/lance.git?tag=v3.0.0-rc.2#3fb3e705b8a25ab1bb0fc9e1e0158e8a13356181" source = "git+https://github.com/lance-format/lance.git?tag=v4.0.0-beta.7#e1e5689e9c1e58df75cfdc64e8f029004e98ec71"
dependencies = [ dependencies = [
"arrow-array", "arrow-array",
"arrow-schema", "arrow-schema",
@@ -4819,7 +4820,7 @@ dependencies = [
"serde", "serde",
"serde_json", "serde_json",
"serde_with", "serde_with",
"snafu", "snafu 0.8.9",
"tempfile", "tempfile",
"test-log", "test-log",
"tokenizers", "tokenizers",
@@ -4855,21 +4856,17 @@ version = "0.30.0-beta.3"
dependencies = [ dependencies = [
"arrow", "arrow",
"async-trait", "async-trait",
"bytes",
"env_logger", "env_logger",
"futures", "futures",
"lance-core", "lance-core",
"lance-io", "lance-io",
"lance-namespace", "lance-namespace",
"lance-namespace-impls",
"lancedb", "lancedb",
"pin-project", "pin-project",
"pyo3", "pyo3",
"pyo3-async-runtimes", "pyo3-async-runtimes",
"pyo3-build-config", "pyo3-build-config",
"serde", "snafu 0.8.9",
"serde_json",
"snafu",
"tokio", "tokio",
] ]
@@ -7781,7 +7778,16 @@ version = "0.8.9"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e84b3f4eacbf3a1ce05eac6763b4d629d60cbc94d632e4092c54ade71f1e1a2" checksum = "6e84b3f4eacbf3a1ce05eac6763b4d629d60cbc94d632e4092c54ade71f1e1a2"
dependencies = [ dependencies = [
"snafu-derive", "snafu-derive 0.8.9",
]
[[package]]
name = "snafu"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d1d4bced6a69f90b2056c03dcff2c4737f98d6fb9e0853493996e1d253ca29c6"
dependencies = [
"snafu-derive 0.9.0",
] ]
[[package]] [[package]]
@@ -7796,6 +7802,18 @@ dependencies = [
"syn 2.0.114", "syn 2.0.114",
] ]
[[package]]
name = "snafu-derive"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "54254b8531cafa275c5e096f62d48c81435d1015405a91198ddb11e967301d40"
dependencies = [
"heck 0.4.1",
"proc-macro2",
"quote",
"syn 2.0.114",
]
[[package]] [[package]]
name = "socket2" name = "socket2"
version = "0.5.10" version = "0.5.10"
+14 -14
View File
@@ -15,20 +15,20 @@ categories = ["database-implementations"]
rust-version = "1.91.0" rust-version = "1.91.0"
[workspace.dependencies] [workspace.dependencies]
lance = { "version" = "=3.0.0-rc.2", default-features = false, "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance = { "version" = "=4.0.0-beta.7", default-features = false, "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-core = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-datagen = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-file = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=3.0.0-rc.2", default-features = false, "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-io = { "version" = "=4.0.0-beta.7", default-features = false, "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-index = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-linalg = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-namespace = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=3.0.0-rc.2", default-features = false, "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-namespace-impls = { "version" = "=4.0.0-beta.7", default-features = false, "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-table = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-testing = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-datafusion = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-encoding = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=3.0.0-rc.2", "tag" = "v3.0.0-rc.2", "git" = "https://github.com/lance-format/lance.git" } lance-arrow = { "version" = "=4.0.0-beta.7", "tag" = "v4.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
ahash = "0.8" ahash = "0.8"
# Note that this one does not include pyarrow # Note that this one does not include pyarrow
arrow = { version = "57.2", optional = false } arrow = { version = "57.2", optional = false }
+3 -10
View File
@@ -52,21 +52,14 @@ plugins:
options: options:
docstring_style: numpy docstring_style: numpy
heading_level: 3 heading_level: 3
show_source: true
show_symbol_type_in_heading: true
show_signature_annotations: true show_signature_annotations: true
show_root_heading: true show_root_heading: true
show_docstring_examples: true
show_docstring_attributes: false
show_docstring_other_parameters: true
show_symbol_type_heading: true
show_labels: false
show_if_no_docstring: true
show_source: false
members_order: source members_order: source
docstring_section_style: list docstring_section_style: list
signature_crossrefs: true signature_crossrefs: true
separate_signature: true separate_signature: true
filters:
- "!^_"
import: import:
# for cross references # for cross references
- https://arrow.apache.org/docs/objects.inv - https://arrow.apache.org/docs/objects.inv
@@ -120,7 +113,7 @@ markdown_extensions:
emoji_index: !!python/name:material.extensions.emoji.twemoji emoji_index: !!python/name:material.extensions.emoji.twemoji
emoji_generator: !!python/name:material.extensions.emoji.to_svg emoji_generator: !!python/name:material.extensions.emoji.to_svg
- markdown.extensions.toc: - markdown.extensions.toc:
toc_depth: 4 toc_depth: 3
permalink: true permalink: true
permalink_title: Anchor link to this section permalink_title: Anchor link to this section
+1 -1
View File
@@ -28,7 +28,7 @@
<properties> <properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<arrow.version>15.0.0</arrow.version> <arrow.version>15.0.0</arrow.version>
<lance-core.version>3.1.0-beta.2</lance-core.version> <lance-core.version>4.0.0-beta.7</lance-core.version>
<spotless.skip>false</spotless.skip> <spotless.skip>false</spotless.skip>
<spotless.version>2.30.0</spotless.version> <spotless.version>2.30.0</spotless.version>
<spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version> <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
+1 -1
View File
@@ -1,5 +1,5 @@
[tool.bumpversion] [tool.bumpversion]
current_version = "0.30.0-beta.4" current_version = "0.30.0-beta.3"
parse = """(?x) parse = """(?x)
(?P<major>0|[1-9]\\d*)\\. (?P<major>0|[1-9]\\d*)\\.
(?P<minor>0|[1-9]\\d*)\\. (?P<minor>0|[1-9]\\d*)\\.
+1 -5
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "lancedb-python" name = "lancedb-python"
version = "0.30.0-beta.4" version = "0.30.0-beta.3"
edition.workspace = true edition.workspace = true
description = "Python bindings for LanceDB" description = "Python bindings for LanceDB"
license.workspace = true license.workspace = true
@@ -16,11 +16,9 @@ crate-type = ["cdylib"]
[dependencies] [dependencies]
arrow = { version = "57.2", features = ["pyarrow"] } arrow = { version = "57.2", features = ["pyarrow"] }
async-trait = "0.1" async-trait = "0.1"
bytes = "1"
lancedb = { path = "../rust/lancedb", default-features = false } lancedb = { path = "../rust/lancedb", default-features = false }
lance-core.workspace = true lance-core.workspace = true
lance-namespace.workspace = true lance-namespace.workspace = true
lance-namespace-impls.workspace = true
lance-io.workspace = true lance-io.workspace = true
env_logger.workspace = true env_logger.workspace = true
pyo3 = { version = "0.26", features = ["extension-module", "abi3-py39"] } pyo3 = { version = "0.26", features = ["extension-module", "abi3-py39"] }
@@ -30,8 +28,6 @@ pyo3-async-runtimes = { version = "0.26", features = [
] } ] }
pin-project = "1.1.5" pin-project = "1.1.5"
futures.workspace = true futures.workspace = true
serde = "1"
serde_json = "1"
snafu.workspace = true snafu.workspace = true
tokio = { version = "1.40", features = ["sync"] } tokio = { version = "1.40", features = ["sync"] }
+3 -3
View File
@@ -45,7 +45,7 @@ repository = "https://github.com/lancedb/lancedb"
[project.optional-dependencies] [project.optional-dependencies]
pylance = [ pylance = [
"pylance>=4.0.0b7", "pylance>=1.0.0b14",
] ]
tests = [ tests = [
"aiohttp", "aiohttp",
@@ -59,9 +59,9 @@ tests = [
"polars>=0.19, <=1.3.0", "polars>=0.19, <=1.3.0",
"tantivy", "tantivy",
"pyarrow-stubs", "pyarrow-stubs",
"pylance>=4.0.0b7", "pylance>=1.0.0b14,<3.0.0",
"requests", "requests",
"datafusion>=52,<53", "datafusion<52",
] ]
dev = [ dev = [
"ruff", "ruff",
+1 -8
View File
@@ -8,7 +8,7 @@ from abc import abstractmethod
from datetime import timedelta from datetime import timedelta
from pathlib import Path from pathlib import Path
import sys import sys
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Literal, Optional, Union from typing import TYPE_CHECKING, Dict, Iterable, List, Literal, Optional, Union
if sys.version_info >= (3, 12): if sys.version_info >= (3, 12):
from typing import override from typing import override
@@ -1541,8 +1541,6 @@ class AsyncConnection(object):
storage_options_provider: Optional["StorageOptionsProvider"] = None, storage_options_provider: Optional["StorageOptionsProvider"] = None,
index_cache_size: Optional[int] = None, index_cache_size: Optional[int] = None,
location: Optional[str] = None, location: Optional[str] = None,
namespace_client: Optional[Any] = None,
managed_versioning: Optional[bool] = None,
) -> AsyncTable: ) -> AsyncTable:
"""Open a Lance Table in the database. """Open a Lance Table in the database.
@@ -1575,9 +1573,6 @@ class AsyncConnection(object):
The explicit location (URI) of the table. If provided, the table will be The explicit location (URI) of the table. If provided, the table will be
opened from this location instead of deriving it from the database URI opened from this location instead of deriving it from the database URI
and table name. and table name.
managed_versioning: bool, optional
Whether managed versioning is enabled for this table. If provided,
avoids a redundant describe_table call when namespace_client is set.
Returns Returns
------- -------
@@ -1592,8 +1587,6 @@ class AsyncConnection(object):
storage_options_provider=storage_options_provider, storage_options_provider=storage_options_provider,
index_cache_size=index_cache_size, index_cache_size=index_cache_size,
location=location, location=location,
namespace_client=namespace_client,
managed_versioning=managed_versioning,
) )
return AsyncTable(table) return AsyncTable(table)
+27 -51
View File
@@ -12,7 +12,7 @@ from __future__ import annotations
import asyncio import asyncio
import sys import sys
from typing import Any, Dict, Iterable, List, Optional, Union from typing import Dict, Iterable, List, Optional, Union
if sys.version_info >= (3, 12): if sys.version_info >= (3, 12):
from typing import override from typing import override
@@ -240,7 +240,7 @@ class LanceNamespaceDBConnection(DBConnection):
session : Optional[Session] session : Optional[Session]
A session to use for this connection A session to use for this connection
""" """
self._namespace_client = namespace self._ns = namespace
self.read_consistency_interval = read_consistency_interval self.read_consistency_interval = read_consistency_interval
self.storage_options = storage_options or {} self.storage_options = storage_options or {}
self.session = session self.session = session
@@ -269,7 +269,7 @@ class LanceNamespaceDBConnection(DBConnection):
if namespace is None: if namespace is None:
namespace = [] namespace = []
request = ListTablesRequest(id=namespace, page_token=page_token, limit=limit) request = ListTablesRequest(id=namespace, page_token=page_token, limit=limit)
response = self._namespace_client.list_tables(request) response = self._ns.list_tables(request)
return response.tables if response.tables else [] return response.tables if response.tables else []
@override @override
@@ -309,9 +309,7 @@ class LanceNamespaceDBConnection(DBConnection):
# Try to describe the table first to see if it exists # Try to describe the table first to see if it exists
try: try:
describe_request = DescribeTableRequest(id=table_id) describe_request = DescribeTableRequest(id=table_id)
describe_response = self._namespace_client.describe_table( describe_response = self._ns.describe_table(describe_request)
describe_request
)
location = describe_response.location location = describe_response.location
namespace_storage_options = describe_response.storage_options namespace_storage_options = describe_response.storage_options
except Exception: except Exception:
@@ -325,7 +323,7 @@ class LanceNamespaceDBConnection(DBConnection):
location=None, location=None,
properties=self.storage_options if self.storage_options else None, properties=self.storage_options if self.storage_options else None,
) )
declare_response = self._namespace_client.declare_table(declare_request) declare_response = self._ns.declare_table(declare_request)
if not declare_response.location: if not declare_response.location:
raise ValueError( raise ValueError(
@@ -355,7 +353,7 @@ class LanceNamespaceDBConnection(DBConnection):
# Only create if namespace returned storage_options (not None) # Only create if namespace returned storage_options (not None)
if storage_options_provider is None and namespace_storage_options is not None: if storage_options_provider is None and namespace_storage_options is not None:
storage_options_provider = LanceNamespaceStorageOptionsProvider( storage_options_provider = LanceNamespaceStorageOptionsProvider(
namespace=self._namespace_client, namespace=self._ns,
table_id=table_id, table_id=table_id,
) )
@@ -373,7 +371,6 @@ class LanceNamespaceDBConnection(DBConnection):
storage_options=merged_storage_options, storage_options=merged_storage_options,
storage_options_provider=storage_options_provider, storage_options_provider=storage_options_provider,
location=location, location=location,
namespace_client=self._namespace_client,
) )
return tbl return tbl
@@ -392,7 +389,7 @@ class LanceNamespaceDBConnection(DBConnection):
namespace = [] namespace = []
table_id = namespace + [name] table_id = namespace + [name]
request = DescribeTableRequest(id=table_id) request = DescribeTableRequest(id=table_id)
response = self._namespace_client.describe_table(request) response = self._ns.describe_table(request)
# Merge storage options: self.storage_options < user options < namespace options # Merge storage options: self.storage_options < user options < namespace options
merged_storage_options = dict(self.storage_options) merged_storage_options = dict(self.storage_options)
@@ -405,14 +402,10 @@ class LanceNamespaceDBConnection(DBConnection):
# Only create if namespace returned storage_options (not None) # Only create if namespace returned storage_options (not None)
if storage_options_provider is None and response.storage_options is not None: if storage_options_provider is None and response.storage_options is not None:
storage_options_provider = LanceNamespaceStorageOptionsProvider( storage_options_provider = LanceNamespaceStorageOptionsProvider(
namespace=self._namespace_client, namespace=self._ns,
table_id=table_id, table_id=table_id,
) )
# Pass managed_versioning to avoid redundant describe_table call in Rust.
# Convert None to False since we already have the answer from describe_table.
managed_versioning = response.managed_versioning is True
return self._lance_table_from_uri( return self._lance_table_from_uri(
name, name,
response.location, response.location,
@@ -420,8 +413,6 @@ class LanceNamespaceDBConnection(DBConnection):
storage_options=merged_storage_options, storage_options=merged_storage_options,
storage_options_provider=storage_options_provider, storage_options_provider=storage_options_provider,
index_cache_size=index_cache_size, index_cache_size=index_cache_size,
namespace_client=self._namespace_client,
managed_versioning=managed_versioning,
) )
@override @override
@@ -431,7 +422,7 @@ class LanceNamespaceDBConnection(DBConnection):
namespace = [] namespace = []
table_id = namespace + [name] table_id = namespace + [name]
request = DropTableRequest(id=table_id) request = DropTableRequest(id=table_id)
self._namespace_client.drop_table(request) self._ns.drop_table(request)
@override @override
def rename_table( def rename_table(
@@ -493,7 +484,7 @@ class LanceNamespaceDBConnection(DBConnection):
request = ListNamespacesRequest( request = ListNamespacesRequest(
id=namespace, page_token=page_token, limit=limit id=namespace, page_token=page_token, limit=limit
) )
response = self._namespace_client.list_namespaces(request) response = self._ns.list_namespaces(request)
return ListNamespacesResponse( return ListNamespacesResponse(
namespaces=response.namespaces if response.namespaces else [], namespaces=response.namespaces if response.namespaces else [],
page_token=response.page_token, page_token=response.page_token,
@@ -529,7 +520,7 @@ class LanceNamespaceDBConnection(DBConnection):
mode=_normalize_create_namespace_mode(mode), mode=_normalize_create_namespace_mode(mode),
properties=properties, properties=properties,
) )
response = self._namespace_client.create_namespace(request) response = self._ns.create_namespace(request)
return CreateNamespaceResponse( return CreateNamespaceResponse(
properties=response.properties if hasattr(response, "properties") else None properties=response.properties if hasattr(response, "properties") else None
) )
@@ -564,7 +555,7 @@ class LanceNamespaceDBConnection(DBConnection):
mode=_normalize_drop_namespace_mode(mode), mode=_normalize_drop_namespace_mode(mode),
behavior=_normalize_drop_namespace_behavior(behavior), behavior=_normalize_drop_namespace_behavior(behavior),
) )
response = self._namespace_client.drop_namespace(request) response = self._ns.drop_namespace(request)
return DropNamespaceResponse( return DropNamespaceResponse(
properties=( properties=(
response.properties if hasattr(response, "properties") else None response.properties if hasattr(response, "properties") else None
@@ -590,7 +581,7 @@ class LanceNamespaceDBConnection(DBConnection):
Response containing the namespace properties. Response containing the namespace properties.
""" """
request = DescribeNamespaceRequest(id=namespace) request = DescribeNamespaceRequest(id=namespace)
response = self._namespace_client.describe_namespace(request) response = self._ns.describe_namespace(request)
return DescribeNamespaceResponse( return DescribeNamespaceResponse(
properties=response.properties if hasattr(response, "properties") else None properties=response.properties if hasattr(response, "properties") else None
) )
@@ -624,7 +615,7 @@ class LanceNamespaceDBConnection(DBConnection):
if namespace is None: if namespace is None:
namespace = [] namespace = []
request = ListTablesRequest(id=namespace, page_token=page_token, limit=limit) request = ListTablesRequest(id=namespace, page_token=page_token, limit=limit)
response = self._namespace_client.list_tables(request) response = self._ns.list_tables(request)
return ListTablesResponse( return ListTablesResponse(
tables=response.tables if response.tables else [], tables=response.tables if response.tables else [],
page_token=response.page_token, page_token=response.page_token,
@@ -639,8 +630,6 @@ class LanceNamespaceDBConnection(DBConnection):
storage_options: Optional[Dict[str, str]] = None, storage_options: Optional[Dict[str, str]] = None,
storage_options_provider: Optional[StorageOptionsProvider] = None, storage_options_provider: Optional[StorageOptionsProvider] = None,
index_cache_size: Optional[int] = None, index_cache_size: Optional[int] = None,
namespace_client: Optional[Any] = None,
managed_versioning: Optional[bool] = None,
) -> LanceTable: ) -> LanceTable:
# Open a table directly from a URI using the location parameter # Open a table directly from a URI using the location parameter
# Note: storage_options should already be merged by the caller # Note: storage_options should already be merged by the caller
@@ -654,8 +643,6 @@ class LanceNamespaceDBConnection(DBConnection):
) )
# Open the table using the temporary connection with the location parameter # Open the table using the temporary connection with the location parameter
# Pass namespace_client to enable managed versioning support
# Pass managed_versioning to avoid redundant describe_table call
return LanceTable.open( return LanceTable.open(
temp_conn, temp_conn,
name, name,
@@ -664,8 +651,6 @@ class LanceNamespaceDBConnection(DBConnection):
storage_options_provider=storage_options_provider, storage_options_provider=storage_options_provider,
index_cache_size=index_cache_size, index_cache_size=index_cache_size,
location=table_uri, location=table_uri,
namespace_client=namespace_client,
managed_versioning=managed_versioning,
) )
@@ -700,7 +685,7 @@ class AsyncLanceNamespaceDBConnection:
session : Optional[Session] session : Optional[Session]
A session to use for this connection A session to use for this connection
""" """
self._namespace_client = namespace self._ns = namespace
self.read_consistency_interval = read_consistency_interval self.read_consistency_interval = read_consistency_interval
self.storage_options = storage_options or {} self.storage_options = storage_options or {}
self.session = session self.session = session
@@ -728,7 +713,7 @@ class AsyncLanceNamespaceDBConnection:
if namespace is None: if namespace is None:
namespace = [] namespace = []
request = ListTablesRequest(id=namespace, page_token=page_token, limit=limit) request = ListTablesRequest(id=namespace, page_token=page_token, limit=limit)
response = self._namespace_client.list_tables(request) response = self._ns.list_tables(request)
return response.tables if response.tables else [] return response.tables if response.tables else []
async def create_table( async def create_table(
@@ -765,9 +750,7 @@ class AsyncLanceNamespaceDBConnection:
# Try to describe the table first to see if it exists # Try to describe the table first to see if it exists
try: try:
describe_request = DescribeTableRequest(id=table_id) describe_request = DescribeTableRequest(id=table_id)
describe_response = self._namespace_client.describe_table( describe_response = self._ns.describe_table(describe_request)
describe_request
)
location = describe_response.location location = describe_response.location
namespace_storage_options = describe_response.storage_options namespace_storage_options = describe_response.storage_options
except Exception: except Exception:
@@ -781,7 +764,7 @@ class AsyncLanceNamespaceDBConnection:
location=None, location=None,
properties=self.storage_options if self.storage_options else None, properties=self.storage_options if self.storage_options else None,
) )
declare_response = self._namespace_client.declare_table(declare_request) declare_response = self._ns.declare_table(declare_request)
if not declare_response.location: if not declare_response.location:
raise ValueError( raise ValueError(
@@ -814,7 +797,7 @@ class AsyncLanceNamespaceDBConnection:
and namespace_storage_options is not None and namespace_storage_options is not None
): ):
provider = LanceNamespaceStorageOptionsProvider( provider = LanceNamespaceStorageOptionsProvider(
namespace=self._namespace_client, namespace=self._ns,
table_id=table_id, table_id=table_id,
) )
else: else:
@@ -834,7 +817,6 @@ class AsyncLanceNamespaceDBConnection:
storage_options=merged_storage_options, storage_options=merged_storage_options,
storage_options_provider=provider, storage_options_provider=provider,
location=location, location=location,
namespace_client=self._namespace_client,
) )
lance_table = await asyncio.to_thread(_create_table) lance_table = await asyncio.to_thread(_create_table)
@@ -855,7 +837,7 @@ class AsyncLanceNamespaceDBConnection:
namespace = [] namespace = []
table_id = namespace + [name] table_id = namespace + [name]
request = DescribeTableRequest(id=table_id) request = DescribeTableRequest(id=table_id)
response = self._namespace_client.describe_table(request) response = self._ns.describe_table(request)
# Merge storage options: self.storage_options < user options < namespace options # Merge storage options: self.storage_options < user options < namespace options
merged_storage_options = dict(self.storage_options) merged_storage_options = dict(self.storage_options)
@@ -867,14 +849,10 @@ class AsyncLanceNamespaceDBConnection:
# Create a storage options provider if not provided by user # Create a storage options provider if not provided by user
if storage_options_provider is None and response.storage_options is not None: if storage_options_provider is None and response.storage_options is not None:
storage_options_provider = LanceNamespaceStorageOptionsProvider( storage_options_provider = LanceNamespaceStorageOptionsProvider(
namespace=self._namespace_client, namespace=self._ns,
table_id=table_id, table_id=table_id,
) )
# Capture managed_versioning from describe response.
# Convert None to False since we already have the answer from describe_table.
managed_versioning = response.managed_versioning is True
# Open table in a thread # Open table in a thread
def _open_table(): def _open_table():
temp_conn = LanceDBConnection( temp_conn = LanceDBConnection(
@@ -892,8 +870,6 @@ class AsyncLanceNamespaceDBConnection:
storage_options_provider=storage_options_provider, storage_options_provider=storage_options_provider,
index_cache_size=index_cache_size, index_cache_size=index_cache_size,
location=response.location, location=response.location,
namespace_client=self._namespace_client,
managed_versioning=managed_versioning,
) )
lance_table = await asyncio.to_thread(_open_table) lance_table = await asyncio.to_thread(_open_table)
@@ -905,7 +881,7 @@ class AsyncLanceNamespaceDBConnection:
namespace = [] namespace = []
table_id = namespace + [name] table_id = namespace + [name]
request = DropTableRequest(id=table_id) request = DropTableRequest(id=table_id)
self._namespace_client.drop_table(request) self._ns.drop_table(request)
async def rename_table( async def rename_table(
self, self,
@@ -967,7 +943,7 @@ class AsyncLanceNamespaceDBConnection:
request = ListNamespacesRequest( request = ListNamespacesRequest(
id=namespace, page_token=page_token, limit=limit id=namespace, page_token=page_token, limit=limit
) )
response = self._namespace_client.list_namespaces(request) response = self._ns.list_namespaces(request)
return ListNamespacesResponse( return ListNamespacesResponse(
namespaces=response.namespaces if response.namespaces else [], namespaces=response.namespaces if response.namespaces else [],
page_token=response.page_token, page_token=response.page_token,
@@ -1002,7 +978,7 @@ class AsyncLanceNamespaceDBConnection:
mode=_normalize_create_namespace_mode(mode), mode=_normalize_create_namespace_mode(mode),
properties=properties, properties=properties,
) )
response = self._namespace_client.create_namespace(request) response = self._ns.create_namespace(request)
return CreateNamespaceResponse( return CreateNamespaceResponse(
properties=response.properties if hasattr(response, "properties") else None properties=response.properties if hasattr(response, "properties") else None
) )
@@ -1036,7 +1012,7 @@ class AsyncLanceNamespaceDBConnection:
mode=_normalize_drop_namespace_mode(mode), mode=_normalize_drop_namespace_mode(mode),
behavior=_normalize_drop_namespace_behavior(behavior), behavior=_normalize_drop_namespace_behavior(behavior),
) )
response = self._namespace_client.drop_namespace(request) response = self._ns.drop_namespace(request)
return DropNamespaceResponse( return DropNamespaceResponse(
properties=( properties=(
response.properties if hasattr(response, "properties") else None response.properties if hasattr(response, "properties") else None
@@ -1063,7 +1039,7 @@ class AsyncLanceNamespaceDBConnection:
Response containing the namespace properties. Response containing the namespace properties.
""" """
request = DescribeNamespaceRequest(id=namespace) request = DescribeNamespaceRequest(id=namespace)
response = self._namespace_client.describe_namespace(request) response = self._ns.describe_namespace(request)
return DescribeNamespaceResponse( return DescribeNamespaceResponse(
properties=response.properties if hasattr(response, "properties") else None properties=response.properties if hasattr(response, "properties") else None
) )
@@ -1096,7 +1072,7 @@ class AsyncLanceNamespaceDBConnection:
if namespace is None: if namespace is None:
namespace = [] namespace = []
request = ListTablesRequest(id=namespace, page_token=page_token, limit=limit) request = ListTablesRequest(id=namespace, page_token=page_token, limit=limit)
response = self._namespace_client.list_tables(request) response = self._ns.list_tables(request)
return ListTablesResponse( return ListTablesResponse(
tables=response.tables if response.tables else [], tables=response.tables if response.tables else [],
page_token=response.page_token, page_token=response.page_token,
-21
View File
@@ -1746,8 +1746,6 @@ class LanceTable(Table):
storage_options_provider: Optional["StorageOptionsProvider"] = None, storage_options_provider: Optional["StorageOptionsProvider"] = None,
index_cache_size: Optional[int] = None, index_cache_size: Optional[int] = None,
location: Optional[str] = None, location: Optional[str] = None,
namespace_client: Optional[Any] = None,
managed_versioning: Optional[bool] = None,
_async: AsyncTable = None, _async: AsyncTable = None,
): ):
if namespace is None: if namespace is None:
@@ -1755,7 +1753,6 @@ class LanceTable(Table):
self._conn = connection self._conn = connection
self._namespace = namespace self._namespace = namespace
self._location = location # Store location for use in _dataset_path self._location = location # Store location for use in _dataset_path
self._namespace_client = namespace_client
if _async is not None: if _async is not None:
self._table = _async self._table = _async
else: else:
@@ -1767,8 +1764,6 @@ class LanceTable(Table):
storage_options_provider=storage_options_provider, storage_options_provider=storage_options_provider,
index_cache_size=index_cache_size, index_cache_size=index_cache_size,
location=location, location=location,
namespace_client=namespace_client,
managed_versioning=managed_versioning,
) )
) )
@@ -1811,8 +1806,6 @@ class LanceTable(Table):
storage_options_provider: Optional["StorageOptionsProvider"] = None, storage_options_provider: Optional["StorageOptionsProvider"] = None,
index_cache_size: Optional[int] = None, index_cache_size: Optional[int] = None,
location: Optional[str] = None, location: Optional[str] = None,
namespace_client: Optional[Any] = None,
managed_versioning: Optional[bool] = None,
): ):
if namespace is None: if namespace is None:
namespace = [] namespace = []
@@ -1824,8 +1817,6 @@ class LanceTable(Table):
storage_options_provider=storage_options_provider, storage_options_provider=storage_options_provider,
index_cache_size=index_cache_size, index_cache_size=index_cache_size,
location=location, location=location,
namespace_client=namespace_client,
managed_versioning=managed_versioning,
) )
# check the dataset exists # check the dataset exists
@@ -1857,16 +1848,6 @@ class LanceTable(Table):
"Please install with `pip install pylance`." "Please install with `pip install pylance`."
) )
if self._namespace_client is not None:
table_id = self._namespace + [self.name]
return lance.dataset(
version=self.version,
storage_options=self._conn.storage_options,
namespace=self._namespace_client,
table_id=table_id,
**kwargs,
)
return lance.dataset( return lance.dataset(
self._dataset_path, self._dataset_path,
version=self.version, version=self.version,
@@ -2732,7 +2713,6 @@ class LanceTable(Table):
data_storage_version: Optional[str] = None, data_storage_version: Optional[str] = None,
enable_v2_manifest_paths: Optional[bool] = None, enable_v2_manifest_paths: Optional[bool] = None,
location: Optional[str] = None, location: Optional[str] = None,
namespace_client: Optional[Any] = None,
): ):
""" """
Create a new table. Create a new table.
@@ -2793,7 +2773,6 @@ class LanceTable(Table):
self._conn = db self._conn = db
self._namespace = namespace self._namespace = namespace
self._location = location self._location = location
self._namespace_client = namespace_client
if data_storage_version is not None: if data_storage_version is not None:
warnings.warn( warnings.warn(
-18
View File
@@ -326,24 +326,6 @@ def test_add_struct(mem_db: DBConnection):
table = mem_db.create_table("test2", schema=schema) table = mem_db.create_table("test2", schema=schema)
table.add(data) table.add(data)
struct_type = pa.struct(
[
("b", pa.int64()),
("a", pa.int64()),
]
)
expected = pa.table(
{
"s_list": [
[
pa.scalar({"b": 1, "a": 2}, type=struct_type),
pa.scalar({"b": 4, "a": None}, type=struct_type),
]
],
}
)
assert table.to_arrow() == expected
def test_add_subschema(mem_db: DBConnection): def test_add_subschema(mem_db: DBConnection):
schema = pa.schema( schema = pa.schema(
+3 -21
View File
@@ -17,8 +17,7 @@ use pyo3::{
use pyo3_async_runtimes::tokio::future_into_py; use pyo3_async_runtimes::tokio::future_into_py;
use crate::{ use crate::{
error::PythonErrorExt, namespace::extract_namespace_arc, error::PythonErrorExt, storage_options::py_object_to_storage_options_provider, table::Table,
storage_options::py_object_to_storage_options_provider, table::Table,
}; };
#[pyclass] #[pyclass]
@@ -183,8 +182,7 @@ impl Connection {
}) })
} }
#[allow(clippy::too_many_arguments)] #[pyo3(signature = (name, namespace=vec![], storage_options = None, storage_options_provider=None, index_cache_size = None, location=None))]
#[pyo3(signature = (name, namespace=vec![], storage_options = None, storage_options_provider=None, index_cache_size = None, location=None, namespace_client=None, managed_versioning=None))]
pub fn open_table( pub fn open_table(
self_: PyRef<'_, Self>, self_: PyRef<'_, Self>,
name: String, name: String,
@@ -193,13 +191,11 @@ impl Connection {
storage_options_provider: Option<Py<PyAny>>, storage_options_provider: Option<Py<PyAny>>,
index_cache_size: Option<u32>, index_cache_size: Option<u32>,
location: Option<String>, location: Option<String>,
namespace_client: Option<Py<PyAny>>,
managed_versioning: Option<bool>,
) -> PyResult<Bound<'_, PyAny>> { ) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone(); let inner = self_.get_inner()?.clone();
let mut builder = inner.open_table(name); let mut builder = inner.open_table(name);
builder = builder.namespace(namespace.clone()); builder = builder.namespace(namespace);
if let Some(storage_options) = storage_options { if let Some(storage_options) = storage_options {
builder = builder.storage_options(storage_options); builder = builder.storage_options(storage_options);
} }
@@ -213,20 +209,6 @@ impl Connection {
if let Some(location) = location { if let Some(location) = location {
builder = builder.location(location); builder = builder.location(location);
} }
// Extract namespace client from Python object if provided
let ns_client = if let Some(ns_obj) = namespace_client {
let py = self_.py();
Some(extract_namespace_arc(py, ns_obj)?)
} else {
None
};
if let Some(ns_client) = ns_client {
builder = builder.namespace_client(ns_client);
}
// Pass managed_versioning if provided to avoid redundant describe_table call
if let Some(enabled) = managed_versioning {
builder = builder.managed_versioning(enabled);
}
future_into_py(self_.py(), async move { future_into_py(self_.py(), async move {
let table = builder.execute().await.infer_error()?; let table = builder.execute().await.infer_error()?;
-1
View File
@@ -23,7 +23,6 @@ pub mod connection;
pub mod error; pub mod error;
pub mod header; pub mod header;
pub mod index; pub mod index;
pub mod namespace;
pub mod permutation; pub mod permutation;
pub mod query; pub mod query;
pub mod session; pub mod session;
-746
View File
@@ -1,746 +0,0 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
//! Namespace utilities for Python bindings
use std::collections::HashMap;
use std::sync::Arc;
use async_trait::async_trait;
use bytes::Bytes;
use lance_namespace::LanceNamespace as LanceNamespaceTrait;
use lance_namespace::models::*;
use pyo3::prelude::*;
use pyo3::types::PyDict;
/// Wrapper that allows any Python object implementing LanceNamespace protocol
/// to be used as a Rust LanceNamespace.
///
/// This is similar to PyLanceNamespace in lance's Python bindings - it wraps a Python
/// object and calls back into Python when namespace methods are invoked.
pub struct PyLanceNamespace {
py_namespace: Arc<Py<PyAny>>,
namespace_id: String,
}
impl PyLanceNamespace {
/// Create a new PyLanceNamespace wrapper around a Python namespace object.
pub fn new(_py: Python<'_>, py_namespace: &Bound<'_, PyAny>) -> PyResult<Self> {
let namespace_id = py_namespace
.call_method0("namespace_id")?
.extract::<String>()?;
Ok(Self {
py_namespace: Arc::new(py_namespace.clone().unbind()),
namespace_id,
})
}
/// Create an Arc<dyn LanceNamespace> from a Python namespace object.
pub fn create_arc(
py: Python<'_>,
py_namespace: &Bound<'_, PyAny>,
) -> PyResult<Arc<dyn LanceNamespaceTrait>> {
let wrapper = Self::new(py, py_namespace)?;
Ok(Arc::new(wrapper))
}
}
impl std::fmt::Debug for PyLanceNamespace {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "PyLanceNamespace {{ id: {} }}", self.namespace_id)
}
}
/// Get or create the DictWithModelDump class in Python.
/// This class acts like a dict but also has model_dump() method.
/// This allows it to work with both:
/// - depythonize (which expects a dict/Mapping)
/// - Python code that calls .model_dump() (like DirectoryNamespace wrapper)
fn get_dict_with_model_dump_class(py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
// Use a module-level cache via __builtins__
let builtins = py.import("builtins")?;
if builtins.hasattr("_DictWithModelDump")? {
return builtins.getattr("_DictWithModelDump");
}
// Create the class using exec
let locals = PyDict::new(py);
py.run(
c"class DictWithModelDump(dict):
def model_dump(self):
return dict(self)",
None,
Some(&locals),
)?;
let class = locals.get_item("DictWithModelDump")?.ok_or_else(|| {
pyo3::exceptions::PyRuntimeError::new_err("Failed to create DictWithModelDump class")
})?;
// Cache it
builtins.setattr("_DictWithModelDump", &class)?;
Ok(class)
}
/// Helper to call a Python namespace method with JSON serialization.
/// For methods that take a request and return a response.
/// Uses DictWithModelDump to pass a dict that also has model_dump() method,
/// making it compatible with both depythonize and Python wrappers.
async fn call_py_method<Req, Resp>(
py_namespace: Arc<Py<PyAny>>,
method_name: &'static str,
request: Req,
) -> lance_core::Result<Resp>
where
Req: serde::Serialize + Send + 'static,
Resp: serde::de::DeserializeOwned + Send + 'static,
{
let request_json = serde_json::to_string(&request).map_err(|e| {
lance_core::Error::io(
format!("Failed to serialize request for {}: {}", method_name, e),
Default::default(),
)
})?;
let response_json = tokio::task::spawn_blocking(move || {
Python::attach(|py| {
let json_module = py.import("json")?;
let request_dict = json_module.call_method1("loads", (&request_json,))?;
// Wrap dict in DictWithModelDump so it works with both depythonize and .model_dump()
let dict_class = get_dict_with_model_dump_class(py)?;
let request_arg = dict_class.call1((request_dict,))?;
// Call the Python method
let result = py_namespace.call_method1(py, method_name, (request_arg,))?;
// Convert response to dict, then to JSON
// Pydantic models have model_dump() method
let result_dict = if result.bind(py).hasattr("model_dump")? {
result.call_method0(py, "model_dump")?
} else {
result
};
let response_json: String = json_module
.call_method1("dumps", (result_dict,))?
.extract()?;
Ok::<_, PyErr>(response_json)
})
})
.await
.map_err(|e| {
lance_core::Error::io(
format!("Task join error for {}: {}", method_name, e),
Default::default(),
)
})?
.map_err(|e: PyErr| {
lance_core::Error::io(
format!("Python error in {}: {}", method_name, e),
Default::default(),
)
})?;
serde_json::from_str(&response_json).map_err(|e| {
lance_core::Error::io(
format!("Failed to deserialize response from {}: {}", method_name, e),
Default::default(),
)
})
}
/// Helper for methods that return () on success
async fn call_py_method_unit<Req>(
py_namespace: Arc<Py<PyAny>>,
method_name: &'static str,
request: Req,
) -> lance_core::Result<()>
where
Req: serde::Serialize + Send + 'static,
{
let request_json = serde_json::to_string(&request).map_err(|e| {
lance_core::Error::io(
format!("Failed to serialize request for {}: {}", method_name, e),
Default::default(),
)
})?;
tokio::task::spawn_blocking(move || {
Python::attach(|py| {
let json_module = py.import("json")?;
let request_dict = json_module.call_method1("loads", (&request_json,))?;
// Wrap dict in DictWithModelDump
let dict_class = get_dict_with_model_dump_class(py)?;
let request_arg = dict_class.call1((request_dict,))?;
// Call the Python method
py_namespace.call_method1(py, method_name, (request_arg,))?;
Ok::<_, PyErr>(())
})
})
.await
.map_err(|e| {
lance_core::Error::io(
format!("Task join error for {}: {}", method_name, e),
Default::default(),
)
})?
.map_err(|e: PyErr| {
lance_core::Error::io(
format!("Python error in {}: {}", method_name, e),
Default::default(),
)
})
}
/// Helper for methods that return a primitive type
async fn call_py_method_primitive<Req, Resp>(
py_namespace: Arc<Py<PyAny>>,
method_name: &'static str,
request: Req,
) -> lance_core::Result<Resp>
where
Req: serde::Serialize + Send + 'static,
Resp: for<'py> pyo3::FromPyObject<'py> + Send + 'static,
{
let request_json = serde_json::to_string(&request).map_err(|e| {
lance_core::Error::io(
format!("Failed to serialize request for {}: {}", method_name, e),
Default::default(),
)
})?;
tokio::task::spawn_blocking(move || {
Python::attach(|py| {
let json_module = py.import("json")?;
let request_dict = json_module.call_method1("loads", (&request_json,))?;
// Wrap dict in DictWithModelDump
let dict_class = get_dict_with_model_dump_class(py)?;
let request_arg = dict_class.call1((request_dict,))?;
// Call the Python method
let result = py_namespace.call_method1(py, method_name, (request_arg,))?;
let value: Resp = result.extract(py)?;
Ok::<_, PyErr>(value)
})
})
.await
.map_err(|e| {
lance_core::Error::io(
format!("Task join error for {}: {}", method_name, e),
Default::default(),
)
})?
.map_err(|e: PyErr| {
lance_core::Error::io(
format!("Python error in {}: {}", method_name, e),
Default::default(),
)
})
}
/// Helper for methods that return Bytes
async fn call_py_method_bytes<Req>(
py_namespace: Arc<Py<PyAny>>,
method_name: &'static str,
request: Req,
) -> lance_core::Result<Bytes>
where
Req: serde::Serialize + Send + 'static,
{
let request_json = serde_json::to_string(&request).map_err(|e| {
lance_core::Error::io(
format!("Failed to serialize request for {}: {}", method_name, e),
Default::default(),
)
})?;
tokio::task::spawn_blocking(move || {
Python::attach(|py| {
let json_module = py.import("json")?;
let request_dict = json_module.call_method1("loads", (&request_json,))?;
// Wrap dict in DictWithModelDump
let dict_class = get_dict_with_model_dump_class(py)?;
let request_arg = dict_class.call1((request_dict,))?;
// Call the Python method
let result = py_namespace.call_method1(py, method_name, (request_arg,))?;
let bytes_data: Vec<u8> = result.extract(py)?;
Ok::<_, PyErr>(Bytes::from(bytes_data))
})
})
.await
.map_err(|e| {
lance_core::Error::io(
format!("Task join error for {}: {}", method_name, e),
Default::default(),
)
})?
.map_err(|e: PyErr| {
lance_core::Error::io(
format!("Python error in {}: {}", method_name, e),
Default::default(),
)
})
}
/// Helper for methods that take request + data and return a response
async fn call_py_method_with_data<Req, Resp>(
py_namespace: Arc<Py<PyAny>>,
method_name: &'static str,
request: Req,
data: Bytes,
) -> lance_core::Result<Resp>
where
Req: serde::Serialize + Send + 'static,
Resp: serde::de::DeserializeOwned + Send + 'static,
{
let request_json = serde_json::to_string(&request).map_err(|e| {
lance_core::Error::io(
format!("Failed to serialize request for {}: {}", method_name, e),
Default::default(),
)
})?;
let response_json = tokio::task::spawn_blocking(move || {
Python::attach(|py| {
let json_module = py.import("json")?;
let request_dict = json_module.call_method1("loads", (&request_json,))?;
// Wrap dict in DictWithModelDump
let dict_class = get_dict_with_model_dump_class(py)?;
let request_arg = dict_class.call1((request_dict,))?;
// Pass request and bytes to Python method
let py_bytes = pyo3::types::PyBytes::new(py, &data);
let result = py_namespace.call_method1(py, method_name, (request_arg, py_bytes))?;
// Convert response dict to JSON
let response_json: String = json_module.call_method1("dumps", (result,))?.extract()?;
Ok::<_, PyErr>(response_json)
})
})
.await
.map_err(|e| {
lance_core::Error::io(
format!("Task join error for {}: {}", method_name, e),
Default::default(),
)
})?
.map_err(|e: PyErr| {
lance_core::Error::io(
format!("Python error in {}: {}", method_name, e),
Default::default(),
)
})?;
serde_json::from_str(&response_json).map_err(|e| {
lance_core::Error::io(
format!("Failed to deserialize response from {}: {}", method_name, e),
Default::default(),
)
})
}
#[async_trait]
impl LanceNamespaceTrait for PyLanceNamespace {
fn namespace_id(&self) -> String {
self.namespace_id.clone()
}
async fn list_namespaces(
&self,
request: ListNamespacesRequest,
) -> lance_core::Result<ListNamespacesResponse> {
call_py_method(self.py_namespace.clone(), "list_namespaces", request).await
}
async fn describe_namespace(
&self,
request: DescribeNamespaceRequest,
) -> lance_core::Result<DescribeNamespaceResponse> {
call_py_method(self.py_namespace.clone(), "describe_namespace", request).await
}
async fn create_namespace(
&self,
request: CreateNamespaceRequest,
) -> lance_core::Result<CreateNamespaceResponse> {
call_py_method(self.py_namespace.clone(), "create_namespace", request).await
}
async fn drop_namespace(
&self,
request: DropNamespaceRequest,
) -> lance_core::Result<DropNamespaceResponse> {
call_py_method(self.py_namespace.clone(), "drop_namespace", request).await
}
async fn namespace_exists(&self, request: NamespaceExistsRequest) -> lance_core::Result<()> {
call_py_method_unit(self.py_namespace.clone(), "namespace_exists", request).await
}
async fn list_tables(
&self,
request: ListTablesRequest,
) -> lance_core::Result<ListTablesResponse> {
call_py_method(self.py_namespace.clone(), "list_tables", request).await
}
async fn describe_table(
&self,
request: DescribeTableRequest,
) -> lance_core::Result<DescribeTableResponse> {
call_py_method(self.py_namespace.clone(), "describe_table", request).await
}
async fn register_table(
&self,
request: RegisterTableRequest,
) -> lance_core::Result<RegisterTableResponse> {
call_py_method(self.py_namespace.clone(), "register_table", request).await
}
async fn table_exists(&self, request: TableExistsRequest) -> lance_core::Result<()> {
call_py_method_unit(self.py_namespace.clone(), "table_exists", request).await
}
async fn drop_table(&self, request: DropTableRequest) -> lance_core::Result<DropTableResponse> {
call_py_method(self.py_namespace.clone(), "drop_table", request).await
}
async fn deregister_table(
&self,
request: DeregisterTableRequest,
) -> lance_core::Result<DeregisterTableResponse> {
call_py_method(self.py_namespace.clone(), "deregister_table", request).await
}
async fn count_table_rows(&self, request: CountTableRowsRequest) -> lance_core::Result<i64> {
call_py_method_primitive(self.py_namespace.clone(), "count_table_rows", request).await
}
async fn create_table(
&self,
request: CreateTableRequest,
request_data: Bytes,
) -> lance_core::Result<CreateTableResponse> {
call_py_method_with_data(
self.py_namespace.clone(),
"create_table",
request,
request_data,
)
.await
}
async fn declare_table(
&self,
request: DeclareTableRequest,
) -> lance_core::Result<DeclareTableResponse> {
call_py_method(self.py_namespace.clone(), "declare_table", request).await
}
async fn insert_into_table(
&self,
request: InsertIntoTableRequest,
request_data: Bytes,
) -> lance_core::Result<InsertIntoTableResponse> {
call_py_method_with_data(
self.py_namespace.clone(),
"insert_into_table",
request,
request_data,
)
.await
}
async fn merge_insert_into_table(
&self,
request: MergeInsertIntoTableRequest,
request_data: Bytes,
) -> lance_core::Result<MergeInsertIntoTableResponse> {
call_py_method_with_data(
self.py_namespace.clone(),
"merge_insert_into_table",
request,
request_data,
)
.await
}
async fn update_table(
&self,
request: UpdateTableRequest,
) -> lance_core::Result<UpdateTableResponse> {
call_py_method(self.py_namespace.clone(), "update_table", request).await
}
async fn delete_from_table(
&self,
request: DeleteFromTableRequest,
) -> lance_core::Result<DeleteFromTableResponse> {
call_py_method(self.py_namespace.clone(), "delete_from_table", request).await
}
async fn query_table(&self, request: QueryTableRequest) -> lance_core::Result<Bytes> {
call_py_method_bytes(self.py_namespace.clone(), "query_table", request).await
}
async fn create_table_index(
&self,
request: CreateTableIndexRequest,
) -> lance_core::Result<CreateTableIndexResponse> {
call_py_method(self.py_namespace.clone(), "create_table_index", request).await
}
async fn list_table_indices(
&self,
request: ListTableIndicesRequest,
) -> lance_core::Result<ListTableIndicesResponse> {
call_py_method(self.py_namespace.clone(), "list_table_indices", request).await
}
async fn describe_table_index_stats(
&self,
request: DescribeTableIndexStatsRequest,
) -> lance_core::Result<DescribeTableIndexStatsResponse> {
call_py_method(
self.py_namespace.clone(),
"describe_table_index_stats",
request,
)
.await
}
async fn describe_transaction(
&self,
request: DescribeTransactionRequest,
) -> lance_core::Result<DescribeTransactionResponse> {
call_py_method(self.py_namespace.clone(), "describe_transaction", request).await
}
async fn alter_transaction(
&self,
request: AlterTransactionRequest,
) -> lance_core::Result<AlterTransactionResponse> {
call_py_method(self.py_namespace.clone(), "alter_transaction", request).await
}
async fn create_table_scalar_index(
&self,
request: CreateTableIndexRequest,
) -> lance_core::Result<CreateTableScalarIndexResponse> {
call_py_method(
self.py_namespace.clone(),
"create_table_scalar_index",
request,
)
.await
}
async fn drop_table_index(
&self,
request: DropTableIndexRequest,
) -> lance_core::Result<DropTableIndexResponse> {
call_py_method(self.py_namespace.clone(), "drop_table_index", request).await
}
async fn list_all_tables(
&self,
request: ListTablesRequest,
) -> lance_core::Result<ListTablesResponse> {
call_py_method(self.py_namespace.clone(), "list_all_tables", request).await
}
async fn restore_table(
&self,
request: RestoreTableRequest,
) -> lance_core::Result<RestoreTableResponse> {
call_py_method(self.py_namespace.clone(), "restore_table", request).await
}
async fn rename_table(
&self,
request: RenameTableRequest,
) -> lance_core::Result<RenameTableResponse> {
call_py_method(self.py_namespace.clone(), "rename_table", request).await
}
async fn list_table_versions(
&self,
request: ListTableVersionsRequest,
) -> lance_core::Result<ListTableVersionsResponse> {
call_py_method(self.py_namespace.clone(), "list_table_versions", request).await
}
async fn create_table_version(
&self,
request: CreateTableVersionRequest,
) -> lance_core::Result<CreateTableVersionResponse> {
call_py_method(self.py_namespace.clone(), "create_table_version", request).await
}
async fn describe_table_version(
&self,
request: DescribeTableVersionRequest,
) -> lance_core::Result<DescribeTableVersionResponse> {
call_py_method(self.py_namespace.clone(), "describe_table_version", request).await
}
async fn batch_delete_table_versions(
&self,
request: BatchDeleteTableVersionsRequest,
) -> lance_core::Result<BatchDeleteTableVersionsResponse> {
call_py_method(
self.py_namespace.clone(),
"batch_delete_table_versions",
request,
)
.await
}
async fn update_table_schema_metadata(
&self,
request: UpdateTableSchemaMetadataRequest,
) -> lance_core::Result<UpdateTableSchemaMetadataResponse> {
call_py_method(
self.py_namespace.clone(),
"update_table_schema_metadata",
request,
)
.await
}
async fn get_table_stats(
&self,
request: GetTableStatsRequest,
) -> lance_core::Result<GetTableStatsResponse> {
call_py_method(self.py_namespace.clone(), "get_table_stats", request).await
}
async fn explain_table_query_plan(
&self,
request: ExplainTableQueryPlanRequest,
) -> lance_core::Result<String> {
call_py_method_primitive(
self.py_namespace.clone(),
"explain_table_query_plan",
request,
)
.await
}
async fn analyze_table_query_plan(
&self,
request: AnalyzeTableQueryPlanRequest,
) -> lance_core::Result<String> {
call_py_method_primitive(
self.py_namespace.clone(),
"analyze_table_query_plan",
request,
)
.await
}
async fn alter_table_add_columns(
&self,
request: AlterTableAddColumnsRequest,
) -> lance_core::Result<AlterTableAddColumnsResponse> {
call_py_method(
self.py_namespace.clone(),
"alter_table_add_columns",
request,
)
.await
}
async fn alter_table_alter_columns(
&self,
request: AlterTableAlterColumnsRequest,
) -> lance_core::Result<AlterTableAlterColumnsResponse> {
call_py_method(
self.py_namespace.clone(),
"alter_table_alter_columns",
request,
)
.await
}
async fn alter_table_drop_columns(
&self,
request: AlterTableDropColumnsRequest,
) -> lance_core::Result<AlterTableDropColumnsResponse> {
call_py_method(
self.py_namespace.clone(),
"alter_table_drop_columns",
request,
)
.await
}
async fn list_table_tags(
&self,
request: ListTableTagsRequest,
) -> lance_core::Result<ListTableTagsResponse> {
call_py_method(self.py_namespace.clone(), "list_table_tags", request).await
}
async fn create_table_tag(
&self,
request: CreateTableTagRequest,
) -> lance_core::Result<CreateTableTagResponse> {
call_py_method(self.py_namespace.clone(), "create_table_tag", request).await
}
async fn delete_table_tag(
&self,
request: DeleteTableTagRequest,
) -> lance_core::Result<DeleteTableTagResponse> {
call_py_method(self.py_namespace.clone(), "delete_table_tag", request).await
}
async fn update_table_tag(
&self,
request: UpdateTableTagRequest,
) -> lance_core::Result<UpdateTableTagResponse> {
call_py_method(self.py_namespace.clone(), "update_table_tag", request).await
}
async fn get_table_tag_version(
&self,
request: GetTableTagVersionRequest,
) -> lance_core::Result<GetTableTagVersionResponse> {
call_py_method(self.py_namespace.clone(), "get_table_tag_version", request).await
}
}
/// Convert Python dict to HashMap<String, String>
#[allow(dead_code)]
fn dict_to_hashmap(dict: &Bound<'_, PyDict>) -> PyResult<HashMap<String, String>> {
let mut map = HashMap::new();
for (key, value) in dict.iter() {
let key_str: String = key.extract()?;
let value_str: String = value.extract()?;
map.insert(key_str, value_str);
}
Ok(map)
}
/// Extract an Arc<dyn LanceNamespace> from a Python namespace object.
///
/// This function wraps any Python namespace object with PyLanceNamespace.
/// The PyLanceNamespace wrapper uses DictWithModelDump to pass requests,
/// which works with both:
/// - Native namespaces (DirectoryNamespace, RestNamespace) that use depythonize (expects dict)
/// - Custom Python implementations that call .model_dump() on the request
pub fn extract_namespace_arc(
py: Python<'_>,
ns: Py<PyAny>,
) -> PyResult<Arc<dyn LanceNamespaceTrait>> {
let ns_ref = ns.bind(py);
PyLanceNamespace::create_arc(py, ns_ref)
}
+16 -21
View File
@@ -66,12 +66,11 @@ impl StorageOptionsProvider for PyStorageOptionsProviderWrapper {
.inner .inner
.bind(py) .bind(py)
.call_method0("fetch_storage_options") .call_method0("fetch_storage_options")
.map_err(|e| lance_core::Error::IO { .map_err(|e| {
source: Box::new(std::io::Error::other(format!( lance_core::Error::io_source(Box::new(std::io::Error::other(format!(
"Failed to call fetch_storage_options: {}", "Failed to call fetch_storage_options: {}",
e e
))), ))))
location: snafu::location!(),
})?; })?;
// If result is None, return None // If result is None, return None
@@ -81,26 +80,25 @@ impl StorageOptionsProvider for PyStorageOptionsProviderWrapper {
// Extract the result dict - should be a flat Map<String, String> // Extract the result dict - should be a flat Map<String, String>
let result_dict = result.downcast::<PyDict>().map_err(|_| { let result_dict = result.downcast::<PyDict>().map_err(|_| {
lance_core::Error::InvalidInput { lance_core::Error::invalid_input(
source: "fetch_storage_options() must return None or a dict of string key-value pairs".into(), "fetch_storage_options() must return None or a dict of string key-value pairs",
location: snafu::location!(), )
}
})?; })?;
// Convert all entries to HashMap<String, String> // Convert all entries to HashMap<String, String>
let mut storage_options = HashMap::new(); let mut storage_options = HashMap::new();
for (key, value) in result_dict.iter() { for (key, value) in result_dict.iter() {
let key_str: String = key.extract().map_err(|e| { let key_str: String = key.extract().map_err(|e| {
lance_core::Error::InvalidInput { lance_core::Error::invalid_input(format!(
source: format!("Storage option key must be a string: {}", e).into(), "Storage option key must be a string: {}",
location: snafu::location!(), e
} ))
})?; })?;
let value_str: String = value.extract().map_err(|e| { let value_str: String = value.extract().map_err(|e| {
lance_core::Error::InvalidInput { lance_core::Error::invalid_input(format!(
source: format!("Storage option value must be a string: {}", e).into(), "Storage option value must be a string: {}",
location: snafu::location!(), e
} ))
})?; })?;
storage_options.insert(key_str, value_str); storage_options.insert(key_str, value_str);
} }
@@ -109,13 +107,10 @@ impl StorageOptionsProvider for PyStorageOptionsProviderWrapper {
}) })
}) })
.await .await
.map_err(|e| lance_core::Error::IO { .map_err(|e| lance_core::Error::io_source(Box::new(std::io::Error::other(format!(
source: Box::new(std::io::Error::other(format!(
"Task join error: {}", "Task join error: {}",
e e
))), )))))?
location: snafu::location!(),
})?
} }
fn provider_id(&self) -> String { fn provider_id(&self) -> String {
+2 -2
View File
@@ -2006,7 +2006,7 @@ requires-dist = [
{ name = "botocore", marker = "extra == 'embeddings'", specifier = ">=1.31.57" }, { name = "botocore", marker = "extra == 'embeddings'", specifier = ">=1.31.57" },
{ name = "cohere", marker = "extra == 'embeddings'" }, { name = "cohere", marker = "extra == 'embeddings'" },
{ name = "colpali-engine", marker = "extra == 'embeddings'", specifier = ">=0.3.10" }, { name = "colpali-engine", marker = "extra == 'embeddings'", specifier = ">=0.3.10" },
{ name = "datafusion", marker = "extra == 'tests'", specifier = "<52" }, { name = "datafusion", marker = "extra == 'tests'" },
{ name = "deprecation" }, { name = "deprecation" },
{ name = "duckdb", marker = "extra == 'tests'" }, { name = "duckdb", marker = "extra == 'tests'" },
{ name = "google-generativeai", marker = "extra == 'embeddings'" }, { name = "google-generativeai", marker = "extra == 'embeddings'" },
@@ -2035,7 +2035,7 @@ requires-dist = [
{ name = "pyarrow-stubs", marker = "extra == 'tests'" }, { name = "pyarrow-stubs", marker = "extra == 'tests'" },
{ name = "pydantic", specifier = ">=1.10" }, { name = "pydantic", specifier = ">=1.10" },
{ name = "pylance", marker = "extra == 'pylance'", specifier = ">=1.0.0b14" }, { name = "pylance", marker = "extra == 'pylance'", specifier = ">=1.0.0b14" },
{ name = "pylance", marker = "extra == 'tests'", specifier = ">=1.0.0b14,<3.0.0" }, { name = "pylance", marker = "extra == 'tests'", specifier = ">=1.0.0b14" },
{ name = "pyright", marker = "extra == 'dev'" }, { name = "pyright", marker = "extra == 'dev'" },
{ name = "pytest", marker = "extra == 'tests'" }, { name = "pytest", marker = "extra == 'tests'" },
{ name = "pytest-asyncio", marker = "extra == 'tests'" }, { name = "pytest-asyncio", marker = "extra == 'tests'" },
+2 -35
View File
@@ -136,7 +136,6 @@ impl OpenTableBuilder {
lance_read_params: None, lance_read_params: None,
location: None, location: None,
namespace_client: None, namespace_client: None,
managed_versioning: None,
}, },
embedding_registry, embedding_registry,
} }
@@ -236,29 +235,6 @@ impl OpenTableBuilder {
self self
} }
/// Set a namespace client for managed versioning support.
///
/// When a namespace client is provided and the table has `managed_versioning` enabled,
/// the table will use the namespace's commit handler to notify the namespace of
/// version changes. This enables features like event emission for table modifications.
pub fn namespace_client(mut self, client: Arc<dyn lance_namespace::LanceNamespace>) -> Self {
self.request.namespace_client = Some(client);
self
}
/// Set whether managed versioning is enabled for this table.
///
/// When set to `Some(true)`, the table will use namespace-managed commits.
/// When set to `Some(false)`, the table will use local commits even if namespace_client is set.
/// When set to `None` (default), the value will be fetched from the namespace if namespace_client is set.
///
/// This is typically set when the caller has already queried the namespace and knows the
/// managed_versioning status, avoiding a redundant describe_table call.
pub fn managed_versioning(mut self, enabled: bool) -> Self {
self.request.managed_versioning = Some(enabled);
self
}
/// Open the table /// Open the table
pub async fn execute(self) -> Result<Table> { pub async fn execute(self) -> Result<Table> {
let table = self.parent.open_table(self.request).await?; let table = self.parent.open_table(self.request).await?;
@@ -318,12 +294,6 @@ impl CloneTableBuilder {
self self
} }
/// Set a namespace client for managed versioning support.
pub fn namespace_client(mut self, client: Arc<dyn lance_namespace::LanceNamespace>) -> Self {
self.request.namespace_client = Some(client);
self
}
/// Execute the clone operation /// Execute the clone operation
pub async fn execute(self) -> Result<Table> { pub async fn execute(self) -> Result<Table> {
let parent = self.parent.clone(); let parent = self.parent.clone();
@@ -596,11 +566,8 @@ pub struct ConnectBuilder {
} }
#[cfg(feature = "remote")] #[cfg(feature = "remote")]
const ENV_VARS_TO_STORAGE_OPTS: [(&str, &str); 3] = [ const ENV_VARS_TO_STORAGE_OPTS: [(&str, &str); 1] =
("AZURE_STORAGE_ACCOUNT_NAME", "azure_storage_account_name"), [("AZURE_STORAGE_ACCOUNT_NAME", "azure_storage_account_name")];
("AZURE_CLIENT_ID", "azure_client_id"),
("AZURE_TENANT_ID", "azure_tenant_id"),
];
impl ConnectBuilder { impl ConnectBuilder {
/// Create a new [`ConnectOptions`] with the given database URI. /// Create a new [`ConnectOptions`] with the given database URI.
-9
View File
@@ -66,10 +66,6 @@ pub struct OpenTableRequest {
/// Optional namespace client for server-side query execution. /// Optional namespace client for server-side query execution.
/// When set, queries will be executed on the namespace server instead of locally. /// When set, queries will be executed on the namespace server instead of locally.
pub namespace_client: Option<Arc<dyn LanceNamespace>>, pub namespace_client: Option<Arc<dyn LanceNamespace>>,
/// Whether managed versioning is enabled for this table.
/// When Some(true), the table will use namespace-managed commits instead of local commits.
/// When None and namespace_client is provided, the value will be fetched from the namespace.
pub managed_versioning: Option<bool>,
} }
impl std::fmt::Debug for OpenTableRequest { impl std::fmt::Debug for OpenTableRequest {
@@ -81,7 +77,6 @@ impl std::fmt::Debug for OpenTableRequest {
.field("lance_read_params", &self.lance_read_params) .field("lance_read_params", &self.lance_read_params)
.field("location", &self.location) .field("location", &self.location)
.field("namespace_client", &self.namespace_client) .field("namespace_client", &self.namespace_client)
.field("managed_versioning", &self.managed_versioning)
.finish() .finish()
} }
} }
@@ -166,9 +161,6 @@ pub struct CloneTableRequest {
/// Whether to perform a shallow clone (true) or deep clone (false). Defaults to true. /// Whether to perform a shallow clone (true) or deep clone (false). Defaults to true.
/// Currently only shallow clone is supported. /// Currently only shallow clone is supported.
pub is_shallow: bool, pub is_shallow: bool,
/// Optional namespace client for managed versioning support.
/// When set, enables the commit handler to track table versions through the namespace.
pub namespace_client: Option<Arc<dyn LanceNamespace>>,
} }
impl CloneTableRequest { impl CloneTableRequest {
@@ -180,7 +172,6 @@ impl CloneTableRequest {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
} }
} }
} }
+1 -19
View File
@@ -669,7 +669,6 @@ impl ListingDatabase {
lance_read_params: None, lance_read_params: None,
location: None, location: None,
namespace_client: None, namespace_client: None,
managed_versioning: None,
}; };
let req = (callback)(req); let req = (callback)(req);
let table = self.open_table(req).await?; let table = self.open_table(req).await?;
@@ -870,7 +869,6 @@ impl Database for ListingDatabase {
Some(write_params), Some(write_params),
self.read_consistency_interval, self.read_consistency_interval,
request.namespace_client, request.namespace_client,
false, // server_side_query_enabled - listing database doesn't support server-side queries
) )
.await .await
{ {
@@ -948,9 +946,7 @@ impl Database for ListingDatabase {
self.store_wrapper.clone(), self.store_wrapper.clone(),
None, None,
self.read_consistency_interval, self.read_consistency_interval,
request.namespace_client, None,
false, // server_side_query_enabled - listing database doesn't support server-side queries
None, // managed_versioning - will be queried if namespace_client is provided
) )
.await?; .await?;
@@ -1026,8 +1022,6 @@ impl Database for ListingDatabase {
Some(read_params), Some(read_params),
self.read_consistency_interval, self.read_consistency_interval,
request.namespace_client, request.namespace_client,
false, // server_side_query_enabled - listing database doesn't support server-side queries
request.managed_versioning, // Pass through managed_versioning from request
) )
.await?, .await?,
); );
@@ -1168,7 +1162,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await .await
.unwrap(); .unwrap();
@@ -1229,7 +1222,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await .await
.unwrap(); .unwrap();
@@ -1289,7 +1281,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await; .await;
@@ -1326,7 +1317,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: false, // Request deep clone is_shallow: false, // Request deep clone
namespace_client: None,
}) })
.await; .await;
@@ -1367,7 +1357,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await; .await;
@@ -1408,7 +1397,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await; .await;
@@ -1428,7 +1416,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await; .await;
@@ -1465,7 +1452,6 @@ mod tests {
source_version: Some(1), source_version: Some(1),
source_tag: Some("v1.0".to_string()), source_tag: Some("v1.0".to_string()),
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await; .await;
@@ -1539,7 +1525,6 @@ mod tests {
source_version: Some(initial_version), source_version: Some(initial_version),
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await .await
.unwrap(); .unwrap();
@@ -1618,7 +1603,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: Some("v1.0".to_string()), source_tag: Some("v1.0".to_string()),
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await .await
.unwrap(); .unwrap();
@@ -1670,7 +1654,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await .await
.unwrap(); .unwrap();
@@ -1763,7 +1746,6 @@ mod tests {
source_version: None, source_version: None,
source_tag: None, source_tag: None,
is_shallow: true, is_shallow: true,
namespace_client: None,
}) })
.await .await
.unwrap(); .unwrap();
+10 -28
View File
@@ -7,7 +7,6 @@ use std::collections::HashMap;
use std::sync::Arc; use std::sync::Arc;
use async_trait::async_trait; use async_trait::async_trait;
use lance::io::commit::namespace_manifest::LanceNamespaceExternalManifestStore;
use lance_io::object_store::{ObjectStoreParams, StorageOptionsAccessor}; use lance_io::object_store::{ObjectStoreParams, StorageOptionsAccessor};
use lance_namespace::{ use lance_namespace::{
LanceNamespace, LanceNamespace,
@@ -19,8 +18,6 @@ use lance_namespace::{
}, },
}; };
use lance_namespace_impls::ConnectBuilder; use lance_namespace_impls::ConnectBuilder;
use lance_table::io::commit::CommitHandler;
use lance_table::io::commit::external_manifest::ExternalManifestCommitHandler;
use crate::database::ReadConsistency; use crate::database::ReadConsistency;
use crate::error::{Error, Result}; use crate::error::{Error, Result};
@@ -208,55 +205,40 @@ impl Database for LanceNamespaceDatabase {
let mut table_id = request.namespace.clone(); let mut table_id = request.namespace.clone();
table_id.push(request.name.clone()); table_id.push(request.name.clone());
// Declare table metadata through lance-namespace before creating data files.
let declare_request = DeclareTableRequest { let declare_request = DeclareTableRequest {
id: Some(table_id.clone()), id: Some(table_id.clone()),
..Default::default() ..Default::default()
}; };
let response = self let declare_response = self
.namespace .namespace
.declare_table(declare_request) .declare_table(declare_request)
.await .await
.map_err(|e| Error::Runtime { .map_err(|e| Error::Runtime {
message: format!("Failed to declare table: {}", e), message: format!("Failed to declare table: {}", e),
})?; })?;
let location = declare_response.location.ok_or_else(|| Error::Runtime {
let location = response.location.ok_or_else(|| Error::Runtime {
message: "Table location is missing from declare_table response".to_string(), message: "Table location is missing from declare_table response".to_string(),
})?; })?;
// Use storage options from response, fall back to self.storage_options // Use storage options from response, fall back to self.storage_options
let initial_storage_options = response let initial_storage_options = declare_response
.storage_options .storage_options
.or_else(|| Some(self.storage_options.clone())) .or_else(|| Some(self.storage_options.clone()))
.filter(|o| !o.is_empty()); .filter(|o| !o.is_empty());
let managed_versioning = response.managed_versioning; let write_params = if let Some(storage_opts) = initial_storage_options {
let mut params = request.write_options.lance_write_params.unwrap_or_default();
// Build write params with storage options and commit handler
let mut params = request.write_options.lance_write_params.unwrap_or_default();
// Set up storage options if provided
if let Some(storage_opts) = initial_storage_options {
let store_params = params let store_params = params
.store_params .store_params
.get_or_insert_with(ObjectStoreParams::default); .get_or_insert_with(ObjectStoreParams::default);
store_params.storage_options_accessor = Some(Arc::new( store_params.storage_options_accessor = Some(Arc::new(
StorageOptionsAccessor::with_static_options(storage_opts), StorageOptionsAccessor::with_static_options(storage_opts),
)); ));
} Some(params)
} else {
// Set up commit handler when managed_versioning is enabled request.write_options.lance_write_params
if managed_versioning == Some(true) { };
let external_store =
LanceNamespaceExternalManifestStore::new(self.namespace.clone(), table_id.clone());
let commit_handler: Arc<dyn CommitHandler> = Arc::new(ExternalManifestCommitHandler {
external_manifest_store: Arc::new(external_store),
});
params.commit_handler = Some(commit_handler);
}
let write_params = Some(params);
let native_table = NativeTable::create_from_namespace( let native_table = NativeTable::create_from_namespace(
self.namespace.clone(), self.namespace.clone(),
+7 -47
View File
@@ -446,23 +446,13 @@ impl<S: HttpSend> RestfulLanceDbClient<S> {
})?, })?,
); );
} }
// Map azure storage options to x-azure-* headers. if let Some(v) = options.0.get("azure_storage_account_name") {
// The option key uses underscores (e.g. "azure_client_id") while the headers.insert(
// header uses hyphens (e.g. "x-azure-client-id"). HeaderName::from_static("x-azure-storage-account-name"),
let azure_opts: [(&str, &str); 3] = [ HeaderValue::from_str(v).map_err(|_| Error::InvalidInput {
("azure_storage_account_name", "x-azure-storage-account-name"), message: format!("non-ascii storage account name '{}' provided", db_name),
("azure_client_id", "x-azure-client-id"), })?,
("azure_tenant_id", "x-azure-tenant-id"), );
];
for (opt_key, header_name) in azure_opts {
if let Some(v) = options.0.get(opt_key) {
headers.insert(
HeaderName::from_static(header_name),
HeaderValue::from_str(v).map_err(|_| Error::InvalidInput {
message: format!("non-ascii value '{}' for option '{}'", v, opt_key),
})?,
);
}
} }
for (key, value) in &config.extra_headers { for (key, value) in &config.extra_headers {
@@ -1085,34 +1075,4 @@ mod tests {
_ => panic!("Expected Runtime error"), _ => panic!("Expected Runtime error"),
} }
} }
#[test]
fn test_default_headers_azure_opts() {
let mut opts = HashMap::new();
opts.insert(
"azure_storage_account_name".to_string(),
"myaccount".to_string(),
);
opts.insert("azure_client_id".to_string(), "my-client-id".to_string());
opts.insert("azure_tenant_id".to_string(), "my-tenant-id".to_string());
let remote_opts = RemoteOptions::new(opts);
let headers = RestfulLanceDbClient::<Sender>::default_headers(
"test-key",
"us-east-1",
"testdb",
false,
&remote_opts,
None,
&ClientConfig::default(),
)
.unwrap();
assert_eq!(
headers.get("x-azure-storage-account-name").unwrap(),
"myaccount"
);
assert_eq!(headers.get("x-azure-client-id").unwrap(), "my-client-id");
assert_eq!(headers.get("x-azure-tenant-id").unwrap(), "my-tenant-id");
}
} }
+1 -7
View File
@@ -464,7 +464,6 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
lance_read_params: None, lance_read_params: None,
location: None, location: None,
namespace_client: None, namespace_client: None,
managed_versioning: None,
}; };
let req = (callback)(req); let req = (callback)(req);
self.open_table(req).await self.open_table(req).await
@@ -778,12 +777,7 @@ impl RemoteOptions {
impl From<StorageOptions> for RemoteOptions { impl From<StorageOptions> for RemoteOptions {
fn from(options: StorageOptions) -> Self { fn from(options: StorageOptions) -> Self {
let supported_opts = vec![ let supported_opts = vec!["account_name", "azure_storage_account_name"];
"account_name",
"azure_storage_account_name",
"azure_client_id",
"azure_tenant_id",
];
let mut filtered = HashMap::new(); let mut filtered = HashMap::new();
for opt in supported_opts { for opt in supported_opts {
if let Some(v) = options.0.get(opt) { if let Some(v) = options.0.get(opt) {
+18 -76
View File
@@ -34,13 +34,9 @@ use lance_index::vector::sq::builder::SQBuildParams;
use lance_io::object_store::{LanceNamespaceStorageOptionsProvider, StorageOptionsAccessor}; use lance_io::object_store::{LanceNamespaceStorageOptionsProvider, StorageOptionsAccessor};
pub use query::AnyQuery; pub use query::AnyQuery;
use lance::io::commit::namespace_manifest::LanceNamespaceExternalManifestStore;
use lance_namespace::LanceNamespace; use lance_namespace::LanceNamespace;
use lance_namespace::models::DescribeTableRequest;
use lance_table::format::Manifest; use lance_table::format::Manifest;
use lance_table::io::commit::CommitHandler;
use lance_table::io::commit::ManifestNamingScheme; use lance_table::io::commit::ManifestNamingScheme;
use lance_table::io::commit::external_manifest::ExternalManifestCommitHandler;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::HashMap; use std::collections::HashMap;
use std::format; use std::format;
@@ -1216,13 +1212,10 @@ pub struct NativeTable {
// This comes from the connection options. We store here so we can pass down // This comes from the connection options. We store here so we can pass down
// to the dataset when we recreate it (for example, in checkout_latest). // to the dataset when we recreate it (for example, in checkout_latest).
read_consistency_interval: Option<std::time::Duration>, read_consistency_interval: Option<std::time::Duration>,
// Optional namespace client for namespace operations (e.g., managed versioning). // Optional namespace client for server-side query execution.
// pub(crate) so query.rs can access the field for server-side query execution. // When set, queries will be executed on the namespace server instead of locally.
// pub (crate) namespace_client so query.rs can access the fields
pub(crate) namespace_client: Option<Arc<dyn LanceNamespace>>, pub(crate) namespace_client: Option<Arc<dyn LanceNamespace>>,
// Whether to enable server-side query execution via the namespace client.
// When true and namespace_client is set, queries will be executed on the
// namespace server instead of locally.
pub(crate) server_side_query_enabled: bool,
} }
impl std::fmt::Debug for NativeTable { impl std::fmt::Debug for NativeTable {
@@ -1234,7 +1227,6 @@ impl std::fmt::Debug for NativeTable {
.field("uri", &self.uri) .field("uri", &self.uri)
.field("read_consistency_interval", &self.read_consistency_interval) .field("read_consistency_interval", &self.read_consistency_interval)
.field("namespace_client", &self.namespace_client) .field("namespace_client", &self.namespace_client)
.field("server_side_query_enabled", &self.server_side_query_enabled)
.finish() .finish()
} }
} }
@@ -1271,7 +1263,7 @@ impl NativeTable {
/// * A [NativeTable] object. /// * A [NativeTable] object.
pub async fn open(uri: &str) -> Result<Self> { pub async fn open(uri: &str) -> Result<Self> {
let name = Self::get_table_name(uri)?; let name = Self::get_table_name(uri)?;
Self::open_with_params(uri, &name, vec![], None, None, None, None, false, None).await Self::open_with_params(uri, &name, vec![], None, None, None, None).await
} }
/// Opens an existing Table /// Opens an existing Table
@@ -1281,10 +1273,7 @@ impl NativeTable {
/// * `base_path` - The base path where the table is located /// * `base_path` - The base path where the table is located
/// * `name` The Table name /// * `name` The Table name
/// * `params` The [ReadParams] to use when opening the table /// * `params` The [ReadParams] to use when opening the table
/// * `namespace_client` - Optional namespace client for namespace operations /// * `namespace_client` - Optional namespace client for server-side query execution
/// * `server_side_query_enabled` - Whether to enable server-side query execution
/// * `managed_versioning` - Whether managed versioning is enabled. If None and namespace_client
/// is provided, the value will be fetched via describe_table.
/// ///
/// # Returns /// # Returns
/// ///
@@ -1298,8 +1287,6 @@ impl NativeTable {
params: Option<ReadParams>, params: Option<ReadParams>,
read_consistency_interval: Option<std::time::Duration>, read_consistency_interval: Option<std::time::Duration>,
namespace_client: Option<Arc<dyn LanceNamespace>>, namespace_client: Option<Arc<dyn LanceNamespace>>,
server_side_query_enabled: bool,
managed_versioning: Option<bool>,
) -> Result<Self> { ) -> Result<Self> {
let params = params.unwrap_or_default(); let params = params.unwrap_or_default();
// patch the params if we have a write store wrapper // patch the params if we have a write store wrapper
@@ -1308,54 +1295,17 @@ impl NativeTable {
None => params, None => params,
}; };
// Build table_id from namespace + name let dataset = DatasetBuilder::from_uri(uri)
let mut table_id = namespace.clone(); .with_read_params(params)
table_id.push(name.to_string()); .load()
.await
// Determine if managed_versioning is enabled .map_err(|e| match e {
// Use the provided value if available, otherwise query the namespace lance::Error::DatasetNotFound { .. } => Error::TableNotFound {
let managed_versioning = match managed_versioning { name: name.to_string(),
Some(value) => value, source: Box::new(e),
None if namespace_client.is_some() => { },
let ns_client = namespace_client.as_ref().unwrap(); e => e.into(),
let describe_request = DescribeTableRequest { })?;
id: Some(table_id.clone()),
..Default::default()
};
let response = ns_client
.describe_table(describe_request)
.await
.map_err(|e| Error::Runtime {
message: format!(
"Failed to describe table via namespace client: {}. \
If you don't need managed versioning, don't pass namespace_client.",
e
),
})?;
response.managed_versioning == Some(true)
}
None => false,
};
let mut builder = DatasetBuilder::from_uri(uri).with_read_params(params);
// Set up commit handler when managed_versioning is enabled
if managed_versioning && let Some(ref ns_client) = namespace_client {
let external_store =
LanceNamespaceExternalManifestStore::new(ns_client.clone(), table_id.clone());
let commit_handler: Arc<dyn CommitHandler> = Arc::new(ExternalManifestCommitHandler {
external_manifest_store: Arc::new(external_store),
});
builder = builder.with_commit_handler(commit_handler);
}
let dataset = builder.load().await.map_err(|e| match e {
lance::Error::DatasetNotFound { .. } => Error::TableNotFound {
name: name.to_string(),
source: Box::new(e),
},
e => e.into(),
})?;
let dataset = DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval); let dataset = DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval);
let id = Self::build_id(&namespace, name); let id = Self::build_id(&namespace, name);
@@ -1368,7 +1318,6 @@ impl NativeTable {
dataset, dataset,
read_consistency_interval, read_consistency_interval,
namespace_client, namespace_client,
server_side_query_enabled,
}) })
} }
@@ -1472,7 +1421,6 @@ impl NativeTable {
dataset, dataset,
read_consistency_interval, read_consistency_interval,
namespace_client: stored_namespace_client, namespace_client: stored_namespace_client,
server_side_query_enabled,
}) })
} }
@@ -1512,8 +1460,7 @@ impl NativeTable {
/// * `namespace` - The namespace path. When non-empty, an explicit URI must be provided. /// * `namespace` - The namespace path. When non-empty, an explicit URI must be provided.
/// * `batches` RecordBatch to be saved in the database. /// * `batches` RecordBatch to be saved in the database.
/// * `params` - Write parameters. /// * `params` - Write parameters.
/// * `namespace_client` - Optional namespace client for namespace operations /// * `namespace_client` - Optional namespace client for server-side query execution
/// * `server_side_query_enabled` - Whether to enable server-side query execution
/// ///
/// # Returns /// # Returns
/// ///
@@ -1528,7 +1475,6 @@ impl NativeTable {
params: Option<WriteParams>, params: Option<WriteParams>,
read_consistency_interval: Option<std::time::Duration>, read_consistency_interval: Option<std::time::Duration>,
namespace_client: Option<Arc<dyn LanceNamespace>>, namespace_client: Option<Arc<dyn LanceNamespace>>,
server_side_query_enabled: bool,
) -> Result<Self> { ) -> Result<Self> {
// Default params uses format v1. // Default params uses format v1.
let params = params.unwrap_or(WriteParams { let params = params.unwrap_or(WriteParams {
@@ -1561,7 +1507,6 @@ impl NativeTable {
dataset: DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval), dataset: DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval),
read_consistency_interval, read_consistency_interval,
namespace_client, namespace_client,
server_side_query_enabled,
}) })
} }
@@ -1575,7 +1520,6 @@ impl NativeTable {
params: Option<WriteParams>, params: Option<WriteParams>,
read_consistency_interval: Option<std::time::Duration>, read_consistency_interval: Option<std::time::Duration>,
namespace_client: Option<Arc<dyn LanceNamespace>>, namespace_client: Option<Arc<dyn LanceNamespace>>,
server_side_query_enabled: bool,
) -> Result<Self> { ) -> Result<Self> {
let data: Box<dyn Scannable> = Box::new(RecordBatch::new_empty(schema)); let data: Box<dyn Scannable> = Box::new(RecordBatch::new_empty(schema));
Self::create( Self::create(
@@ -1587,7 +1531,6 @@ impl NativeTable {
params, params,
read_consistency_interval, read_consistency_interval,
namespace_client, namespace_client,
server_side_query_enabled,
) )
.await .await
} }
@@ -1691,7 +1634,6 @@ impl NativeTable {
dataset: DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval), dataset: DatasetConsistencyWrapper::new_latest(dataset, read_consistency_interval),
read_consistency_interval, read_consistency_interval,
namespace_client: stored_namespace_client, namespace_client: stored_namespace_client,
server_side_query_enabled,
}) })
} }
@@ -2683,7 +2625,7 @@ mod tests {
vec![Ok(batch.clone())], vec![Ok(batch.clone())],
batch.schema(), batch.schema(),
)); ));
let table = NativeTable::create(uri, "test", vec![], reader, None, None, None, None, false) let table = NativeTable::create(uri, "test", vec![], reader, None, None, None, None)
.await .await
.unwrap(); .unwrap();
+36 -159
View File
@@ -3,13 +3,12 @@
use std::sync::Arc; use std::sync::Arc;
use arrow_cast::can_cast_types;
use arrow_schema::{DataType, Field, FieldRef, Fields, Schema}; use arrow_schema::{DataType, Field, FieldRef, Fields, Schema};
use datafusion::functions::core::{get_field, named_struct}; use datafusion::functions::core::{get_field, named_struct};
use datafusion_common::ScalarValue; use datafusion_common::ScalarValue;
use datafusion_common::config::ConfigOptions; use datafusion_common::config::ConfigOptions;
use datafusion_physical_expr::ScalarFunctionExpr; use datafusion_physical_expr::ScalarFunctionExpr;
use datafusion_physical_expr::expressions::{CastExpr, Literal}; use datafusion_physical_expr::expressions::{Literal, cast};
use datafusion_physical_plan::expressions::Column; use datafusion_physical_plan::expressions::Column;
use datafusion_physical_plan::projection::ProjectionExec; use datafusion_physical_plan::projection::ProjectionExec;
use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr}; use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr};
@@ -26,9 +25,12 @@ pub fn cast_to_table_schema(
return Ok(input); return Ok(input);
} }
let exprs = build_field_exprs(input_schema.fields(), table_schema.fields(), &|idx| { let exprs = build_field_exprs(
Arc::new(Column::new(input_schema.field(idx).name(), idx)) as Arc<dyn PhysicalExpr> input_schema.fields(),
})?; table_schema.fields(),
&|idx| Arc::new(Column::new(input_schema.field(idx).name(), idx)) as Arc<dyn PhysicalExpr>,
&input_schema,
)?;
let exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = exprs let exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = exprs
.into_iter() .into_iter()
@@ -49,6 +51,7 @@ fn build_field_exprs(
input_fields: &Fields, input_fields: &Fields,
table_fields: &Fields, table_fields: &Fields,
get_input_expr: &dyn Fn(usize) -> Arc<dyn PhysicalExpr>, get_input_expr: &dyn Fn(usize) -> Arc<dyn PhysicalExpr>,
input_schema: &Schema,
) -> Result<Vec<(Arc<dyn PhysicalExpr>, FieldRef)>> { ) -> Result<Vec<(Arc<dyn PhysicalExpr>, FieldRef)>> {
let config = Arc::new(ConfigOptions::default()); let config = Arc::new(ConfigOptions::default());
let mut result = Vec::new(); let mut result = Vec::new();
@@ -69,19 +72,24 @@ fn build_field_exprs(
(DataType::Struct(in_children), DataType::Struct(tbl_children)) (DataType::Struct(in_children), DataType::Struct(tbl_children))
if in_children != tbl_children => if in_children != tbl_children =>
{ {
let sub_exprs = build_field_exprs(in_children, tbl_children, &|child_idx| { let sub_exprs = build_field_exprs(
let child_name = in_children[child_idx].name(); in_children,
Arc::new(ScalarFunctionExpr::new( tbl_children,
&format!("get_field({child_name})"), &|child_idx| {
get_field(), let child_name = in_children[child_idx].name();
vec![ Arc::new(ScalarFunctionExpr::new(
input_expr.clone(), &format!("get_field({child_name})"),
Arc::new(Literal::new(ScalarValue::from(child_name.as_str()))), get_field(),
], vec![
Arc::new(in_children[child_idx].as_ref().clone()), input_expr.clone(),
config.clone(), Arc::new(Literal::new(ScalarValue::from(child_name.as_str()))),
)) as Arc<dyn PhysicalExpr> ],
})?; Arc::new(in_children[child_idx].as_ref().clone()),
config.clone(),
)) as Arc<dyn PhysicalExpr>
},
input_schema,
)?;
let output_struct_fields: Fields = sub_exprs let output_struct_fields: Fields = sub_exprs
.iter() .iter()
@@ -117,21 +125,17 @@ fn build_field_exprs(
// Types match: pass through. // Types match: pass through.
(inp, tbl) if inp == tbl => input_expr, (inp, tbl) if inp == tbl => input_expr,
// Types differ: cast. // Types differ: cast.
// safe: false (the default) means overflow/truncation errors surface at execution time. _ => cast(input_expr, input_schema, table_field.data_type().clone()).map_err(|e| {
(_, _) if can_cast_types(input_field.data_type(), table_field.data_type()) => Arc::new( Error::InvalidInput {
CastExpr::new(input_expr, table_field.data_type().clone(), None),
)
as Arc<dyn PhysicalExpr>,
(inp, tbl) => {
return Err(Error::InvalidInput {
message: format!( message: format!(
"cannot cast field '{}' from {} to {}", "cannot cast field '{}' from {} to {}: {}",
table_field.name(), table_field.name(),
inp, input_field.data_type(),
tbl, table_field.data_type(),
e
), ),
}); }
} })?,
}; };
let output_field = Arc::new(Field::new( let output_field = Arc::new(Field::new(
@@ -149,12 +153,10 @@ fn build_field_exprs(
mod tests { mod tests {
use std::sync::Arc; use std::sync::Arc;
use arrow::buffer::OffsetBuffer;
use arrow_array::{ use arrow_array::{
Array, Float32Array, Float64Array, Int32Array, Int64Array, ListArray, RecordBatch, Float32Array, Float64Array, Int32Array, Int64Array, RecordBatch, StringArray, StructArray,
StringArray, StructArray, UInt32Array, UInt64Array,
}; };
use arrow_schema::{DataType, Field, Fields, Schema}; use arrow_schema::{DataType, Field, Schema};
use datafusion::prelude::SessionContext; use datafusion::prelude::SessionContext;
use datafusion_catalog::MemTable; use datafusion_catalog::MemTable;
use futures::TryStreamExt; use futures::TryStreamExt;
@@ -493,129 +495,4 @@ mod tests {
assert_eq!(b.value(0), "hello"); assert_eq!(b.value(0), "hello");
assert_eq!(b.value(1), "world"); assert_eq!(b.value(1), "world");
} }
#[tokio::test]
async fn test_narrowing_numeric_cast_success() {
let input_batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("a", DataType::UInt64, false)])),
vec![Arc::new(UInt64Array::from(vec![1u64, 2, 3]))],
)
.unwrap();
let table_schema = Schema::new(vec![Field::new("a", DataType::UInt32, false)]);
let plan = plan_from_batch(input_batch).await;
let casted = cast_to_table_schema(plan, &table_schema).unwrap();
let result = collect(casted).await;
assert_eq!(result.schema().field(0).data_type(), &DataType::UInt32);
let a: &UInt32Array = result.column(0).as_any().downcast_ref().unwrap();
assert_eq!(a.values(), &[1u32, 2, 3]);
}
#[tokio::test]
async fn test_narrowing_numeric_cast_overflow_errors() {
let overflow_val = u32::MAX as u64 + 1;
let input_batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("a", DataType::UInt64, false)])),
vec![Arc::new(UInt64Array::from(vec![overflow_val]))],
)
.unwrap();
let table_schema = Schema::new(vec![Field::new("a", DataType::UInt32, false)]);
let plan = plan_from_batch(input_batch).await;
// Planning succeeds — the overflow is only detected at execution time.
let casted = cast_to_table_schema(plan, &table_schema).unwrap();
let ctx = SessionContext::new();
let stream = casted.execute(0, ctx.task_ctx()).unwrap();
let result: Result<Vec<RecordBatch>, _> = stream.try_collect().await;
assert!(result.is_err(), "expected overflow error at execution time");
}
#[tokio::test]
async fn test_list_struct_field_reorder() {
// list<struct<a: Int32, b: Int32>> → list<struct<b: Int64, a: Int64>>
// Tests both reordering (a,b → b,a) and element-type widening (Int32 → Int64).
let inner_fields: Fields = vec![
Field::new("a", DataType::Int32, true),
Field::new("b", DataType::Int32, true),
]
.into();
let struct_array = StructArray::from(vec![
(
Arc::new(inner_fields[0].as_ref().clone()),
Arc::new(Int32Array::from(vec![1, 3])) as _,
),
(
Arc::new(inner_fields[1].as_ref().clone()),
Arc::new(Int32Array::from(vec![2, 4])) as _,
),
]);
// Offsets: one list element containing two struct rows (0..2).
let offsets = OffsetBuffer::from_lengths(vec![2]);
let list_array = ListArray::try_new(
Arc::new(Field::new("item", DataType::Struct(inner_fields), true)),
offsets,
Arc::new(struct_array),
None,
)
.unwrap();
let input_batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new(
"s_list",
list_array.data_type().clone(),
false,
)])),
vec![Arc::new(list_array)],
)
.unwrap();
let table_inner: Fields = vec![
Field::new("b", DataType::Int64, true),
Field::new("a", DataType::Int64, true),
]
.into();
let table_schema = Schema::new(vec![Field::new(
"s_list",
DataType::List(Arc::new(Field::new(
"item",
DataType::Struct(table_inner),
true,
))),
false,
)]);
let plan = plan_from_batch(input_batch).await;
let casted = cast_to_table_schema(plan, &table_schema).unwrap();
let result = collect(casted).await;
let list_col = result
.column(0)
.as_any()
.downcast_ref::<ListArray>()
.unwrap();
let struct_col = list_col
.values()
.as_any()
.downcast_ref::<StructArray>()
.unwrap();
assert_eq!(struct_col.num_columns(), 2);
let b: &Int64Array = struct_col
.column_by_name("b")
.unwrap()
.as_any()
.downcast_ref()
.unwrap();
assert_eq!(b.values(), &[2, 4]);
let a: &Int64Array = struct_col
.column_by_name("a")
.unwrap()
.as_any()
.downcast_ref()
.unwrap();
assert_eq!(a.values(), &[1, 3]);
}
} }
+1 -2
View File
@@ -1,4 +1,3 @@
use futures::FutureExt;
// SPDX-License-Identifier: Apache-2.0 // SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors // SPDX-FileCopyrightText: Copyright The LanceDB Authors
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@@ -24,7 +23,7 @@ pub struct DeleteResult {
pub(crate) async fn execute_delete(table: &NativeTable, predicate: &str) -> Result<DeleteResult> { pub(crate) async fn execute_delete(table: &NativeTable, predicate: &str) -> Result<DeleteResult> {
table.dataset.ensure_mutable()?; table.dataset.ensure_mutable()?;
let mut dataset = (*table.dataset.get().await?).clone(); let mut dataset = (*table.dataset.get().await?).clone();
let delete_result = dataset.delete(predicate).boxed().await?; let delete_result = dataset.delete(predicate).await?;
let num_deleted_rows = delete_result.num_deleted_rows; let num_deleted_rows = delete_result.num_deleted_rows;
let version = dataset.version().version; let version = dataset.version().version;
table.dataset.update(dataset); table.dataset.update(dataset);
+2 -4
View File
@@ -40,10 +40,8 @@ pub async fn execute_query(
query: &AnyQuery, query: &AnyQuery,
options: QueryExecutionOptions, options: QueryExecutionOptions,
) -> Result<DatasetRecordBatchStream> { ) -> Result<DatasetRecordBatchStream> {
// If server-side query is enabled and namespace client is configured, use server-side query execution // If namespace client is configured, use server-side query execution
if table.server_side_query_enabled if let Some(ref namespace_client) = table.namespace_client {
&& let Some(ref namespace_client) = table.namespace_client
{
return execute_namespace_query(table, namespace_client.clone(), query, options).await; return execute_namespace_query(table, namespace_client.clone(), query, options).await;
} }
execute_generic_query(table, query, options).await execute_generic_query(table, query, options).await