feat(python): expose AsyncTable.to_lance (#3730)

## Summary

- expose the existing async Lance dataset conversion as
`AsyncTable.to_lance`
- preserve table version, branch, and refreshed storage options when
opening the dataset
- route internal async pandas/query paths through the public API
- cover normal tables, checked-out versions, branches, and forwarded
dataset options

## Testing

- `cd python && uv run --no-sync pytest python/tests/test_table.py -q`
- `cd python && uv run --no-sync pytest python/tests/test_query.py -q`
- `cd python && uv run --no-sync pytest --doctest-modules
python/lancedb/table.py -q`
- `uv run --project python --no-sync ruff format --check
python/python/lancedb/table.py python/python/lancedb/query.py
python/python/tests/test_table.py`
- `uv run --project python --no-sync ruff check .`

Fixes #1387

python/python/lancedb/query.py

@@ -3041,7 +3041,7 @@ class AsyncQueryBase(object):
             if blob_mode == "bytes"
             else {}
         )
-        dataset = await self._table._to_lance()
+        dataset = await self._table.to_lance()
         scanner = dataset.scanner(
             **_scanner_kwargs_for_query(
                 query,

python/python/lancedb/table.py

@@ -4678,7 +4678,24 @@ class AsyncTable:
         """
         return AsyncQuery(self._inner.query(), self)
 
-    async def _to_lance(self, **kwargs) -> lance.LanceDataset:
+    async def to_lance(self, **kwargs) -> lance.LanceDataset:
+        """Return the Lance dataset backing this table.
+
+        Parameters
+        ----------
+        **kwargs
+            Forwarded to [`lance.dataset`][lance.dataset].
+
+        Returns
+        -------
+        lance.LanceDataset
+            The Lance dataset at this table handle's version and branch.
+
+        Examples
+        --------
+        >>> async def get_lance_dataset(table):
+        ...     return await table.to_lance()
+        """
         try:
             import lance
         except ImportError:
@@ -4728,7 +4745,7 @@ class AsyncTable:
             return (await self.to_arrow()).to_pandas(**kwargs)
         if blob_mode == "bytes" and blob_v2_column_paths(schema):
             return await self.query().to_pandas(blob_mode=blob_mode, **kwargs)
-        return (await self._to_lance()).to_pandas(blob_mode=blob_mode, **kwargs)
+        return (await self.to_lance()).to_pandas(blob_mode=blob_mode, **kwargs)
 
     async def to_arrow(self) -> pa.Table:
         """Return the table as a pyarrow Table.

python/python/tests/test_table.py

@@ -1257,6 +1257,53 @@ def test_branch_to_lance_targets_branch(tmp_path):
     assert table.to_lance().count_rows() == 1
 
 
+@pytest.mark.asyncio
+async def test_async_to_lance(tmp_path):
+    pytest.importorskip("lance")
+    db = await lancedb.connect_async(tmp_path)
+    table = await db.create_table("t", [{"i": 1}])
+
+    dataset = await table.to_lance()
+
+    assert dataset.count_rows() == 1
+
+
+@pytest.mark.asyncio
+async def test_async_branch_to_lance_targets_branch(tmp_path):
+    pytest.importorskip("lance")
+    db = await lancedb.connect_async(tmp_path)
+    table = await db.create_table("t", [{"i": 1}])
+    branch = await table.branches.create("exp")
+    await branch.add([{"i": 2}])
+
+    assert (await branch.to_lance()).count_rows() == 2
+    assert (await table.to_lance()).count_rows() == 1
+
+
+@pytest.mark.asyncio
+async def test_async_to_lance_targets_checked_out_version(tmp_path):
+    pytest.importorskip("lance")
+    db = await lancedb.connect_async(tmp_path)
+    table = await db.create_table("t", [{"i": 1}])
+    version = await table.version()
+    await table.add([{"i": 2}])
+    checked_out = await db.open_table("t", version=version)
+
+    assert (await checked_out.to_lance()).count_rows() == 1
+    assert (await table.to_lance()).count_rows() == 2
+
+
+@pytest.mark.asyncio
+async def test_async_to_lance_forwards_dataset_options(tmp_path):
+    pytest.importorskip("lance")
+    db = await lancedb.connect_async(tmp_path)
+    table = await db.create_table("t", [{"i": 1}])
+
+    dataset = await table.to_lance(default_scan_options={"with_row_id": True})
+
+    assert "_rowid" in dataset.schema.names
+
+
 @pytest.mark.asyncio
 async def test_async_branches(tmp_path):
     db = await lancedb.connect_async(tmp_path)