docs: update OAuth refresh buffer guidance

feat: add native OAuth/OIDC authentication support
Add OAuthConfig and OAuthHeaderProvider to the Rust core with support for five OAuth flows: ClientCredentials, AuthorizationCodePKCE, DeviceCode, AzureManagedIdentity, and WorkloadIdentity. Token acquisition and auto-refresh happen entirely in Rust. Python and TypeScript expose OAuthConfig as a plain config object that maps to the Rust header provider via FFI — no dynamic callbacks cross the language boundary. ConnectBuilder gains an oauth_config() method that replaces the API key requirement when OAuth is configured.
2026-06-24 14:40:41 +00:00 · 2026-06-23 00:18:07 -07:00 · 2026-06-22 22:14:00 -07:00
33 changed files with 1857 additions and 1101 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1472,9 +1472,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"

 [[package]]
 name = "bytes"
-version = "1.12.0"
+version = "1.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ae3f5d315924270530207e2a68396c3cc547f6dca3fbdca317cfb1a51edb593"
+checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"

 [[package]]
 name = "bytes-utils"
@@ -3432,8 +3432,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow-array",
 "rand 0.9.4",
@@ -4735,8 +4735,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

 [[package]]
 name = "lance"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4810,8 +4810,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 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.2#23211989de648fefc4454f5eee09ec176f0a465b"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 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.2#23211989de648fefc4454f5eee09ec176f0a465b"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4855,8 +4855,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrayref",
 "paste",
@@ -4865,8 +4865,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4904,8 +4904,8 @@ dependencies = [

 [[package]]
 name = "lance-datafusion"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4935,8 +4935,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4953,8 +4953,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -4963,8 +4963,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -4999,8 +4999,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5030,8 +5030,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5096,8 +5096,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow",
 "arrow-arith",
@@ -5138,8 +5138,8 @@ dependencies = [

 [[package]]
 name = "lance-linalg"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5150,13 +5150,12 @@ dependencies = [
 "lance-core",
 "num-traits",
 "rand 0.9.4",
- "rayon",
 ]

 [[package]]
 name = "lance-namespace"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5168,8 +5167,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace-impls"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5223,8 +5222,8 @@ dependencies = [

 [[package]]
 name = "lance-select"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5239,8 +5238,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5279,8 +5278,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5293,8 +5292,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "9.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.2#23211989de648fefc4454f5eee09ec176f0a465b"
+version = "8.0.0-rc.1"
+source = "git+https://github.com/lance-format/lance.git?tag=v8.0.0-rc.1#eea4095b188bf2ba2fa95d934a2f5d6c2c9e661c"
 dependencies = [
 "icu_segmenter",
 "jieba-rs",
@@ -5384,6 +5383,7 @@ dependencies = [
 "tokenizers",
 "tokio",
 "url",
+ "urlencoding",
 "uuid",
 "walkdir",
 ]
@@ -5958,9 +5958,9 @@ dependencies = [

 [[package]]
 name = "napi"
-version = "3.9.3"
+version = "3.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fbd9f9295f3ff5921e78a71222c3361a8216f7760b1a99a6ad4e8441de18bbb9"
+checksum = "ad513ff22558f1830b595ea6eb4091da48145d09a222ce157e781896f78be0b9"
 dependencies = [
 "bitflags 2.11.1",
 "chrono",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=9.0.0-beta.2", default-features = false, "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=9.0.0-beta.2", default-features = false, "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=9.0.0-beta.2", default-features = false, "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=9.0.0-beta.2", "tag" = "v9.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=8.0.0-rc.1", default-features = false, "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=8.0.0-rc.1", default-features = false, "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=8.0.0-rc.1", default-features = false, "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=8.0.0-rc.1", "tag" = "v8.0.0-rc.1", "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 }
--- a/docs/src/js/enumerations/OAuthFlowType.md
+++ b/docs/src/js/enumerations/OAuthFlowType.md
@@ -0,0 +1,29 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / OAuthFlowType
+
+# Enumeration: OAuthFlowType
+
+OAuth authentication flow types.
+
+## Enumeration Members
+
+### AzureManagedIdentity
+
+```ts
+AzureManagedIdentity: "azure_managed_identity";
+```
+
+Azure Managed Identity via IMDS.
+
+***
+
+### ClientCredentials
+
+```ts
+ClientCredentials: "client_credentials";
+```
+
+Client Credentials grant (service-to-service / M2M).
--- a/docs/src/js/globals.md
+++ b/docs/src/js/globals.md
@@ -12,6 +12,7 @@
 ## Enumerations

 - [FullTextQueryType](enumerations/FullTextQueryType.md)
+- [OAuthFlowType](enumerations/OAuthFlowType.md)
 - [Occur](enumerations/Occur.md)
 - [Operator](enumerations/Operator.md)

@@ -85,6 +86,8 @@
 - [ListNamespacesResponse](interfaces/ListNamespacesResponse.md)
 - [LsmWriteSpec](interfaces/LsmWriteSpec.md)
 - [MergeResult](interfaces/MergeResult.md)
+- [NativeOAuthConfig](interfaces/NativeOAuthConfig.md)
+- [OAuthConfig](interfaces/OAuthConfig.md)
 - [OpenTableOptions](interfaces/OpenTableOptions.md)
 - [OptimizeOptions](interfaces/OptimizeOptions.md)
 - [OptimizeStats](interfaces/OptimizeStats.md)
--- a/docs/src/js/interfaces/ConnectionOptions.md
+++ b/docs/src/js/interfaces/ConnectionOptions.md
@@ -64,6 +64,19 @@ client used by manifest-enabled native connections.

 ***

+### oauthConfig?
+
+```ts
+optional oauthConfig: NativeOAuthConfig;
+```
+
+(For LanceDB cloud only): OAuth configuration for IdP-based
+authentication (e.g., Azure Entra ID). When set, token acquisition
+and refresh are handled entirely in Rust. TypeScript users should pass
+the public `OAuthConfig` type exported from `@lancedb/lancedb`.
+
+***
+
 ### readConsistencyInterval?

 ```ts
--- a/docs/src/js/interfaces/NativeOAuthConfig.md
+++ b/docs/src/js/interfaces/NativeOAuthConfig.md
@@ -0,0 +1,88 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / NativeOAuthConfig
+
+# Interface: NativeOAuthConfig
+
+OAuth configuration for LanceDB authentication.
+
+This is the generated napi-rs binding shape. TypeScript users should prefer
+the public `OAuthConfig` type exported from `@lancedb/lancedb`.
+
+All token acquisition and refresh is handled in the Rust layer.
+
+## Properties
+
+### clientId
+
+```ts
+clientId: string;
+```
+
+Application / Client ID.
+
+***
+
+### clientSecret?
+
+```ts
+optional clientSecret: string;
+```
+
+Client secret (required for client_credentials).
+
+***
+
+### flow?
+
+```ts
+optional flow: string;
+```
+
+Authentication flow: "client_credentials" or "azure_managed_identity"
+
+***
+
+### issuerUrl
+
+```ts
+issuerUrl: string;
+```
+
+OIDC issuer URL or OAuth authority URL.
+For Azure: `https://login.microsoftonline.com/{tenant_id}/v2.0`
+
+***
+
+### managedIdentityClientId?
+
+```ts
+optional managedIdentityClientId: string;
+```
+
+Client ID for user-assigned managed identity (azure_managed_identity).
+
+***
+
+### refreshBufferSecs?
+
+```ts
+optional refreshBufferSecs: number;
+```
+
+Seconds before 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.
+
+***
+
+### scopes
+
+```ts
+scopes: string[];
+```
+
+OAuth scopes to request. For Azure managed identity, exactly one scope
+or resource is required. For example: `["api://{app_id}/.default"]`
--- a/docs/src/js/interfaces/OAuthConfig.md
+++ b/docs/src/js/interfaces/OAuthConfig.md
@@ -0,0 +1,111 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / OAuthConfig
+
+# Interface: OAuthConfig
+
+OAuth configuration for LanceDB authentication.
+
+This is the public TypeScript OAuth configuration type. The generated
+`NativeOAuthConfig` type has the same runtime shape but is an implementation
+detail of the napi-rs binding.
+
+All token acquisition and refresh is handled in the Rust layer.
+This config is passed through to Rust via napi-rs.
+
+## Examples
+
+```typescript
+const config: OAuthConfig = {
+  issuerUrl: "https://login.microsoftonline.com/{tenant}/v2.0",
+  clientId: "app-id",
+  clientSecret: "secret",
+  scopes: ["api://lancedb-api/.default"],
+};
+```
+
+```typescript
+const config: OAuthConfig = {
+  issuerUrl: "https://login.microsoftonline.com/{tenant}/v2.0",
+  clientId: "app-id",
+  scopes: ["api://lancedb-api/.default"],
+  flow: OAuthFlowType.AzureManagedIdentity,
+};
+```
+
+## Properties
+
+### clientId
+
+```ts
+clientId: string;
+```
+
+Application / Client ID.
+
+***
+
+### clientSecret?
+
+```ts
+optional clientSecret: string;
+```
+
+Client secret (required for ClientCredentials).
+
+***
+
+### flow?
+
+```ts
+optional flow: OAuthFlowType;
+```
+
+Authentication flow (default: ClientCredentials).
+
+***
+
+### issuerUrl
+
+```ts
+issuerUrl: string;
+```
+
+OIDC issuer URL or OAuth authority URL.
+For Azure: `https://login.microsoftonline.com/{tenant_id}/v2.0`
+
+***
+
+### managedIdentityClientId?
+
+```ts
+optional managedIdentityClientId: string;
+```
+
+Client ID for user-assigned managed identity (AzureManagedIdentity).
+
+***
+
+### refreshBufferSecs?
+
+```ts
+optional refreshBufferSecs: number;
+```
+
+Seconds before 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.
+
+***
+
+### scopes
+
+```ts
+scopes: string[];
+```
+
+OAuth scopes to request.
+For Azure managed identity, exactly one scope or resource is required.
+For example: `["api://{app_id}/.default"]`
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>9.0.0-beta.2</lance-core.version>
+        <lance-core.version>8.0.0-rc.1</lance-core.version>
        <spotless.skip>false</spotless.skip>
        <spotless.version>2.30.0</spotless.version>
        <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
--- a/nodejs/lancedb/index.ts
+++ b/nodejs/lancedb/index.ts
@@ -52,6 +52,7 @@ export {
  SplitHashOptions,
  SplitSequentialOptions,
  ShuffleOptions,
+  OAuthConfig as NativeOAuthConfig,
 } from "./native.js";

 export {
@@ -130,6 +131,8 @@ export {
  TokenResponse,
 } from "./header";

+export { OAuthConfig, OAuthFlowType } from "./oauth";
+
 export { MergeInsertBuilder, WriteExecutionOptions } from "./merge";

 export * as embedding from "./embedding";
--- a/nodejs/lancedb/oauth.ts
+++ b/nodejs/lancedb/oauth.ts
@@ -0,0 +1,76 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+/**
+ * OAuth authentication flow types.
+ */
+export enum OAuthFlowType {
+  /** Client Credentials grant (service-to-service / M2M). */
+  ClientCredentials = "client_credentials",
+  /** Azure Managed Identity via IMDS. */
+  AzureManagedIdentity = "azure_managed_identity",
+}
+
+/**
+ * OAuth configuration for LanceDB authentication.
+ *
+ * This is the public TypeScript OAuth configuration type. The generated
+ * `NativeOAuthConfig` type has the same runtime shape but is an implementation
+ * detail of the napi-rs binding.
+ *
+ * All token acquisition and refresh is handled in the Rust layer.
+ * This config is passed through to Rust via napi-rs.
+ *
+ * @example Client Credentials (service-to-service):
+ * ```typescript
+ * const config: OAuthConfig = {
+ *   issuerUrl: "https://login.microsoftonline.com/{tenant}/v2.0",
+ *   clientId: "app-id",
+ *   clientSecret: "secret",
+ *   scopes: ["api://lancedb-api/.default"],
+ * };
+ * ```
+ *
+ * @example Azure Managed Identity:
+ * ```typescript
+ * const config: OAuthConfig = {
+ *   issuerUrl: "https://login.microsoftonline.com/{tenant}/v2.0",
+ *   clientId: "app-id",
+ *   scopes: ["api://lancedb-api/.default"],
+ *   flow: OAuthFlowType.AzureManagedIdentity,
+ * };
+ * ```
+ */
+export interface OAuthConfig {
+  /**
+   * OIDC issuer URL or OAuth authority URL.
+   * For Azure: `https://login.microsoftonline.com/{tenant_id}/v2.0`
+   */
+  issuerUrl: string;
+
+  /** Application / Client ID. */
+  clientId: string;
+
+  /**
+   * OAuth scopes to request.
+   * For Azure managed identity, exactly one scope or resource is required.
+   * For example: `["api://{app_id}/.default"]`
+   */
+  scopes: string[];
+
+  /** Authentication flow (default: ClientCredentials). */
+  flow?: OAuthFlowType;
+
+  /** Client secret (required for ClientCredentials). */
+  clientSecret?: string;
+
+  /** Client ID for user-assigned managed identity (AzureManagedIdentity). */
+  managedIdentityClientId?: string;
+
+  /**
+   * Seconds before 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.
+   */
+  refreshBufferSecs?: number;
+}
--- a/nodejs/src/connection.rs
+++ b/nodejs/src/connection.rs
@@ -112,6 +112,12 @@ impl Connection {

        builder = builder.client_config(rust_config);

+        if let Some(oauth_config) = options.oauth_config {
+            let config: lancedb::remote::oauth::OAuthConfig =
+                oauth_config.try_into().default_error()?;
+            builder = builder.oauth_config(config);
+        }
+
        if let Some(api_key) = options.api_key {
            builder = builder.api_key(&api_key);
        }
--- a/nodejs/src/lib.rs
+++ b/nodejs/src/lib.rs
@@ -65,6 +65,11 @@ pub struct ConnectionOptions {
    /// (For LanceDB cloud only): the host to use for LanceDB cloud. Used
    /// for testing purposes.
    pub host_override: Option<String>,
+    /// (For LanceDB cloud only): OAuth configuration for IdP-based
+    /// authentication (e.g., Azure Entra ID). When set, token acquisition
+    /// and refresh are handled entirely in Rust. TypeScript users should pass
+    /// the public `OAuthConfig` type exported from `@lancedb/lancedb`.
+    pub oauth_config: Option<remote::OAuthConfig>,
 }

 #[napi(object)]
--- a/nodejs/src/remote.rs
+++ b/nodejs/src/remote.rs
@@ -3,6 +3,7 @@

 use std::collections::HashMap;

+use lancedb::error::Error;
 use napi_derive::*;

 /// Timeout configuration for remote HTTP client.
@@ -140,6 +141,84 @@ impl From<TlsConfig> for lancedb::remote::TlsConfig {
    }
 }

+/// OAuth configuration for LanceDB authentication.
+///
+/// This is the generated napi-rs binding shape. TypeScript users should prefer
+/// the public `OAuthConfig` type exported from `@lancedb/lancedb`.
+///
+/// All token acquisition and refresh is handled in the Rust layer.
+#[napi(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,
+    /// 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: "client_credentials" or "azure_managed_identity"
+    pub flow: Option<String>,
+    /// Client secret (required for client_credentials).
+    pub client_secret: Option<String>,
+    /// Client ID for user-assigned managed identity (azure_managed_identity).
+    pub managed_identity_client_id: Option<String>,
+    /// Seconds before 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<u32>,
+}
+
+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("scopes", &self.scopes)
+            .field("flow", &self.flow)
+            .field(
+                "client_secret",
+                &self.client_secret.as_deref().map(|_| "<redacted>"),
+            )
+            .field(
+                "managed_identity_client_id",
+                &self.managed_identity_client_id,
+            )
+            .field("refresh_buffer_secs", &self.refresh_buffer_secs)
+            .finish()
+    }
+}
+
+impl TryFrom<OAuthConfig> for lancedb::remote::oauth::OAuthConfig {
+    type Error = Error;
+
+    fn try_from(config: OAuthConfig) -> Result<Self, Self::Error> {
+        use lancedb::remote::oauth::OAuthFlow;
+
+        let flow = match config.flow.as_deref().unwrap_or("client_credentials") {
+            "client_credentials" => OAuthFlow::ClientCredentials,
+            "azure_managed_identity" => OAuthFlow::AzureManagedIdentity {
+                client_id: config.managed_identity_client_id,
+            },
+            other => {
+                return Err(Error::InvalidInput {
+                    message: format!("Unknown OAuth flow type: {other}"),
+                });
+            }
+        };
+
+        Ok(Self {
+            issuer_url: config.issuer_url,
+            client_id: config.client_id,
+            client_secret: config.client_secret,
+            scopes: config.scopes,
+            flow,
+            refresh_buffer_secs: config.refresh_buffer_secs.map(|v| v as u64),
+        })
+    }
+}
+
 impl From<ClientConfig> for lancedb::remote::ClientConfig {
    fn from(config: ClientConfig) -> Self {
        Self {
@@ -156,3 +235,45 @@ impl From<ClientConfig> for lancedb::remote::ClientConfig {
        }
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_unknown_oauth_flow_returns_invalid_input() {
+        let config = OAuthConfig {
+            issuer_url: "https://issuer.example.com".to_string(),
+            client_id: "client-id".to_string(),
+            scopes: vec!["scope".to_string()],
+            flow: Some("typo".to_string()),
+            client_secret: None,
+            managed_identity_client_id: None,
+            refresh_buffer_secs: None,
+        };
+
+        let err = lancedb::remote::oauth::OAuthConfig::try_from(config).unwrap_err();
+        assert!(matches!(
+            err,
+            Error::InvalidInput { message }
+                if message == "Unknown OAuth flow type: typo"
+        ));
+    }
+
+    #[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(),
+            scopes: vec!["scope".to_string()],
+            flow: Some("client_credentials".to_string()),
+            client_secret: Some("super-secret".to_string()),
+            managed_identity_client_id: None,
+            refresh_buffer_secs: None,
+        };
+
+        let debug = format!("{config:?}");
+        assert!(!debug.contains("super-secret"));
+        assert!(debug.contains("client_secret: Some(\"<redacted>\")"));
+    }
+}
--- a/python/.bumpversion.toml
+++ b/python/.bumpversion.toml
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.34.0-beta.2"
+current_version = "0.34.0-beta.1"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
--- a/python/Cargo.toml
+++ b/python/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.34.0-beta.2"
+version = "0.34.0-beta.1"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
--- a/python/python/lancedb/init.py
+++ b/python/python/lancedb/init.py
@@ -89,6 +89,8 @@ def connect(
        If presented, connect to LanceDB cloud.
        Otherwise, connect to a database on file system or cloud storage.
        Can be set via environment variable `LANCEDB_API_KEY`.
+        OAuth configuration is currently supported only by ``connect_async``;
+        synchronous LanceDB Cloud connections require an API key.
    region: str, default "us-east-1"
        The region to use for LanceDB Cloud.
    host_override: str, optional
@@ -340,6 +342,7 @@ async def connect_async(
    session: Optional[Session] = None,
    manifest_enabled: bool = False,
    namespace_client_properties: Optional[Dict[str, str]] = None,
+    oauth_config=None,
 ) -> AsyncConnection:
    """Connect to a LanceDB database.

@@ -389,6 +392,10 @@ async def connect_async(
    namespace_client_properties : dict, optional
        Additional directory namespace client properties to use with
        ``manifest_enabled=True``.
+    oauth_config : OAuthConfig, optional
+        OAuth configuration for LanceDB Cloud/Enterprise. This is supported by
+        ``connect_async`` only; synchronous ``connect`` uses API key
+        authentication for ``db://`` URIs.

    Examples
    --------
@@ -435,6 +442,7 @@ async def connect_async(
            session,
            manifest_enabled,
            namespace_client_properties,
+            oauth_config,
        )
    )

--- a/python/python/lancedb/_lancedb.pyi
+++ b/python/python/lancedb/_lancedb.pyi
@@ -280,6 +280,7 @@ async def connect(
    session: Optional[Session],
    manifest_enabled: bool = False,
    namespace_client_properties: Optional[Dict[str, str]] = None,
+    oauth_config: Optional[Any] = None,
 ) -> Connection: ...

 class RecordBatchStream:
--- a/python/python/lancedb/remote/init.py
+++ b/python/python/lancedb/remote/init.py
@@ -9,6 +9,7 @@ from typing import List, Optional
 from lancedb import __version__

 from .header import HeaderProvider
+from .oauth import OAuthConfig, OAuthFlowType

 __all__ = [
    "TimeoutConfig",
@@ -16,6 +17,8 @@ __all__ = [
    "TlsConfig",
    "ClientConfig",
    "HeaderProvider",
+    "OAuthConfig",
+    "OAuthFlowType",
 ]


--- a/python/python/lancedb/remote/oauth.py
+++ b/python/python/lancedb/remote/oauth.py
@@ -0,0 +1,75 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import List, Optional
+
+
+class OAuthFlowType(str, Enum):
+    """OAuth authentication flow types."""
+
+    CLIENT_CREDENTIALS = "client_credentials"
+    """Client Credentials grant (service-to-service / M2M)."""
+
+    AZURE_MANAGED_IDENTITY = "azure_managed_identity"
+    """Azure Managed Identity via IMDS."""
+
+
+@dataclass
+class OAuthConfig:
+    """OAuth configuration for LanceDB authentication.
+
+    All token acquisition and refresh is handled in the Rust layer.
+    This config is passed through to Rust via PyO3.
+
+    Parameters
+    ----------
+    issuer_url : str
+        OIDC issuer URL or OAuth authority URL.
+        For Azure: ``https://login.microsoftonline.com/{tenant_id}/v2.0``
+    client_id : str
+        Application / Client ID.
+    scopes : List[str]
+        OAuth scopes to request.
+        For Azure managed identity, exactly one scope or resource is required.
+        For example: ``["api://{app_id}/.default"]``
+    flow : OAuthFlowType
+        Authentication flow to use. Default: CLIENT_CREDENTIALS.
+    client_secret : Optional[str]
+        Client secret (required for CLIENT_CREDENTIALS).
+    managed_identity_client_id : Optional[str]
+        Client ID for user-assigned managed identity (AZURE_MANAGED_IDENTITY).
+    refresh_buffer_secs : Optional[int]
+        Seconds before 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.
+
+    Examples
+    --------
+    Client Credentials (service-to-service):
+
+    >>> config = OAuthConfig(
+    ...     issuer_url="https://login.microsoftonline.com/{tenant}/v2.0",
+    ...     client_id="app-id",
+    ...     client_secret="secret",
+    ...     scopes=["api://lancedb-api/.default"],
+    ... )
+
+    Azure Managed Identity:
+
+    >>> config = OAuthConfig(
+    ...     issuer_url="https://login.microsoftonline.com/{tenant}/v2.0",
+    ...     client_id="app-id",
+    ...     scopes=["api://lancedb-api/.default"],
+    ...     flow=OAuthFlowType.AZURE_MANAGED_IDENTITY,
+    ... )
+    """
+
+    issuer_url: str
+    client_id: str
+    scopes: List[str]
+    flow: OAuthFlowType = OAuthFlowType.CLIENT_CREDENTIALS
+    client_secret: Optional[str] = field(default=None, repr=False)
+    managed_identity_client_id: Optional[str] = None
+    refresh_buffer_secs: Optional[int] = None
--- a/python/src/connection.rs
+++ b/python/src/connection.rs
@@ -539,7 +539,7 @@ impl Connection {
 }

 #[pyfunction]
-#[pyo3(signature = (uri, api_key=None, region=None, host_override=None, read_consistency_interval=None, client_config=None, storage_options=None, session=None, manifest_enabled=false, namespace_client_properties=None))]
+#[pyo3(signature = (uri, api_key=None, region=None, host_override=None, read_consistency_interval=None, client_config=None, storage_options=None, session=None, manifest_enabled=false, namespace_client_properties=None, oauth_config=None))]
 #[allow(clippy::too_many_arguments)]
 pub fn connect(
    py: Python<'_>,
@@ -553,6 +553,7 @@ pub fn connect(
    session: Option<crate::session::Session>,
    manifest_enabled: bool,
    namespace_client_properties: Option<HashMap<String, String>>,
+    oauth_config: Option<crate::oauth::PyOAuthConfig>,
 ) -> PyResult<Bound<'_, PyAny>> {
    future_into_py(py, async move {
        let mut builder = lancedb::connect(&uri);
@@ -582,6 +583,11 @@ pub fn connect(
        if let Some(client_config) = client_config {
            builder = builder.client_config(client_config.into());
        }
+        if let Some(oauth_config) = oauth_config {
+            let config: lancedb::remote::oauth::OAuthConfig =
+                oauth_config.try_into().infer_error()?;
+            builder = builder.oauth_config(config);
+        }
        if let Some(session) = session {
            builder = builder.session(session.inner.clone());
        }
--- a/python/src/lib.rs
+++ b/python/src/lib.rs
@@ -26,6 +26,7 @@ pub mod expr;
 pub mod header;
 pub mod index;
 pub mod namespace;
+pub mod oauth;
 pub mod permutation;
 pub mod query;
 pub mod runtime;
--- a/python/src/oauth.rs
+++ b/python/src/oauth.rs
@@ -0,0 +1,72 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+use pyo3::FromPyObject;
+
+use lancedb::error::Error;
+use lancedb::remote::oauth::{OAuthConfig, OAuthFlow};
+
+/// Python-side OAuth configuration, extracted via FromPyObject.
+/// Maps to `lancedb.remote.oauth.OAuthConfig` Python dataclass.
+#[derive(FromPyObject)]
+pub struct PyOAuthConfig {
+    pub issuer_url: String,
+    pub client_id: String,
+    pub scopes: Vec<String>,
+    pub flow: String,
+    pub client_secret: Option<String>,
+    pub managed_identity_client_id: Option<String>,
+    pub refresh_buffer_secs: Option<u64>,
+}
+
+impl TryFrom<PyOAuthConfig> for OAuthConfig {
+    type Error = Error;
+
+    fn try_from(py: PyOAuthConfig) -> Result<Self, Self::Error> {
+        let flow = match py.flow.as_str() {
+            "client_credentials" => OAuthFlow::ClientCredentials,
+            "azure_managed_identity" => OAuthFlow::AzureManagedIdentity {
+                client_id: py.managed_identity_client_id,
+            },
+            other => {
+                return Err(Error::InvalidInput {
+                    message: format!("Unknown OAuth flow type: {other}"),
+                });
+            }
+        };
+
+        Ok(Self {
+            issuer_url: py.issuer_url,
+            client_id: py.client_id,
+            client_secret: py.client_secret,
+            scopes: py.scopes,
+            flow,
+            refresh_buffer_secs: py.refresh_buffer_secs,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_unknown_oauth_flow_returns_invalid_input() {
+        let config = PyOAuthConfig {
+            issuer_url: "https://issuer.example.com".to_string(),
+            client_id: "client-id".to_string(),
+            scopes: vec!["scope".to_string()],
+            flow: "typo".to_string(),
+            client_secret: None,
+            managed_identity_client_id: None,
+            refresh_buffer_secs: None,
+        };
+
+        let err = OAuthConfig::try_from(config).unwrap_err();
+        assert!(matches!(
+            err,
+            Error::InvalidInput { message }
+                if message == "Unknown OAuth flow type: typo"
+        ));
+    }
+}
--- a/python/tests/test_oauth.py
+++ b/python/tests/test_oauth.py
@@ -0,0 +1,33 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import importlib.util
+import sys
+from pathlib import Path
+
+
+def _load_oauth_module():
+    oauth_path = (
+        Path(__file__).parents[1] / "python" / "lancedb" / "remote" / "oauth.py"
+    )
+    spec = importlib.util.spec_from_file_location("lancedb_remote_oauth", oauth_path)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_oauth_config_repr_redacts_client_secret():
+    oauth = _load_oauth_module()
+
+    config = oauth.OAuthConfig(
+        issuer_url="https://issuer.example.com",
+        client_id="client-id",
+        scopes=["scope"],
+        client_secret="super-secret",
+    )
+
+    rendered = repr(config)
+    assert "super-secret" not in rendered
+    assert "client_secret" not in rendered
--- a/rust/lancedb/Cargo.toml
+++ b/rust/lancedb/Cargo.toml
@@ -75,6 +75,8 @@ reqwest = { version = "0.12.0", default-features = false, features = [
    "stream",
 ], optional = true }
 http = { version = "1", optional = true } # Matching what is in reqwest
+# OAuth dependencies (used by remote feature)
+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 }
@@ -93,6 +95,7 @@ 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"] }
 uuid = { version = "1.7.0", features = ["v4"] }
 walkdir = "2"
 aws-sdk-dynamodb = { version = "1.55.0" }
@@ -129,7 +132,7 @@ huggingface = [
    "lance-namespace-impls/dir-huggingface",
 ]
 dynamodb = ["lance/dynamodb", "aws"]
-remote = ["dep:reqwest", "dep:http", "lance-namespace-impls/rest", "lance-namespace-impls/rest-adapter"]
+remote = ["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"]
--- a/rust/lancedb/src/blob.rs
+++ b/rust/lancedb/src/blob.rs
@@ -9,28 +9,16 @@
 //!
 //! Blob tables require Lance file format >= 2.2 and stable row ids at create.

-use std::sync::Arc;
-
-use arrow_array::builder::LargeBinaryBuilder;
-use arrow_array::{Array, LargeBinaryArray, RecordBatch, StructArray, UInt8Array, UInt64Array};
-use arrow_schema::{DataType, Field, Schema};
-use lance::dataset::{Dataset, WriteParams};
+use arrow_schema::{Field, Schema};
+use lance::dataset::WriteParams;
 use lance_arrow::FieldExt;
-use lance_core::datatypes::parse_field_path;
 use lance_encoding::version::LanceFileVersion;

-use crate::error::{Error, Result};
-
-pub use lance::dataset::BlobFile;
-
 /// Creates an Arrow field for a Lance blob v2 column.
 ///
 /// `Struct<data, uri>` with the `lance.blob.v2` marker. Same layout Lance
 /// expects on write.
 ///
-/// A blob column may be top-level or nested inside a struct or list. Nested
-/// blobs are addressed by a dotted path (e.g. `info.blob`) in the read APIs.
-///
 /// ```
 /// use arrow_schema::{DataType, Field, Schema};
 ///
@@ -39,71 +27,15 @@ pub use lance::dataset::BlobFile;
 ///     lancedb::blob("image", true),
 /// ]);
 /// ```
+///
+/// Blob tables use Lance file format >= 2.2 and stable row ids at create.
 pub fn blob(name: impl AsRef<str>, nullable: bool) -> Field {
    lance::blob::blob_field(name.as_ref(), nullable)
 }

-/// Returns true if `field` is a blob v2 column.
-///
-/// ```
-/// let field = lancedb::blob("image", true);
-/// assert!(lancedb::blob::is_blob(&field));
-/// ```
-pub fn is_blob(field: &Field) -> bool {
-    field.is_blob_v2()
-}
-
-/// Returns true if `field`, or any field nested under it, is a blob v2 column.
-fn field_tree_has_blob_v2(field: &Field) -> bool {
-    if field.is_blob_v2() {
-        return true;
-    }
-    match field.data_type() {
-        DataType::Struct(children) => children.iter().any(|c| field_tree_has_blob_v2(c)),
-        DataType::List(child) | DataType::LargeList(child) | DataType::FixedSizeList(child, _) => {
-            field_tree_has_blob_v2(child)
-        }
-        _ => false,
-    }
-}
-
-/// Collects the dotted paths of blob v2 columns under `field`, into `paths`.
-fn collect_blob_paths(field: &Field, prefix: &str, paths: &mut Vec<String>) {
-    let path = if prefix.is_empty() {
-        field.name().clone()
-    } else {
-        format!("{prefix}.{}", field.name())
-    };
-    if field.is_blob_v2() {
-        paths.push(path);
-        return;
-    }
-    match field.data_type() {
-        DataType::Struct(children) => {
-            for child in children {
-                collect_blob_paths(child, &path, paths);
-            }
-        }
-        DataType::List(child) | DataType::LargeList(child) | DataType::FixedSizeList(child, _) => {
-            collect_blob_paths(child, &path, paths)
-        }
-        _ => {}
-    }
-}
-
-/// Returns true if `schema` declares any blob v2 column, including nested ones.
+/// Returns true if `schema` declares any blob v2 column.
 pub(crate) fn has_blob_columns(schema: &Schema) -> bool {
-    schema.fields().iter().any(|f| field_tree_has_blob_v2(f))
-}
-
-/// Blob v2 column paths in `schema`, declaration order preserved. Nested blobs
-/// are dotted paths (e.g. `info.blob`).
-pub(crate) fn blob_column_names(schema: &Schema) -> Vec<String> {
-    let mut paths = Vec::new();
-    for field in schema.fields() {
-        collect_blob_paths(field, "", &mut paths);
-    }
-    paths
+    schema.fields().iter().any(|field| field.is_blob_v2())
 }

 /// Bumps storage format to at least [`LanceFileVersion::V2_2`] for blob schemas.
@@ -121,206 +53,6 @@ pub(crate) fn ensure_blob_storage_version(schema: &Schema, params: &mut WritePar
    }
 }

-/// Validate that `column` exists and is a blob v2 column.
-///
-/// Legacy v1 columns (`lance-encoding:blob`) error with a migration hint.
-pub(crate) fn ensure_blob_v2_column(
-    schema: &lance_core::datatypes::Schema,
-    column: &str,
-) -> Result<()> {
-    match schema.field(column) {
-        Some(field) if field.is_blob_v2() => Ok(()),
-        Some(field) if field.is_blob() => Err(Error::InvalidInput {
-            message: format!(
-                "column '{column}' is a legacy blob column; blob APIs require blob v2 columns \
-                 (ARROW:extension:name = \"lance.blob.v2\")"
-            ),
-        }),
-        Some(_) => Err(Error::InvalidInput {
-            message: format!("column '{column}' is not a blob column"),
-        }),
-        None => Err(Error::InvalidInput {
-            message: format!("no column named '{column}' in this table"),
-        }),
-    }
-}
-
-/// Returns the leaf descriptor `StructArray` for `column` in a descriptor batch.
-fn leaf_descriptor_struct<'a>(batch: &'a RecordBatch, column: &str) -> Result<&'a StructArray> {
-    let path = parse_field_path(column).map_err(|e| Error::InvalidInput {
-        message: format!("invalid blob column path '{column}': {e}"),
-    })?;
-    let not_struct = || Error::Runtime {
-        message: format!("blob column '{column}' did not read back as a descriptor struct"),
-    };
-    let mut current = batch
-        .column_by_name(&path[0])
-        .and_then(|c| c.as_any().downcast_ref::<StructArray>())
-        .ok_or_else(not_struct)?;
-    for segment in &path[1..] {
-        current = current
-            .column_by_name(segment)
-            .and_then(|c| c.as_any().downcast_ref::<StructArray>())
-            .ok_or_else(not_struct)?;
-    }
-    Ok(current)
-}
-
-/// Null rows in `row_ids`, from a descriptor take.
-///
-/// Lance `read_blobs` / `take_blobs` skip null rows (`kind == 0 && position == 0 && size == 0`).
-/// TODO(lance): aligned read API would drop this pass.
-async fn blob_null_mask(
-    dataset: &Arc<Dataset>,
-    column: &str,
-    row_ids: &[u64],
-) -> Result<Vec<bool>> {
-    let projection = dataset.schema().project(&[column])?;
-    let descriptors = dataset.take_builder(row_ids, projection)?.execute().await?;
-    if descriptors.num_rows() != row_ids.len() {
-        return Err(Error::InvalidInput {
-            message: format!(
-                "blob take for column '{column}' requested {} row ids but only {} exist in the \
-                 table; pass row ids collected from this table",
-                row_ids.len(),
-                descriptors.num_rows()
-            ),
-        });
-    }
-    let descriptor_struct = leaf_descriptor_struct(&descriptors, column)?;
-    let child = |name: &str| {
-        descriptor_struct
-            .column_by_name(name)
-            .ok_or_else(|| Error::Runtime {
-                message: format!("blob descriptor for '{column}' is missing the '{name}' field"),
-            })
-    };
-    let kinds = child("kind")?
-        .as_any()
-        .downcast_ref::<UInt8Array>()
-        .ok_or_else(|| Error::Runtime {
-            message: format!("blob descriptor 'kind' for '{column}' is not a UInt8 array"),
-        })?;
-    let positions = child("position")?
-        .as_any()
-        .downcast_ref::<UInt64Array>()
-        .ok_or_else(|| Error::Runtime {
-            message: format!("blob descriptor 'position' for '{column}' is not a UInt64 array"),
-        })?;
-    let sizes = child("size")?
-        .as_any()
-        .downcast_ref::<UInt64Array>()
-        .ok_or_else(|| Error::Runtime {
-            message: format!("blob descriptor 'size' for '{column}' is not a UInt64 array"),
-        })?;
-
-    // Match Lance `collect_blob_entries_v2` skip condition (`BlobKind::Inline` == 0).
-    Ok((0..descriptor_struct.len())
-        .map(|i| {
-            descriptor_struct.is_null(i)
-                || kinds.is_null(i)
-                || (kinds.value(i) == 0 && positions.value(i) == 0 && sizes.value(i) == 0)
-        })
-        .collect())
-}
-
-fn non_null_row_ids(row_ids: &[u64], null_mask: &[bool]) -> Vec<u64> {
-    row_ids
-        .iter()
-        .zip(null_mask)
-        .filter_map(|(row_id, is_null)| (!is_null).then_some(*row_id))
-        .collect()
-}
-
-/// Materialize blob bytes for `row_ids` (same length and order, nulls preserved).
-pub(crate) async fn take_blobs_aligned(
-    dataset: &Arc<Dataset>,
-    column: &str,
-    row_ids: &[u64],
-) -> Result<LargeBinaryArray> {
-    ensure_blob_v2_column(dataset.schema(), column)?;
-    if row_ids.is_empty() {
-        return Ok(LargeBinaryBuilder::new().finish());
-    }
-
-    let null_mask = blob_null_mask(dataset, column, row_ids).await?;
-    let non_null_row_ids = non_null_row_ids(row_ids, &null_mask);
-    let non_null_count = non_null_row_ids.len();
-    let payloads = if non_null_count == 0 {
-        Vec::new()
-    } else {
-        dataset
-            .read_blobs(column)?
-            .with_row_ids(non_null_row_ids)
-            .preserve_order(true)
-            .execute()
-            .await?
-    };
-
-    if payloads.len() != non_null_count {
-        return Err(Error::Runtime {
-            message: format!(
-                "blob read for column '{column}' returned {} payloads for {} non-null rows",
-                payloads.len(),
-                non_null_count
-            ),
-        });
-    }
-
-    let mut builder = LargeBinaryBuilder::new();
-    let mut payload_idx = 0;
-    for is_null in &null_mask {
-        if *is_null {
-            builder.append_null();
-        } else {
-            builder.append_value(payloads[payload_idx].data.as_ref());
-            payload_idx += 1;
-        }
-    }
-    Ok(builder.finish())
-}
-
-/// Open lazy [`BlobFile`] handles for `row_ids` (same length and order, nulls as `None`).
-pub(crate) async fn take_blob_files_aligned(
-    dataset: &Arc<Dataset>,
-    column: &str,
-    row_ids: &[u64],
-) -> Result<Vec<Option<BlobFile>>> {
-    ensure_blob_v2_column(dataset.schema(), column)?;
-    if row_ids.is_empty() {
-        return Ok(Vec::new());
-    }
-
-    let null_mask = blob_null_mask(dataset, column, row_ids).await?;
-    let non_null_row_ids = non_null_row_ids(row_ids, &null_mask);
-    let handles = if non_null_row_ids.is_empty() {
-        Vec::new()
-    } else {
-        dataset.take_blobs(&non_null_row_ids, column).await?
-    };
-    if handles.len() != non_null_row_ids.len() {
-        return Err(Error::Runtime {
-            message: format!(
-                "blob take for column '{column}' returned {} handles for {} non-null rows",
-                handles.len(),
-                non_null_row_ids.len()
-            ),
-        });
-    }
-
-    let mut handles = handles.into_iter();
-    Ok(null_mask
-        .iter()
-        .map(|is_null| {
-            if *is_null {
-                None
-            } else {
-                Some(handles.next().unwrap())
-            }
-        })
-        .collect())
-}
-
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -384,47 +116,6 @@ mod tests {
        assert_eq!(params.data_storage_version.unwrap(), LanceFileVersion::V2_3);
    }

-    #[test]
-    fn legacy_v1_blob_column_is_rejected_with_migration_hint() {
-        let legacy = Field::new("image", DataType::LargeBinary, true).with_metadata(
-            std::collections::HashMap::from([(
-                "lance-encoding:blob".to_string(),
-                "true".to_string(),
-            )]),
-        );
-        let arrow_schema = Schema::new(vec![legacy]);
-        let lance_schema = lance_core::datatypes::Schema::try_from(&arrow_schema).unwrap();
-
-        let err = ensure_blob_v2_column(&lance_schema, "image").unwrap_err();
-        assert!(matches!(err, Error::InvalidInput { .. }));
-        assert!(err.to_string().contains("legacy blob column"));
-        assert!(err.to_string().contains("lance.blob.v2"));
-    }
-
-    #[test]
-    fn non_blob_and_unknown_columns_are_rejected_by_name() {
-        let arrow_schema = Schema::new(vec![Field::new("id", DataType::Int64, false)]);
-        let lance_schema = lance_core::datatypes::Schema::try_from(&arrow_schema).unwrap();
-
-        let err = ensure_blob_v2_column(&lance_schema, "id").unwrap_err();
-        assert!(err.to_string().contains("'id' is not a blob column"));
-
-        let err = ensure_blob_v2_column(&lance_schema, "missing").unwrap_err();
-        assert!(err.to_string().contains("no column named 'missing'"));
-    }
-
-    #[test]
-    fn blob_column_names_includes_nested_path() {
-        let blob_field = blob("blob", true);
-        let info = Field::new(
-            "info",
-            DataType::Struct(vec![Field::new("name", DataType::Utf8, false), blob_field].into()),
-            true,
-        );
-        let schema = Schema::new(vec![Field::new("id", DataType::Int64, false), info]);
-        assert_eq!(blob_column_names(&schema), vec!["info.blob"]);
-    }
-
    #[test]
    fn storage_version_noop_without_blob_columns() {
        let schema = Schema::new(vec![Field::new("id", DataType::Int64, false)]);
--- a/rust/lancedb/src/connection.rs
+++ b/rust/lancedb/src/connection.rs
@@ -576,6 +576,10 @@ impl Connection {
    /// For LanceNamespaceDatabase, it is the underlying LanceNamespace.
    /// For ListingDatabase, it is the equivalent DirectoryNamespace.
    /// For RemoteDatabase, it is the equivalent RestNamespace.
+    ///
+    /// Remote connections using dynamic headers, including OAuth, are not
+    /// currently supported because the namespace client only accepts static
+    /// headers.
    pub async fn namespace_client(&self) -> Result<Arc<dyn lance_namespace::LanceNamespace>> {
        self.internal.namespace_client().await
    }
@@ -584,6 +588,10 @@ impl Connection {
    /// Returns (impl_type, properties) where:
    /// - impl_type: "dir" for DirectoryNamespace, "rest" for RestNamespace
    /// - properties: configuration properties for the namespace
+    ///
+    /// Remote connections using dynamic headers, including OAuth, are not
+    /// currently supported because the namespace client config only carries
+    /// static headers.
    pub async fn namespace_client_config(
        &self,
    ) -> Result<(String, std::collections::HashMap<String, String>)> {
@@ -661,6 +669,8 @@ pub struct ConnectRequest {
 pub struct ConnectBuilder {
    request: ConnectRequest,
    embedding_registry: Option<Arc<dyn EmbeddingRegistry>>,
+    #[cfg(feature = "remote")]
+    oauth_config: Option<crate::remote::oauth::OAuthConfig>,
 }

 #[cfg(feature = "remote")]
@@ -682,6 +692,8 @@ impl ConnectBuilder {
                session: None,
            },
            embedding_registry: None,
+            #[cfg(feature = "remote")]
+            oauth_config: None,
        }
    }

@@ -770,6 +782,19 @@ 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.
+    ///
+    /// Token acquisition and refresh are handled entirely in Rust.
+    #[cfg(feature = "remote")]
+    pub fn oauth_config(mut self, config: crate::remote::oauth::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);
@@ -915,9 +940,42 @@ 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 = options.api_key.ok_or_else(|| Error::InvalidInput {
-            message: "An api_key is required when connecting to LanceDb Cloud".to_string(),
-        })?;
+
+        // When OAuth is configured, api_key is not required
+        let api_key = match (&self.oauth_config, &options.api_key) {
+            (Some(_), None) => String::new(),
+            (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;
+
+        // Apply OAuth header provider if configured
+        if let Some(oauth_config) = self.oauth_config {
+            let provider = crate::remote::oauth::OAuthHeaderProvider::new(oauth_config)?;
+            client_config.header_provider =
+                Some(Arc::new(provider) as Arc<dyn crate::remote::client::HeaderProvider>);
+        }

        let storage_options = StorageOptions(options.storage_options.clone());
        let internal = Arc::new(crate::remote::db::RemoteDatabase::try_new(
@@ -925,7 +983,7 @@ impl ConnectBuilder {
            &api_key,
            &region,
            options.host_override,
-            self.request.client_config,
+            client_config,
            storage_options.into(),
            self.request.read_consistency_interval,
        )?);
@@ -1234,6 +1292,83 @@ 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::oauth::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::oauth::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::oauth::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::oauth::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() {
--- a/rust/lancedb/src/lib.rs
+++ b/rust/lancedb/src/lib.rs
@@ -189,7 +189,7 @@ use std::{fmt::Display, str::FromStr};

 use serde::{Deserialize, Serialize};

-pub use blob::{blob, is_blob};
+pub use blob::blob;
 pub use connection::{ConnectNamespaceBuilder, Connection};
 pub use error::{Error, Result};
 use lance_index::vector::ApproxMode as LanceApproxMode;
--- a/rust/lancedb/src/remote.rs
+++ b/rust/lancedb/src/remote.rs
@@ -8,6 +8,7 @@

 pub(crate) mod client;
 pub(crate) mod db;
+pub mod oauth;
 mod retry;
 pub(crate) mod table;
 pub(crate) mod util;
@@ -20,3 +21,4 @@ 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};
--- a/rust/lancedb/src/remote/client.rs
+++ b/rust/lancedb/src/remote/client.rs
@@ -459,12 +459,14 @@ impl<S: HttpSend> RestfulLanceDbClient<S> {
        config: &ClientConfig,
    ) -> Result<HeaderMap> {
        let mut headers = HeaderMap::new();
-        headers.insert(
-            HeaderName::from_static("x-api-key"),
-            HeaderValue::from_str(api_key).map_err(|_| Error::InvalidInput {
-                message: "non-ascii api key provided".to_string(),
-            })?,
-        );
+        if !api_key.is_empty() {
+            headers.insert(
+                HeaderName::from_static("x-api-key"),
+                HeaderValue::from_str(api_key).map_err(|_| Error::InvalidInput {
+                    message: "non-ascii api key provided".to_string(),
+                })?,
+            );
+        }
        if region == "local" {
            let host = format!("{}.local.api.lancedb.com", db_name);
            headers.insert(
@@ -1037,6 +1039,35 @@ mod tests {
        }
    }

+    #[test]
+    fn test_default_headers_skip_empty_api_key() {
+        let headers = RestfulLanceDbClient::<Sender>::default_headers(
+            "",
+            "us-east-1",
+            "db",
+            false,
+            &RemoteOptions(HashMap::new()),
+            None,
+            &ClientConfig::default(),
+        )
+        .unwrap();
+
+        assert!(!headers.contains_key("x-api-key"));
+
+        let headers = RestfulLanceDbClient::<Sender>::default_headers(
+            "api-key",
+            "us-east-1",
+            "db",
+            false,
+            &RemoteOptions(HashMap::new()),
+            None,
+            &ClientConfig::default(),
+        )
+        .unwrap();
+
+        assert_eq!(headers.get("x-api-key").unwrap(), "api-key");
+    }
+
    #[tokio::test]
    async fn test_client_config_with_header_provider() {
        let mut headers = HashMap::new();
--- a/rust/lancedb/src/remote/db.rs
+++ b/rust/lancedb/src/remote/db.rs
@@ -194,6 +194,7 @@ pub struct RemoteDatabase<S: HttpSend = Sender> {
    uri: String,
    /// Headers to pass to the namespace client for authentication
    namespace_headers: HashMap<String, String>,
+    has_dynamic_header_provider: bool,
    /// TLS configuration for mTLS support
    tls_config: Option<super::client::TlsConfig>,
 }
@@ -247,6 +248,7 @@ impl RemoteDatabase {
            table_cache,
            uri: uri.to_owned(),
            namespace_headers,
+            has_dynamic_header_provider: client_config.header_provider.is_some(),
            tls_config: client_config.tls_config,
        })
    }
@@ -271,6 +273,7 @@ mod test_utils {
                table_cache: Cache::new(0),
                uri: "http://localhost".to_string(),
                namespace_headers: HashMap::new(),
+                has_dynamic_header_provider: false,
                tls_config: None,
            }
        }
@@ -286,6 +289,7 @@ mod test_utils {
                table_cache: Cache::new(0),
                uri: "http://localhost".to_string(),
                namespace_headers: config.extra_headers.clone(),
+                has_dynamic_header_provider: config.header_provider.is_some(),
                tls_config: config.tls_config.clone(),
            }
        }
@@ -756,10 +760,17 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
    }

    async fn namespace_client(&self) -> Result<Arc<dyn lance_namespace::LanceNamespace>> {
+        if self.has_dynamic_header_provider {
+            return Err(Error::NotSupported {
+                message:
+                    "Cannot create a namespace client when dynamic headers are configured; use LanceDB connection namespace methods instead"
+                        .to_string(),
+            });
+        }
+
        // Create a RestNamespace pointing to the same remote host with the same authentication headers
        let mut builder = lance_namespace_impls::RestNamespaceBuilder::new(self.client.host())
            .delimiter(&self.client.id_delimiter)
-            // TODO: support header provider
            .headers(self.namespace_headers.clone());

        // Apply mTLS configuration if present
@@ -781,6 +792,14 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
    }

    async fn namespace_client_config(&self) -> Result<(String, HashMap<String, String>)> {
+        if self.has_dynamic_header_provider {
+            return Err(Error::NotSupported {
+                message:
+                    "Cannot export a namespace client config when dynamic headers are configured; use LanceDB connection namespace methods instead"
+                        .to_string(),
+            });
+        }
+
        let mut properties = HashMap::new();
        properties.insert("uri".to_string(), self.client.host().to_string());
        properties.insert("delimiter".to_string(), self.client.id_delimiter.clone());
@@ -1702,6 +1721,51 @@ mod tests {
        assert!(namespace_client.is_ok());
    }

+    #[tokio::test]
+    async fn test_namespace_client_rejects_dynamic_headers() {
+        #[derive(Debug)]
+        struct TestHeaderProvider;
+
+        #[async_trait::async_trait]
+        impl HeaderProvider for TestHeaderProvider {
+            async fn get_headers(&self) -> crate::Result<HashMap<String, String>> {
+                Ok(HashMap::from([(
+                    "authorization".to_string(),
+                    "Bearer token".to_string(),
+                )]))
+            }
+        }
+
+        let client_config = ClientConfig {
+            header_provider: Some(Arc::new(TestHeaderProvider) as Arc<dyn HeaderProvider>),
+            ..Default::default()
+        };
+
+        let conn = Connection::new_with_handler_and_config(
+            |_| {
+                http::Response::builder()
+                    .status(200)
+                    .body(r#"{"tables": []}"#)
+                    .unwrap()
+            },
+            client_config,
+        );
+
+        match conn.namespace_client().await {
+            Err(Error::NotSupported { message })
+                if message.contains("dynamic headers are configured") => {}
+            Err(err) => panic!("expected NotSupported, got {err:?}"),
+            Ok(_) => panic!("expected namespace_client to reject dynamic headers"),
+        }
+
+        match conn.namespace_client_config().await {
+            Err(Error::NotSupported { message })
+                if message.contains("dynamic headers are configured") => {}
+            Err(err) => panic!("expected NotSupported, got {err:?}"),
+            Ok(_) => panic!("expected namespace_client_config to reject dynamic headers"),
+        }
+    }
+
    /// Integration tests using RestAdapter to run RemoteDatabase against a real namespace server
    mod rest_adapter_integration {
        use super::*;
--- a/rust/lancedb/src/remote/oauth.rs
+++ b/rust/lancedb/src/remote/oauth.rs
@@ -0,0 +1,857 @@
+// 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 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.
+    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(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));
+    }
+}
+
+/// 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 {
+    config: OAuthConfig,
+    http_client: Client,
+    token_state: Arc<RwLock<TokenState>>,
+    /// Cached OIDC discovery document
+    discovery: Arc<RwLock<Option<OidcDiscovery>>>,
+    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("issuer_url", &self.config.issuer_url)
+            .field("client_id", &self.config.client_id)
+            .field("flow", &self.config.flow)
+            .finish()
+    }
+}
+
+impl OAuthHeaderProvider {
+    /// Create a new OAuth header provider from configuration.
+    pub fn new(config: OAuthConfig) -> Result<Self> {
+        // Validate config upfront
+        if matches!(config.flow, OAuthFlow::ClientCredentials) && config.client_secret.is_none() {
+            return Err(Error::InvalidInput {
+                message: "client_secret is required for ClientCredentials flow".to_string(),
+            });
+        }
+        if config.scopes.is_empty() {
+            return Err(Error::InvalidInput {
+                message: "At least one OAuth scope is required".to_string(),
+            });
+        }
+        if matches!(config.flow, OAuthFlow::AzureManagedIdentity { .. }) && config.scopes.len() != 1
+        {
+            return Err(Error::InvalidInput {
+                message: "AzureManagedIdentity flow requires exactly one OAuth scope or resource"
+                    .to_string(),
+            });
+        }
+        Self::validate_issuer_transport(&config)?;
+
+        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}"),
+            })?;
+
+        let refresh_buffer = Duration::from_secs(
+            config
+                .refresh_buffer_secs
+                .unwrap_or(DEFAULT_REFRESH_BUFFER_SECS),
+        );
+
+        Ok(Self {
+            config,
+            http_client,
+            token_state: Arc::new(RwLock::new(TokenState::new())),
+            discovery: Arc::new(RwLock::new(None)),
+            refresh_buffer,
+        })
+    }
+
+    fn validate_issuer_transport(config: &OAuthConfig) -> Result<()> {
+        if !matches!(config.flow, OAuthFlow::ClientCredentials) {
+            return Ok(());
+        }
+
+        let issuer = url::Url::parse(&config.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)
+    }
+
+    /// 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 {:?} flow", self.config.flow);
+        let resp = self.acquire_token().await?;
+
+        state.update(&resp);
+        Ok(resp.access_token)
+    }
+
+    /// Acquire a new token using the configured flow.
+    async fn acquire_token(&self) -> Result<TokenResponse> {
+        match &self.config.flow {
+            OAuthFlow::ClientCredentials => self.acquire_client_credentials().await,
+            OAuthFlow::AzureManagedIdentity { client_id } => {
+                self.acquire_managed_identity(client_id.as_deref()).await
+            }
+        }
+    }
+
+    // -- OIDC Discovery --
+
+    async fn get_discovery(&self) -> Result<OidcDiscovery> {
+        {
+            let cached = self.discovery.read().await;
+            if let Some(ref disc) = *cached {
+                return Ok(OidcDiscovery {
+                    token_endpoint: disc.token_endpoint.clone(),
+                });
+            }
+        }
+
+        let mut cache = self.discovery.write().await;
+        // Double-check
+        if let Some(ref disc) = *cache {
+            return Ok(OidcDiscovery {
+                token_endpoint: disc.token_endpoint.clone(),
+            });
+        }
+
+        let discovery_url = format!(
+            "{}/.well-known/openid-configuration",
+            self.config.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 = OidcDiscovery {
+            token_endpoint: disc.token_endpoint.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.config.scopes.join(" ")
+    }
+
+    fn managed_identity_resource(&self) -> Result<String> {
+        let [scope] = self.config.scopes.as_slice() 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())
+    }
+
+    // -- Client Credentials Flow --
+
+    async fn acquire_client_credentials(&self) -> Result<TokenResponse> {
+        let client_secret = self
+            .config
+            .client_secret
+            .as_ref()
+            .ok_or(Error::InvalidInput {
+                message: "client_secret is required for ClientCredentials flow".to_string(),
+            })?;
+
+        let token_endpoint = self.get_token_endpoint().await?;
+
+        let params = [
+            ("grant_type", "client_credentials"),
+            ("client_id", &self.config.client_id),
+            ("client_secret", client_secret),
+            ("scope", &self.scopes_string()),
+        ];
+
+        self.post_token_request(&token_endpoint, &params).await
+    }
+
+    // -- Azure Managed Identity Flow --
+
+    async fn acquire_managed_identity(&self, mi_client_id: Option<&str>) -> Result<TokenResponse> {
+        let resource = self.managed_identity_resource()?;
+
+        let mut url = format!(
+            "{AZURE_IMDS_ENDPOINT}?api-version={AZURE_IMDS_API_VERSION}&resource={}",
+            urlencoding::encode(&resource),
+        );
+        if let Some(cid) = mi_client_id {
+            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}"),
+        })
+    }
+
+    // -- Shared Helpers --
+
+    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::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 config = OAuthConfig {
+            issuer_url: "https://login.microsoftonline.com/tenant/v2.0".to_string(),
+            client_id: "app-id".to_string(),
+            client_secret: Some("secret".to_string()),
+            scopes: vec!["scope1".to_string(), "scope2".to_string()],
+            flow: OAuthFlow::ClientCredentials,
+            refresh_buffer_secs: None,
+        };
+        let provider = OAuthHeaderProvider::new(config).unwrap();
+        assert_eq!(provider.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_managed_identity_resource_from_default_scope() {
+        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/.default".to_string()],
+            flow: OAuthFlow::AzureManagedIdentity { client_id: None },
+            refresh_buffer_secs: None,
+        };
+        let provider = OAuthHeaderProvider::new(config).unwrap();
+        assert_eq!(provider.managed_identity_resource().unwrap(), "api://test");
+    }
+
+    #[test]
+    fn test_managed_identity_resource_without_default_suffix() {
+        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".to_string()],
+            flow: OAuthFlow::AzureManagedIdentity { client_id: None },
+            refresh_buffer_secs: None,
+        };
+        let provider = OAuthHeaderProvider::new(config).unwrap();
+        assert_eq!(provider.managed_identity_resource().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 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: None,
+        };
+        let provider = OAuthHeaderProvider::new(config).unwrap();
+
+        let err = provider.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();
+    }
+}
--- a/rust/lancedb/src/table.rs
+++ b/rust/lancedb/src/table.rs
@@ -3,7 +3,7 @@

 //! LanceDB Table APIs

-use arrow_array::{LargeBinaryArray, RecordBatch, RecordBatchReader};
+use arrow_array::{RecordBatch, RecordBatchReader};
 use arrow_schema::{Schema, SchemaRef};
 use async_trait::async_trait;
 use datafusion_execution::TaskContext;
@@ -12,7 +12,6 @@ use datafusion_physical_plan::ExecutionPlan;
 use datafusion_physical_plan::display::DisplayableExecutionPlan;
 use futures::StreamExt;
 use futures::stream::FuturesUnordered;
-use lance::dataset::BlobFile;
 pub use lance::dataset::ColumnAlteration;
 pub use lance::dataset::NewColumnTransform;
 pub use lance::dataset::ReadParams;
@@ -588,28 +587,6 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
    async fn close_lsm_writers(&self) -> Result<()> {
        Ok(())
    }
-    /// Names of the blob v2 columns in this table, in declaration order.
-    async fn blob_columns(&self) -> Result<Vec<String>> {
-        Err(Error::NotSupported {
-            message: "blob_columns is not supported on this table type".into(),
-        })
-    }
-    /// Materialize blob bytes for the given row ids. See [`Table::fetch_blobs`].
-    async fn fetch_blobs(&self, _column: &str, _row_ids: &[u64]) -> Result<LargeBinaryArray> {
-        Err(Error::NotSupported {
-            message: "fetch_blobs is not supported on this table type".into(),
-        })
-    }
-    /// Open lazy blob handles for the given row ids. See [`Table::fetch_blob_files`].
-    async fn fetch_blob_files(
-        &self,
-        _column: &str,
-        _row_ids: &[u64],
-    ) -> Result<Vec<Option<BlobFile>>> {
-        Err(Error::NotSupported {
-            message: "fetch_blob_files is not supported on this table type".into(),
-        })
-    }
    /// Gets the table tag manager.
    async fn tags(&self) -> Result<Box<dyn Tags + '_>>;
    /// Optimize the dataset.
@@ -950,76 +927,6 @@ impl Table {
        self.inner.count_rows(filter.map(Filter::Sql)).await
    }

-    /// Names of the blob v2 columns in this table, in declaration order.
-    ///
-    /// Nested blobs use dotted paths (e.g. `info.blob`). Returns
-    /// [`Error::NotSupported`] on table types without blob support.
-    pub async fn blob_columns(&self) -> Result<Vec<String>> {
-        self.inner.blob_columns().await
-    }
-
-    /// Materialize blob bytes for the given row ids.
-    ///
-    /// Output matches `row_ids` in length and order. Null and zero-length rows
-    /// are null. Prefer [`Self::fetch_blob_files`] for large selections.
-    ///
-    /// ```
-    /// use arrow_array::UInt64Array;
-    /// use futures::TryStreamExt;
-    /// use lancedb::query::{ExecutableQuery, QueryBase};
-    ///
-    /// # use lancedb::Table;
-    /// # async fn materialize(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
-    /// let mut stream = table.query().with_row_id().limit(10).execute().await?;
-    /// while let Some(batch) = stream.try_next().await? {
-    ///     let row_ids = batch
-    ///         .column_by_name("_rowid")
-    ///         .unwrap()
-    ///         .as_any()
-    ///         .downcast_ref::<UInt64Array>()
-    ///         .unwrap();
-    ///     let images = table.fetch_blobs("image", row_ids.values()).await?;
-    ///     let _ = images;
-    /// }
-    /// # Ok(())
-    /// # }
-    /// ```
-    ///
-    /// Returns [`Error::InvalidInput`] when the column does not exist or is
-    /// not a blob v2 column, and [`Error::NotSupported`] on table types
-    /// without blob support.
-    pub async fn fetch_blobs(
-        &self,
-        column: impl AsRef<str>,
-        row_ids: &[u64],
-    ) -> Result<LargeBinaryArray> {
-        self.inner.fetch_blobs(column.as_ref(), row_ids).await
-    }
-
-    /// Open lazy [`BlobFile`] handles for the given row ids.
-    ///
-    /// Same length and order as `row_ids`. Null rows are `None`. Bytes are not
-    /// read from disk until a call to [`BlobFile::read`].
-    ///
-    /// ```
-    /// # use lancedb::Table;
-    /// # async fn lazy_read(table: &Table, row_ids: &[u64]) -> Result<(), Box<dyn std::error::Error>> {
-    /// let handles = table.fetch_blob_files("image", row_ids).await?;
-    /// if let Some(Some(first)) = handles.first() {
-    ///     let bytes = first.read().await?;
-    ///     println!("first blob is {} bytes", bytes.len());
-    /// }
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub async fn fetch_blob_files(
-        &self,
-        column: impl AsRef<str>,
-        row_ids: &[u64],
-    ) -> Result<Vec<Option<BlobFile>>> {
-        self.inner.fetch_blob_files(column.as_ref(), row_ids).await
-    }
-
    /// Insert new records into this Table
    ///
    /// # Arguments
@@ -2854,25 +2761,6 @@ impl BaseTable for NativeTable {
        merge::lsm::close_lsm_writers(self).await
    }

-    async fn blob_columns(&self) -> Result<Vec<String>> {
-        let schema = self.schema().await?;
-        Ok(crate::blob::blob_column_names(schema.as_ref()))
-    }
-
-    async fn fetch_blobs(&self, column: &str, row_ids: &[u64]) -> Result<LargeBinaryArray> {
-        let dataset = self.dataset.get().await?;
-        crate::blob::take_blobs_aligned(&dataset, column, row_ids).await
-    }
-
-    async fn fetch_blob_files(
-        &self,
-        column: &str,
-        row_ids: &[u64],
-    ) -> Result<Vec<Option<BlobFile>>> {
-        let dataset = self.dataset.get().await?;
-        crate::blob::take_blob_files_aligned(&dataset, column, row_ids).await
-    }
-
    /// Delete rows from the table
    async fn delete(&self, predicate: Predicate<'_>) -> Result<DeleteResult> {
        let result = delete::execute_delete(self, predicate).await?;
--- a/rust/lancedb/tests/blob_integration.rs
+++ b/rust/lancedb/tests/blob_integration.rs
@@ -1,22 +1,17 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

+//! Integration tests for blob v2 columns.
+
 use std::sync::Arc;

-use arrow_array::{
-    Array, ArrayRef, BinaryArray, Int64Array, LargeBinaryArray, RecordBatch, StringArray,
-    StructArray, UInt64Array,
-};
-use arrow_schema::{DataType, Field, Fields, Schema};
+use arrow_array::{Array, BinaryArray, Int64Array, LargeBinaryArray, RecordBatch, StructArray};
+use arrow_schema::{DataType, Field, Schema};
 use futures::TryStreamExt;
 use lance_encoding::version::LanceFileVersion;
 use lancedb::{
-    Connection, Error, Result, Table,
-    blob::blob,
-    connect, connect_namespace,
-    database::listing::OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS,
-    query::{ExecutableQuery, QueryBase},
-    table::{AddDataMode, CompactionOptions, OptimizeAction},
+    Connection, Result, Table, blob::blob, connect,
+    database::listing::OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, query::ExecutableQuery,
 };
 use tempfile::tempdir;

@@ -96,7 +91,7 @@ async fn query_image_struct(table: &Table) -> StructArray {
        .expect("image column present")
        .as_any()
        .downcast_ref::<StructArray>()
-        .expect("image column is a descriptor struct")
+        .expect("blob column reads back as a descriptor struct")
        .clone()
 }

@@ -124,7 +119,10 @@ async fn explicit_stable_row_id_setting_wins_over_blob_default() -> Result<()> {
        .execute()
        .await?;

-    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
+    assert!(
+        storage_format_version(&table).await >= LanceFileVersion::V2_2,
+        "format bump still applies; the schema cannot be written below 2.2"
+    );
    assert!(!uses_stable_row_ids(&table).await);
    Ok(())
 }
@@ -146,6 +144,7 @@ async fn creating_with_blob_data_bumps_format() -> Result<()> {
    let tmp = tempdir().unwrap();
    let db = connect(tmp.path().to_str().unwrap()).execute().await?;

+    // Batch already declares the blob field (pre-built struct).
    let blob_field = blob("image", true);
    let DataType::Struct(children) = blob_field.data_type().clone() else {
        unreachable!("blob field is a struct")
@@ -154,7 +153,7 @@ async fn creating_with_blob_data_bumps_format() -> Result<()> {
        children,
        vec![
            Arc::new(LargeBinaryArray::from_iter_values([b"payload".as_slice()])),
-            Arc::new(StringArray::from(vec![None::<&str>])),
+            Arc::new(arrow_array::StringArray::from(vec![None::<&str>])),
        ],
        None,
    );
@@ -185,6 +184,7 @@ async fn add_coerces_large_binary_into_blob_column() -> Result<()> {
    assert_eq!(table.count_rows(None).await?, 2);
    let image = query_image_struct(&table).await;
    assert_eq!(image.len(), 2);
+    // Table schema still has the blob marker after append.
    let schema = table.schema().await?;
    let field = schema.field_with_name("image").unwrap();
    assert_eq!(
@@ -257,12 +257,12 @@ async fn add_rejects_uncoercible_blob_input() -> Result<()> {
        ])),
        vec![
            Arc::new(Int64Array::from(vec![1])),
-            Arc::new(StringArray::from(vec!["not bytes"])),
+            Arc::new(arrow_array::StringArray::from(vec!["not bytes"])),
        ],
    )
    .unwrap();
    let err = table.add(batch).execute().await.unwrap_err();
-    assert!(err.to_string().contains("image"));
+    assert!(err.to_string().contains("image"), "got: {err}");
    Ok(())
 }

@@ -288,7 +288,9 @@ async fn namespace_create_applies_blob_defaults() -> Result<()> {
    let tmp = tempdir().unwrap();
    let mut properties = std::collections::HashMap::new();
    properties.insert("root".to_string(), tmp.path().to_str().unwrap().to_string());
-    let db = connect_namespace("dir", properties).execute().await?;
+    let db = lancedb::connect_namespace("dir", properties)
+        .execute()
+        .await?;
    let table = db
        .create_empty_table("t", blob_table_schema())
        .execute()
@@ -299,14 +301,17 @@ async fn namespace_create_applies_blob_defaults() -> Result<()> {
    Ok(())
 }

-// Overwrite takes the input schema as-is. A raw-binary overwrite drops the blob
-// marker; re-declaring blob v2 in the input restores it.
+// Overwrite takes the input schema as-is (same as cast skip). Raw binary
+// overwrite drops the blob marker unless the input declares blob v2.
 #[tokio::test]
 async fn overwrite_replaces_blob_schema_with_input_schema() -> Result<()> {
+    use lancedb::table::AddDataMode;
+
    let tmp = tempdir().unwrap();
    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
    let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"blob".as_slice())]).await?;

+    // Raw binary overwrite. Plain LargeBinary replaces the blob declaration.
    let raw_schema = Arc::new(Schema::new(vec![
        Field::new("id", DataType::Int64, false),
        Field::new("image", DataType::LargeBinary, true),
@@ -331,9 +336,11 @@ async fn overwrite_replaces_blob_schema_with_input_schema() -> Result<()> {
            .field_with_name("image")
            .unwrap()
            .metadata()
-            .contains_key("ARROW:extension:name")
+            .contains_key("ARROW:extension:name"),
+        "raw binary overwrite leaves a plain binary column"
    );

+    // Overwrite with a declared blob struct keeps the blob column.
    let blob_field = blob("image", true);
    let DataType::Struct(children) = blob_field.data_type().clone() else {
        unreachable!("blob field is a struct")
@@ -342,7 +349,7 @@ async fn overwrite_replaces_blob_schema_with_input_schema() -> Result<()> {
        children,
        vec![
            Arc::new(LargeBinaryArray::from_iter_values([b"declared".as_slice()])),
-            Arc::new(StringArray::from(vec![None::<&str>])),
+            Arc::new(arrow_array::StringArray::from(vec![None::<&str>])),
        ],
        None,
    );
@@ -371,579 +378,3 @@ async fn overwrite_replaces_blob_schema_with_input_schema() -> Result<()> {
    );
    Ok(())
 }
-
-async fn collect_row_ids(table: &Table) -> Result<Vec<u64>> {
-    let batches = table
-        .query()
-        .with_row_id()
-        .execute()
-        .await?
-        .try_collect::<Vec<_>>()
-        .await?;
-    let batch = arrow_select::concat::concat_batches(&batches[0].schema(), &batches).unwrap();
-    Ok(batch
-        .column_by_name("_rowid")
-        .unwrap()
-        .as_any()
-        .downcast_ref::<UInt64Array>()
-        .unwrap()
-        .values()
-        .to_vec())
-}
-
-async fn collect_id_rowid(table: &Table) -> Result<Vec<(i64, u64)>> {
-    let batches = table
-        .query()
-        .with_row_id()
-        .execute()
-        .await?
-        .try_collect::<Vec<_>>()
-        .await?;
-    let batch = arrow_select::concat::concat_batches(&batches[0].schema(), &batches).unwrap();
-    let ids = batch
-        .column_by_name("id")
-        .unwrap()
-        .as_any()
-        .downcast_ref::<Int64Array>()
-        .unwrap();
-    let row_ids = batch
-        .column_by_name("_rowid")
-        .unwrap()
-        .as_any()
-        .downcast_ref::<UInt64Array>()
-        .unwrap();
-    Ok(ids
-        .values()
-        .iter()
-        .copied()
-        .zip(row_ids.values().iter().copied())
-        .collect())
-}
-
-#[tokio::test]
-async fn fetch_blobs_round_trips_bytes() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let payload: &[u8] = b"blob-round-trip-payload";
-    let table = create_inline_blob_table(&db, "t", &[1], &[Some(payload)]).await?;
-
-    let ids = collect_row_ids(&table).await?;
-    let bytes = table.fetch_blobs("image", &ids).await?;
-    assert_eq!(bytes.len(), 1);
-    assert_eq!(bytes.value(0), payload);
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_round_trips_nested_blob_column() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-
-    let blob_field = blob("blob", true);
-    let DataType::Struct(blob_children) = blob_field.data_type().clone() else {
-        unreachable!("blob field is a struct")
-    };
-    let blob_array = StructArray::new(
-        blob_children,
-        vec![
-            Arc::new(LargeBinaryArray::from_iter_values([
-                b"hello".as_slice(),
-                b"world".as_slice(),
-            ])) as ArrayRef,
-            Arc::new(StringArray::from(vec![None::<&str>, None::<&str>])) as ArrayRef,
-        ],
-        None,
-    );
-    let info_fields: Fields = vec![Field::new("name", DataType::Utf8, false), blob_field].into();
-    let info_array = StructArray::new(
-        info_fields.clone(),
-        vec![
-            Arc::new(StringArray::from(vec!["a", "b"])) as ArrayRef,
-            Arc::new(blob_array) as ArrayRef,
-        ],
-        None,
-    );
-    let schema = Arc::new(Schema::new(vec![Field::new(
-        "info",
-        DataType::Struct(info_fields),
-        true,
-    )]));
-    let batch = RecordBatch::try_new(schema, vec![Arc::new(info_array) as ArrayRef]).unwrap();
-    let table = db.create_table("t", batch).execute().await?;
-
-    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
-    assert!(uses_stable_row_ids(&table).await);
-
-    let ids = collect_row_ids(&table).await?;
-    let bytes = table.fetch_blobs("info.blob", &ids).await?;
-    assert_eq!(bytes.len(), 2);
-    let values: std::collections::HashSet<&[u8]> =
-        (0..bytes.len()).map(|i| bytes.value(i)).collect();
-    assert!(values.contains(b"hello".as_slice()));
-    assert!(values.contains(b"world".as_slice()));
-    Ok(())
-}
-
-#[tokio::test]
-async fn blob_columns_lists_nested_dotted_paths() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let blob_field = blob("blob", true);
-    let info = Field::new(
-        "info",
-        DataType::Struct(vec![Field::new("name", DataType::Utf8, false), blob_field].into()),
-        true,
-    );
-    let schema = Arc::new(Schema::new(vec![
-        blob("thumbnail", true),
-        Field::new("id", DataType::Int64, false),
-        info,
-    ]));
-    let table = db.create_empty_table("t", schema).execute().await?;
-    assert_eq!(table.blob_columns().await?, vec!["thumbnail", "info.blob"]);
-    Ok(())
-}
-
-#[tokio::test]
-async fn blob_columns_lists_blob_fields_in_order() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let schema = Arc::new(Schema::new(vec![
-        blob("thumbnail", true),
-        Field::new("id", DataType::Int64, false),
-        blob("image", true),
-    ]));
-    let table = db.create_empty_table("t", schema).execute().await?;
-    assert_eq!(table.blob_columns().await?, vec!["thumbnail", "image"]);
-
-    let plain = db
-        .create_empty_table(
-            "plain",
-            Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])),
-        )
-        .execute()
-        .await?;
-    assert!(plain.blob_columns().await?.is_empty());
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_preserves_null_alignment() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(
-        &db,
-        "t",
-        &[1, 2, 3, 4],
-        &[Some(b"a".as_slice()), None, Some(b"c"), None],
-    )
-    .await?;
-
-    let pairs = collect_id_rowid(&table).await?;
-    let ids: Vec<u64> = pairs.iter().map(|(_, rowid)| *rowid).collect();
-    let bytes = table.fetch_blobs("image", &ids).await?;
-    assert_eq!(bytes.len(), ids.len());
-    for (i, (id, _)) in pairs.iter().enumerate() {
-        match id {
-            1 => assert_eq!(bytes.value(i), b"a"),
-            2 | 4 => assert!(bytes.is_null(i)),
-            3 => assert_eq!(bytes.value(i), b"c"),
-            _ => unreachable!(),
-        }
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_all_null_column_returns_all_nulls() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(&db, "t", &[1, 2], &[None, None]).await?;
-
-    let ids = collect_row_ids(&table).await?;
-    let bytes = table.fetch_blobs("image", &ids).await?;
-    assert_eq!(bytes.len(), 2);
-    assert_eq!(bytes.null_count(), 2);
-
-    let files = table.fetch_blob_files("image", &ids).await?;
-    assert_eq!(files.len(), 2);
-    assert!(files.iter().all(Option::is_none));
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_aligns_with_reordered_and_duplicate_ids() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(
-        &db,
-        "t",
-        &[1, 2, 3],
-        &[Some(b"one".as_slice()), Some(b"two"), Some(b"three")],
-    )
-    .await?;
-
-    let pairs = collect_id_rowid(&table).await?;
-    let by_id = |want: i64| pairs.iter().find(|(id, _)| *id == want).unwrap().1;
-    let request = vec![by_id(3), by_id(1), by_id(3), by_id(2)];
-    let bytes = table.fetch_blobs("image", &request).await?;
-    assert_eq!(bytes.len(), 4);
-    assert_eq!(bytes.value(0), b"three");
-    assert_eq!(bytes.value(1), b"one");
-    assert_eq!(bytes.value(2), b"three");
-    assert_eq!(bytes.value(3), b"two");
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_empty_ids_returns_empty() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"x".as_slice())]).await?;
-
-    assert_eq!(table.fetch_blobs("image", &[]).await?.len(), 0);
-    assert!(table.fetch_blob_files("image", &[]).await?.is_empty());
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_out_of_range_id_errors_without_panic() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"x".as_slice())]).await?;
-
-    let err = table.fetch_blobs("image", &[u64::MAX]).await.unwrap_err();
-    assert!(err.to_string().contains("row ids"));
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_rejects_non_blob_column() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"x".as_slice())]).await?;
-
-    let err = table.fetch_blobs("id", &[0]).await.unwrap_err();
-    assert!(matches!(err, Error::InvalidInput { .. }));
-    assert!(err.to_string().contains("'id' is not a blob column"));
-
-    let err = table.fetch_blob_files("id", &[0]).await.unwrap_err();
-    assert!(err.to_string().contains("'id' is not a blob column"));
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_rejects_unknown_column() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"x".as_slice())]).await?;
-
-    let err = table.fetch_blobs("missing", &[0]).await.unwrap_err();
-    assert!(err.to_string().contains("no column named 'missing'"));
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_rejects_legacy_v1_blob_column() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let legacy = Field::new("image", DataType::LargeBinary, true).with_metadata(
-        std::collections::HashMap::from([("lance-encoding:blob".to_string(), "true".to_string())]),
-    );
-    let schema = Arc::new(Schema::new(vec![
-        Field::new("id", DataType::Int64, false),
-        legacy,
-    ]));
-    let table = db.create_empty_table("t", schema).execute().await?;
-
-    let err = table.fetch_blobs("image", &[0]).await.unwrap_err();
-    assert!(err.to_string().contains("legacy blob column"));
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blob_files_reads_lazily_and_aligns_nulls() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table =
-        create_inline_blob_table(&db, "t", &[1, 2], &[Some(b"lazy-bytes".as_slice()), None])
-            .await?;
-
-    let pairs = collect_id_rowid(&table).await?;
-    let ids: Vec<u64> = pairs.iter().map(|(_, rowid)| *rowid).collect();
-    let files = table.fetch_blob_files("image", &ids).await?;
-    assert_eq!(files.len(), 2);
-    for ((id, _), file) in pairs.iter().zip(&files) {
-        match id {
-            1 => {
-                let handle = file.as_ref().unwrap();
-                assert_eq!(handle.read().await.unwrap().as_ref(), b"lazy-bytes");
-            }
-            2 => assert!(file.is_none()),
-            _ => unreachable!(),
-        }
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_reads_multiple_blob_columns_independently() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let schema = Arc::new(Schema::new(vec![
-        Field::new("id", DataType::Int64, false),
-        blob("image", true),
-        blob("thumbnail", true),
-    ]));
-    let table = db.create_empty_table("t", schema).execute().await?;
-    let batch = RecordBatch::try_new(
-        Arc::new(Schema::new(vec![
-            Field::new("id", DataType::Int64, false),
-            Field::new("image", DataType::LargeBinary, true),
-            Field::new("thumbnail", DataType::LargeBinary, true),
-        ])),
-        vec![
-            Arc::new(Int64Array::from(vec![1, 2])),
-            Arc::new(LargeBinaryArray::from_iter(vec![
-                Some(b"image-1".as_slice()),
-                None,
-            ])),
-            Arc::new(LargeBinaryArray::from_iter(vec![
-                None,
-                Some(b"thumb-2".as_slice()),
-            ])),
-        ],
-    )
-    .unwrap();
-    table.add(batch).execute().await?;
-
-    let pairs = collect_id_rowid(&table).await?;
-    let ids: Vec<u64> = pairs.iter().map(|(_, rowid)| *rowid).collect();
-    let images = table.fetch_blobs("image", &ids).await?;
-    let thumbs = table.fetch_blobs("thumbnail", &ids).await?;
-    for (i, (id, _)) in pairs.iter().enumerate() {
-        match id {
-            1 => {
-                assert_eq!(images.value(i), b"image-1");
-                assert!(thumbs.is_null(i));
-            }
-            2 => {
-                assert!(images.is_null(i));
-                assert_eq!(thumbs.value(i), b"thumb-2");
-            }
-            _ => unreachable!(),
-        }
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_spans_fragments() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"frag-one".as_slice())]).await?;
-    table
-        .add(binary_input_batch(&[2], &[Some(b"frag-two".as_slice())]))
-        .execute()
-        .await?;
-
-    let pairs = collect_id_rowid(&table).await?;
-    let ids: Vec<u64> = pairs.iter().map(|(_, rowid)| *rowid).collect();
-    let bytes = table.fetch_blobs("image", &ids).await?;
-    for (i, (id, _)) in pairs.iter().enumerate() {
-        match id {
-            1 => assert_eq!(bytes.value(i), b"frag-one"),
-            2 => assert_eq!(bytes.value(i), b"frag-two"),
-            _ => unreachable!(),
-        }
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_packed_payload_round_trip() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let big = vec![0xAB_u8; 100 * 1024];
-    let small = b"small".to_vec();
-    let table = create_inline_blob_table(
-        &db,
-        "t",
-        &[1, 2],
-        &[Some(big.as_slice()), Some(small.as_slice())],
-    )
-    .await?;
-
-    let pairs = collect_id_rowid(&table).await?;
-    let ids: Vec<u64> = pairs.iter().map(|(_, rowid)| *rowid).collect();
-    let bytes = table.fetch_blobs("image", &ids).await?;
-    for (i, (id, _)) in pairs.iter().enumerate() {
-        match id {
-            1 => assert_eq!(bytes.value(i), big.as_slice()),
-            2 => assert_eq!(bytes.value(i), small.as_slice()),
-            _ => unreachable!(),
-        }
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_after_delete() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(
-        &db,
-        "t",
-        &[1, 2, 3],
-        &[Some(b"one".as_slice()), Some(b"two"), Some(b"three")],
-    )
-    .await?;
-
-    table.delete("id = 2").await?;
-    let pairs = collect_id_rowid(&table).await?;
-    assert_eq!(pairs.len(), 2);
-    let ids: Vec<u64> = pairs.iter().map(|(_, rowid)| *rowid).collect();
-    let bytes = table.fetch_blobs("image", &ids).await?;
-    for (i, (id, _)) in pairs.iter().enumerate() {
-        match id {
-            1 => assert_eq!(bytes.value(i), b"one"),
-            3 => assert_eq!(bytes.value(i), b"three"),
-            _ => unreachable!(),
-        }
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blobs_with_precompaction_row_ids_survives_compaction() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"frag-one".as_slice())]).await?;
-    table
-        .add(binary_input_batch(&[2], &[Some(b"frag-two".as_slice())]))
-        .execute()
-        .await?;
-
-    let pairs_before = collect_id_rowid(&table).await?;
-    let ids_before: Vec<u64> = pairs_before.iter().map(|(_, rowid)| *rowid).collect();
-
-    table
-        .optimize(OptimizeAction::Compact {
-            options: CompactionOptions::default(),
-            remap_options: None,
-        })
-        .await?;
-
-    let bytes_after = table.fetch_blobs("image", &ids_before).await?;
-    assert_eq!(bytes_after.len(), 2);
-    for (i, (id, _)) in pairs_before.iter().enumerate() {
-        match id {
-            1 => assert_eq!(bytes_after.value(i), b"frag-one"),
-            2 => assert_eq!(bytes_after.value(i), b"frag-two"),
-            _ => unreachable!(),
-        }
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn zero_length_blob_reads_back_as_null() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"".as_slice())]).await?;
-
-    let ids = collect_row_ids(&table).await?;
-    let bytes = table.fetch_blobs("image", &ids).await?;
-    assert_eq!(bytes.len(), 1);
-    assert!(bytes.is_null(0));
-    Ok(())
-}
-
-const DEDICATED_BLOB_LEN: usize = 64 * 1024;
-const SCRAMBLED_LOGICAL_IDS: [i64; 7] = [6, 3, 1, 4, 6, 2, 5];
-
-fn dedicated_blob_bytes(tag: u8) -> Vec<u8> {
-    vec![tag; DEDICATED_BLOB_LEN]
-}
-
-async fn multi_fragment_dedicated_blob_table(db: &Connection) -> Result<Table> {
-    let rows: [(i64, Option<u8>); 6] = [
-        (1, Some(1)),
-        (2, Some(2)),
-        (3, None),
-        (4, Some(4)),
-        (5, None),
-        (6, Some(6)),
-    ];
-    let mut table: Option<Table> = None;
-    for (logical_id, blob_tag) in rows {
-        let bytes = blob_tag.map(dedicated_blob_bytes);
-        let image = [bytes.as_deref()];
-        table = Some(match table {
-            None => create_inline_blob_table(db, "t", &[logical_id], &image).await?,
-            Some(t) => {
-                t.add(binary_input_batch(&[logical_id], &image))
-                    .execute()
-                    .await?;
-                t
-            }
-        });
-    }
-    Ok(table.unwrap())
-}
-
-async fn row_ids_for_logical(table: &Table, logical_ids: &[i64]) -> Result<Vec<u64>> {
-    let id_rowid = collect_id_rowid(table).await?;
-    Ok(logical_ids
-        .iter()
-        .map(|logical_id| {
-            id_rowid
-                .iter()
-                .find(|(id, _)| id == logical_id)
-                .map(|(_, row_id)| *row_id)
-                .unwrap()
-        })
-        .collect())
-}
-
-#[tokio::test]
-async fn fetch_blobs_aligns_across_fragments_with_nulls_and_dups() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = multi_fragment_dedicated_blob_table(&db).await?;
-    let row_ids = row_ids_for_logical(&table, &SCRAMBLED_LOGICAL_IDS).await?;
-
-    let bytes = table.fetch_blobs("image", &row_ids).await?;
-    assert_eq!(bytes.len(), SCRAMBLED_LOGICAL_IDS.len());
-    for (slot, logical_id) in SCRAMBLED_LOGICAL_IDS.iter().enumerate() {
-        match logical_id {
-            3 | 5 => assert!(bytes.is_null(slot)),
-            id => assert_eq!(
-                bytes.value(slot),
-                dedicated_blob_bytes(*id as u8).as_slice()
-            ),
-        }
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn fetch_blob_files_aligns_across_fragments_with_nulls_and_dups() -> Result<()> {
-    let tmp = tempdir().unwrap();
-    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
-    let table = multi_fragment_dedicated_blob_table(&db).await?;
-    let row_ids = row_ids_for_logical(&table, &SCRAMBLED_LOGICAL_IDS).await?;
-
-    let files = table.fetch_blob_files("image", &row_ids).await?;
-    assert_eq!(files.len(), SCRAMBLED_LOGICAL_IDS.len());
-    for (slot, logical_id) in SCRAMBLED_LOGICAL_IDS.iter().enumerate() {
-        match logical_id {
-            3 | 5 => assert!(files[slot].is_none()),
-            id => {
-                let payload = files[slot].as_ref().unwrap().read().await?;
-                assert_eq!(payload.as_ref(), dedicated_blob_bytes(*id as u8).as_slice());
-            }
-        }
-    }
-    Ok(())
-}