fix(storage): recover mirrored manifest creates

2026-09-04 12:38:38 +00:00 · 2026-08-11 02:18:01 +08:00
19 changed files with 360 additions and 2323 deletions
@@ -69,16 +69,6 @@ jobs:
        uses: actions/setup-python@v6
        with:
          python-version: "3.10"
-      - name: Add swap for Arm fat LTO
-        if: matrix.config.platform == 'aarch64'
-        shell: bash
-        run: |
-          swap_file="$RUNNER_TEMP/lancedb-swap"
-          sudo fallocate --length 16G "$swap_file"
-          sudo chmod 600 "$swap_file"
-          sudo mkswap "$swap_file"
-          sudo swapon "$swap_file"
-          free -h
      - uses: ./.github/workflows/build_linux_wheel
        with:
          python-minor-version: 10
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "rand 0.9.5",
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

 [[package]]
 name = "lance"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4832,6 +4832,7 @@ dependencies = [
 "async-recursion",
 "async-trait",
 "async_cell",
+ "aws-credential-types",
 "aws-sdk-dynamodb",
 "byteorder",
 "bytes",
@@ -4847,6 +4848,7 @@ dependencies = [
 "either",
 "fst",
 "futures",
+ "half",
 "humantime",
 "itertools 0.14.0",
 "lance-arrow",
@@ -4888,8 +4890,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4911,7 +4913,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4925,7 +4927,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4934,8 +4936,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4945,8 +4947,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4954,10 +4956,12 @@ dependencies = [
 "arrow-schema",
 "async-trait",
 "blake3",
+ "byteorder",
 "bytes",
 "datafusion-common",
 "datafusion-sql",
 "futures",
+ "itertools 0.14.0",
 "lance-arrow",
 "lance-derive",
 "libc",
@@ -4975,6 +4979,7 @@ dependencies = [
 "snafu 0.9.0",
 "tempfile",
 "tokio",
+ "tokio-stream",
 "tokio-util",
 "tracing",
 "twox-hash",
@@ -4983,8 +4988,8 @@ dependencies = [

 [[package]]
 name = "lance-datafusion"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5014,8 +5019,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5032,8 +5037,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5042,8 +5047,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5068,6 +5073,7 @@ dependencies = [
 "num-traits",
 "prost",
 "prost-build",
+ "rand 0.9.5",
 "tokio",
 "tracing",
 "xxhash-rust",
@@ -5076,8 +5082,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5108,8 +5114,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5124,6 +5130,7 @@ dependencies = [
 "async-trait",
 "bitvec",
 "bytes",
+ "chrono",
 "crossbeam-queue",
 "datafusion",
 "datafusion-common",
@@ -5141,6 +5148,7 @@ dependencies = [
 "lance-bitpacking",
 "lance-core",
 "lance-datafusion",
+ "lance-datagen",
 "lance-encoding",
 "lance-file",
 "lance-index-core",
@@ -5169,12 +5177,13 @@ dependencies = [
 "tempfile",
 "tokio",
 "tracing",
+ "uuid",
 ]

 [[package]]
 name = "lance-index-core"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5196,8 +5205,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5211,6 +5220,7 @@ dependencies = [
 "futures",
 "http 1.5.0",
 "io-uring",
+ "lance-arrow",
 "lance-core",
 "lance-namespace",
 "log",
@@ -5228,28 +5238,29 @@ dependencies = [
 "tokio",
 "tracing",
 "url",
- "uuid",
 ]

 [[package]]
 name = "lance-linalg"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
+ "arrow-buffer",
 "arrow-schema",
 "cc",
 "half",
 "lance-arrow",
 "lance-core",
 "num-traits",
+ "rand 0.9.5",
 "rayon",
 ]

 [[package]]
 name = "lance-namespace"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5261,8 +5272,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace-impls"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5315,13 +5326,14 @@ dependencies = [

 [[package]]
 name = "lance-select"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
 "arrow-schema",
 "byteorder",
+ "bytes",
 "itertools 0.14.0",
 "lance-core",
 "roaring",
@@ -5330,8 +5342,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5371,8 +5383,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5385,8 +5397,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "11.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.8#9acbac748e8a7d616146dac8b06da42d8e7c6b62"
+version = "11.0.0-beta.3"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.3#f7d475539cefbd140cc46a828f3d843e68cd10f1"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=11.0.0-beta.8", default-features = false, "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=11.0.0-beta.8", default-features = false, "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=11.0.0-beta.8", default-features = false, "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=11.0.0-beta.8", "tag" = "v11.0.0-beta.8", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=11.0.0-beta.3", default-features = false, "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=11.0.0-beta.3", "tag" = "v11.0.0-beta.3", "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 }
@@ -101,13 +101,6 @@ ignore = [
    # https://rustsec.org/advisories/RUSTSEC-2026-0195
    { id = "RUSTSEC-2026-0194", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
    { id = "RUSTSEC-2026-0195", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
-    # smartstring: unmaintained — the repository was archived by its author on
-    # 2026-05-03. Not a vulnerability. Reached only transitively through polars
-    # (polars-core/-io/-ops/-time/-utils); nothing in LanceDB depends on it directly.
-    # The advisory states no safe upgrade is available: upstream recommends
-    # compact_str/smol_str, so clearing this requires polars to migrate.
-    # https://rustsec.org/advisories/RUSTSEC-2026-0249
-    { id = "RUSTSEC-2026-0249", reason = "smartstring unmaintained via polars; no fixed upstream release" },
 ]

 # ---------------------------------------------------------------------------
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>11.0.0-beta.8</lance-core.version>
+        <lance-core.version>11.0.0-beta.3</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>
@@ -716,9 +716,20 @@ class LanceDBConnection(DBConnection):
        if not isinstance(uri, Path):
            scheme = get_uri_scheme(uri)
        is_local = isinstance(uri, Path) or scheme == "file"
-        is_file_uri = isinstance(uri, str) and uri.lower().startswith("file:")
-        if is_local and not is_file_uri:
+        if is_local:
            if isinstance(uri, str):
+                # Strip file:// or file:/ scheme if present
+                # file:///path becomes file:/path after URL normalization
+                if uri.startswith("file://"):
+                    uri = uri[7:]  # Remove "file://"
+                elif uri.startswith("file:/"):
+                    uri = uri[5:]  # Remove "file:"
+
+                if sys.platform == "win32":
+                    # On Windows, a path like /C:/path should become C:/path
+                    if len(uri) >= 3 and uri[0] == "/" and uri[2] == ":":
+                        uri = uri[1:]
+
                uri = Path(uri)
            uri = uri.expanduser().absolute()
            Path(uri).mkdir(parents=True, exist_ok=True)
@@ -919,7 +919,17 @@ class LanceNamespaceDBConnection(DBConnection):
            The namespace client for this connection.
        """
        if self._namespace_client is None:
-            self._namespace_client = LOOP.run(self._inner.namespace_client())
+            if (
+                self._namespace_client_impl is None
+                or self._namespace_client_properties is None
+            ):
+                raise ValueError(
+                    "Cannot construct a Python namespace client without "
+                    "namespace implementation properties"
+                )
+            self._namespace_client = namespace_connect(
+                self._namespace_client_impl, self._namespace_client_properties
+            )
        return self._namespace_client


@@ -1360,7 +1370,17 @@ class AsyncLanceNamespaceDBConnection:
            The namespace client for this connection.
        """
        if self._namespace_client is None:
-            self._namespace_client = await self._inner.namespace_client()
+            if (
+                self._namespace_client_impl is None
+                or self._namespace_client_properties is None
+            ):
+                raise ValueError(
+                    "Cannot construct a Python namespace client without "
+                    "namespace implementation properties"
+                )
+            self._namespace_client = namespace_connect(
+                self._namespace_client_impl, self._namespace_client_properties
+            )
        return self._namespace_client


@@ -89,32 +89,6 @@ def test_sync_debugger_inspection_does_not_use_background_loop(tmp_path, monkeyp
    assert repr(table) == f"LanceTable(name='test', _conn={db!r})"


-def test_connect_preserves_file_uri_authority(monkeypatch):
-    uri = "file://server/share/database"
-    received = []
-
-    async def fake_connect(passed_uri, *_args):
-        received.append(passed_uri)
-        return SimpleNamespace(uri=passed_uri)
-
-    monkeypatch.setattr("lancedb.db.lancedb_connect", fake_connect)
-    db = lancedb.connect(uri)
-
-    assert received == [uri]
-    assert db.uri == uri
-
-
-def test_connect_file_uri_lifecycle(tmp_path):
-    uri = (tmp_path / "sync").as_uri()
-    db = lancedb.connect(uri)
-
-    db.create_table("test", data=[{"id": 1}])
-    assert db.table_names() == ["test"]
-    assert db.open_table("test").count_rows() == 1
-    db.drop_table("test")
-    assert db.table_names() == []
-
-
 def test_read_consistency_interval_does_not_use_background_loop(tmp_path, monkeypatch):
    from lancedb.background_loop import LOOP
    from lancedb.db import LanceDBConnection
@@ -431,35 +405,6 @@ async def test_connect(tmp_path):
    assert str(db) == f"ListingDatabase(uri={tmp_path}, read_consistency_interval=5s)"


-@pytest.mark.asyncio
-async def test_connect_async_preserves_file_uri_authority(monkeypatch):
-    uri = "file://server/share/database"
-    received = []
-
-    async def fake_connect(passed_uri, *_args):
-        received.append(passed_uri)
-        return SimpleNamespace(uri=passed_uri)
-
-    monkeypatch.setattr(lancedb, "lancedb_connect", fake_connect)
-    db = await lancedb.connect_async(uri)
-
-    assert received == [uri]
-    assert db.uri == uri
-
-
-@pytest.mark.asyncio
-async def test_connect_async_file_uri_lifecycle(tmp_path):
-    uri = (tmp_path / "async").as_uri()
-    db = await lancedb.connect_async(uri)
-
-    await db.create_table("test", data=[{"id": 1}])
-    assert await db.table_names() == ["test"]
-    table = await db.open_table("test")
-    assert await table.count_rows() == 1
-    await db.drop_table("test")
-    assert await db.table_names() == []
-
-
@pytest.mark.asyncio
 async def test_close(mem_db_async: lancedb.AsyncConnection):
    assert mem_db_async.is_open()
@@ -1235,40 +1180,6 @@ def test_clone_table_deep_clone_fails(tmp_path):
        db.clone_table("cloned", source_uri, is_shallow=False)


-class _UnsupportedNamespaceConfig:
-    async def namespace_client_config(self):
-        raise RuntimeError("UNC namespace client export is not supported")
-
-
-def test_sync_namespace_client_propagates_export_guard(monkeypatch):
-    from lancedb.db import AsyncConnection, LanceDBConnection
-
-    monkeypatch.setattr(
-        "lancedb.db.namespace_connect",
-        lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
-    )
-    db = LanceDBConnection.__new__(LanceDBConnection)
-    db._conn = AsyncConnection(_UnsupportedNamespaceConfig())
-    db._cached_namespace_client = None
-
-    with pytest.raises(RuntimeError, match="UNC namespace client export"):
-        db.namespace_client()
-
-
-@pytest.mark.asyncio
-async def test_async_namespace_client_propagates_export_guard(monkeypatch):
-    from lancedb.db import AsyncConnection
-
-    monkeypatch.setattr(
-        "lancedb.db.namespace_connect",
-        lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
-    )
-    db = AsyncConnection(_UnsupportedNamespaceConfig())
-
-    with pytest.raises(RuntimeError, match="UNC namespace client export"):
-        await db.namespace_client()
-
-
@pytest.mark.skipif(sys.platform == "win32", reason="Namespace client issues")
 def test_namespace_client_native_storage(tmp_path):
    """Test namespace_client() returns DirectoryNamespace for native storage."""
@@ -60,11 +60,6 @@ class _NamespaceClient:
        return _ipc_file()


-class _UnsupportedNamespaceConfig:
-    async def namespace_client_config(self):
-        raise RuntimeError("UNC namespace client export is not supported")
-
-
 def _namespace_lance_table(namespace_client: _NamespaceClient) -> LanceTable:
    table = LanceTable.__new__(LanceTable)
    table._table = _FailingSyncInner()
@@ -143,24 +138,6 @@ class TestNamespaceConnection:
        db.drop_namespace(["test_ns"])
        assert "test_ns" not in db.list_namespaces().namespaces

-    def test_sync_namespace_client_propagates_export_guard(self, monkeypatch):
-        from lancedb.db import AsyncConnection
-
-        monkeypatch.setattr(
-            "lancedb.namespace.namespace_connect",
-            lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
-        )
-        db = lancedb.LanceNamespaceDBConnection.__new__(
-            lancedb.LanceNamespaceDBConnection
-        )
-        db._namespace_client = None
-        db._namespace_client_impl = "dir"
-        db._namespace_client_properties = {"root": "file://server/share/database"}
-        db._inner = AsyncConnection(_UnsupportedNamespaceConfig())
-
-        with pytest.raises(RuntimeError, match="UNC namespace client export"):
-            db.namespace_client()
-
    def test_create_table_through_namespace(self):
        """Test creating a table through namespace."""
        db = lancedb.connect_namespace("dir", {"root": self.temp_dir})
@@ -662,24 +639,6 @@ class TestAsyncNamespaceConnection:
        await db.drop_namespace(["test_ns"])
        assert "test_ns" not in (await db.list_namespaces()).namespaces

-    async def test_async_namespace_client_propagates_export_guard(self, monkeypatch):
-        from lancedb.db import AsyncConnection
-
-        monkeypatch.setattr(
-            "lancedb.namespace.namespace_connect",
-            lambda *_args, **_kwargs: pytest.fail("guarded config was reconstructed"),
-        )
-        db = lancedb.AsyncLanceNamespaceDBConnection.__new__(
-            lancedb.AsyncLanceNamespaceDBConnection
-        )
-        db._namespace_client = None
-        db._namespace_client_impl = "dir"
-        db._namespace_client_properties = {"root": "file://server/share/database"}
-        db._inner = AsyncConnection(_UnsupportedNamespaceConfig())
-
-        with pytest.raises(RuntimeError, match="UNC namespace client export"):
-            await db.namespace_client()
-
    async def test_async_namespace_client_is_lazy(self):
        """namespace_client() should still return the backing client on demand."""
        pytest.importorskip("lance")
@@ -188,9 +188,6 @@ required-features = ["bedrock"]
 [[example]]
 name = "bench_streaming_dataloader"

-[[example]]
-name = "bench_open_missing_table"
-
 [[example]]
 name = "simple"

@@ -1,150 +0,0 @@
-// SPDX-License-Identifier: Apache-2.0
-// SPDX-FileCopyrightText: Copyright The LanceDB Authors
-
-// Release benchmark for opening a missing table as sibling-table cardinality grows.
-//
-// The fixture uses real `.lance` directories and marker files. Fixture creation is
-// outside the timed section. Defaults intentionally cover 1k, 10k, and 100k siblings
-// with 10 warmups and 100 distinct missing-table opens per scale:
-//
-// ```text
-// cargo run --release -p lancedb --example bench_open_missing_table
-// ```
-//
-// `BENCH_SIBLINGS`, `BENCH_WARMUPS`, and `BENCH_TRIALS` override those defaults.
-// Reduced settings are useful only as a smoke test. Performance comparisons require
-// the same machine, filesystem, fixture sizes, settings, lockfile, and alternating
-// baseline/candidate execution order.
-
-use std::time::{Duration, Instant};
-
-use anyhow::{Context, Result, bail};
-use lancedb::connection::Connection;
-use lancedb::{Error, connect};
-use object_store::ObjectStoreExt as _;
-use object_store::path::Path;
-
-const MAX_SIBLINGS: usize = 1_000_000;
-const MAX_WARMUPS: usize = 10_000;
-const MAX_TRIALS: usize = 100_000;
-
-fn env_usize(key: &str, default: usize, max: usize) -> Result<usize> {
-    let value = match std::env::var(key) {
-        Ok(value) => value
-            .parse()
-            .with_context(|| format!("invalid {key} value: {value}"))?,
-        Err(std::env::VarError::NotPresent) => default,
-        Err(error) => return Err(error).with_context(|| format!("reading {key}")),
-    };
-    if value == 0 || value > max {
-        bail!("{key} must be between 1 and {max}");
-    }
-    Ok(value)
-}
-
-fn sibling_counts() -> Result<Vec<usize>> {
-    let raw = std::env::var("BENCH_SIBLINGS").unwrap_or_else(|_| "1000,10000,100000".into());
-    let mut counts = raw
-        .split(',')
-        .map(|value| {
-            value
-                .trim()
-                .parse::<usize>()
-                .with_context(|| format!("invalid BENCH_SIBLINGS value: {value}"))
-        })
-        .collect::<Result<Vec<_>>>()?;
-    counts.sort_unstable();
-    counts.dedup();
-    if counts.is_empty() || counts[0] == 0 || counts[counts.len() - 1] > MAX_SIBLINGS {
-        bail!("BENCH_SIBLINGS values must be between 1 and {MAX_SIBLINGS}");
-    }
-    Ok(counts)
-}
-
-async fn add_siblings(
-    store: &object_store::local::LocalFileSystem,
-    start: usize,
-    end: usize,
-) -> Result<()> {
-    for index in start..end {
-        let marker = Path::from(format!("sibling_{index:06}.lance/_marker"));
-        store
-            .put(&marker, bytes::Bytes::new().into())
-            .await
-            .with_context(|| format!("creating benchmark marker {marker}"))?;
-    }
-    Ok(())
-}
-
-async fn time_missing_open(db: &Connection, name: &str) -> Result<Duration> {
-    let started = Instant::now();
-    let result = db.open_table(name).execute().await;
-    let elapsed = started.elapsed();
-    match result {
-        Err(Error::TableNotFound { .. }) => Ok(elapsed),
-        Err(error) => bail!("expected TableNotFound for {name}, got {error:?}"),
-        Ok(_) => bail!("benchmark missing-table name unexpectedly exists: {name}"),
-    }
-}
-
-fn percentile(sorted: &[Duration], percentile: usize) -> Duration {
-    let rank = (sorted.len() * percentile).div_ceil(100).saturating_sub(1);
-    sorted[rank]
-}
-
-#[tokio::main]
-async fn main() -> Result<()> {
-    let counts = sibling_counts()?;
-    let warmups = env_usize("BENCH_WARMUPS", 10, MAX_WARMUPS)?;
-    let trials = env_usize("BENCH_TRIALS", 100, MAX_TRIALS)?;
-
-    let fixture = tempfile::tempdir().context("creating benchmark fixture")?;
-    let database_path = fixture.path();
-    let fixture_store = object_store::local::LocalFileSystem::new_with_prefix(database_path)
-        .context("creating benchmark object store")?;
-    let db = connect(database_path.to_str().context("non-UTF-8 fixture path")?)
-        .execute()
-        .await?;
-
-    println!(
-        "config: siblings={counts:?} warmups={warmups} trials={trials} profile={} os={} arch={}",
-        if cfg!(debug_assertions) {
-            "debug"
-        } else {
-            "release"
-        },
-        std::env::consts::OS,
-        std::env::consts::ARCH,
-    );
-    println!("lower is better; fixture setup and teardown are excluded");
-    println!("| siblings | samples | p50 | p95 | max |");
-    println!("| ---: | ---: | ---: | ---: | ---: |");
-
-    let mut created = 0;
-    for sibling_count in counts {
-        add_siblings(&fixture_store, created, sibling_count).await?;
-        created = sibling_count;
-
-        for index in 0..warmups {
-            let name = format!("__missing_warmup_{sibling_count}_{index}");
-            let _ = time_missing_open(&db, &name).await?;
-        }
-
-        let mut samples = Vec::with_capacity(trials);
-        for index in 0..trials {
-            let name = format!("__missing_trial_{sibling_count}_{index}");
-            samples.push(time_missing_open(&db, &name).await?);
-        }
-        samples.sort_unstable();
-
-        println!(
-            "| {sibling_count} | {} | {:?} | {:?} | {:?} |",
-            samples.len(),
-            percentile(&samples, 50),
-            percentile(&samples, 95),
-            samples[samples.len() - 1],
-        );
-    }
-
-    Ok(())
-}
@@ -17,7 +17,7 @@ use arrow_array::builder::LargeBinaryBuilder;
 use arrow_schema::{DataType, Field, Schema};
 use lance::dataset::{BlobRangeRequest as LanceBlobRangeRequest, Dataset, WriteParams};
 use lance_arrow::FieldExt;
-use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
+use lance_file::version::LanceFileVersion;
 use lance_io::object_store::ObjectStore;
 use object_store::path::Path;

@@ -333,10 +333,7 @@ pub(crate) fn ensure_blob_storage_version(schema: &Schema, params: &mut WritePar
        .data_storage_version
        .unwrap_or(LanceFileVersion::Stable)
        .resolve();
-    if matches!(
-        resolved,
-        ConcreteFileVersion::V1 | ConcreteFileVersion::V2_0 | ConcreteFileVersion::V2_1
-    ) {
+    if resolved < LanceFileVersion::V2_2 {
        params.data_storage_version = Some(LanceFileVersion::V2_2);
    }
 }
@@ -502,7 +499,7 @@ mod tests {
        ensure_blob_storage_version(&blob_schema(), &mut params);
        assert_eq!(
            params.data_storage_version.unwrap().resolve(),
-            ConcreteFileVersion::V2_2
+            LanceFileVersion::V2_2
        );
    }

@@ -515,7 +512,7 @@ mod tests {
        ensure_blob_storage_version(&blob_schema(), &mut params);
        assert_eq!(
            params.data_storage_version.unwrap().resolve(),
-            ConcreteFileVersion::V2_2
+            LanceFileVersion::V2_2
        );
    }

@@ -409,11 +409,6 @@ impl Connection {
    ///
    /// The names will be returned in lexicographical order (ascending)
    ///
-    /// Listing databases discover physical `*.lance` entries without opening every
-    /// dataset. The result is a point-in-time discovery snapshot: an entry may still be
-    /// under creation, may contain only uncommitted storage, or may be concurrently
-    /// dropped before it is opened.
-    ///
    /// The parameters `page_token` and `limit` can be used to paginate the results
    pub fn table_names(&self) -> TableNamesBuilder {
        TableNamesBuilder::new(self.internal.clone())
@@ -461,9 +456,10 @@ impl Connection {
    ///
    /// # Returns
    /// Created [`TableRef`], or [`Error::TableNotFound`] if the table does not exist.
-    /// On listing databases, a committed Lance manifest is authoritative for table
-    /// existence. Uncommitted files or a physical `<name>.lance` directory alone do not
-    /// make a table openable.
+    /// If the table's storage is present but holds no readable dataset (for example a
+    /// `<name>.lance` directory left behind by an interrupted drop and re-create, which
+    /// [`Self::table_names`] still lists) this returns [`Error::TableCorrupted`]
+    /// instead.
    pub fn open_table(&self, name: impl Into<String>) -> OpenTableBuilder {
        OpenTableBuilder::new(
            self.internal.clone(),
@@ -438,9 +438,10 @@ mod tests {
            .await
            .unwrap()
            .data_storage_format
-            .lance_file_format();
+            .lance_file_version()
+            .unwrap();
        // Compare resolved versions since Stable/Next are aliases that resolve at storage time
-        assert_eq!(storage_format, data_storage_version.resolve());
+        assert_eq!(storage_format.resolve(), data_storage_version.resolve());
    }

    #[tokio::test]
@@ -59,46 +59,6 @@ fn is_table_already_exists_namespace_error(err: &lance::Error) -> bool {
 /// via the `delimiter` property.
 const DEFAULT_NAMESPACE_DELIMITER: &str = "$";

-#[cfg(any(windows, test))]
-fn is_unc_root(root: &str) -> bool {
-    let normalized = root.replace('\\', "/");
-    let lowercase = normalized.to_ascii_lowercase();
-
-    if lowercase.starts_with("//?/unc/") {
-        return true;
-    }
-    if lowercase.starts_with("//?/") || lowercase.starts_with("//./") {
-        return false;
-    }
-    if let Some(authority) = normalized.strip_prefix("//") {
-        return !authority.is_empty() && !authority.starts_with('/');
-    }
-
-    normalized.split_once("://").is_some_and(|(scheme, path)| {
-        scheme.eq_ignore_ascii_case("file") && !path.is_empty() && !path.starts_with('/')
-    })
-}
-
-#[cfg(any(windows, test))]
-fn is_directory_unc_root(ns_impl: &str, ns_properties: &HashMap<String, String>) -> bool {
-    ns_impl.eq_ignore_ascii_case("dir")
-        && ns_properties
-            .get("root")
-            .is_some_and(|root| is_unc_root(root))
-}
-
-#[cfg(any(windows, test))]
-#[derive(Debug)]
-struct UnavailableUncDirectoryNamespace;
-
-#[cfg(any(windows, test))]
-#[async_trait]
-impl LanceNamespace for UnavailableUncDirectoryNamespace {
-    fn namespace_id(&self) -> String {
-        "unavailable-unc-directory".to_string()
-    }
-}
-
 /// A database implementation that uses lance-namespace for table management
 pub struct LanceNamespaceDatabase {
    namespace: Arc<dyn LanceNamespace>,
@@ -132,17 +92,6 @@ fn resolve_delimiter(ns_properties: &HashMap<String, String>) -> String {
 }

 impl LanceNamespaceDatabase {
-    fn ensure_storage_supported(&self) -> Result<()> {
-        #[cfg(any(windows, test))]
-        if is_directory_unc_root(&self.ns_impl, &self.ns_properties) {
-            return Err(Error::NotSupported {
-                message: "Directory namespace operations are not supported for UNC roots because the namespace manifest resolver does not preserve file URI authorities; use flat root tables or a non-UNC namespace root"
-                    .to_string(),
-            });
-        }
-        Ok(())
-    }
-
    pub fn from_namespace_client(
        namespace_client: Arc<dyn LanceNamespace>,
        namespace_client_impl: String,
@@ -204,25 +153,6 @@ impl LanceNamespaceDatabase {
        pushdown_operations: HashSet<NamespaceClientPushdownOperation>,
        new_table_config: NewTableConfig,
    ) -> Result<Self> {
-        #[cfg(any(windows, test))]
-        if is_directory_unc_root(ns_impl, &ns_properties) {
-            let freshness_baselines: FreshnessBaselines = Arc::new(Mutex::new(HashMap::new()));
-            let delimiter = resolve_delimiter(&ns_properties);
-            return Ok(Self {
-                namespace: Arc::new(UnavailableUncDirectoryNamespace),
-                storage_options,
-                read_consistency_interval,
-                session,
-                uri: format!("namespace://{}", ns_impl),
-                pushdown_operations,
-                ns_impl: ns_impl.to_string(),
-                ns_properties,
-                new_table_config,
-                freshness_baselines,
-                delimiter,
-            });
-        }
-
        let mut builder = ConnectBuilder::new(ns_impl);
        for (key, value) in ns_properties.clone() {
            builder = builder.property(key, value);
@@ -380,7 +310,6 @@ impl Database for LanceNamespaceDatabase {
        &self,
        request: ListNamespacesRequest,
    ) -> Result<ListNamespacesResponse> {
-        self.ensure_storage_supported()?;
        Ok(self.namespace.list_namespaces(request).await?)
    }

@@ -388,12 +317,10 @@ impl Database for LanceNamespaceDatabase {
        &self,
        request: CreateNamespaceRequest,
    ) -> Result<CreateNamespaceResponse> {
-        self.ensure_storage_supported()?;
        Ok(self.namespace.create_namespace(request).await?)
    }

    async fn drop_namespace(&self, request: DropNamespaceRequest) -> Result<DropNamespaceResponse> {
-        self.ensure_storage_supported()?;
        Ok(self.namespace.drop_namespace(request).await?)
    }

@@ -401,12 +328,10 @@ impl Database for LanceNamespaceDatabase {
        &self,
        request: DescribeNamespaceRequest,
    ) -> Result<DescribeNamespaceResponse> {
-        self.ensure_storage_supported()?;
        Ok(self.namespace.describe_namespace(request).await?)
    }

    async fn table_names(&self, request: TableNamesRequest) -> Result<Vec<String>> {
-        self.ensure_storage_supported()?;
        let ns_request = ListTablesRequest {
            id: Some(request.namespace_path),
            page_token: request.start_after,
@@ -420,12 +345,10 @@ impl Database for LanceNamespaceDatabase {
    }

    async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
-        self.ensure_storage_supported()?;
        Ok(self.namespace.list_tables(request).await?)
    }

    async fn create_table(&self, request: DbCreateTableRequest) -> Result<Arc<dyn BaseTable>> {
-        self.ensure_storage_supported()?;
        let mut table_id = request.namespace_path.clone();
        table_id.push(request.name.clone());
        let mut existing_table = None;
@@ -595,7 +518,6 @@ impl Database for LanceNamespaceDatabase {
    }

    async fn open_table(&self, request: OpenTableRequest) -> Result<Arc<dyn BaseTable>> {
-        self.ensure_storage_supported()?;
        let native_table = NativeTable::open_from_namespace(
            self.namespace.clone(),
            &request.name,
@@ -625,7 +547,6 @@ impl Database for LanceNamespaceDatabase {
        cur_namespace_path: &[String],
        new_namespace_path: &[String],
    ) -> Result<()> {
-        self.ensure_storage_supported()?;
        let mut cur_table_id = cur_namespace_path.to_vec();
        cur_table_id.push(cur_name.to_string());

@@ -652,7 +573,6 @@ impl Database for LanceNamespaceDatabase {
    }

    async fn drop_table(&self, name: &str, namespace_path: &[String]) -> Result<()> {
-        self.ensure_storage_supported()?;
        let mut table_id = namespace_path.to_vec();
        table_id.push(name.to_string());

@@ -692,12 +612,10 @@ impl Database for LanceNamespaceDatabase {
    }

    async fn namespace_client(&self) -> Result<Arc<dyn LanceNamespace>> {
-        self.ensure_storage_supported()?;
        Ok(self.namespace.clone())
    }

    async fn namespace_client_config(&self) -> Result<(String, HashMap<String, String>)> {
-        self.ensure_storage_supported()?;
        Ok((self.ns_impl.clone(), self.ns_properties.clone()))
    }
 }
@@ -726,56 +644,6 @@ mod tests {
        RecordBatch::try_new(schema, vec![Arc::new(id_array), Arc::new(name_array)]).unwrap()
    }

-    #[test]
-    fn detects_unc_namespace_roots_without_misclassifying_drives() {
-        for root in [
-            "file://server/share/database",
-            "FILE://server/share/database",
-            r"\\server\share\database",
-            r"\\?\UNC\server\share\database",
-        ] {
-            assert!(is_unc_root(root), "expected an UNC root: {root}");
-        }
-        for root in [
-            "file:///C:/database",
-            r"C:\database",
-            r"\\?\C:\database",
-            "/var/lib/database",
-        ] {
-            assert!(!is_unc_root(root), "expected a non-UNC root: {root}");
-        }
-    }
-
-    #[tokio::test]
-    async fn directory_namespace_fails_closed_before_accessing_an_unc_root() {
-        let mut properties = HashMap::new();
-        properties.insert(
-            "root".to_string(),
-            "file://server/share/database".to_string(),
-        );
-        let db = LanceNamespaceDatabase::connect(
-            "dir",
-            properties,
-            HashMap::new(),
-            None,
-            None,
-            HashSet::new(),
-        )
-        .await
-        .unwrap();
-
-        let error = db
-            .list_tables(ListTablesRequest::default())
-            .await
-            .unwrap_err();
-        assert!(matches!(error, Error::NotSupported { .. }));
-        assert!(error.to_string().contains("UNC roots"));
-
-        let error = db.namespace_client_config().await.unwrap_err();
-        assert!(matches!(error, Error::NotSupported { .. }));
-        assert!(error.to_string().contains("UNC roots"));
-    }
-
    #[tokio::test]
    async fn test_namespace_connection_simple() {
        // Test that namespace connections work with simple connect_namespace(impl_type, properties)
@@ -1,821 +1,23 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

-//! Object-store providers and adapters used by LanceDB.
+//! A mirroring object store that mirror writes to a secondary object store

 use std::{fmt::Formatter, sync::Arc};

-#[cfg(any(windows, test))]
-use futures::TryStreamExt;
 use futures::{StreamExt, TryFutureExt, stream::BoxStream};
 use lance::io::WrappingObjectStore;
-#[cfg(any(windows, test))]
-use lance_table::{
-    format::{IndexMetadata, Manifest, Transaction},
-    io::commit::{
-        CommitError, CommitHandler, ManifestLocation, ManifestNamingScheme, ManifestWriter,
-        RenameCommitHandler,
-    },
-};
 use object_store::{
-    CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
+    CopyMode, CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
    ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
    UploadPart, path::Path,
 };

 use async_trait::async_trait;

-#[cfg(any(windows, test))]
-use lance_core::{Error as LanceError, Result as LanceResult};
-#[cfg(test)]
-use lance_io::object_store::ObjectStoreRegistry;
-#[cfg(any(windows, test))]
-use lance_io::object_store::{
-    DEFAULT_LOCAL_IO_PARALLELISM, ObjectStoreParams, ObjectStoreProvider, StorageOptions,
-};
-#[cfg(any(windows, test))]
-use object_store::local::LocalFileSystem;
-#[cfg(any(windows, test))]
-use url::Url;
-
 #[cfg(test)]
 pub mod io_tracking;

-/// A local commit handler that resolves the latest manifest through the object store.
-///
-/// Lance's native local shortcut reconstructs the selected manifest from a
-/// filesystem path. On Windows that conversion drops the authority from a UNC
-/// path. Listing through the already-rooted object store preserves the structural
-/// server/share prefix while retaining the normal atomic-rename commit behavior.
-#[derive(Debug)]
-#[cfg(any(windows, test))]
-struct RootedFileCommitHandler;
-
-#[cfg(any(windows, test))]
-#[async_trait]
-impl CommitHandler for RootedFileCommitHandler {
-    fn is_version_not_found_definitive(&self) -> bool {
-        RenameCommitHandler.is_version_not_found_definitive()
-    }
-
-    fn propagate_commit_error_after_success(&self) -> bool {
-        RenameCommitHandler.propagate_commit_error_after_success()
-    }
-
-    async fn resolve_latest_location(
-        &self,
-        base_path: &Path,
-        object_store: &lance::io::ObjectStore,
-    ) -> LanceResult<ManifestLocation> {
-        self.list_manifest_locations(base_path, object_store, true)
-            .try_next()
-            .await?
-            .ok_or_else(|| LanceError::not_found(base_path.to_string()))
-    }
-
-    async fn commit(
-        &self,
-        manifest: &mut Manifest,
-        indices: Option<Vec<IndexMetadata>>,
-        base_path: &Path,
-        object_store: &lance::io::ObjectStore,
-        manifest_writer: ManifestWriter,
-        naming_scheme: ManifestNamingScheme,
-        transaction: Option<Transaction>,
-    ) -> std::result::Result<ManifestLocation, CommitError> {
-        RenameCommitHandler
-            .commit(
-                manifest,
-                indices,
-                base_path,
-                object_store,
-                manifest_writer,
-                naming_scheme,
-                transaction,
-            )
-            .await
-    }
-}
-
-#[cfg(any(windows, test))]
-pub(crate) fn rooted_file_commit_handler() -> Arc<dyn CommitHandler> {
-    Arc::new(RootedFileCommitHandler)
-}
-
-/// A file-store provider that anchors each request at its filesystem root.
-///
-/// On Windows, an unprefixed [`LocalFileSystem`] cannot service UNC paths. Its
-/// conversion to an object-store [`Path`] drops the UNC host, so subsequent I/O
-/// is directed at a different local path. Anchoring the store at the drive or
-/// UNC-share root keeps the UNC authority in the filesystem prefix and exposes
-/// only paths relative to that prefix to `object_store`.
-///
-/// Extracted paths retain the native drive or UNC-share root as a structural
-/// first component. This keeps Lance's local classification and optimized I/O
-/// safe without recovering absolute path provenance from ambiguous path text.
-#[cfg(any(windows, test))]
-#[derive(Debug)]
-struct PrefixedFileStoreProvider;
-
-#[cfg(any(windows, test))]
-impl PrefixedFileStoreProvider {
-    fn root_and_relative_path(url: &Url) -> LanceResult<(std::path::PathBuf, Path)> {
-        let filesystem_path = url.to_file_path().map_err(|_| {
-            LanceError::invalid_input(format!("Unable to convert URL '{url}' to a local path"))
-        })?;
-
-        let mut root = std::path::PathBuf::new();
-        for component in filesystem_path.components() {
-            match component {
-                std::path::Component::Prefix(_) | std::path::Component::RootDir => {
-                    root.push(component.as_os_str());
-                }
-                _ => break,
-            }
-        }
-        if root.as_os_str().is_empty() {
-            return Err(LanceError::invalid_input(format!(
-                "Local path '{}' has no filesystem root",
-                filesystem_path.display()
-            )));
-        }
-
-        let relative = filesystem_path.strip_prefix(&root).map_err(|_| {
-            LanceError::invalid_input(format!(
-                "Local path '{}' is not beneath store root '{}'",
-                filesystem_path.display(),
-                root.display()
-            ))
-        })?;
-        let relative = relative
-            .components()
-            .filter_map(|component| match component {
-                std::path::Component::Normal(part) => Some(part),
-                _ => None,
-            })
-            .map(|part| {
-                part.to_str().ok_or_else(|| {
-                    LanceError::invalid_input(format!(
-                        "Local path '{}' is not valid UTF-8",
-                        filesystem_path.display()
-                    ))
-                })
-            })
-            .collect::<LanceResult<Vec<_>>>()?
-            .join("/");
-
-        Ok((root, Path::parse(relative)?))
-    }
-
-    /// Preserve the native filesystem root as a structural path component.
-    ///
-    /// `object_store::Path::from_absolute_path` converts a UNC path to its URL
-    /// path and loses the server. Keeping `C:` or `\\server\share` as the first
-    /// component distinguishes an absolute path from every relative path while
-    /// remaining directly usable by Windows filesystem APIs.
-    fn rooted_path(root: &std::path::Path, relative: &Path) -> LanceResult<Path> {
-        let root = root.to_string_lossy();
-        let root = root.trim_end_matches(['/', '\\']);
-        if root.is_empty() {
-            return Ok(relative.clone());
-        }
-        Ok(relative
-            .parts()
-            .fold(Path::parse(root)?, |path, part| path.join(part)))
-    }
-}
-
-/// A local store rooted at a Windows drive or UNC share.
-///
-/// Most calls use paths returned by [`PrefixedFileStoreProvider`], which are
-/// relative to `root`. Some Lance operations retain an existing object store
-/// while independently re-extracting a Windows file URI with the default file
-/// provider. Those paths include the drive (`C:/...`) or UNC share
-/// (`share/...`) again. Normalize that absolute alias before delegating so the
-/// filesystem prefix is never applied twice.
-#[cfg(any(windows, test))]
-#[derive(Debug, Clone)]
-struct RootedLocalFileSystem {
-    inner: Arc<LocalFileSystem>,
-    root: std::path::PathBuf,
-    absolute_alias: Path,
-}
-
-#[cfg(any(windows, test))]
-impl RootedLocalFileSystem {
-    fn new(root: std::path::PathBuf) -> LanceResult<Self> {
-        let absolute_alias = PrefixedFileStoreProvider::rooted_path(&root, &Path::default())?;
-        Ok(Self {
-            inner: Arc::new(LocalFileSystem::new_with_prefix(&root)?),
-            root,
-            absolute_alias,
-        })
-    }
-
-    fn path_from_parts<'a>(parts: impl Iterator<Item = object_store::path::PathPart<'a>>) -> Path {
-        parts.fold(Path::default(), |path, part| path.join(part))
-    }
-
-    fn normalize(&self, path: &Path) -> Path {
-        if self.absolute_alias.as_ref().is_empty() {
-            return path.clone();
-        }
-        let Some(suffix) = path.prefix_match(&self.absolute_alias) else {
-            return path.clone();
-        };
-        Self::path_from_parts(suffix)
-    }
-
-    fn restore_prefix(&self, path: Path, requested: &Path, normalized: &Path) -> Path {
-        if requested == normalized {
-            return path;
-        }
-        path.prefix_match(normalized)
-            .map(|suffix| suffix.fold(requested.clone(), |path, part| path.join(part)))
-            .unwrap_or(path)
-    }
-}
-
-#[cfg(any(windows, test))]
-impl std::fmt::Display for RootedLocalFileSystem {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "RootedLocalFileSystem({})", self.root.display())
-    }
-}
-
-#[cfg(any(windows, test))]
-#[async_trait]
-impl ObjectStore for RootedLocalFileSystem {
-    async fn put_opts(
-        &self,
-        location: &Path,
-        payload: PutPayload,
-        options: PutOptions,
-    ) -> Result<PutResult> {
-        self.inner
-            .put_opts(&self.normalize(location), payload, options)
-            .await
-    }
-
-    async fn put_multipart_opts(
-        &self,
-        location: &Path,
-        options: PutMultipartOptions,
-    ) -> Result<Box<dyn MultipartUpload>> {
-        self.inner
-            .put_multipart_opts(&self.normalize(location), options)
-            .await
-    }
-
-    async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
-        let normalized = self.normalize(location);
-        let mut result = self.inner.get_opts(&normalized, options).await?;
-        result.meta.location = location.clone();
-        Ok(result)
-    }
-
-    fn delete_stream(
-        &self,
-        locations: BoxStream<'static, Result<Path>>,
-    ) -> BoxStream<'static, Result<Path>> {
-        let store = self.clone();
-        locations
-            .map(move |location| {
-                let store = store.clone();
-                async move {
-                    let location = location?;
-                    let normalized = store.normalize(&location);
-                    store.inner.delete(&normalized).await?;
-                    Ok(location)
-                }
-            })
-            .buffered(10)
-            .boxed()
-    }
-
-    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
-        let requested = prefix.cloned().unwrap_or_default();
-        let normalized = self.normalize(&requested);
-        let store = self.clone();
-        self.inner
-            .list(prefix.map(|_| &normalized))
-            .map(move |result| {
-                result.map(|mut meta| {
-                    meta.location = store.restore_prefix(meta.location, &requested, &normalized);
-                    meta
-                })
-            })
-            .boxed()
-    }
-
-    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
-        let requested = prefix.cloned().unwrap_or_default();
-        let normalized = self.normalize(&requested);
-        let mut result = self
-            .inner
-            .list_with_delimiter(prefix.map(|_| &normalized))
-            .await?;
-        for meta in &mut result.objects {
-            meta.location = self.restore_prefix(meta.location.clone(), &requested, &normalized);
-        }
-        for path in &mut result.common_prefixes {
-            *path = self.restore_prefix(path.clone(), &requested, &normalized);
-        }
-        Ok(result)
-    }
-
-    async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
-        self.inner
-            .copy_opts(&self.normalize(from), &self.normalize(to), options)
-            .await
-    }
-}
-
-#[cfg(any(windows, test))]
-#[async_trait]
-impl ObjectStoreProvider for PrefixedFileStoreProvider {
-    async fn new_store(
-        &self,
-        base_path: Url,
-        params: &ObjectStoreParams,
-    ) -> LanceResult<lance::io::ObjectStore> {
-        let (root, _) = Self::root_and_relative_path(&base_path)?;
-        let raw_store: Arc<dyn ObjectStore> = Arc::new(RootedLocalFileSystem::new(root)?);
-        // Native local shortcuts are safe only when the rooted filesystem is
-        // the final store. An arbitrary wrapper can redirect I/O, so use
-        // Lance's non-cloud object-store route in that case. This keeps local
-        // scan planning without enabling native copy/delete or the io_uring
-        // scheduler, all of which would bypass the wrapper.
-        let location = if params.object_store_wrapper.is_some() {
-            Url::parse("memory:///").expect("static URL must be valid")
-        } else {
-            Url::parse("file:///").expect("static URL must be valid")
-        };
-        let storage_options =
-            StorageOptions::new(params.storage_options().cloned().unwrap_or_default());
-
-        // ObjectStore::new initializes the private local-store fields. The
-        // registry owns tracing, custom wrapper, and I/O tracker installation,
-        // so restore the raw store before returning to avoid applying them twice.
-        let mut store = lance::io::ObjectStore::new(
-            raw_store.clone(),
-            location,
-            Some(params.block_size.unwrap_or(4 * 1024)),
-            None,
-            false,
-            false,
-            DEFAULT_LOCAL_IO_PARALLELISM,
-            storage_options.download_retry_count(),
-            params.storage_options(),
-        );
-        store.inner = raw_store;
-        store.store_prefix =
-            self.calculate_object_store_prefix(&base_path, params.storage_options())?;
-        Ok(store)
-    }
-
-    fn extract_path(&self, url: &Url) -> LanceResult<Path> {
-        let (root, relative) = Self::root_and_relative_path(url)?;
-        Self::rooted_path(&root, &relative)
-    }
-
-    fn calculate_object_store_prefix(
-        &self,
-        url: &Url,
-        _storage_options: Option<&std::collections::HashMap<String, String>>,
-    ) -> LanceResult<String> {
-        let (root, _) = Self::root_and_relative_path(url)?;
-        let root = root.canonicalize()?;
-        // One store per drive or UNC share keeps registry and metrics
-        // cardinality bounded. Absolute-vs-relative provenance is carried by
-        // the extracted path instead of the cache key.
-        Ok(format!("file${}", root.display()))
-    }
-}
-
-/// Build LanceDB's default session with the Windows file fallback installed.
-///
-/// Callers that provide a Session retain its registry unchanged.
-pub(crate) fn new_default_session() -> Arc<lance::session::Session> {
-    let session = Arc::new(lance::session::Session::default());
-    #[cfg(windows)]
-    session
-        .store_registry()
-        .insert("file", Arc::new(PrefixedFileStoreProvider));
-    session
-}
-
-#[cfg(test)]
-pub(crate) fn new_prefixed_file_session() -> Arc<lance::session::Session> {
-    let session = Arc::new(lance::session::Session::default());
-    session
-        .store_registry()
-        .insert("file", Arc::new(PrefixedFileStoreProvider));
-    session
-}
-
-#[cfg(test)]
-mod prefixed_file_store_test {
-    use super::*;
-    use object_store::memory::InMemory;
-    use std::sync::atomic::{AtomicUsize, Ordering};
-
-    #[derive(Debug, Default)]
-    struct CountingWrapper {
-        calls: AtomicUsize,
-    }
-
-    impl WrappingObjectStore for CountingWrapper {
-        fn wrap(
-            &self,
-            _store_prefix: &str,
-            original: Arc<dyn ObjectStore>,
-        ) -> Arc<dyn ObjectStore> {
-            self.calls.fetch_add(1, Ordering::Relaxed);
-            original
-        }
-    }
-
-    #[derive(Debug, Default)]
-    struct MemoryRedirectWrapper {
-        store: Arc<InMemory>,
-    }
-
-    impl WrappingObjectStore for MemoryRedirectWrapper {
-        fn wrap(
-            &self,
-            _store_prefix: &str,
-            _original: Arc<dyn ObjectStore>,
-        ) -> Arc<dyn ObjectStore> {
-            self.store.clone()
-        }
-    }
-
-    #[test]
-    fn commit_handler_forwards_commit_outcome_capabilities() {
-        let handler = RootedFileCommitHandler;
-
-        assert!(handler.is_version_not_found_definitive());
-        assert!(!handler.propagate_commit_error_after_success());
-    }
-
-    #[tokio::test]
-    async fn anchors_new_and_existing_directories_at_a_filesystem_prefix() {
-        let tempdir = tempfile::tempdir().unwrap();
-        let database_path = tempdir.path().join("database");
-        std::fs::create_dir(&database_path).unwrap();
-        let table_path = database_path.join("test.lance");
-        let table_url = Url::from_directory_path(&table_path).unwrap();
-
-        let registry = Arc::new(ObjectStoreRegistry::default());
-        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
-
-        // The extracted path remains absolute for Lance's local fast paths,
-        // while the inner store strips the structural root before delegation.
-        let (store, base_path) = lance::io::ObjectStore::from_uri_and_params(
-            registry.clone(),
-            table_url.as_str(),
-            &ObjectStoreParams::default(),
-        )
-        .await
-        .unwrap();
-        assert_eq!(store.scheme(), "file");
-        assert!(store.is_local());
-        assert!(!store.is_cloud());
-        assert_eq!(store.block_size(), 4 * 1024);
-        assert_eq!(store.io_parallelism(), DEFAULT_LOCAL_IO_PARALLELISM);
-        assert_eq!(base_path.filename(), Some("test.lance"));
-        let initial_base_path = base_path.clone();
-
-        let marker = base_path.join("marker");
-        store
-            .inner
-            .put(&marker, bytes::Bytes::from_static(b"new").into())
-            .await
-            .unwrap();
-        assert_eq!(std::fs::read(table_path.join("marker")).unwrap(), b"new");
-
-        // Once the table directory exists, a fresh store uses the same stable
-        // root and object-store path.
-        drop(store);
-        let (store, base_path) = lance::io::ObjectStore::from_uri_and_params(
-            registry,
-            table_url.as_str(),
-            &ObjectStoreParams::default(),
-        )
-        .await
-        .unwrap();
-        assert_eq!(base_path, initial_base_path);
-        let contents = store
-            .inner
-            .get(&base_path.join("marker"))
-            .await
-            .unwrap()
-            .bytes()
-            .await
-            .unwrap();
-        assert_eq!(contents.as_ref(), b"new");
-    }
-
-    #[tokio::test]
-    async fn applies_wrapper_and_io_tracking_once() {
-        let tempdir = tempfile::tempdir().unwrap();
-        let table_url = Url::from_directory_path(tempdir.path().join("test.lance")).unwrap();
-        let registry = Arc::new(ObjectStoreRegistry::default());
-        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
-        let wrapper = Arc::new(CountingWrapper::default());
-        let params = ObjectStoreParams {
-            object_store_wrapper: Some(wrapper.clone()),
-            ..Default::default()
-        };
-
-        let (store, base_path) =
-            lance::io::ObjectStore::from_uri_and_params(registry, table_url.as_str(), &params)
-                .await
-                .unwrap();
-        assert_eq!(wrapper.calls.load(Ordering::Relaxed), 1);
-        assert_eq!(store.scheme(), "memory");
-        assert!(!store.is_local());
-        assert!(!store.is_cloud());
-        assert!(!store.prefers_lite_scheduler());
-
-        store
-            .inner
-            .put(
-                &base_path.join("marker"),
-                bytes::Bytes::from_static(b"tracked").into(),
-            )
-            .await
-            .unwrap();
-        let stats = store.io_tracker().stats();
-        assert_eq!(stats.write_iops, 1);
-        assert_eq!(stats.written_bytes, 7);
-    }
-
-    #[tokio::test]
-    async fn wrapped_store_cleanup_does_not_bypass_the_wrapper() {
-        let tempdir = tempfile::tempdir().unwrap();
-        let table_path = tempdir.path().join("test.lance");
-        std::fs::create_dir(&table_path).unwrap();
-        std::fs::write(table_path.join("native-marker"), b"native").unwrap();
-
-        let table_url = Url::from_directory_path(&table_path).unwrap();
-        let registry = Arc::new(ObjectStoreRegistry::default());
-        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
-        let wrapper = Arc::new(MemoryRedirectWrapper::default());
-        let params = ObjectStoreParams {
-            object_store_wrapper: Some(wrapper.clone()),
-            ..Default::default()
-        };
-        let (store, base_path) =
-            lance::io::ObjectStore::from_uri_and_params(registry, table_url.as_str(), &params)
-                .await
-                .unwrap();
-        let wrapped_marker = base_path.clone().join("wrapped-marker");
-        store
-            .inner
-            .put(
-                &wrapped_marker,
-                bytes::Bytes::from_static(b"wrapped").into(),
-            )
-            .await
-            .unwrap();
-
-        store.remove_dir_all(base_path).await.unwrap();
-
-        assert!(table_path.join("native-marker").exists());
-        assert!(matches!(
-            wrapper.store.head(&wrapped_marker).await,
-            Err(object_store::Error::NotFound { .. })
-        ));
-    }
-
-    #[tokio::test]
-    async fn rooted_commit_handler_uses_store_paths_for_latest_manifest() {
-        let tempdir = tempfile::tempdir().unwrap();
-        let versions = tempdir.path().join("_versions");
-        std::fs::create_dir(&versions).unwrap();
-        std::fs::write(versions.join("2.manifest"), b"native").unwrap();
-
-        let base_path = Path::from_absolute_path(tempdir.path()).unwrap();
-        let raw_store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
-        raw_store
-            .put(
-                &base_path.clone().join("_versions").join("1.manifest"),
-                bytes::Bytes::from_static(b"wrapped").into(),
-            )
-            .await
-            .unwrap();
-        let mut store = lance::io::ObjectStore::new(
-            raw_store.clone(),
-            Url::parse("file:///").unwrap(),
-            Some(4 * 1024),
-            None,
-            false,
-            false,
-            DEFAULT_LOCAL_IO_PARALLELISM,
-            0,
-            None,
-        );
-        store.inner = raw_store;
-
-        let native = RenameCommitHandler
-            .resolve_latest_location(&base_path, &store)
-            .await
-            .unwrap();
-        assert_eq!(native.version, 2);
-
-        let rooted = rooted_file_commit_handler()
-            .resolve_latest_location(&base_path, &store)
-            .await
-            .unwrap();
-        assert_eq!(rooted.version, 1);
-        assert_eq!(
-            rooted.path,
-            base_path.clone().join("_versions").join("1.manifest")
-        );
-    }
-
-    #[tokio::test]
-    async fn reuses_store_cache_across_database_paths() {
-        let tempdir = tempfile::tempdir().unwrap();
-        let first_url = Url::from_directory_path(tempdir.path().join("database")).unwrap();
-        let second_url = Url::from_directory_path(tempdir.path().join("share/database")).unwrap();
-        let registry = Arc::new(ObjectStoreRegistry::default());
-        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
-
-        let (first, _) = lance::io::ObjectStore::from_uri_and_params(
-            registry.clone(),
-            first_url.as_str(),
-            &ObjectStoreParams::default(),
-        )
-        .await
-        .unwrap();
-        let (second, _) = lance::io::ObjectStore::from_uri_and_params(
-            registry.clone(),
-            second_url.as_str(),
-            &ObjectStoreParams::default(),
-        )
-        .await
-        .unwrap();
-
-        assert!(Arc::ptr_eq(&first, &second));
-        assert_eq!(registry.stats().misses, 1);
-        assert_eq!(registry.stats().hits, 1);
-    }
-
-    #[tokio::test]
-    async fn reuses_database_store_for_table_and_clone_targets() {
-        let tempdir = tempfile::tempdir().unwrap();
-        let database_url = Url::from_directory_path(tempdir.path().join("database")).unwrap();
-        let source_url =
-            Url::from_directory_path(tempdir.path().join("database/source.lance")).unwrap();
-        let target_url =
-            Url::from_directory_path(tempdir.path().join("database/target.lance")).unwrap();
-        let registry = Arc::new(ObjectStoreRegistry::default());
-        registry.insert("file", Arc::new(PrefixedFileStoreProvider));
-
-        let (database, _) = lance::io::ObjectStore::from_uri_and_params(
-            registry.clone(),
-            database_url.as_str(),
-            &ObjectStoreParams::default(),
-        )
-        .await
-        .unwrap();
-        let (source, _) = lance::io::ObjectStore::from_uri_and_params(
-            registry.clone(),
-            source_url.as_str(),
-            &ObjectStoreParams::default(),
-        )
-        .await
-        .unwrap();
-        let (target, _) = lance::io::ObjectStore::from_uri_and_params(
-            registry.clone(),
-            target_url.as_str(),
-            &ObjectStoreParams::default(),
-        )
-        .await
-        .unwrap();
-
-        assert!(Arc::ptr_eq(&database, &source));
-        assert!(Arc::ptr_eq(&database, &target));
-        assert_eq!(registry.stats().misses, 1);
-        assert_eq!(registry.stats().hits, 2);
-    }
-
-    #[test]
-    fn normalizes_absolute_drive_and_unc_aliases() {
-        let tempdir = tempfile::tempdir().unwrap();
-        let mut drive_store = RootedLocalFileSystem::new(tempdir.path().to_path_buf()).unwrap();
-        drive_store.absolute_alias = Path::from("C:");
-        assert_eq!(
-            drive_store.normalize(&Path::from("C:/Users/db/table.lance")),
-            Path::from("Users/db/table.lance")
-        );
-
-        let mut unc_store = RootedLocalFileSystem::new(tempdir.path().to_path_buf()).unwrap();
-        unc_store.absolute_alias = Path::parse(r"\\server\share").unwrap();
-        assert_eq!(
-            unc_store.normalize(&Path::parse(r"\\server\share/share/db/table.lance").unwrap()),
-            Path::from("share/db/table.lance")
-        );
-        assert_eq!(
-            unc_store.normalize(&Path::from("share/db/table.lance")),
-            Path::from("share/db/table.lance")
-        );
-    }
-
-    #[test]
-    fn relative_unc_alias_does_not_cross_database_roots() {
-        let tempdir = tempfile::tempdir().unwrap();
-        let mut store = RootedLocalFileSystem::new(tempdir.path().to_path_buf()).unwrap();
-        store.absolute_alias = Path::parse(r"\\server\share").unwrap();
-
-        let relative = Path::from("share/database/table.lance/marker");
-        assert_eq!(store.normalize(&relative), relative);
-        assert_eq!(
-            store.normalize(
-                &Path::parse(r"\\server\share/share/database/table.lance/marker").unwrap()
-            ),
-            relative
-        );
-    }
-
-    #[tokio::test]
-    async fn routes_unc_alias_lifecycle_through_the_prefix() {
-        let tempdir = tempfile::tempdir().unwrap();
-        let mut store = RootedLocalFileSystem::new(tempdir.path().to_path_buf()).unwrap();
-        store.absolute_alias = Path::parse(r"\\server\share").unwrap();
-        let table = Path::parse(r"\\server\share/share/db/test.lance").unwrap();
-        let marker = table.clone().join("marker");
-
-        store
-            .put(&marker, bytes::Bytes::from_static(b"unc").into())
-            .await
-            .unwrap();
-        assert_eq!(
-            std::fs::read(tempdir.path().join("share/db/test.lance/marker")).unwrap(),
-            b"unc"
-        );
-
-        let listed = store.list(Some(&table)).collect::<Vec<_>>().await;
-        assert_eq!(listed.len(), 1);
-        assert_eq!(listed[0].as_ref().unwrap().location, marker);
-        assert_eq!(
-            store
-                .get(&marker)
-                .await
-                .unwrap()
-                .bytes()
-                .await
-                .unwrap()
-                .as_ref(),
-            b"unc"
-        );
-
-        store.delete(&marker).await.unwrap();
-        assert!(!tempdir.path().join("share/db/test.lance/marker").exists());
-    }
-
-    #[cfg(windows)]
-    #[test]
-    fn extracts_drive_and_unc_share_roots() {
-        let (root, path) = PrefixedFileStoreProvider::root_and_relative_path(
-            &Url::parse("file:///C:/database").unwrap(),
-        )
-        .unwrap();
-        assert_eq!(root, std::path::PathBuf::from(r"C:\"));
-        assert_eq!(path, Path::from("database"));
-
-        let (root, path) = PrefixedFileStoreProvider::root_and_relative_path(
-            &Url::parse("file://server/share/database").unwrap(),
-        )
-        .unwrap();
-        assert_eq!(root, std::path::PathBuf::from(r"\\server\share\"));
-        assert_eq!(path, Path::from("database"));
-    }
-
-    #[test]
-    fn preserves_drive_and_unc_roots_in_extracted_paths() {
-        assert_eq!(
-            PrefixedFileStoreProvider::rooted_path(
-                std::path::Path::new(r"C:\"),
-                &Path::from("database/table.lance")
-            )
-            .unwrap(),
-            Path::from("C:/database/table.lance")
-        );
-        assert_eq!(
-            PrefixedFileStoreProvider::rooted_path(
-                std::path::Path::new(r"\\server\share\"),
-                &Path::from("database/table.lance")
-            )
-            .unwrap(),
-            Path::parse(r"\\server\share/database/table.lance").unwrap()
-        );
-    }
-}
-
 #[derive(Debug)]
 struct MirroringObjectStore {
    primary: Arc<dyn ObjectStore>,
@@ -850,7 +52,7 @@ impl PrimaryOnly for Path {
 /// store. We have primary store that is durable but slow, and a secondary
 /// store that is fast but not asdurable
 ///
-/// Note: this object store does not mirror writes to *.manifest files
+/// Note: this object store does not mirror writes to `_latest.manifest`.
 #[async_trait]
 impl ObjectStore for MirroringObjectStore {
    async fn put_opts(
@@ -935,6 +137,16 @@ impl ObjectStore for MirroringObjectStore {
            // or may be evicted before the copy begins.
            match self.secondary.copy_opts(from, to, options.clone()).await {
                Ok(()) | Err(Error::NotFound { .. }) => {}
+                // The secondary is a non-authoritative cache. A process can
+                // leave an orphaned manifest there if it exits before creating
+                // the durable primary object, so let the primary decide the
+                // outcome of create-only manifest copies.
+                Err(Error::AlreadyExists { .. } | Error::Precondition { .. })
+                    if options.mode == CopyMode::Create
+                        && to
+                            .filename()
+                            .map(|name| name.ends_with(".manifest"))
+                            .unwrap_or(false) => {}
                Err(err) => return Err(err),
            }
            self.primary.copy_opts(from, to, options).await
@@ -1138,6 +350,64 @@ mod test {
        ));
    }

+    #[tokio::test]
+    async fn test_create_manifest_recovers_from_orphaned_secondary() {
+        let primary: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+        let secondary: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+        let store = MirroringObjectStore {
+            primary: primary.clone(),
+            secondary: secondary.clone(),
+        };
+        let staging = Path::from("_versions/1.manifest-staging");
+        let finalized = Path::from("_versions/1.manifest");
+
+        primary
+            .put(&staging, "manifest contents".into())
+            .await
+            .unwrap();
+        secondary
+            .put(&staging, "manifest contents".into())
+            .await
+            .unwrap();
+        secondary
+            .copy_if_not_exists(&staging, &finalized)
+            .await
+            .expect("simulate a crash after secondary create and before primary create");
+
+        store
+            .copy_if_not_exists(&staging, &finalized)
+            .await
+            .expect("an orphaned secondary manifest must not block primary creation");
+
+        let copied = primary
+            .get(&finalized)
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+        assert_eq!(copied, "manifest contents");
+
+        assert!(matches!(
+            store.copy_if_not_exists(&staging, &finalized).await,
+            Err(Error::AlreadyExists { .. } | Error::Precondition { .. })
+        ));
+
+        let non_manifest = Path::from("data/existing.lance");
+        secondary
+            .copy_if_not_exists(&staging, &non_manifest)
+            .await
+            .unwrap();
+        assert!(matches!(
+            store.copy_if_not_exists(&staging, &non_manifest).await,
+            Err(Error::AlreadyExists { .. } | Error::Precondition { .. })
+        ));
+        assert!(matches!(
+            primary.head(&non_manifest).await,
+            Err(Error::NotFound { .. })
+        ));
+    }
+
    // This test is ignored because lance 3.0 introduced LocalWriter optimization
    // that bypasses the object store wrapper for local writes. The mirroring feature
    // still works for remote/cloud storage, but can't be tested with local storage.
@@ -50,6 +50,7 @@ use crate::DistanceType;
 use crate::blob::BlobRangeRequest;
 use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions};
 use crate::database::Database;
+use crate::database::listing::LANCE_FILE_EXTENSION;
 use crate::database::read_freshness::TableFreshness;
 use crate::embeddings::{EmbeddingDefinition, EmbeddingRegistry, MemoryRegistry};
 use crate::error::{Error, Result};
@@ -151,6 +152,55 @@ pub(crate) fn map_namespace_lance_error(err: lance::Error, table_name: &str) ->
    }
 }

+/// Map a `lance::Error::DatasetNotFound` for the table at `uri` into a `lancedb::Error`.
+///
+/// Lance reports "there is nothing at this location" and "there is a table directory
+/// here but nothing loadable inside it" with the same error. Only the first is a
+/// `TableNotFound`: a `<name>.lance` directory left behind by an interrupted drop and
+/// re-create is still reported by `Connection::table_names`, so callers need to be able
+/// to tell "never existed" from "exists but is broken".
+///
+/// See <https://github.com/lancedb/lancedb/issues/3127>.
+async fn map_dataset_not_found(
+    uri: &str,
+    name: &str,
+    params: ReadParams,
+    err: lance::Error,
+) -> Error {
+    let name = name.to_string();
+    let source = Box::new(err);
+    if table_dir_exists(uri, params).await.unwrap_or(false) {
+        Error::TableCorrupted { name, source }
+    } else {
+        Error::TableNotFound { name, source }
+    }
+}
+
+/// Whether a table directory is present at `uri`, even though no dataset could be
+/// loaded from it.
+///
+/// This looks for a `<name>.lance` entry in the parent directory, which is exactly what
+/// `ListingDatabase::table_names` lists, so the two APIs agree on whether a table is
+/// present. Probing `uri` itself would not work: object stores have no empty
+/// directories to probe, and on a local filesystem the interesting case is precisely an
+/// empty directory.
+async fn table_dir_exists(uri: &str, params: ReadParams) -> Result<bool> {
+    let (object_store, path, _) = DatasetBuilder::from_uri(uri)
+        .with_read_params(params)
+        .build_object_store()
+        .await?;
+    // Only `*.lance` entries are ever reported as tables, so nothing else can produce
+    // the list-then-open mismatch this guards against.
+    if path.extension() != Some(LANCE_FILE_EXTENSION) {
+        return Ok(false);
+    }
+    let (Some(parent), Some(dir_name)) = (path.parent(), path.filename()) else {
+        return Ok(false);
+    };
+    let entries = object_store.read_dir(parent).await?;
+    Ok(entries.iter().any(|entry| entry.as_str() == dir_name))
+}
+
 /// Defines the type of column
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum ColumnKind {
@@ -2370,6 +2420,8 @@ impl NativeTable {
            None => false,
        };

+        // Kept so that a `DatasetNotFound` can be re-checked against storage below.
+        let recovery_params = params.clone();
        let mut builder = DatasetBuilder::from_uri(uri).with_read_params(params);

        // Set up commit handler when managed_versioning is enabled
@@ -2388,12 +2440,7 @@ impl NativeTable {
        let dataset = match builder.load().await {
            Ok(dataset) => dataset,
            Err(e @ lance::Error::DatasetNotFound { .. }) => {
-                // The manifest load is the existence check. A physical prefix may be
-                // from a concurrent or abandoned create, so it cannot refine this error.
-                return Err(Error::TableNotFound {
-                    name: name.to_string(),
-                    source: Box::new(e),
-                });
+                return Err(map_dataset_not_found(uri, name, recovery_params, e).await);
            }
            Err(e) => return Err(e.into()),
        };
@@ -3661,7 +3708,7 @@ pub struct FragmentSummaryStats {
 #[allow(deprecated)]
 mod tests {
    use std::sync::Arc;
-    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Duration;

    use arrow_array::{
@@ -3743,50 +3790,73 @@ mod tests {
        );
    }

-    #[tokio::test]
-    async fn test_open_not_found_when_empty_directory_exists() {
-        let tmp_dir = tempdir().unwrap();
-        let dataset_path = tmp_dir.path().join("test.lance");
-        std::fs::create_dir(&dataset_path).unwrap();
+    /// Write a table and then break it, leaving the `<name>.lance` directory in place.
+    ///
+    /// `remove_all` reproduces an interrupted drop + re-create (the directory is left
+    /// empty); otherwise only the manifests are removed, leaving the data files behind.
+    async fn write_then_corrupt_table(dir: &std::path::Path, remove_all: bool) -> String {
+        let dataset_path = dir.join("test.lance");
+        let uri = dataset_path.to_str().unwrap().to_string();

-        let err = NativeTable::open(dataset_path.to_str().unwrap())
-            .await
-            .unwrap_err();
+        let batch = make_test_batches();
+        let reader = RecordBatchIterator::new(vec![Ok(batch.clone())], batch.schema());
+        Dataset::write(reader, &uri, None).await.unwrap();
+
+        if remove_all {
+            for entry in std::fs::read_dir(&dataset_path).unwrap() {
+                let entry = entry.unwrap();
+                if entry.file_type().unwrap().is_dir() {
+                    std::fs::remove_dir_all(entry.path()).unwrap();
+                } else {
+                    std::fs::remove_file(entry.path()).unwrap();
+                }
+            }
+            assert_eq!(std::fs::read_dir(&dataset_path).unwrap().count(), 0);
+        } else {
+            let versions = dataset_path.join("_versions");
+            assert!(versions.is_dir(), "expected manifests under {versions:?}");
+            std::fs::remove_dir_all(&versions).unwrap();
+            assert!(std::fs::read_dir(&dataset_path).unwrap().count() > 0);
+        }
+
+        uri
+    }
+
+    #[tokio::test]
+    async fn test_open_corrupt_empty_dir() {
+        let tmp_dir = tempdir().unwrap();
+        let uri = write_then_corrupt_table(tmp_dir.path(), true).await;
+
+        let err = NativeTable::open(&uri).await.unwrap_err();
        assert!(
-            matches!(&err, Error::TableNotFound { name, .. } if name == "test"),
+            matches!(&err, Error::TableCorrupted { name, .. } if name == "test"),
            "got {err:?}"
        );
    }

    #[tokio::test]
-    async fn test_open_not_found_when_only_uncommitted_storage_exists() {
+    async fn test_open_corrupt_missing_manifest() {
        let tmp_dir = tempdir().unwrap();
-        let dataset_path = tmp_dir.path().join("test.lance");
-        let data_dir = dataset_path.join("data");
-        std::fs::create_dir_all(&data_dir).unwrap();
-        std::fs::write(data_dir.join("orphan.lance"), b"uncommitted").unwrap();
+        let uri = write_then_corrupt_table(tmp_dir.path(), false).await;

-        let err = NativeTable::open(dataset_path.to_str().unwrap())
-            .await
-            .unwrap_err();
+        let err = NativeTable::open(&uri).await.unwrap_err();
        assert!(
-            matches!(&err, Error::TableNotFound { name, .. } if name == "test"),
+            matches!(&err, Error::TableCorrupted { name, .. } if name == "test"),
            "got {err:?}"
        );
    }

-    /// Listing databases discover physical `*.lance` entries. That snapshot is not an
-    /// authoritative table-existence check: only a committed manifest makes a table
-    /// openable, and the entry could also be concurrently created or dropped.
+    /// A table listed by `table_names()` must not be reported as missing by
+    /// `open_table()`. See <https://github.com/lancedb/lancedb/issues/3127>.
    #[tokio::test]
-    async fn test_table_names_may_include_uncommitted_storage() {
+    async fn test_open_table_corrupt_is_still_listed() {
        let tmp_dir = tempdir().unwrap();
        let db = connect(tmp_dir.path().to_str().unwrap())
            .execute()
            .await
            .unwrap();

-        std::fs::create_dir(tmp_dir.path().join("test.lance")).unwrap();
+        write_then_corrupt_table(tmp_dir.path(), true).await;

        assert_eq!(
            db.table_names().execute().await.unwrap(),
@@ -3794,177 +3864,12 @@ mod tests {
        );
        let err = db.open_table("test").execute().await.unwrap_err();
        assert!(
-            matches!(&err, Error::TableNotFound { name, .. } if name == "test"),
-            "physical storage without a committed manifest is not a table: {err:?}"
-        );
-    }
-
-    #[derive(Debug)]
-    struct ParentListGuardStore {
-        inner: Arc<dyn object_store::ObjectStore>,
-        parent: object_store::path::Path,
-        parent_list_calls: Arc<AtomicUsize>,
-    }
-
-    impl std::fmt::Display for ParentListGuardStore {
-        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-            f.write_str("ParentListGuardStore")
-        }
-    }
-
-    #[async_trait::async_trait]
-    #[deny(clippy::missing_trait_methods)]
-    impl object_store::ObjectStore for ParentListGuardStore {
-        async fn put_opts(
-            &self,
-            location: &object_store::path::Path,
-            payload: object_store::PutPayload,
-            opts: object_store::PutOptions,
-        ) -> object_store::Result<object_store::PutResult> {
-            self.inner.put_opts(location, payload, opts).await
-        }
-
-        async fn put_multipart_opts(
-            &self,
-            location: &object_store::path::Path,
-            opts: object_store::PutMultipartOptions,
-        ) -> object_store::Result<Box<dyn object_store::MultipartUpload>> {
-            self.inner.put_multipart_opts(location, opts).await
-        }
-
-        async fn get_opts(
-            &self,
-            location: &object_store::path::Path,
-            options: object_store::GetOptions,
-        ) -> object_store::Result<object_store::GetResult> {
-            self.inner.get_opts(location, options).await
-        }
-
-        async fn get_ranges(
-            &self,
-            location: &object_store::path::Path,
-            ranges: &[std::ops::Range<u64>],
-        ) -> object_store::Result<Vec<bytes::Bytes>> {
-            self.inner.get_ranges(location, ranges).await
-        }
-
-        fn delete_stream(
-            &self,
-            locations: futures::stream::BoxStream<
-                'static,
-                object_store::Result<object_store::path::Path>,
-            >,
-        ) -> futures::stream::BoxStream<'static, object_store::Result<object_store::path::Path>>
-        {
-            self.inner.delete_stream(locations)
-        }
-
-        fn list(
-            &self,
-            prefix: Option<&object_store::path::Path>,
-        ) -> futures::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>>
-        {
-            if prefix == Some(&self.parent) {
-                self.parent_list_calls.fetch_add(1, Ordering::Relaxed);
-            }
-            self.inner.list(prefix)
-        }
-
-        fn list_with_offset(
-            &self,
-            prefix: Option<&object_store::path::Path>,
-            offset: &object_store::path::Path,
-        ) -> futures::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>>
-        {
-            if prefix == Some(&self.parent) {
-                self.parent_list_calls.fetch_add(1, Ordering::Relaxed);
-            }
-            self.inner.list_with_offset(prefix, offset)
-        }
-
-        async fn list_with_delimiter(
-            &self,
-            prefix: Option<&object_store::path::Path>,
-        ) -> object_store::Result<object_store::ListResult> {
-            if prefix == Some(&self.parent) {
-                self.parent_list_calls.fetch_add(1, Ordering::Relaxed);
-            }
-            self.inner.list_with_delimiter(prefix).await
-        }
-
-        async fn copy_opts(
-            &self,
-            from: &object_store::path::Path,
-            to: &object_store::path::Path,
-            options: object_store::CopyOptions,
-        ) -> object_store::Result<()> {
-            self.inner.copy_opts(from, to, options).await
-        }
-
-        async fn rename_opts(
-            &self,
-            from: &object_store::path::Path,
-            to: &object_store::path::Path,
-            options: object_store::RenameOptions,
-        ) -> object_store::Result<()> {
-            self.inner.rename_opts(from, to, options).await
-        }
-    }
-
-    #[derive(Debug)]
-    struct ParentListGuardWrapper {
-        parent_list_calls: Arc<AtomicUsize>,
-    }
-
-    impl WrappingObjectStore for ParentListGuardWrapper {
-        fn wrap(
-            &self,
-            _store_prefix: &str,
-            inner: Arc<dyn object_store::ObjectStore>,
-        ) -> Arc<dyn object_store::ObjectStore> {
-            Arc::new(ParentListGuardStore {
-                inner,
-                parent: object_store::path::Path::from("database"),
-                parent_list_calls: self.parent_list_calls.clone(),
-            })
-        }
-    }
-
-    #[tokio::test]
-    async fn test_open_missing_never_lists_database_parent() {
-        let parent_list_calls = Arc::new(AtomicUsize::new(0));
-        let params = ReadParams {
-            store_options: Some(ObjectStoreParams {
-                object_store_wrapper: Some(Arc::new(ParentListGuardWrapper {
-                    parent_list_calls: parent_list_calls.clone(),
-                })),
-                ..Default::default()
-            }),
-            ..Default::default()
-        };
-
-        let err = NativeTable::open_with_params(
-            "memory:///database/missing.lance",
-            "missing",
-            Vec::new(),
-            None,
-            Some(params),
-            None,
-            None,
-            HashSet::new(),
-            None,
-        )
-        .await
-        .unwrap_err();
-
-        assert!(
-            matches!(&err, Error::TableNotFound { name, .. } if name == "missing"),
+            matches!(&err, Error::TableCorrupted { name, .. } if name == "test"),
            "got {err:?}"
        );
-        assert_eq!(
-            parent_list_calls.load(Ordering::Relaxed),
-            0,
-            "opening one missing table must not enumerate sibling tables"
+        assert!(
+            err.to_string().contains("exists but could not be loaded"),
+            "got {err}"
        );
    }

@@ -5434,7 +5339,7 @@ mod tests {
    pub async fn test_stats_includes_index_and_overlay_files() {
        use lance::dataset::WriteDestination;
        use lance::dataset::transaction::{DataOverlayGroup, Operation};
-        use lance_file::version::stable_file_version;
+        use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
        use lance_file::writer::FileWriterOptions;
        use lance_io::utils::CachedFileSize;
        use lance_table::format::DataFile;
@@ -5500,7 +5405,7 @@ mod tests {
        let fragment_id = dataset.get_fragments()[0].id() as u64;
        let foo_field_id = dataset.schema().field("foo").unwrap().id;
        let overlay_schema = dataset.schema().project_by_ids(&[foo_field_id], true);
-        let file_version = stable_file_version();
+        let file_version = ConcreteFileVersion::from(LanceFileVersion::Stable);

        let filename = "overlay.lance".to_string();
        let store = dataset.object_store(None).await.unwrap();
@@ -10,7 +10,7 @@ use arrow_array::{
 use arrow_schema::{DataType, Field, Fields, Schema};
 use futures::TryStreamExt;
 use lance::Dataset;
-use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
+use lance_file::version::LanceFileVersion;
 use lancedb::{
    Connection, Error, Result, Table,
    blob::{BlobRangeRequest, blob},
@@ -61,7 +61,7 @@ async fn create_inline_blob_table(
    Ok(table)
 }

-async fn storage_format_version(table: &Table) -> ConcreteFileVersion {
+async fn storage_format_version(table: &Table) -> LanceFileVersion {
    table
        .as_native()
        .unwrap()
@@ -69,14 +69,9 @@ async fn storage_format_version(table: &Table) -> ConcreteFileVersion {
        .await
        .unwrap()
        .data_storage_format
-        .lance_file_format()
-}
-
-fn supports_blob_v2(version: ConcreteFileVersion) -> bool {
-    matches!(
-        version,
-        ConcreteFileVersion::V2_2 | ConcreteFileVersion::V2_3
-    )
+        .lance_file_version()
+        .unwrap()
+        .resolve()
 }

 async fn uses_stable_row_ids(table: &Table) -> bool {
@@ -117,7 +112,7 @@ async fn declaring_blob_column_bumps_format_and_enables_stable_row_ids() -> Resu
        .execute()
        .await?;

-    assert!(supports_blob_v2(storage_format_version(&table).await));
+    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
    assert!(uses_stable_row_ids(&table).await);
    Ok(())
 }
@@ -132,7 +127,7 @@ async fn explicit_stable_row_id_setting_wins_over_blob_default() -> Result<()> {
        .execute()
        .await?;

-    assert!(supports_blob_v2(storage_format_version(&table).await));
+    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
    assert!(!uses_stable_row_ids(&table).await);
    Ok(())
 }
@@ -144,7 +139,7 @@ async fn non_blob_table_keeps_default_format_and_row_id_setting() -> Result<()>
    let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
    let table = db.create_empty_table("t", schema).execute().await?;

-    assert!(!supports_blob_v2(storage_format_version(&table).await));
+    assert!(storage_format_version(&table).await < LanceFileVersion::V2_2);
    assert!(!uses_stable_row_ids(&table).await);
    Ok(())
 }
@@ -176,7 +171,7 @@ async fn creating_with_blob_data_bumps_format() -> Result<()> {
    .unwrap();
    let table = db.create_table("t", batch).execute().await?;

-    assert!(supports_blob_v2(storage_format_version(&table).await));
+    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
    assert!(uses_stable_row_ids(&table).await);
    assert_eq!(table.count_rows(None).await?, 1);
    Ok(())
@@ -286,7 +281,7 @@ async fn connection_level_stable_row_id_setting_wins_over_blob_default() -> Resu
        .execute()
        .await?;

-    assert!(supports_blob_v2(storage_format_version(&table).await));
+    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
    assert!(!uses_stable_row_ids(&table).await);
    Ok(())
 }
@@ -302,7 +297,7 @@ async fn namespace_create_applies_blob_defaults() -> Result<()> {
        .execute()
        .await?;

-    assert!(supports_blob_v2(storage_format_version(&table).await));
+    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
    assert!(uses_stable_row_ids(&table).await);
    Ok(())
 }
@@ -479,7 +474,7 @@ async fn fetch_blobs_round_trips_nested_blob_column() -> Result<()> {
    let batch = RecordBatch::try_new(schema, vec![Arc::new(info_array) as ArrayRef]).unwrap();
    let table = db.create_table("t", batch).execute().await?;

-    assert!(supports_blob_v2(storage_format_version(&table).await));
+    assert!(storage_format_version(&table).await >= LanceFileVersion::V2_2);
    assert!(uses_stable_row_ids(&table).await);

    let ids = collect_row_ids(&table).await?;
@@ -1310,7 +1305,7 @@ async fn optimize_preserves_blob_v2_null_and_empty_distinction() -> Result<()> {
        .await?;
    table.add(null_empty_input_batch()).execute().await?;
    assert!(
-        supports_blob_v2(storage_format_version(&table).await),
+        storage_format_version(&table).await >= LanceFileVersion::V2_2,
        "blob v2 columns require storage >= 2.2"
    );