fix(python): honor namespace table storage options (#4483)

Follow-up on current main `7b3ba4c4` (merge commit `0b0475c1`): resolved
the test import conflict, preserving both upstream imports and the
original `warnings` import. The final PR diff is still limited to the
same storage-option helper/call sites and regression class. AST
comparisons preserve every upstream test definition and the original
regression class/helper; both changed files compile and pass Ruff 0.15.1
check/format.

Fresh locked-environment runtime validation was attempted with `uv run
--locked --extra tests --extra dev pytest python/tests/test_namespace.py
-q`, but dependency setup failed downloading the locked
`pylance==14.0.0b12` wheel due a network connection error. No
current-main native test result is claimed. Repository-wide Ruff reports
49 findings and 11 formatting differences in unchanged upstream
docs/web-tests files; these were left outside this PR. CI approval and a
human review are still pending.

---

Original implementation and historical validation (before the main
merge):

Namespace-backed `create_table` accepts `data_storage_version` and
`enable_v2_manifest_paths`, but neither sync nor async creation forwards
them.
The requested format/manifest settings are silently ignored.

Normalize these deprecated arguments into the existing `new_table_*`
storage
options, matching ordinary connections. Explicit arguments take
precedence;
the caller's dictionary is not mutated. No Rust or native interface
changes.

Regression tests inspect persisted tables, including sync/async
creation,
populated/empty tables, stable/legacy formats, both manifest settings
and
explicit-option precedence.

Validation on Linux/Python 3.12, based on current main `d3390a9`:
- Rebuilt the exact-main native extension and reused that identical
extension
  in separate baseline/candidate environments; only the candidate Python
namespace wrapper differs. Dependencies came from the unchanged
`uv.lock`.
- Same complete `test_namespace.py`: baseline 16 regression failures and
39
  passing controls; candidate **55 passed**, no skips/errors.
- Ruff check/format and diff whitespace checks passed.

This validates the local directory-namespace path, not a live Cloud
service or
the complete repository CI suite.

Signed-off-by: Xucheng Zhou <aden1350@outlook.com>

python/python/lancedb/namespace.py

@@ -11,6 +11,7 @@ through a namespace abstraction.
 from __future__ import annotations
 
 import sys
+import warnings
 from typing import Any, Dict, Iterable, List, Optional, Union
 from uuid import UUID
 
@@ -82,6 +83,31 @@ from ._lancedb import Session
 _MAX_QUERY_K = 2**31 - 1
 
 
+def _create_table_storage_options(
+    storage_options: Optional[Dict[str, str]],
+    data_storage_version: Optional[str],
+    enable_v2_manifest_paths: Optional[bool],
+) -> Optional[Dict[str, str]]:
+    if data_storage_version is None and enable_v2_manifest_paths is None:
+        return storage_options
+    options = dict(storage_options or {})
+    for name, value in (
+        ("data_storage_version", data_storage_version),
+        ("enable_v2_manifest_paths", enable_v2_manifest_paths),
+    ):
+        if value is not None:
+            warnings.warn(
+                f"setting {name} directly on create_table is deprecated. "
+                "Use storage_options instead.",
+                DeprecationWarning,
+                stacklevel=3,
+            )
+            options[f"new_table_{name}"] = (
+                str(value).lower() if isinstance(value, bool) else value
+            )
+    return options
+
+
 def _query_to_namespace_request(
     table_id: List[str],
     query: "Query",
@@ -587,6 +613,9 @@ class LanceNamespaceDBConnection(DBConnection):
         if mode.lower() not in ["create", "overwrite"]:
             raise ValueError("mode must be either 'create' or 'overwrite'")
         validate_table_name(name)
+        storage_options = _create_table_storage_options(
+            storage_options, data_storage_version, enable_v2_manifest_paths
+        )
         async_table = LOOP.run(
             self._inner.create_table(
                 name,
@@ -1230,6 +1259,9 @@ class AsyncLanceNamespaceDBConnection:
         if mode.lower() not in ["create", "overwrite"]:
             raise ValueError("mode must be either 'create' or 'overwrite'")
         validate_table_name(name)
+        storage_options = _create_table_storage_options(
+            storage_options, data_storage_version, enable_v2_manifest_paths
+        )
         table = await self._inner.create_table(
             name,
             data,

python/python/tests/test_namespace.py

@@ -6,6 +6,7 @@
 import tempfile
 import shutil
 import importlib
+import warnings
 import http.server
 import json
 import threading
@@ -25,6 +26,125 @@ PUSHDOWN_DATA = pa.table(
 )
 
 
+class TestNamespaceCreateTableOptions:
+    @pytest.mark.parametrize("version", ["legacy", "stable"])
+    @pytest.mark.parametrize("empty", [False, True])
+    def test_deprecated_storage_version(self, tmp_path, version, empty):
+        db = lancedb.connect_namespace("dir", {"root": str(tmp_path / "namespace")})
+        data = None if empty else [{"id": 1}]
+        schema = pa.schema([("id", pa.int64())])
+        other_version = "stable" if version == "legacy" else "legacy"
+        options = {"new_table_data_storage_version": other_version}
+        control = lancedb.connect(tmp_path / "control").create_table(
+            "test",
+            data=data,
+            schema=schema,
+            storage_options={"new_table_data_storage_version": version},
+        )
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always", DeprecationWarning)
+            table = db.create_table(
+                "test",
+                data=data,
+                schema=schema,
+                storage_options=options,
+                data_storage_version=version,
+            )
+
+        assert (
+            table.to_lance().data_storage_version
+            == control.to_lance().data_storage_version
+        )
+        assert table.to_arrow().to_pylist() == ([] if empty else data)
+        assert options == {"new_table_data_storage_version": other_version}
+        assert any("data_storage_version" in str(w.message) for w in caught)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("version", ["legacy", "stable"])
+    @pytest.mark.parametrize("empty", [False, True])
+    async def test_deprecated_storage_version_async(self, tmp_path, version, empty):
+        root = str(tmp_path / "namespace")
+        db = lancedb.connect_namespace_async("dir", {"root": root})
+        data = None if empty else [{"id": 1}]
+        schema = pa.schema([("id", pa.int64())])
+        other_version = "stable" if version == "legacy" else "legacy"
+        options = {"new_table_data_storage_version": other_version}
+        control = lancedb.connect(tmp_path / "control").create_table(
+            "test",
+            data=data,
+            schema=schema,
+            storage_options={"new_table_data_storage_version": version},
+        )
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always", DeprecationWarning)
+            table = await db.create_table(
+                "test",
+                data=data,
+                schema=schema,
+                storage_options=options,
+                data_storage_version=version,
+            )
+
+        reopened = lancedb.connect_namespace("dir", {"root": root}).open_table("test")
+        assert (
+            reopened.to_lance().data_storage_version
+            == control.to_lance().data_storage_version
+        )
+        assert (await table.to_arrow()).to_pylist() == ([] if empty else data)
+        assert options == {"new_table_data_storage_version": other_version}
+        assert any("data_storage_version" in str(w.message) for w in caught)
+
+    @pytest.mark.parametrize("enabled", [False, True])
+    @pytest.mark.parametrize("empty", [False, True])
+    def test_deprecated_manifest_paths(self, tmp_path, enabled, empty):
+        db = lancedb.connect_namespace("dir", {"root": str(tmp_path)})
+        data = None if empty else [{"id": 1}]
+        options = {"new_table_enable_v2_manifest_paths": str(not enabled).lower()}
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always", DeprecationWarning)
+            table = db.create_table(
+                "test",
+                data=data,
+                schema=pa.schema([("id", pa.int64())]),
+                storage_options=options,
+                enable_v2_manifest_paths=enabled,
+            )
+
+        assert table.uses_v2_manifest_paths() == enabled
+        assert table.to_arrow().to_pylist() == ([] if empty else data)
+        assert db.open_table("test").uses_v2_manifest_paths() == enabled
+        assert options == {
+            "new_table_enable_v2_manifest_paths": str(not enabled).lower()
+        }
+        assert any("enable_v2_manifest_paths" in str(w.message) for w in caught)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("enabled", [False, True])
+    @pytest.mark.parametrize("empty", [False, True])
+    async def test_deprecated_manifest_paths_async(self, tmp_path, enabled, empty):
+        db = lancedb.connect_namespace_async("dir", {"root": str(tmp_path)})
+        data = None if empty else [{"id": 1}]
+        options = {"new_table_enable_v2_manifest_paths": str(not enabled).lower()}
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always", DeprecationWarning)
+            table = await db.create_table(
+                "test",
+                data=data,
+                schema=pa.schema([("id", pa.int64())]),
+                storage_options=options,
+                enable_v2_manifest_paths=enabled,
+            )
+
+        assert await table.uses_v2_manifest_paths() == enabled
+        assert (await table.to_arrow()).to_pylist() == ([] if empty else data)
+        reopened = await db.open_table("test")
+        assert await reopened.uses_v2_manifest_paths() == enabled
+        assert options == {
+            "new_table_enable_v2_manifest_paths": str(not enabled).lower()
+        }
+        assert any("enable_v2_manifest_paths" in str(w.message) for w in caught)
+
+
 def _ipc_file(table: pa.Table = PUSHDOWN_DATA) -> bytes:
     sink = pa.BufferOutputStream()
     with pa.ipc.new_file(sink, table.schema) as writer: