feat(listing): soft-delete root tables via an embedded V1 namespace

Root-table lifecycle in a ListingDatabase now flows through an embedded V1
(manifest-disabled) DirectoryNamespace, so a root drop is a soft-delete rather than
an immediate remove_dir_all:

- drop_table writes a delete marker and leaves the data for a later purge (TTL).
- create_table on a soft-deleted name revives it (clears the marker under the
  namespace's lifecycle lock, then overwrites via the native create path).
- open_table / table_names / list_tables treat soft-deleted tables as absent.
- table listing is now a single O(1) read_dir in the namespace instead of a
  per-table probe here.

A ListingDatabase now holds two embedded namespaces: the existing manifest-backed
`namespace_database` for child namespaces (multi-level table ids), and a new
`root_namespace_database` (V1, manifest-off) that owns root soft-delete/purge/
table_status. `namespace_client()` still returns the manifest namespace so child
namespace ops are unaffected.

Also preserves TableNotFound through LanceNamespaceDatabase::drop_table instead of
flattening it to a generic Runtime error, so dropping a missing table reports the
right error.

Removes the now-dead native drop_tables / commit-handler path and the
object_store/base_path fields it needed.

Depends on lance-format/lance#7541.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.31.0-beta.3"
+current_version = "0.31.0-beta.2"
 parse = """(?x)
     (?P<major>0|[1-9]\\d*)\\.
     (?P<minor>0|[1-9]\\d*)\\.

Cargo.lock

@@ -3433,7 +3433,7 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 [[package]]
 name = "fsst"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-array",
  "rand 0.9.4",
@@ -4736,7 +4736,7 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
 [[package]]
 name = "lance"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arc-swap",
  "arrow",
@@ -4811,7 +4811,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4832,7 +4832,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4846,7 +4846,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -4856,7 +4856,7 @@ dependencies = [
 [[package]]
 name = "lance-bitpacking"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrayref",
  "paste",
@@ -4866,7 +4866,7 @@ dependencies = [
 [[package]]
 name = "lance-core"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4905,7 +4905,7 @@ dependencies = [
 [[package]]
 name = "lance-datafusion"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -4936,7 +4936,7 @@ dependencies = [
 [[package]]
 name = "lance-datagen"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -4954,7 +4954,7 @@ dependencies = [
 [[package]]
 name = "lance-derive"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -4964,7 +4964,7 @@ dependencies = [
 [[package]]
 name = "lance-encoding"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -5000,7 +5000,7 @@ dependencies = [
 [[package]]
 name = "lance-file"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -5031,7 +5031,7 @@ dependencies = [
 [[package]]
 name = "lance-index"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arc-swap",
  "arrow",
@@ -5097,7 +5097,7 @@ dependencies = [
 [[package]]
 name = "lance-io"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow",
  "arrow-arith",
@@ -5139,7 +5139,7 @@ dependencies = [
 [[package]]
 name = "lance-linalg"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5156,7 +5156,7 @@ dependencies = [
 [[package]]
 name = "lance-namespace"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow",
  "async-trait",
@@ -5169,7 +5169,7 @@ dependencies = [
 [[package]]
 name = "lance-namespace-impls"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow",
  "arrow-ipc",
@@ -5224,7 +5224,7 @@ dependencies = [
 [[package]]
 name = "lance-select"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5240,7 +5240,7 @@ dependencies = [
 [[package]]
 name = "lance-table"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5280,7 +5280,7 @@ dependencies = [
 [[package]]
 name = "lance-testing"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -5294,7 +5294,7 @@ dependencies = [
 [[package]]
 name = "lance-tokenizer"
 version = "9.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.8#71c4aa2174971e98acb7e256fde1e1589024f5bc"
+source = "git+https://github.com/lance-format/lance.git?rev=4acefffd5d38f88003fce681ae1d0871077ce5e7#4acefffd5d38f88003fce681ae1d0871077ce5e7"
 dependencies = [
  "icu_segmenter",
  "jieba-rs",
@@ -5307,7 +5307,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb"
-version = "0.31.0-beta.3"
+version = "0.31.0-beta.2"
 dependencies = [
  "ahash",
  "anyhow",
@@ -5384,14 +5384,13 @@ dependencies = [
  "tokenizers",
  "tokio",
  "url",
- "urlencoding",
  "uuid",
  "walkdir",
 ]
 
 [[package]]
 name = "lancedb-nodejs"
-version = "0.31.0-beta.3"
+version = "0.31.0-beta.2"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5416,7 +5415,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb-python"
-version = "0.34.0-beta.3"
+version = "0.34.0-beta.2"
 dependencies = [
  "arrow",
  "async-trait",

Cargo.toml

@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"
 
 [workspace.dependencies]
-lance = { "version" = "=9.0.0-beta.8", default-features = false, "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=9.0.0-beta.8", default-features = false, "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=9.0.0-beta.8", default-features = false, "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=9.0.0-beta.8", default-features = false, "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=9.0.0-beta.8", default-features = false, "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=9.0.0-beta.8", default-features = false, "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=9.0.0-beta.8", "tag" = "v9.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=9.0.0-beta.8", "rev" = "4acefffd5d38f88003fce681ae1d0871077ce5e7", "git" = "https://github.com/lance-format/lance.git" }
 ahash = "0.8"
 # Note that this one does not include pyarrow
 arrow = { version = "58.0.0", optional = false }

docs/src/java/java.md

@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-core</artifactId>
-    <version>0.31.0-beta.3</version>
+    <version>0.31.0-beta.2</version>
 </dependency>
 ```
 

java/lancedb-core/pom.xml

@@ -8,7 +8,7 @@
     <parent>
       <groupId>com.lancedb</groupId>
       <artifactId>lancedb-parent</artifactId>
-      <version>0.31.0-beta.3</version>
+      <version>0.31.0-beta.2</version>
       <relativePath>../pom.xml</relativePath>
     </parent>
 

java/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-parent</artifactId>
-    <version>0.31.0-beta.3</version>
+    <version>0.31.0-beta.2</version>
     <packaging>pom</packaging>
     <name>${project.artifactId}</name>
     <description>LanceDB Java SDK Parent POM</description>

nodejs/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.31.0-beta.3"
+version = "0.31.0-beta.2"
 publish = false
 license.workspace = true
 description.workspace = true

nodejs/npm/darwin-arm64/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.31.0-beta.3",
+	"version": "0.31.0-beta.2",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",

nodejs/npm/linux-arm64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.31.0-beta.3",
+	"version": "0.31.0-beta.2",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",

nodejs/npm/linux-arm64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.31.0-beta.3",
+	"version": "0.31.0-beta.2",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",

nodejs/npm/linux-x64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.31.0-beta.3",
+	"version": "0.31.0-beta.2",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",

nodejs/npm/linux-x64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.31.0-beta.3",
+	"version": "0.31.0-beta.2",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",

nodejs/npm/win32-arm64-msvc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.31.0-beta.3",
+  "version": "0.31.0-beta.2",
   "os": [
     "win32"
   ],

nodejs/npm/win32-x64-msvc/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.31.0-beta.3",
+	"version": "0.31.0-beta.2",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",

nodejs/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@lancedb/lancedb",
-  "version": "0.31.0-beta.3",
+  "version": "0.31.0-beta.2",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@lancedb/lancedb",
-      "version": "0.31.0-beta.3",
+      "version": "0.31.0-beta.2",
       "cpu": [
         "x64",
         "arm64"

nodejs/package.json

@@ -11,7 +11,7 @@
     "ann"
   ],
   "private": false,
-  "version": "0.31.0-beta.3",
+  "version": "0.31.0-beta.2",
   "main": "dist/index.js",
   "exports": {
     ".": "./dist/index.js",

python/.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.34.0-beta.4"
+current_version = "0.34.0-beta.2"
 parse = """(?x)
     (?P<major>0|[1-9]\\d*)\\.
     (?P<minor>0|[1-9]\\d*)\\.

python/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.34.0-beta.4"
+version = "0.34.0-beta.2"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"

python/python/lancedb/namespace.py

@@ -373,19 +373,6 @@ def _convert_pyarrow_schema_to_json(schema: pa.Schema) -> JsonArrowSchema:
     return JsonArrowSchema(fields=fields, metadata=meta)
 
 
-def _builds_namespace_natively(
-    namespace_client_impl: Optional[str],
-    namespace_client_properties: Optional[Dict[str, str]],
-) -> bool:
-    """Whether ``connect_namespace_client`` builds the namespace client natively
-    in Rust (installing the read-freshness context provider) rather than wrapping
-    the pre-built Python client.
-
-    Must mirror Rust ``build_namespace_natively`` in ``python/src/connection.rs``.
-    """
-    return namespace_client_impl == "rest" and bool(namespace_client_properties)
-
-
 class LanceNamespaceDBConnection(DBConnection):
     """
     A LanceDB connection that uses a namespace for table management.
@@ -445,13 +432,6 @@ class LanceNamespaceDBConnection(DBConnection):
         )
         self._namespace_client_impl = namespace_client_impl
         self._namespace_client_properties = namespace_client_properties
-        # When the namespace client is built natively (see Rust
-        # ``build_namespace_natively``), the underlying Rust table performs
-        # QueryTable pushdown through the read-freshness context provider, which
-        # the pure-Python ``query_table`` path bypasses.
-        self._route_pushdown_to_rust = _builds_namespace_natively(
-            namespace_client_impl, namespace_client_properties
-        )
         self._inner = AsyncConnection(
             _connect_namespace_client(
                 namespace_client,
@@ -563,7 +543,6 @@ class LanceNamespaceDBConnection(DBConnection):
             namespace_path=namespace_path,
             namespace_client=self._namespace_client,
             pushdown_operations=self._namespace_client_pushdown_operations,
-            route_pushdown_to_rust=self._route_pushdown_to_rust,
             _async=async_table,
         )
 
@@ -601,7 +580,6 @@ class LanceNamespaceDBConnection(DBConnection):
             namespace_path=namespace_path,
             namespace_client=self._namespace_client,
             pushdown_operations=self._namespace_client_pushdown_operations,
-            route_pushdown_to_rust=self._route_pushdown_to_rust,
             _async=async_table,
         )
         if branch is not None:
@@ -897,8 +875,6 @@ class AsyncLanceNamespaceDBConnection:
         storage_options: Optional[Dict[str, str]] = None,
         session: Optional[Session] = None,
         namespace_client_pushdown_operations: Optional[List[str]] = None,
-        namespace_client_impl: Optional[str] = None,
-        namespace_client_properties: Optional[Dict[str, str]] = None,
     ):
         """
         Initialize an async namespace-based LanceDB connection.
@@ -924,12 +900,6 @@ class AsyncLanceNamespaceDBConnection:
               namespace.create_table() instead of using declare_table + local write.
 
             Default is None (no pushdown, all operations run locally).
-        namespace_client_impl : Optional[str]
-            The namespace implementation name used to create this connection.
-            Required (with ``namespace_client_properties``) for the Rust client to
-            be built natively and install the read-freshness provider.
-        namespace_client_properties : Optional[Dict[str, str]]
-            The namespace properties used to create this connection.
         """
         self._namespace_client = namespace_client
         self.read_consistency_interval = read_consistency_interval
@@ -938,14 +908,6 @@ class AsyncLanceNamespaceDBConnection:
         self._namespace_client_pushdown_operations = set(
             namespace_client_pushdown_operations or []
         )
-        self._namespace_client_impl = namespace_client_impl
-        self._namespace_client_properties = namespace_client_properties
-        # See LanceNamespaceDBConnection: when built natively the Rust table runs
-        # QueryTable pushdown through the read-freshness provider, so defer to it
-        # rather than the urllib3 client (which omits x-lancedb-min-timestamp).
-        self._route_pushdown_to_rust = _builds_namespace_natively(
-            namespace_client_impl, namespace_client_properties
-        )
         self._inner = AsyncConnection(
             _connect_namespace_client(
                 namespace_client,
@@ -959,8 +921,8 @@ class AsyncLanceNamespaceDBConnection:
                 namespace_client_pushdown_operations=(
                     list(self._namespace_client_pushdown_operations)
                 ),
-                namespace_client_impl=namespace_client_impl,
+                namespace_client_impl=None,
-                namespace_client_properties=namespace_client_properties,
+                namespace_client_properties=None,
             )
         )
 
@@ -1030,7 +992,6 @@ class AsyncLanceNamespaceDBConnection:
             namespace_path=namespace_path,
             namespace_client=self._namespace_client,
             pushdown_operations=self._namespace_client_pushdown_operations,
-            route_pushdown_to_rust=self._route_pushdown_to_rust,
         )
 
     async def open_table(
@@ -1068,7 +1029,6 @@ class AsyncLanceNamespaceDBConnection:
             namespace_path=namespace_path,
             namespace_client=self._namespace_client,
             pushdown_operations=self._namespace_client_pushdown_operations,
-            route_pushdown_to_rust=self._route_pushdown_to_rust,
         )
 
     async def drop_table(self, name: str, namespace_path: Optional[List[str]] = None):
@@ -1427,6 +1387,4 @@ def connect_namespace_async(
         storage_options=storage_options,
         session=session,
         namespace_client_pushdown_operations=namespace_client_pushdown_operations,
-        namespace_client_impl=namespace_client_impl,
-        namespace_client_properties=namespace_client_properties,
     )

python/python/lancedb/table.py

@@ -2022,7 +2022,6 @@ class LanceTable(Table):
         namespace_client: Optional[Any] = None,
         managed_versioning: Optional[bool] = None,
         pushdown_operations: Optional[set] = None,
-        route_pushdown_to_rust: bool = False,
         _async: AsyncTable = None,
     ):
         if namespace_path is None:
@@ -2032,14 +2031,6 @@ class LanceTable(Table):
         self._location = location  # Store location for use in _dataset_path
         self._namespace_client = namespace_client
         self._pushdown_operations = pushdown_operations or set()
-        # When the connection built the namespace client natively (e.g. an
-        # enterprise "rest" connection), the underlying Rust table already
-        # executes QueryTable pushdown itself -- and, unlike this Python urllib3
-        # path, it routes through the read-freshness context provider that emits
-        # the ``x-lancedb-min-timestamp`` header. So we must defer pushdown to
-        # Rust instead of calling the Python ``namespace_client.query_table``
-        # directly, or reads silently bypass read-freshness (stale results).
-        self._route_pushdown_to_rust = route_pushdown_to_rust
         if _async is not None:
             self._table = _async
         else:
@@ -2250,7 +2241,6 @@ class LanceTable(Table):
             namespace_path=self._namespace_path,
             namespace_client=self._namespace_client,
             pushdown_operations=self._pushdown_operations,
-            route_pushdown_to_rust=self._route_pushdown_to_rust,
             location=self._location,
             _async=async_table,
         )
@@ -2401,11 +2391,8 @@ class LanceTable(Table):
         Returns
         -------
         pa.Table"""
-        if (
+        if _should_push_down_query_table(
-            _should_push_down_query_table(
             self._namespace_client, self._pushdown_operations
-            )
-            and not self._route_pushdown_to_rust
         ):
             return self._execute_query(Query()).read_all()
 
@@ -3357,7 +3344,6 @@ class LanceTable(Table):
         location: Optional[str] = None,
         namespace_client: Optional[Any] = None,
         pushdown_operations: Optional[set] = None,
-        route_pushdown_to_rust: bool = False,
     ):
         """
         Create a new table.
@@ -3420,7 +3406,6 @@ class LanceTable(Table):
         self._location = location
         self._namespace_client = namespace_client
         self._pushdown_operations = pushdown_operations or set()
-        self._route_pushdown_to_rust = route_pushdown_to_rust
 
         if data_storage_version is not None:
             warnings.warn(
@@ -3534,7 +3519,6 @@ class LanceTable(Table):
             _should_push_down_query_table(
                 self._namespace_client, self._pushdown_operations
             )
-            and not self._route_pushdown_to_rust
             and self.current_branch() is None
         ):
             from lancedb.namespace import _execute_server_side_query
@@ -4276,7 +4260,6 @@ class AsyncTable:
         namespace_path: Optional[List[str]] = None,
         namespace_client: Optional[Any] = None,
         pushdown_operations: Optional[set] = None,
-        route_pushdown_to_rust: bool = False,
     ):
         """Create a new AsyncTable object.
 
@@ -4289,9 +4272,6 @@ class AsyncTable:
         self._namespace_path = namespace_path or []
         self._namespace_client = namespace_client
         self._pushdown_operations = pushdown_operations or set()
-        # See LanceTable.__init__: defer QueryTable pushdown to Rust (which emits
-        # the read-freshness header) for natively-built namespace clients.
-        self._route_pushdown_to_rust = route_pushdown_to_rust
 
     def _set_namespace_context(
         self,
@@ -4299,12 +4279,10 @@ class AsyncTable:
         namespace_path: Optional[List[str]] = None,
         namespace_client: Optional[Any] = None,
         pushdown_operations: Optional[set] = None,
-        route_pushdown_to_rust: bool = False,
     ) -> "AsyncTable":
         self._namespace_path = namespace_path or []
         self._namespace_client = namespace_client
         self._pushdown_operations = pushdown_operations or set()
-        self._route_pushdown_to_rust = route_pushdown_to_rust
         return self
 
     def __repr__(self):
@@ -4514,11 +4492,8 @@ class AsyncTable:
         -------
         pa.Table
         """
-        if (
+        if _should_push_down_query_table(
-            _should_push_down_query_table(
             self._namespace_client, self._pushdown_operations
-            )
-            and not self._route_pushdown_to_rust
         ):
             return (await self._execute_query(Query())).read_all()
 
@@ -5202,11 +5177,8 @@ class AsyncTable:
         batch_size: Optional[int] = None,
         timeout: Optional[timedelta] = None,
     ) -> pa.RecordBatchReader:
-        if (
+        if _should_push_down_query_table(
-            _should_push_down_query_table(
             self._namespace_client, self._pushdown_operations
-            )
-            and not self._route_pushdown_to_rust
         ):
             from lancedb.namespace import _execute_server_side_query
 

python/python/tests/test_namespace.py

@@ -65,9 +65,6 @@ def _namespace_lance_table(namespace_client: _NamespaceClient) -> LanceTable:
     table._namespace_path = ["geneva"]
     table._namespace_client = namespace_client
     table._pushdown_operations = {"QueryTable"}
-    # This test exercises the Python-side pushdown path (non-native client), so
-    # pushdown is not routed to Rust.
-    table._route_pushdown_to_rust = False
     return table
 
 
@@ -808,37 +805,6 @@ class TestPushdownOperations:
         db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
         assert len(db._namespace_client_pushdown_operations) == 0
 
-    def test_route_pushdown_to_rust_for_native_rest(self):
-        """A natively-built rest connection must defer QueryTable pushdown to
-        Rust so reads carry the x-lancedb-min-timestamp read-freshness header."""
-        db = lancedb.connect_namespace(
-            "rest",
-            {"uri": "http://localhost:12345"},
-            namespace_client_pushdown_operations=["QueryTable"],
-        )
-        assert db._route_pushdown_to_rust is True
-
-    def test_route_pushdown_to_rust_false_for_dir(self):
-        """A non-native (dir) connection keeps the Python pushdown path."""
-        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
-        assert db._route_pushdown_to_rust is False
-
-    def test_async_route_pushdown_to_rust_for_native_rest(self):
-        """The async connection must not silently bypass the read-freshness fix:
-        a natively-built rest connection defers pushdown to Rust (regression test
-        for the async path omitting the freshness header)."""
-        db = lancedb.connect_namespace_async(
-            "rest",
-            {"uri": "http://localhost:12345"},
-            namespace_client_pushdown_operations=["QueryTable"],
-        )
-        assert db._route_pushdown_to_rust is True
-
-    def test_async_route_pushdown_to_rust_false_for_dir(self):
-        """The async non-native (dir) connection keeps the Python pushdown path."""
-        db = lancedb.connect_namespace_async("dir", {"root": self.temp_dir})
-        assert db._route_pushdown_to_rust is False
-
     def test_lance_table_to_arrow_uses_query_pushdown(self):
         namespace_client = _NamespaceClient()
         table = _namespace_lance_table(namespace_client)

python/src/connection.rs

@@ -610,38 +610,24 @@ pub fn connect_namespace_client(
     namespace_client_impl: Option<String>,
     namespace_client_properties: Option<HashMap<String, String>>,
 ) -> PyResult<Connection> {
+    let namespace_client = extract_namespace_arc(py, namespace_client)?;
     let read_consistency_interval = read_consistency_interval.map(Duration::from_secs_f64);
     let namespace_client_pushdown_operations =
         parse_namespace_client_pushdown_operations(namespace_client_pushdown_operations)?;
+    let ns_impl = namespace_client_impl.unwrap_or_else(|| "python".to_string());
     let ns_properties = namespace_client_properties.unwrap_or_default();
     let storage_options = storage_options.unwrap_or_default();
     let session = session.map(|s| s.inner.clone());
 
-    // Prefer building the namespace natively from (impl, properties) so the
+    let database = LanceNamespaceDatabase::from_namespace_client(
-    // read-freshness provider installed
-    let database = if build_namespace_natively(namespace_client_impl.as_deref(), &ns_properties) {
-        let ns_impl = namespace_client_impl.expect("impl present per build_namespace_natively");
-        crate::runtime::block_on(LanceNamespaceDatabase::connect(
-            &ns_impl,
-            ns_properties,
-            storage_options,
-            read_consistency_interval,
-            session,
-            namespace_client_pushdown_operations,
-        ))
-        .infer_error()?
-    } else {
-        let namespace_client = extract_namespace_arc(py, namespace_client)?;
-        LanceNamespaceDatabase::from_namespace_client(
         namespace_client,
-            namespace_client_impl.unwrap_or_else(|| "python".to_string()),
+        ns_impl,
         ns_properties,
         storage_options,
         read_consistency_interval,
         session,
         namespace_client_pushdown_operations,
-        )
+    );
-    };
 
     Ok(Connection::new(LanceConnection::new(
         Arc::new(database),
@@ -649,16 +635,6 @@ pub fn connect_namespace_client(
     )))
 }
 
-/// Whether to build the namespace natively (from impl + properties) instead of
-/// wrapping a pre-built client. Native construction is required for the
-/// read-freshness provider to be installed
-fn build_namespace_natively(
-    namespace_client_impl: Option<&str>,
-    namespace_client_properties: &HashMap<String, String>,
-) -> bool {
-    matches!(namespace_client_impl, Some("rest")) && !namespace_client_properties.is_empty()
-}
-
 #[derive(FromPyObject)]
 pub struct PyClientConfig {
     user_agent: String,
@@ -757,36 +733,3 @@ impl From<PyClientConfig> for lancedb::remote::ClientConfig {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    fn props(pairs: &[(&str, &str)]) -> HashMap<String, String> {
-        pairs
-            .iter()
-            .map(|(k, v)| (k.to_string(), v.to_string()))
-            .collect()
-    }
-
-    #[test]
-    fn native_build_only_for_rest_with_properties() {
-        let rest = props(&[("uri", "http://localhost:10024")]);
-
-        // rest + non-empty properties -> build natively (installs the
-        // read-freshness provider so checkout_latest() busts the server cache).
-        assert!(build_namespace_natively(Some("rest"), &rest));
-
-        // dir is local (no server cache) -> wrap the pre-built client unchanged.
-        assert!(!build_namespace_natively(
-            Some("dir"),
-            &props(&[("root", "/tmp")])
-        ));
-
-        // No impl: only a pre-built client was handed in -> wrap it as-is.
-        assert!(!build_namespace_natively(None, &rest));
-
-        // rest but no properties: nothing to build a connection from -> wrap.
-        assert!(!build_namespace_natively(Some("rest"), &HashMap::new()));
-    }
-}

python/src/runtime.rs

@@ -56,15 +56,6 @@ fn get_runtime() -> &'static runtime::Runtime {
     unsafe { &*new_ptr }
 }
 
-/// Block the current thread on a future using the shared runtime.
-///
-/// For sync `#[pyfunction]`s that need to drive an async operation (e.g.
-/// building a namespace client). Must not be called from within the runtime's
-/// own worker threads.
-pub fn block_on<F: std::future::Future>(fut: F) -> F::Output {
-    get_runtime().block_on(fut)
-}
-
 /// Runs in async-signal context after `fork()` in the child.  We can only
 /// touch atomics here; we deliberately leak the previous runtime because
 /// dropping a tokio `Runtime` would try to join its (now-dead) worker

rust/lancedb/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.31.0-beta.3"
+version = "0.31.0-beta.2"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -50,7 +50,7 @@ lance-namespace = { workspace = true }
 lance-namespace-impls = { workspace = true }
 moka = { workspace = true }
 pin-project = { workspace = true }
-tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] }
+tokio = { version = "1.23", features = ["rt-multi-thread"] }
 log.workspace = true
 async-trait = "0"
 bytes = "1"
@@ -75,7 +75,6 @@ reqwest = { version = "0.12.0", default-features = false, features = [
     "stream",
 ], optional = true }
 http = { version = "1", optional = true } # Matching what is in reqwest
-urlencoding = { version = "2", optional = true }
 uuid = { version = "1.7.0", features = ["v4", "v5"] }
 polars-arrow = { version = ">=0.37,<0.40.0", optional = true }
 polars = { version = ">=0.37,<0.40.0", optional = true }
@@ -94,7 +93,6 @@ semver = { workspace = true }
 anyhow = "1"
 tempfile = "3.5.0"
 random_word = { version = "0.4.3", features = ["en"] }
-tokio = { version = "1.23", features = ["io-util", "macros", "net", "rt-multi-thread", "sync"] }
 uuid = { version = "1.7.0", features = ["v4"] }
 walkdir = "2"
 aws-sdk-dynamodb = { version = "1.55.0" }
@@ -131,13 +129,7 @@ huggingface = [
     "lance-namespace-impls/dir-huggingface",
 ]
 dynamodb = ["lance/dynamodb", "aws"]
-remote = [
+remote = ["dep:reqwest", "dep:http", "lance-namespace-impls/rest", "lance-namespace-impls/rest-adapter"]
-    "dep:reqwest",
-    "dep:http",
-    "dep:urlencoding",
-    "lance-namespace-impls/rest",
-    "lance-namespace-impls/rest-adapter",
-]
 fp16kernels = ["lance-linalg/fp16kernels"]
 s3-test = []
 bedrock = ["dep:aws-sdk-bedrockruntime"]

rust/lancedb/src/connection.rs

@@ -667,8 +667,6 @@ pub struct ConnectRequest {
 pub struct ConnectBuilder {
     request: ConnectRequest,
     embedding_registry: Option<Arc<dyn EmbeddingRegistry>>,
-    #[cfg(feature = "remote")]
-    oauth_config: Option<crate::remote::OAuthConfig>,
 }
 
 #[cfg(feature = "remote")]
@@ -690,8 +688,6 @@ impl ConnectBuilder {
                 session: None,
             },
             embedding_registry: None,
-            #[cfg(feature = "remote")]
-            oauth_config: None,
         }
     }
 
@@ -780,19 +776,6 @@ impl ConnectBuilder {
         self
     }
 
-    /// Configure OAuth authentication for LanceDB Cloud/Enterprise.
-    ///
-    /// This creates an [`OAuthHeaderProvider`](crate::remote::OAuthHeaderProvider)
-    /// from the given config and sets it as the header provider. OAuth cannot
-    /// be combined with an API key or another header provider.
-    ///
-    /// Token acquisition and refresh are handled in Rust.
-    #[cfg(feature = "remote")]
-    pub fn oauth_config(mut self, config: crate::remote::OAuthConfig) -> Self {
-        self.oauth_config = Some(config);
-        self
-    }
-
     /// Provide a custom [`EmbeddingRegistry`] to use for this connection.
     pub fn embedding_registry(mut self, registry: Arc<dyn EmbeddingRegistry>) -> Self {
         self.embedding_registry = Some(registry);
@@ -938,40 +921,9 @@ impl ConnectBuilder {
         let region = options.region.ok_or_else(|| Error::InvalidInput {
             message: "A region is required when connecting to LanceDb Cloud".to_string(),
         })?;
-        let api_key = match (&self.oauth_config, &options.api_key) {
+        let api_key = options.api_key.ok_or_else(|| Error::InvalidInput {
-            (Some(_), None) => String::new(),
+            message: "An api_key is required when connecting to LanceDb Cloud".to_string(),
-            (Some(_), Some(_)) => {
+        })?;
-                return Err(Error::InvalidInput {
-                    message:
-                        "api_key and oauth_config cannot both be set when connecting to LanceDb Cloud"
-                            .to_string(),
-                });
-            }
-            (None, Some(key)) => key.clone(),
-            (None, None) => {
-                return Err(Error::InvalidInput {
-                    message:
-                        "An api_key or oauth_config is required when connecting to LanceDb Cloud"
-                            .to_string(),
-                });
-            }
-        };
-
-        if self.oauth_config.is_some() && self.request.client_config.header_provider.is_some() {
-            return Err(Error::InvalidInput {
-                message:
-                    "oauth_config and client_config.header_provider cannot both be set when connecting to LanceDb Cloud"
-                        .to_string(),
-            });
-        }
-
-        let mut client_config = self.request.client_config;
-
-        if let Some(oauth_config) = self.oauth_config {
-            let provider = crate::remote::OAuthHeaderProvider::new(oauth_config)?;
-            client_config.header_provider =
-                Some(Arc::new(provider) as Arc<dyn crate::remote::HeaderProvider>);
-        }
 
         let storage_options = StorageOptions(options.storage_options.clone());
         let internal = Arc::new(crate::remote::db::RemoteDatabase::try_new(
@@ -979,7 +931,7 @@ impl ConnectBuilder {
             &api_key,
             &region,
             options.host_override,
-            client_config,
+            self.request.client_config,
             storage_options.into(),
             self.request.read_consistency_interval,
         )?);
@@ -1288,83 +1240,6 @@ mod tests {
         assert_eq!(Some(&"EXPLICIT-VALUE".to_string()), options.get(opts_key));
     }
 
-    #[cfg(feature = "remote")]
-    #[tokio::test]
-    async fn test_connect_rejects_api_key_with_oauth_config() {
-        let oauth_config = crate::remote::OAuthConfig {
-            issuer_url: "https://issuer.example.com".to_string(),
-            client_id: "client-id".to_string(),
-            client_secret: Some("secret".to_string()),
-            scopes: vec!["scope".to_string()],
-            flow: crate::remote::OAuthFlow::ClientCredentials,
-            refresh_buffer_secs: None,
-        };
-
-        let result = ConnectBuilder::new("db://my-container/my-prefix")
-            .region("us-east-1")
-            .api_key("my-api-key")
-            .oauth_config(oauth_config)
-            .execute()
-            .await;
-
-        match result {
-            Err(Error::InvalidInput { message })
-                if message
-                    == "api_key and oauth_config cannot both be set when connecting to LanceDb Cloud" =>
-                {}
-            Err(err) => panic!("expected InvalidInput, got {err:?}"),
-            Ok(_) => panic!("expected api_key and oauth_config to be rejected"),
-        }
-    }
-
-    #[cfg(feature = "remote")]
-    #[tokio::test]
-    async fn test_connect_rejects_header_provider_with_oauth_config() {
-        #[derive(Debug)]
-        struct TestHeaderProvider;
-
-        #[async_trait::async_trait]
-        impl crate::remote::HeaderProvider for TestHeaderProvider {
-            async fn get_headers(&self) -> Result<HashMap<String, String>> {
-                Ok(HashMap::from([(
-                    "authorization".to_string(),
-                    "Bearer token".to_string(),
-                )]))
-            }
-        }
-
-        let oauth_config = crate::remote::OAuthConfig {
-            issuer_url: "https://issuer.example.com".to_string(),
-            client_id: "client-id".to_string(),
-            client_secret: Some("secret".to_string()),
-            scopes: vec!["scope".to_string()],
-            flow: crate::remote::OAuthFlow::ClientCredentials,
-            refresh_buffer_secs: None,
-        };
-        let client_config = crate::remote::ClientConfig {
-            header_provider: Some(
-                Arc::new(TestHeaderProvider) as Arc<dyn crate::remote::HeaderProvider>
-            ),
-            ..Default::default()
-        };
-
-        let result = ConnectBuilder::new("db://my-container/my-prefix")
-            .region("us-east-1")
-            .client_config(client_config)
-            .oauth_config(oauth_config)
-            .execute()
-            .await;
-
-        match result {
-            Err(Error::InvalidInput { message })
-                if message
-                    == "oauth_config and client_config.header_provider cannot both be set when connecting to LanceDb Cloud" =>
-                {}
-            Err(err) => panic!("expected InvalidInput, got {err:?}"),
-            Ok(_) => panic!("expected header_provider and oauth_config to be rejected"),
-        }
-    }
-
     #[cfg(not(windows))]
     #[tokio::test]
     async fn test_connect_relative() {

rust/lancedb/src/database/listing.rs

@@ -14,7 +14,6 @@ use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore};
 use lance_datafusion::utils::StreamingWriteSource;
 use lance_encoding::version::LanceFileVersion;
 use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
-use lance_table::io::commit::commit_handler_from_url;
 use object_store::local::LocalFileSystem;
 use snafu::ResultExt;
 
@@ -235,11 +234,9 @@ impl ListingDatabaseOptionsBuilder {
 /// We will have two tables named `table1` and `table2`.
 #[derive(Debug)]
 pub struct ListingDatabase {
-    object_store: Arc<ObjectStore>,
     query_string: Option<String>,
 
     pub(crate) uri: String,
-    pub(crate) base_path: object_store::path::Path,
 
     // the object store wrapper to use on write path
     pub(crate) store_wrapper: Option<Arc<dyn WrappingObjectStore>>,
@@ -258,8 +255,13 @@ pub struct ListingDatabase {
     // Session for object stores and caching
     session: Arc<lance::session::Session>,
 
-    // Namespace-backed database for child namespace operations
+    // Namespace-backed database for child namespace operations (manifest mode).
     namespace_database: Arc<LanceNamespaceDatabase>,
+
+    // V1 (manifest-disabled) directory namespace for root table lifecycle, so root
+    // drops are soft-deletes and purge/table_status are available. Shares the same root
+    // as `namespace_database` but in directory mode.
+    root_namespace_database: Arc<LanceNamespaceDatabase>,
 }
 
 impl std::fmt::Display for ListingDatabase {
@@ -280,7 +282,6 @@ impl std::fmt::Display for ListingDatabase {
     }
 }
 
-const LANCE_EXTENSION: &str = "lance";
 const ENGINE: &str = "engine";
 const MIRRORED_STORE: &str = "mirroredStore";
 
@@ -342,6 +343,39 @@ impl ListingDatabase {
         ))
     }
 
+    /// Build the V1 (manifest-disabled) directory namespace used for *root* table
+    /// lifecycle ops.
+    ///
+    /// Root tables in a listing database are flat `<name>.lance` directories; soft-delete
+    /// (drop/purge/TTL) is a V1-only mechanism, so root ops go through this namespace.
+    /// Child namespaces are manifest-backed and handled by the separate
+    /// (manifest-enabled) `namespace_database`.
+    async fn connect_root_namespace_database(
+        uri: &str,
+        storage_options: HashMap<String, String>,
+        namespace_client_properties: HashMap<String, String>,
+        read_consistency_interval: Option<std::time::Duration>,
+        session: Arc<lance::session::Session>,
+    ) -> Result<Arc<LanceNamespaceDatabase>> {
+        let mut ns_properties = Self::build_namespace_client_properties(
+            uri,
+            &storage_options,
+            namespace_client_properties,
+        );
+        ns_properties.insert("manifest_enabled".to_string(), "false".to_string());
+        Ok(Arc::new(
+            LanceNamespaceDatabase::connect(
+                "dir",
+                ns_properties,
+                storage_options,
+                read_consistency_interval,
+                Some(session),
+                HashSet::new(),
+            )
+            .await?,
+        ))
+    }
+
     async fn prepare_namespace_root(
         uri: &str,
         storage_options: &HashMap<String, String>,
@@ -548,7 +582,7 @@ impl ListingDatabase {
                     },
                     ..Default::default()
                 };
-                let (object_store, base_path) = ObjectStore::from_uri_and_params(
+                let (object_store, _base_path) = ObjectStore::from_uri_and_params(
                     session.store_registry(),
                     &plain_uri,
                     &os_params,
@@ -577,12 +611,18 @@ impl ListingDatabase {
                     session.clone(),
                 )
                 .await?;
+                let root_namespace_database = Self::connect_root_namespace_database(
+                    &table_base_uri,
+                    options.storage_options.clone(),
+                    request.namespace_client_properties.clone(),
+                    request.read_consistency_interval,
+                    session.clone(),
+                )
+                .await?;
 
                 Ok(Self {
                     uri: table_base_uri,
                     query_string,
-                    base_path,
-                    object_store,
                     store_wrapper: write_store_wrapper,
                     read_consistency_interval: request.read_consistency_interval,
                     storage_options: options.storage_options,
@@ -590,6 +630,7 @@ impl ListingDatabase {
                     new_table_config: options.new_table_config,
                     session,
                     namespace_database,
+                    root_namespace_database,
                 })
             }
             Err(_) => {
@@ -613,7 +654,7 @@ impl ListingDatabase {
         session: Option<Arc<lance::session::Session>>,
     ) -> Result<Self> {
         let session = session.unwrap_or_else(|| Arc::new(lance::session::Session::default()));
-        let (object_store, base_path) = ObjectStore::from_uri_and_params(
+        let (object_store, _base_path) = ObjectStore::from_uri_and_params(
             session.store_registry(),
             path,
             &ObjectStoreParams::default(),
@@ -624,6 +665,14 @@ impl ListingDatabase {
         }
 
         let namespace_database = Self::connect_namespace_database(
+            path,
+            HashMap::new(),
+            namespace_client_properties.clone(),
+            read_consistency_interval,
+            session.clone(),
+        )
+        .await?;
+        let root_namespace_database = Self::connect_root_namespace_database(
             path,
             HashMap::new(),
             namespace_client_properties,
@@ -635,8 +684,6 @@ impl ListingDatabase {
         Ok(Self {
             uri: path.to_string(),
             query_string: None,
-            base_path,
-            object_store,
             store_wrapper: None,
             read_consistency_interval,
             storage_options: HashMap::new(),
@@ -644,6 +691,7 @@ impl ListingDatabase {
             new_table_config,
             session,
             namespace_database,
+            root_namespace_database,
         })
     }
 
@@ -705,42 +753,10 @@ impl ListingDatabase {
         self.namespace_database.clone()
     }
 
-    async fn drop_tables(&self, names: Vec<String>) -> Result<()> {
+    /// The V1 directory namespace used for root table lifecycle (soft-delete drop, purge,
-        let object_store_params = ObjectStoreParams {
+    /// table_status, O(1) listing).
-            storage_options_accessor: if self.storage_options.is_empty() {
+    fn root_namespace_database(&self) -> Arc<LanceNamespaceDatabase> {
-                None
+        self.root_namespace_database.clone()
-            } else {
-                Some(Arc::new(StorageOptionsAccessor::with_static_options(
-                    self.storage_options.clone(),
-                )))
-            },
-            ..Default::default()
-        };
-        let mut uri = self.uri.clone();
-        if let Some(query_string) = &self.query_string {
-            uri.push_str(&format!("?{}", query_string));
-        }
-        let commit_handler = commit_handler_from_url(&uri, &Some(object_store_params)).await?;
-        for name in names {
-            let dir_name = format!("{}.{}", name, LANCE_EXTENSION);
-            let full_path = self.base_path.clone().join(dir_name.clone());
-
-            commit_handler.delete(&full_path).await?;
-
-            self.object_store
-                .remove_dir_all(full_path.clone())
-                .await
-                .map_err(|err| match err {
-                    // this error is not lance::Error::DatasetNotFound, as the method
-                    // `remove_dir_all` may be used to remove something not be a dataset
-                    lance::Error::NotFound { .. } => Error::TableNotFound {
-                        name: name.clone(),
-                        source: Box::new(err),
-                    },
-                    _ => Error::from(err),
-                })?;
-        }
-        Ok(())
     }
 
     /// Inherit storage options from the connection into the target map
@@ -946,88 +962,43 @@ impl Database for ListingDatabase {
         if !request.namespace_path.is_empty() {
             return self.namespace_database().table_names(request).await;
         }
-        let mut f = self
+        // Root tables: the V1 namespace lists them in a single read_dir (O(1) requests)
-            .object_store
+        // and excludes soft-deleted tables, instead of a per-table probe here.
-            .read_dir(self.base_path.clone())
+        self.root_namespace_database().table_names(request).await
-            .await?
-            .iter()
-            .map(Path::new)
-            .filter(|path| {
-                let is_lance = path
-                    .extension()
-                    .and_then(|e| e.to_str())
-                    .map(|e| e == LANCE_EXTENSION);
-                is_lance.unwrap_or(false)
-            })
-            .filter_map(|p| p.file_stem().and_then(|s| s.to_str().map(String::from)))
-            .collect::<Vec<String>>();
-        f.sort();
-        if let Some(start_after) = request.start_after {
-            let index = f
-                .iter()
-                .position(|name| name.as_str() > start_after.as_str())
-                .unwrap_or(f.len());
-            f.drain(0..index);
-        }
-        if let Some(limit) = request.limit {
-            f.truncate(limit as usize);
-        }
-        Ok(f)
     }
 
     async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
         if request.id.as_ref().map(|v| !v.is_empty()).unwrap_or(false) {
             return self.namespace_database().list_tables(request).await;
         }
-        let mut f = self
+        self.root_namespace_database().list_tables(request).await
-            .object_store
-            .read_dir(self.base_path.clone())
-            .await?
-            .iter()
-            .map(Path::new)
-            .filter(|path| {
-                let is_lance = path
-                    .extension()
-                    .and_then(|e| e.to_str())
-                    .map(|e| e == LANCE_EXTENSION);
-                is_lance.unwrap_or(false)
-            })
-            .filter_map(|p| p.file_stem().and_then(|s| s.to_str().map(String::from)))
-            .collect::<Vec<String>>();
-        f.sort();
-
-        // Handle pagination with page_token
-        if let Some(ref page_token) = request.page_token {
-            let index = f
-                .iter()
-                .position(|name| name.as_str() > page_token.as_str())
-                .unwrap_or(f.len());
-            f.drain(0..index);
-        }
-
-        // Determine if there's a next page
-        let next_page_token = if let Some(limit) = request.limit {
-            if f.len() > limit as usize {
-                let token = f[limit as usize].clone();
-                f.truncate(limit as usize);
-                Some(token)
-            } else {
-                None
-            }
-        } else {
-            None
-        };
-
-        Ok(ListTablesResponse {
-            tables: f,
-            page_token: next_page_token,
-        })
     }
 
     async fn create_table(&self, request: CreateTableRequest) -> Result<Arc<dyn BaseTable>> {
         if !request.namespace_path.is_empty() {
             return self.namespace_database().create_table(request).await;
         }
+        let mut request = request;
+        // Re-creating a soft-deleted table is a revive: clear the delete marker (via the
+        // V1 root namespace, under its lifecycle lock so a concurrent purge can't race),
+        // making the table live again, then overwrite its data through the native create
+        // path below (preserving lineage as a new version). A plain native create would
+        // leave the marker in place, keeping the table hidden.
+        if matches!(
+            self.root_namespace_database()
+                .namespace_client()
+                .await?
+                .table_status(Some(vec![request.name.clone()]))
+                .await?,
+            lance_namespace::TableLifecycle::SoftDeleted { .. }
+        ) {
+            self.root_namespace_database()
+                .namespace_client()
+                .await?
+                .undelete_table(Some(vec![request.name.clone()]))
+                .await?;
+            request.mode = CreateTableMode::Overwrite;
+        }
         // Use provided location if available, otherwise derive from table name
         let table_uri = request
             .location
@@ -1146,6 +1117,19 @@ impl Database for ListingDatabase {
         if !request.namespace_path.is_empty() {
             return self.namespace_database().open_table(request).await;
         }
+        // A soft-deleted (dropped-but-not-purged) table must read as absent even though
+        // its data still exists on disk. Consult the V1 root namespace (which owns the
+        // marker); if soft-deleted, route to it so the open surfaces TableNotFound.
+        if matches!(
+            self.root_namespace_database()
+                .namespace_client()
+                .await?
+                .table_status(Some(vec![request.name.clone()]))
+                .await?,
+            lance_namespace::TableLifecycle::SoftDeleted { .. }
+        ) {
+            return self.root_namespace_database().open_table(request).await;
+        }
         // Use provided location if available, otherwise derive from table name
         let table_uri = request
             .location
@@ -1245,20 +1229,23 @@ impl Database for ListingDatabase {
                 .drop_table(name, namespace_path)
                 .await;
         }
-        self.drop_tables(vec![name.to_string()]).await
+        // Root table: route through the V1 namespace so the drop is a soft-delete (writes
+        // a marker, leaves data for later purge) rather than an immediate remove_dir_all.
+        self.root_namespace_database()
+            .drop_table(name, namespace_path)
+            .await
     }
 
-    #[allow(deprecated)]
     async fn drop_all_tables(&self, namespace_path: &[String]) -> Result<()> {
-        // Check if namespace parameter is provided
         if !namespace_path.is_empty() {
             return self
                 .namespace_database()
                 .drop_all_tables(namespace_path)
                 .await;
         }
-        let tables = self.table_names(TableNamesRequest::default()).await?;
+        self.root_namespace_database()
-        self.drop_tables(tables).await
+            .drop_all_tables(namespace_path)
+            .await
     }
 
     fn as_any(&self) -> &dyn std::any::Any {
@@ -1266,6 +1253,9 @@ impl Database for ListingDatabase {
     }
 
     async fn namespace_client(&self) -> Result<Arc<dyn lance_namespace::LanceNamespace>> {
+        // Returns the manifest-backed namespace so callers can operate on child
+        // namespaces (multi-level table ids) through the client. Root-table soft-delete
+        // lifecycle (table_status/purge) is reached via the V1 root namespace internally.
         self.namespace_database.namespace_client().await
     }
 
@@ -2615,4 +2605,154 @@ mod tests {
             .unwrap();
         assert!(post_drop.tables.is_empty());
     }
+
+    /// Root-table drop is a soft-delete routed through the V1 namespace: the table is
+    /// hidden from listing/open but its data survives until purged, and re-creating it
+    /// revives it. Verifies the consolidation end-to-end at the ListingDatabase level.
+    #[tokio::test]
+    async fn test_root_table_soft_delete_lifecycle() {
+        let (_tempdir, db) = setup_database().await;
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let create = |name: &str| CreateTableRequest {
+            name: name.to_string(),
+            namespace_path: vec![],
+            data: Box::new(RecordBatch::new_empty(schema.clone())) as Box<dyn Scannable>,
+            mode: CreateTableMode::Create,
+            write_options: Default::default(),
+            location: None,
+            namespace_client: None,
+        };
+        let open = |name: &str| OpenTableRequest {
+            name: name.to_string(),
+            namespace_path: vec![],
+            index_cache_size: None,
+            lance_read_params: None,
+            location: None,
+            namespace_client: None,
+            managed_versioning: None,
+        };
+
+        db.create_table(create("t")).await.unwrap();
+        db.drop_table("t", &[]).await.unwrap();
+
+        // Hidden from listing and not openable...
+        #[allow(deprecated)]
+        let names = db.table_names(TableNamesRequest::default()).await.unwrap();
+        assert!(!names.contains(&"t".to_string()));
+        assert!(matches!(
+            db.open_table(open("t")).await,
+            Err(Error::TableNotFound { .. })
+        ));
+
+        // ...but data survives: it shows up as purgable via the V1 root namespace.
+        let root_ns = db
+            .root_namespace_database()
+            .namespace_client()
+            .await
+            .unwrap();
+        let purgable = root_ns.list_purgable_tables(None).await.unwrap();
+        assert_eq!(purgable.len(), 1);
+        assert_eq!(purgable[0].id, vec!["t".to_string()]);
+
+        // Re-creating revives it.
+        db.create_table(create("t")).await.unwrap();
+        db.open_table(open("t")).await.unwrap();
+        #[allow(deprecated)]
+        let names = db.table_names(TableNamesRequest::default()).await.unwrap();
+        assert!(names.contains(&"t".to_string()));
+        assert!(root_ns.list_purgable_tables(None).await.unwrap().is_empty());
+
+        // Drop then purge reclaims it for good.
+        db.drop_table("t", &[]).await.unwrap();
+        let purged = root_ns.purge_tables(None).await.unwrap();
+        assert_eq!(purged, vec![vec!["t".to_string()]]);
+        assert!(root_ns.list_purgable_tables(None).await.unwrap().is_empty());
+    }
+
+    /// Migration: a table written by the *old* path (a plain `<root>/<name>.lance`
+    /// dataset directory with no namespace markers — exactly what a pre-existing listing
+    /// database looks like on disk) is fully usable by the new embedded-namespace routing.
+    /// It lists, opens, and reads normally, and dropping it is now a soft-delete (data
+    /// survives on disk) rather than an immediate removal.
+    #[tokio::test]
+    async fn test_legacy_directory_layout_migrates() {
+        use arrow_array::RecordBatchIterator;
+        use lance::dataset::{Dataset, WriteParams};
+
+        let tempdir = tempdir().unwrap();
+        let root = tempdir.path().to_str().unwrap().to_string();
+
+        // Write a dataset the way the pre-namespace path did: raw lance, no declare/reserve
+        // marker, no manifest — just `<root>/legacy_table.lance`.
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
+        )
+        .unwrap();
+        let legacy_uri = format!("{}/legacy_table.lance", root);
+        Dataset::write(
+            RecordBatchIterator::new(vec![Ok(batch)], schema.clone()),
+            &legacy_uri,
+            Some(WriteParams::default()),
+        )
+        .await
+        .unwrap();
+
+        // Connect the new ListingDatabase over that pre-existing directory.
+        let request = ConnectRequest {
+            uri: root.clone(),
+            #[cfg(feature = "remote")]
+            client_config: Default::default(),
+            options: Default::default(),
+            namespace_client_properties: Default::default(),
+            manifest_enabled: false,
+            read_consistency_interval: None,
+            session: None,
+        };
+        let db = ListingDatabase::connect_with_options(&request)
+            .await
+            .unwrap();
+
+        let open = |name: &str| OpenTableRequest {
+            name: name.to_string(),
+            namespace_path: vec![],
+            index_cache_size: None,
+            lance_read_params: None,
+            location: None,
+            namespace_client: None,
+            managed_versioning: None,
+        };
+
+        // The legacy table is listed, opens, and reads its rows.
+        #[allow(deprecated)]
+        let names = db.table_names(TableNamesRequest::default()).await.unwrap();
+        assert!(names.contains(&"legacy_table".to_string()));
+        let table = db.open_table(open("legacy_table")).await.unwrap();
+        assert_eq!(table.count_rows(None).await.unwrap(), 3);
+
+        // Dropping it is now a soft-delete: hidden from listing / not openable, but the
+        // dataset directory still exists on disk (reclaimed only by a later purge).
+        db.drop_table("legacy_table", &[]).await.unwrap();
+        #[allow(deprecated)]
+        let names = db.table_names(TableNamesRequest::default()).await.unwrap();
+        assert!(!names.contains(&"legacy_table".to_string()));
+        assert!(db.open_table(open("legacy_table")).await.is_err());
+        assert!(
+            PathBuf::from(&legacy_uri).exists(),
+            "soft-delete must leave the legacy dataset directory intact"
+        );
+
+        // And it can be revived by re-creating the name.
+        let root_ns = db
+            .root_namespace_database()
+            .namespace_client()
+            .await
+            .unwrap();
+        assert_eq!(
+            root_ns.list_purgable_tables(None).await.unwrap().len(),
+            1,
+            "the dropped legacy table is purgable"
+        );
+    }
 }

rust/lancedb/src/database/namespace.rs

@@ -583,9 +583,9 @@ impl Database for LanceNamespaceDatabase {
         self.namespace
             .drop_table(drop_request)
             .await
-            .map_err(|e| Error::Runtime {
+            // Preserve TableNotFound (e.g. dropping a non-existent table) rather than
-                message: format!("Failed to drop table: {}", e),
+            // flattening every failure to a generic Runtime error.
-            })?;
+            .map_err(|e| map_namespace_lance_error(e, name))?;
 
         Ok(())
     }

rust/lancedb/src/remote.rs

@@ -8,7 +8,6 @@
 
 pub(crate) mod client;
 pub(crate) mod db;
-pub mod oauth;
 mod retry;
 pub(crate) mod table;
 pub(crate) mod util;
@@ -21,4 +20,3 @@ const JSON_CONTENT_TYPE: &str = "application/json";
 
 pub use client::{ClientConfig, HeaderProvider, RetryConfig, TimeoutConfig, TlsConfig};
 pub use db::{RemoteDatabaseOptions, RemoteDatabaseOptionsBuilder};
-pub use oauth::{OAuthConfig, OAuthFlow, OAuthHeaderProvider};

rust/lancedb/src/remote/oauth.rs

@@ -1,907 +0,0 @@
-// SPDX-License-Identifier: Apache-2.0
-// SPDX-FileCopyrightText: Copyright The LanceDB Authors
-
-use std::collections::HashMap;
-use std::net::IpAddr;
-use std::sync::Arc;
-use std::time::{Duration, Instant};
-
-use async_trait::async_trait;
-use log::debug;
-use reqwest::Client;
-use serde::Deserialize;
-use tokio::sync::RwLock;
-
-use crate::error::{Error, Result};
-use crate::remote::client::HeaderProvider;
-
-const DEFAULT_REFRESH_BUFFER_SECS: u64 = 300;
-const DEFAULT_TOKEN_TTL_SECS: u64 = 3600;
-const AZURE_IMDS_ENDPOINT: &str = "http://169.254.169.254/metadata/identity/oauth2/token";
-const AZURE_IMDS_API_VERSION: &str = "2018-02-01";
-
-/// OAuth authentication flow configuration.
-#[derive(Debug, Clone)]
-pub enum OAuthFlow {
-    /// Client Credentials grant (service-to-service / M2M).
-    /// Requires `client_secret` in [`OAuthConfig`].
-    ClientCredentials,
-
-    /// Azure Managed Identity via IMDS.
-    /// Works on Azure VMs, AKS, App Service, and Azure Functions.
-    /// IMDS requests bypass proxy settings because the endpoint is link-local.
-    AzureManagedIdentity {
-        /// Client ID for user-assigned managed identity.
-        /// Omit for system-assigned managed identity.
-        client_id: Option<String>,
-    },
-}
-
-/// OAuth configuration for LanceDB authentication.
-///
-/// All token acquisition and refresh is handled in the Rust layer.
-/// Python and TypeScript bindings expose this as a plain config object.
-#[derive(Clone)]
-pub struct OAuthConfig {
-    /// OIDC issuer URL or OAuth authority URL.
-    /// For Azure: `https://login.microsoftonline.com/{tenant_id}/v2.0`
-    pub issuer_url: String,
-
-    /// Application / Client ID.
-    pub client_id: String,
-
-    /// Client secret (required for `ClientCredentials`, optional for others).
-    pub client_secret: Option<String>,
-
-    /// OAuth scopes to request.
-    /// For Azure managed identity, exactly one scope or resource is required.
-    /// For example: `["api://{app_id}/.default"]`
-    pub scopes: Vec<String>,
-
-    /// Authentication flow to use.
-    pub flow: OAuthFlow,
-
-    /// Seconds before token expiry to trigger proactive refresh (default: 300).
-    /// Keep this well below the token TTL; if it is greater than or equal to
-    /// the TTL, each request refreshes the token.
-    pub refresh_buffer_secs: Option<u64>,
-}
-
-impl std::fmt::Debug for OAuthConfig {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("OAuthConfig")
-            .field("issuer_url", &self.issuer_url)
-            .field("client_id", &self.client_id)
-            .field(
-                "client_secret",
-                &self.client_secret.as_deref().map(|_| "<redacted>"),
-            )
-            .field("scopes", &self.scopes)
-            .field("flow", &self.flow)
-            .field("refresh_buffer_secs", &self.refresh_buffer_secs)
-            .finish()
-    }
-}
-
-// -- OIDC Discovery --
-
-#[derive(Clone, Debug, Deserialize)]
-struct OidcDiscovery {
-    token_endpoint: String,
-}
-
-// -- Token Response --
-
-#[derive(Deserialize)]
-struct TokenResponse {
-    access_token: String,
-    /// Token lifetime in seconds.
-    /// Some providers (Azure IMDS) return this as a string, so we accept both.
-    #[serde(default, deserialize_with = "deserialize_optional_u64_or_string")]
-    expires_in: Option<u64>,
-    #[serde(default)]
-    #[allow(dead_code)]
-    token_type: Option<String>,
-}
-
-impl std::fmt::Debug for TokenResponse {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("TokenResponse")
-            .field("access_token", &"<redacted>")
-            .field("expires_in", &self.expires_in)
-            .field("token_type", &self.token_type)
-            .finish()
-    }
-}
-
-fn deserialize_optional_u64_or_string<'de, D>(
-    deserializer: D,
-) -> std::result::Result<Option<u64>, D::Error>
-where
-    D: serde::Deserializer<'de>,
-{
-    use serde::de;
-
-    struct U64OrString;
-    impl<'de> de::Visitor<'de> for U64OrString {
-        type Value = Option<u64>;
-
-        fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
-            formatter.write_str("an integer, an integer-valued float, a numeric string, or null")
-        }
-
-        fn visit_u64<E: de::Error>(self, v: u64) -> std::result::Result<Self::Value, E> {
-            Ok(Some(v))
-        }
-
-        fn visit_i64<E: de::Error>(self, v: i64) -> std::result::Result<Self::Value, E> {
-            if v < 0 {
-                return Err(E::custom(format!("invalid expires_in value: {v}")));
-            }
-            Ok(Some(v as u64))
-        }
-
-        fn visit_f64<E: de::Error>(self, v: f64) -> std::result::Result<Self::Value, E> {
-            if !v.is_finite() || v < 0.0 || v.fract() != 0.0 || v > u64::MAX as f64 {
-                return Err(E::custom(format!("invalid expires_in value: {v}")));
-            }
-            Ok(Some(v as u64))
-        }
-
-        fn visit_str<E: de::Error>(self, v: &str) -> std::result::Result<Self::Value, E> {
-            v.parse::<u64>().map(Some).map_err(de::Error::custom)
-        }
-
-        fn visit_none<E: de::Error>(self) -> std::result::Result<Self::Value, E> {
-            Ok(None)
-        }
-
-        fn visit_unit<E: de::Error>(self) -> std::result::Result<Self::Value, E> {
-            Ok(None)
-        }
-    }
-
-    deserializer.deserialize_any(U64OrString)
-}
-
-// -- Internal Token State --
-
-struct TokenState {
-    access_token: Option<String>,
-    expires_at: Option<Instant>,
-}
-
-impl TokenState {
-    fn new() -> Self {
-        Self {
-            access_token: None,
-            expires_at: None,
-        }
-    }
-
-    fn is_expired(&self, buffer: Duration) -> bool {
-        match (self.access_token.as_ref(), self.expires_at) {
-            (Some(_), Some(expires_at)) => Instant::now() + buffer >= expires_at,
-            (None, _) => true,
-            (Some(_), None) => true,
-        }
-    }
-
-    fn update(&mut self, resp: &TokenResponse) {
-        self.access_token = Some(resp.access_token.clone());
-        let expires_in = resp.expires_in.unwrap_or(DEFAULT_TOKEN_TTL_SECS);
-        self.expires_at = Some(Instant::now() + Duration::from_secs(expires_in));
-    }
-}
-
-#[async_trait]
-trait TokenSource: Send + Sync + std::fmt::Debug {
-    async fn fetch_token(&self) -> Result<TokenResponse>;
-}
-
-struct ClientCredentialsSource {
-    issuer_url: String,
-    client_id: String,
-    client_secret: String,
-    scopes: Vec<String>,
-    http_client: Client,
-    discovery: RwLock<Option<OidcDiscovery>>,
-}
-
-impl std::fmt::Debug for ClientCredentialsSource {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("ClientCredentialsSource")
-            .field("issuer_url", &self.issuer_url)
-            .field("client_id", &self.client_id)
-            .field("client_secret", &"<redacted>")
-            .field("scopes", &self.scopes)
-            .finish()
-    }
-}
-
-impl ClientCredentialsSource {
-    fn new(
-        issuer_url: String,
-        client_id: String,
-        client_secret: Option<String>,
-        scopes: Vec<String>,
-    ) -> Result<Self> {
-        let client_secret = client_secret.ok_or(Error::InvalidInput {
-            message: "client_secret is required for ClientCredentials flow".to_string(),
-        })?;
-        Self::validate_issuer_transport(&issuer_url)?;
-
-        let http_client = Client::builder()
-            .timeout(Duration::from_secs(30))
-            .build()
-            .map_err(|e| Error::Runtime {
-                message: format!("Failed to create HTTP client for OAuth: {e}"),
-            })?;
-
-        Ok(Self {
-            issuer_url,
-            client_id,
-            client_secret,
-            scopes,
-            http_client,
-            discovery: RwLock::new(None),
-        })
-    }
-
-    fn validate_issuer_transport(issuer_url: &str) -> Result<()> {
-        let issuer = url::Url::parse(issuer_url).map_err(|e| Error::InvalidInput {
-            message: format!("Invalid OAuth issuer_url: {e}"),
-        })?;
-
-        match issuer.scheme() {
-            "https" => Ok(()),
-            "http" if Self::is_loopback_issuer(&issuer) => Ok(()),
-            _ => Err(Error::InvalidInput {
-                message:
-                    "ClientCredentials OAuth issuer_url must use https, except for loopback hosts"
-                        .to_string(),
-            }),
-        }
-    }
-
-    fn is_loopback_issuer(issuer: &url::Url) -> bool {
-        let Some(host) = issuer.host_str() else {
-            return false;
-        };
-
-        host.eq_ignore_ascii_case("localhost")
-            || host
-                .parse::<IpAddr>()
-                .map(|addr| addr.is_loopback())
-                .unwrap_or(false)
-    }
-
-    async fn get_discovery(&self) -> Result<OidcDiscovery> {
-        {
-            let cached = self.discovery.read().await;
-            if let Some(ref disc) = *cached {
-                return Ok(disc.clone());
-            }
-        }
-
-        let mut cache = self.discovery.write().await;
-        // Double-check
-        if let Some(ref disc) = *cache {
-            return Ok(disc.clone());
-        }
-
-        let discovery_url = format!(
-            "{}/.well-known/openid-configuration",
-            self.issuer_url.trim_end_matches('/')
-        );
-
-        debug!("Fetching OIDC discovery from {}", discovery_url);
-
-        let resp = self
-            .http_client
-            .get(&discovery_url)
-            .send()
-            .await
-            .map_err(|e| Error::Runtime {
-                message: format!("Failed to fetch OIDC discovery document: {e}"),
-            })?;
-
-        if !resp.status().is_success() {
-            return Err(Error::Runtime {
-                message: format!(
-                    "OIDC discovery failed with status {}: {}",
-                    resp.status(),
-                    resp.text().await.unwrap_or_default()
-                ),
-            });
-        }
-
-        let disc: OidcDiscovery = resp.json().await.map_err(|e| Error::Runtime {
-            message: format!("Failed to parse OIDC discovery document: {e}"),
-        })?;
-
-        let result = disc.clone();
-
-        *cache = Some(disc);
-        Ok(result)
-    }
-
-    async fn get_token_endpoint(&self) -> Result<String> {
-        self.get_discovery().await.map(|disc| disc.token_endpoint)
-    }
-
-    fn scopes_string(&self) -> String {
-        self.scopes.join(" ")
-    }
-
-    async fn post_token_request(
-        &self,
-        endpoint: &str,
-        params: &[(&str, &str)],
-    ) -> Result<TokenResponse> {
-        let resp = self
-            .http_client
-            .post(endpoint)
-            .form(params)
-            .send()
-            .await
-            .map_err(|e| Error::Runtime {
-                message: format!("Token request to {endpoint} failed: {e}"),
-            })?;
-
-        if !resp.status().is_success() {
-            return Err(Error::Runtime {
-                message: format!(
-                    "Token request failed with status {}: {}",
-                    resp.status(),
-                    resp.text().await.unwrap_or_default()
-                ),
-            });
-        }
-
-        resp.json().await.map_err(|e| Error::Runtime {
-            message: format!("Failed to parse token response: {e}"),
-        })
-    }
-}
-
-#[async_trait]
-impl TokenSource for ClientCredentialsSource {
-    async fn fetch_token(&self) -> Result<TokenResponse> {
-        let token_endpoint = self.get_token_endpoint().await?;
-        let scope = self.scopes_string();
-        let params = [
-            ("grant_type", "client_credentials"),
-            ("client_id", self.client_id.as_str()),
-            ("client_secret", self.client_secret.as_str()),
-            ("scope", scope.as_str()),
-        ];
-
-        self.post_token_request(&token_endpoint, &params).await
-    }
-}
-
-struct AzureImdsSource {
-    client_id: Option<String>,
-    resource: String,
-    http_client: Client,
-}
-
-impl std::fmt::Debug for AzureImdsSource {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("AzureImdsSource")
-            .field("client_id", &self.client_id)
-            .field("resource", &self.resource)
-            .finish()
-    }
-}
-
-impl AzureImdsSource {
-    fn new(scopes: Vec<String>, client_id: Option<String>) -> Result<Self> {
-        let resource = Self::resource_from_scopes(&scopes)?;
-        let http_client = Client::builder()
-            .timeout(Duration::from_secs(30))
-            .no_proxy()
-            .build()
-            .map_err(|e| Error::Runtime {
-                message: format!("Failed to create HTTP client for Azure IMDS OAuth: {e}"),
-            })?;
-
-        Ok(Self {
-            client_id,
-            resource,
-            http_client,
-        })
-    }
-
-    fn resource_from_scopes(scopes: &[String]) -> Result<String> {
-        let [scope] = scopes else {
-            return Err(Error::InvalidInput {
-                message: "AzureManagedIdentity flow requires exactly one OAuth scope or resource"
-                    .to_string(),
-            });
-        };
-
-        Ok(scope.strip_suffix("/.default").unwrap_or(scope).to_string())
-    }
-}
-
-#[async_trait]
-impl TokenSource for AzureImdsSource {
-    async fn fetch_token(&self) -> Result<TokenResponse> {
-        let mut url = format!(
-            "{AZURE_IMDS_ENDPOINT}?api-version={AZURE_IMDS_API_VERSION}&resource={}",
-            urlencoding::encode(&self.resource),
-        );
-        if let Some(cid) = self.client_id.as_deref() {
-            url.push_str(&format!("&client_id={}", urlencoding::encode(cid)));
-        }
-
-        let resp = self
-            .http_client
-            .get(&url)
-            .header("Metadata", "true")
-            .send()
-            .await
-            .map_err(|e| Error::Runtime {
-                message: format!("Azure IMDS request failed: {e}"),
-            })?;
-
-        if !resp.status().is_success() {
-            return Err(Error::Runtime {
-                message: format!(
-                    "Azure IMDS returned status {}: {}",
-                    resp.status(),
-                    resp.text().await.unwrap_or_default()
-                ),
-            });
-        }
-
-        resp.json().await.map_err(|e| Error::Runtime {
-            message: format!("Failed to parse IMDS token response: {e}"),
-        })
-    }
-}
-
-/// OAuth header provider that manages the full token lifecycle.
-///
-/// Implements [`HeaderProvider`] to inject `Authorization: Bearer <token>`
-/// headers into every LanceDB request, with automatic token refresh.
-pub struct OAuthHeaderProvider {
-    token_source: Box<dyn TokenSource>,
-    token_state: Arc<RwLock<TokenState>>,
-    refresh_buffer: Duration,
-}
-
-impl std::fmt::Debug for OAuthHeaderProvider {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("OAuthHeaderProvider")
-            .field("token_source", &self.token_source)
-            .finish()
-    }
-}
-
-impl OAuthHeaderProvider {
-    /// Create a new OAuth header provider from configuration.
-    pub fn new(config: OAuthConfig) -> Result<Self> {
-        let OAuthConfig {
-            issuer_url,
-            client_id,
-            client_secret,
-            scopes,
-            flow,
-            refresh_buffer_secs,
-        } = config;
-
-        if scopes.is_empty() {
-            return Err(Error::InvalidInput {
-                message: "At least one OAuth scope is required".to_string(),
-            });
-        }
-
-        let refresh_buffer =
-            Duration::from_secs(refresh_buffer_secs.unwrap_or(DEFAULT_REFRESH_BUFFER_SECS));
-        let token_source: Box<dyn TokenSource> = match flow {
-            OAuthFlow::ClientCredentials => Box::new(ClientCredentialsSource::new(
-                issuer_url,
-                client_id,
-                client_secret,
-                scopes,
-            )?),
-            OAuthFlow::AzureManagedIdentity { client_id } => {
-                Box::new(AzureImdsSource::new(scopes, client_id)?)
-            }
-        };
-
-        Ok(Self {
-            token_source,
-            token_state: Arc::new(RwLock::new(TokenState::new())),
-            refresh_buffer,
-        })
-    }
-
-    /// Get a valid access token, refreshing if necessary.
-    async fn get_valid_token(&self) -> Result<String> {
-        // Fast path: check if current token is still valid
-        {
-            let state = self.token_state.read().await;
-            if !state.is_expired(self.refresh_buffer)
-                && let Some(ref token) = state.access_token
-            {
-                return Ok(token.clone());
-            }
-        }
-
-        // Slow path: acquire or refresh token
-        let mut state = self.token_state.write().await;
-
-        // Double-check after acquiring write lock
-        if !state.is_expired(self.refresh_buffer)
-            && let Some(ref token) = state.access_token
-        {
-            return Ok(token.clone());
-        }
-
-        debug!("Acquiring new OAuth token via {:?}", self.token_source);
-        let resp = self.token_source.fetch_token().await?;
-
-        state.update(&resp);
-        Ok(resp.access_token)
-    }
-}
-
-#[async_trait]
-impl HeaderProvider for OAuthHeaderProvider {
-    async fn get_headers(&self) -> Result<HashMap<String, String>> {
-        let token = self.get_valid_token().await?;
-        Ok(HashMap::from([(
-            "authorization".to_string(),
-            format!("Bearer {token}"),
-        )]))
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::sync::atomic::{AtomicUsize, Ordering};
-
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    use tokio::net::{TcpListener, TcpStream};
-    use tokio::task::JoinHandle;
-
-    #[test]
-    fn test_token_state_expiry() {
-        let mut state = TokenState::new();
-        assert!(state.is_expired(Duration::from_secs(0)));
-
-        state.access_token = Some("tok".to_string());
-        state.expires_at = Some(Instant::now() + Duration::from_secs(600));
-        assert!(!state.is_expired(Duration::from_secs(300)));
-        assert!(state.is_expired(Duration::from_secs(601)));
-
-        state.expires_at = None;
-        assert!(state.is_expired(Duration::from_secs(0)));
-    }
-
-    #[test]
-    fn test_token_state_uses_default_expiry() {
-        let mut state = TokenState::new();
-        let response = TokenResponse {
-            access_token: "tok".to_string(),
-            expires_in: None,
-            token_type: None,
-        };
-
-        state.update(&response);
-
-        assert!(!state.is_expired(Duration::from_secs(DEFAULT_TOKEN_TTL_SECS - 1)));
-        assert!(state.is_expired(Duration::from_secs(DEFAULT_TOKEN_TTL_SECS + 1)));
-    }
-
-    #[test]
-    fn test_token_response_accepts_float_expires_in() {
-        let response: TokenResponse =
-            serde_json::from_str(r#"{"access_token":"tok","expires_in":3600.0}"#).unwrap();
-
-        assert_eq!(response.expires_in, Some(3600));
-    }
-
-    #[test]
-    fn test_token_response_rejects_negative_expires_in() {
-        let err =
-            serde_json::from_str::<TokenResponse>(r#"{"access_token":"tok","expires_in":-1}"#)
-                .unwrap_err();
-
-        assert!(err.to_string().contains("invalid expires_in value: -1"));
-    }
-
-    #[test]
-    fn test_token_response_debug_redacts_access_token() {
-        let response = TokenResponse {
-            access_token: "secret-token".to_string(),
-            expires_in: Some(3600),
-            token_type: Some("Bearer".to_string()),
-        };
-
-        let debug = format!("{response:?}");
-        assert!(!debug.contains("secret-token"));
-        assert!(debug.contains("access_token: \"<redacted>\""));
-    }
-
-    #[test]
-    fn test_scopes_string() {
-        let source = ClientCredentialsSource::new(
-            "https://login.microsoftonline.com/tenant/v2.0".to_string(),
-            "app-id".to_string(),
-            Some("secret".to_string()),
-            vec!["scope1".to_string(), "scope2".to_string()],
-        )
-        .unwrap();
-
-        assert_eq!(source.scopes_string(), "scope1 scope2");
-    }
-
-    #[test]
-    fn test_oauth_config_debug_redacts_client_secret() {
-        let config = OAuthConfig {
-            issuer_url: "https://issuer.example.com".to_string(),
-            client_id: "client-id".to_string(),
-            client_secret: Some("super-secret".to_string()),
-            scopes: vec!["scope".to_string()],
-            flow: OAuthFlow::ClientCredentials,
-            refresh_buffer_secs: None,
-        };
-
-        let debug = format!("{config:?}");
-        assert!(!debug.contains("super-secret"));
-        assert!(debug.contains("client_secret: Some(\"<redacted>\")"));
-    }
-
-    #[test]
-    fn test_oauth_header_provider_debug_redacts_client_secret() {
-        let config = OAuthConfig {
-            issuer_url: "https://issuer.example.com".to_string(),
-            client_id: "client-id".to_string(),
-            client_secret: Some("super-secret".to_string()),
-            scopes: vec!["scope".to_string()],
-            flow: OAuthFlow::ClientCredentials,
-            refresh_buffer_secs: None,
-        };
-
-        let provider = OAuthHeaderProvider::new(config).unwrap();
-        let debug = format!("{provider:?}");
-        assert!(!debug.contains("super-secret"));
-        assert!(debug.contains("client_secret: \"<redacted>\""));
-    }
-
-    #[test]
-    fn test_managed_identity_resource_from_default_scope() {
-        assert_eq!(
-            AzureImdsSource::resource_from_scopes(&["api://test/.default".to_string()]).unwrap(),
-            "api://test"
-        );
-    }
-
-    #[test]
-    fn test_managed_identity_resource_without_default_suffix() {
-        assert_eq!(
-            AzureImdsSource::resource_from_scopes(&["api://test".to_string()]).unwrap(),
-            "api://test"
-        );
-    }
-
-    #[test]
-    fn test_managed_identity_rejects_multiple_scopes() {
-        let config = OAuthConfig {
-            issuer_url: "https://login.microsoftonline.com/tenant/v2.0".to_string(),
-            client_id: "app-id".to_string(),
-            client_secret: None,
-            scopes: vec![
-                "api://test-a/.default".to_string(),
-                "api://test-b/.default".to_string(),
-            ],
-            flow: OAuthFlow::AzureManagedIdentity { client_id: None },
-            refresh_buffer_secs: None,
-        };
-        assert!(OAuthHeaderProvider::new(config).is_err());
-    }
-
-    #[tokio::test]
-    async fn test_token_endpoint_requires_discovery_success() {
-        let (issuer_url, server) = spawn_discovery_error_server().await;
-        let source = ClientCredentialsSource::new(
-            issuer_url,
-            "client-id".to_string(),
-            Some("secret".to_string()),
-            vec!["scope".to_string()],
-        )
-        .unwrap();
-
-        let err = source.get_token_endpoint().await.unwrap_err();
-        assert!(matches!(
-            err,
-            Error::Runtime { message }
-                if message.contains("OIDC discovery failed with status 503")
-        ));
-        server.await.unwrap();
-    }
-
-    #[test]
-    fn test_client_credentials_requires_secret() {
-        let config = OAuthConfig {
-            issuer_url: "https://login.microsoftonline.com/tenant/v2.0".to_string(),
-            client_id: "app-id".to_string(),
-            client_secret: None,
-            scopes: vec!["scope".to_string()],
-            flow: OAuthFlow::ClientCredentials,
-            refresh_buffer_secs: None,
-        };
-        assert!(OAuthHeaderProvider::new(config).is_err());
-    }
-
-    #[test]
-    fn test_client_credentials_rejects_insecure_non_loopback_issuer() {
-        let config = OAuthConfig {
-            issuer_url: "http://issuer.example.com".to_string(),
-            client_id: "app-id".to_string(),
-            client_secret: Some("secret".to_string()),
-            scopes: vec!["scope".to_string()],
-            flow: OAuthFlow::ClientCredentials,
-            refresh_buffer_secs: None,
-        };
-
-        let err = OAuthHeaderProvider::new(config).unwrap_err();
-        assert!(matches!(
-            err,
-            Error::InvalidInput { message }
-                if message == "ClientCredentials OAuth issuer_url must use https, except for loopback hosts"
-        ));
-    }
-
-    #[test]
-    fn test_empty_scopes_rejected() {
-        let config = OAuthConfig {
-            issuer_url: "https://login.microsoftonline.com/tenant/v2.0".to_string(),
-            client_id: "app-id".to_string(),
-            client_secret: None,
-            scopes: vec![],
-            flow: OAuthFlow::AzureManagedIdentity { client_id: None },
-            refresh_buffer_secs: None,
-        };
-        assert!(OAuthHeaderProvider::new(config).is_err());
-    }
-
-    #[tokio::test]
-    async fn test_client_credentials_token_lifecycle() {
-        let (issuer_url, token_requests, server) = spawn_oauth_server().await;
-        let config = OAuthConfig {
-            issuer_url,
-            client_id: "client-id".to_string(),
-            client_secret: Some("secret".to_string()),
-            scopes: vec!["scope".to_string()],
-            flow: OAuthFlow::ClientCredentials,
-            refresh_buffer_secs: Some(0),
-        };
-        let provider = OAuthHeaderProvider::new(config).unwrap();
-
-        let headers = provider.get_headers().await.unwrap();
-        assert_eq!(headers.get("authorization").unwrap(), "Bearer token-1");
-        assert_eq!(token_requests.load(Ordering::SeqCst), 1);
-
-        let headers = provider.get_headers().await.unwrap();
-        assert_eq!(headers.get("authorization").unwrap(), "Bearer token-1");
-        assert_eq!(token_requests.load(Ordering::SeqCst), 1);
-
-        provider.token_state.write().await.expires_at =
-            Some(Instant::now() - Duration::from_secs(1));
-
-        let headers = provider.get_headers().await.unwrap();
-        assert_eq!(headers.get("authorization").unwrap(), "Bearer token-2");
-        assert_eq!(token_requests.load(Ordering::SeqCst), 2);
-
-        server.await.unwrap();
-    }
-
-    async fn spawn_oauth_server() -> (String, Arc<AtomicUsize>, JoinHandle<()>) {
-        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-        let addr = listener.local_addr().unwrap();
-        let issuer_url = format!("http://{addr}");
-        let token_requests = Arc::new(AtomicUsize::new(0));
-        let server_token_requests = Arc::clone(&token_requests);
-
-        let server = tokio::spawn(async move {
-            for _ in 0..3 {
-                let (mut stream, _) = listener.accept().await.unwrap();
-                let (request_line, body) = read_http_request(&mut stream).await;
-
-                if request_line.starts_with("GET /.well-known/openid-configuration ") {
-                    let discovery = format!(r#"{{"token_endpoint":"http://{addr}/token"}}"#);
-                    write_json_response(&mut stream, "200 OK", &discovery).await;
-                } else if request_line.starts_with("POST /token ") {
-                    assert!(body.contains("grant_type=client_credentials"));
-                    assert!(body.contains("client_id=client-id"));
-                    assert!(body.contains("client_secret=secret"));
-                    assert!(body.contains("scope=scope"));
-
-                    let token_num = server_token_requests.fetch_add(1, Ordering::SeqCst) + 1;
-                    let token = format!(
-                        r#"{{"access_token":"token-{token_num}","expires_in":3600,"token_type":"Bearer"}}"#
-                    );
-                    write_json_response(&mut stream, "200 OK", &token).await;
-                } else {
-                    write_json_response(&mut stream, "404 Not Found", "{}").await;
-                }
-            }
-        });
-
-        (issuer_url, token_requests, server)
-    }
-
-    async fn spawn_discovery_error_server() -> (String, JoinHandle<()>) {
-        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-        let addr = listener.local_addr().unwrap();
-        let issuer_url = format!("http://{addr}");
-
-        let server = tokio::spawn(async move {
-            let (mut stream, _) = listener.accept().await.unwrap();
-            let (request_line, _) = read_http_request(&mut stream).await;
-            assert!(request_line.starts_with("GET /.well-known/openid-configuration "));
-            write_json_response(&mut stream, "503 Service Unavailable", "{}").await;
-        });
-
-        (issuer_url, server)
-    }
-
-    async fn read_http_request(stream: &mut TcpStream) -> (String, String) {
-        let mut buffer = Vec::new();
-        let mut header_end = None;
-
-        while header_end.is_none() {
-            let mut chunk = [0; 1024];
-            let read = stream.read(&mut chunk).await.unwrap();
-            assert_ne!(read, 0, "connection closed before request headers");
-            buffer.extend_from_slice(&chunk[..read]);
-            header_end = find_subsequence(&buffer, b"\r\n\r\n").map(|pos| pos + 4);
-        }
-
-        let header_end = header_end.unwrap();
-        let headers = String::from_utf8_lossy(&buffer[..header_end]).to_string();
-        let request_line = headers.lines().next().unwrap_or_default().to_string();
-        let content_length = headers
-            .lines()
-            .find_map(|line| {
-                let (name, value) = line.split_once(':')?;
-                name.eq_ignore_ascii_case("content-length")
-                    .then(|| value.trim().parse::<usize>().ok())
-                    .flatten()
-            })
-            .unwrap_or(0);
-
-        while buffer.len() < header_end + content_length {
-            let mut chunk = [0; 1024];
-            let read = stream.read(&mut chunk).await.unwrap();
-            assert_ne!(read, 0, "connection closed before request body");
-            buffer.extend_from_slice(&chunk[..read]);
-        }
-
-        let body =
-            String::from_utf8_lossy(&buffer[header_end..header_end + content_length]).to_string();
-
-        (request_line, body)
-    }
-
-    fn find_subsequence(haystack: &[u8], needle: &[u8]) -> Option<usize> {
-        haystack
-            .windows(needle.len())
-            .position(|window| window == needle)
-    }
-
-    async fn write_json_response(stream: &mut TcpStream, status: &str, body: &str) {
-        let response = format!(
-            "HTTP/1.1 {status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
-            body.len()
-        );
-        stream.write_all(response.as_bytes()).await.unwrap();
-    }
-}

rust/lancedb/src/table/query.rs

@@ -579,45 +579,24 @@ fn array_to_f32_vec(arr: &Arc<dyn arrow_array::Array>) -> Result<Vec<f32>> {
     })
 }
 
-/// Magic bytes that prefix (and suffix) the Arrow IPC *file* format.
-const ARROW_IPC_FILE_MAGIC: &[u8] = b"ARROW1";
-
 /// Parse Arrow IPC response from the namespace server.
-///
-/// The server may return either the Arrow IPC *file* format or the *stream*
-/// format. REST/phalanx returns the file format (it begins with the `ARROW1`
-/// magic); reading that with a `StreamReader` fails with "failed to fill whole
-/// buffer". Detect the magic and pick the matching reader so both are handled.
 async fn parse_arrow_ipc_response(bytes: bytes::Bytes) -> Result<DatasetRecordBatchStream> {
-    use arrow_ipc::reader::{FileReader, StreamReader};
+    use arrow_ipc::reader::StreamReader;
     use std::io::Cursor;
 
-    let (schema, batches) = if bytes.starts_with(ARROW_IPC_FILE_MAGIC) {
+    let cursor = Cursor::new(bytes);
-        let reader = FileReader::try_new(Cursor::new(bytes), None).map_err(|e| Error::Runtime {
+    let reader = StreamReader::try_new(cursor, None).map_err(|e| Error::Runtime {
-            message: format!("Failed to parse Arrow IPC file response: {}", e),
-        })?;
-        let schema = reader.schema();
-        let batches = reader
-            .into_iter()
-            .collect::<std::result::Result<Vec<_>, _>>()
-            .map_err(|e| Error::Runtime {
-                message: format!("Failed to read Arrow IPC file batches: {}", e),
-            })?;
-        (schema, batches)
-    } else {
-        let reader =
-            StreamReader::try_new(Cursor::new(bytes), None).map_err(|e| Error::Runtime {
         message: format!("Failed to parse Arrow IPC response: {}", e),
     })?;
+
+    // Collect all record batches
     let schema = reader.schema();
-        let batches = reader
+    let batches: Vec<_> = reader
         .into_iter()
         .collect::<std::result::Result<Vec<_>, _>>()
         .map_err(|e| Error::Runtime {
             message: format!("Failed to read Arrow IPC batches: {}", e),
         })?;
-        (schema, batches)
-    };
 
     // Create a stream from the batches
     let stream = futures::stream::iter(batches.into_iter().map(Ok));
@@ -645,59 +624,6 @@ mod tests {
         FixedSizeListArray::try_new_from_values(Float32Array::from(values), dimension).unwrap()
     }
 
-    #[tokio::test]
-    async fn test_parse_arrow_ipc_response_handles_file_and_stream() {
-        use arrow_array::{Int32Array, RecordBatch};
-        use arrow_ipc::writer::{FileWriter, StreamWriter};
-        use arrow_schema::{DataType, Field, Schema};
-
-        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
-        let batch = RecordBatch::try_new(
-            schema.clone(),
-            vec![Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef],
-        )
-        .unwrap();
-
-        // Arrow IPC *file* format -- what REST/phalanx returns. Previously this
-        // failed with "failed to fill whole buffer" because we used a StreamReader.
-        let mut file_buf = Vec::new();
-        {
-            let mut writer = FileWriter::try_new(&mut file_buf, &schema).unwrap();
-            writer.write(&batch).unwrap();
-            writer.finish().unwrap();
-        }
-        assert!(file_buf.starts_with(ARROW_IPC_FILE_MAGIC));
-        let rows: usize = parse_arrow_ipc_response(bytes::Bytes::from(file_buf))
-            .await
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap()
-            .iter()
-            .map(|b| b.num_rows())
-            .sum();
-        assert_eq!(rows, 3);
-
-        // Arrow IPC *stream* format must still parse.
-        let mut stream_buf = Vec::new();
-        {
-            let mut writer = StreamWriter::try_new(&mut stream_buf, &schema).unwrap();
-            writer.write(&batch).unwrap();
-            writer.finish().unwrap();
-        }
-        assert!(!stream_buf.starts_with(ARROW_IPC_FILE_MAGIC));
-        let rows: usize = parse_arrow_ipc_response(bytes::Bytes::from(stream_buf))
-            .await
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap()
-            .iter()
-            .map(|b| b.num_rows())
-            .sum();
-        assert_eq!(rows, 3);
-    }
-
     #[test]
     fn test_convert_to_namespace_query_vector() {
         let query_vector = Arc::new(Float32Array::from(vec![1.0, 2.0, 3.0, 4.0]));