feat: add get_lsm_write_spec to read the installed LSM write spec

Adds `Table::get_lsm_write_spec` returning `Option<LsmWriteSpec>`, the read
counterpart to `set_lsm_write_spec` / `unset_lsm_write_spec`.

- Rust core: `NativeTable` reconstructs the spec from the MemWAL index
  details (`mem_wal_index_details`), resolving the shard column via the
  dataset schema. Read-only, so unaffected by describe_indices filtering.
- Remote: `RemoteTable` reads the `__lance_mem_wal` system index via
  `index/list` with `include_system: true` (kept out of the curated
  `list_indices` surface) and parses its `details` JSON, using the
  server-resolved `column` name.
- Python and TypeScript bindings, sync + async, with round-trip tests.

The remote path depends on lance emitting the resolved shard column name in
the MemWAL index details JSON, and on the server honoring `include_system`
on `index/list`.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

python/python/lancedb/_lancedb.pyi

@@ -226,6 +226,7 @@ class Table:
     async def set_unenforced_primary_key(self, columns: List[str]) -> None: ...
     async def set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ...
     async def unset_lsm_write_spec(self) -> None: ...
+    async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ...
     async def close_lsm_writers(self) -> None: ...
     @property
     def tags(self) -> Tags: ...

python/python/lancedb/remote/table.py

@@ -912,6 +912,10 @@ class RemoteTable(Table):
         """Not supported on LanceDB Cloud."""
         return LOOP.run(self._table.unset_lsm_write_spec())
 
+    def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
+        """Read the installed LsmWriteSpec, or ``None``."""
+        return LOOP.run(self._table.get_lsm_write_spec())
+
     def close_lsm_writers(self) -> None:
         """No-op on LanceDB Cloud (no local shard writers)."""
         return LOOP.run(self._table.close_lsm_writers())

python/python/lancedb/table.py

@@ -3742,6 +3742,11 @@ class LanceTable(Table):
         [`AsyncTable.unset_lsm_write_spec`][lancedb.AsyncTable.unset_lsm_write_spec]."""
         return LOOP.run(self._table.unset_lsm_write_spec())
 
+    def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
+        """Read the installed LsmWriteSpec, or ``None``. See
+        [`AsyncTable.get_lsm_write_spec`][lancedb.AsyncTable.get_lsm_write_spec]."""
+        return LOOP.run(self._table.get_lsm_write_spec())
+
     def close_lsm_writers(self) -> None:
         """Close cached MemWAL shard writers. See
         [`AsyncTable.close_lsm_writers`][lancedb.AsyncTable.close_lsm_writers]."""
@@ -4391,6 +4396,17 @@ class AsyncTable:
         """
         await self._inner.unset_lsm_write_spec()
 
+    async def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
+        """Read the LsmWriteSpec currently installed on this table.
+
+        Returns ``None`` when the MemWAL LSM write path is not enabled (no
+        spec has been set, or it was removed with `unset_lsm_write_spec`).
+        The returned spec — including its ``maintained_indexes`` and
+        ``writer_config_defaults`` — mirrors what was passed to
+        `set_lsm_write_spec`.
+        """
+        return await self._inner.get_lsm_write_spec()
+
     async def close_lsm_writers(self) -> None:
         """Drain and close any cached MemWAL shard writers for this table.
 

python/python/tests/test_lsm_write_spec.py

@@ -136,3 +136,62 @@ def test_lsm_write_spec_identity_and_writer_config_defaults():
     s = s.with_writer_config_defaults({"durable_write": "false"})
     assert s.writer_config_defaults == {"durable_write": "false"}
     assert "durable_write" in repr(s)
+
+
+def test_get_lsm_write_spec(tmp_path):
+    _db, table = _make_table(tmp_path)
+    table.set_unenforced_primary_key("id")
+
+    # None when nothing is installed.
+    assert table.get_lsm_write_spec() is None
+
+    # Bucket spec round-trips, including writer config defaults.
+    table.set_lsm_write_spec(
+        LsmWriteSpec.bucket("id", 4).with_writer_config_defaults(
+            {"durable_write": "false"}
+        )
+    )
+    spec = table.get_lsm_write_spec()
+    assert spec is not None
+    assert spec.spec_type == "bucket"
+    assert spec.column == "id"
+    assert spec.num_buckets == 4
+    assert spec.writer_config_defaults == {"durable_write": "false"}
+
+    # After unset, None again.
+    table.unset_lsm_write_spec()
+    assert table.get_lsm_write_spec() is None
+
+    # Identity round-trips (column recovered from the schema).
+    table.set_lsm_write_spec(LsmWriteSpec.identity("id"))
+    spec = table.get_lsm_write_spec()
+    assert spec.spec_type == "identity"
+    assert spec.column == "id"
+    table.unset_lsm_write_spec()
+
+    # Unsharded round-trips (no routing column).
+    table.set_lsm_write_spec(LsmWriteSpec.unsharded())
+    spec = table.get_lsm_write_spec()
+    assert spec.spec_type == "unsharded"
+    assert spec.column is None
+
+
+@pytest.mark.asyncio
+async def test_async_get_lsm_write_spec(tmp_path):
+    db = await lancedb.connect_async(
+        tmp_path, read_consistency_interval=timedelta(seconds=0)
+    )
+    table = await db.create_table(
+        "t",
+        pa.RecordBatchReader.from_batches(SCHEMA, [_batch(["seed"], [0])]),
+    )
+
+    assert await table.get_lsm_write_spec() is None
+    await table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 8))
+    spec = await table.get_lsm_write_spec()
+    assert spec is not None
+    assert spec.spec_type == "bucket"
+    assert spec.column == "id"
+    assert spec.num_buckets == 8
+    await table.unset_lsm_write_spec()
+    assert await table.get_lsm_write_spec() is None

python/src/table.rs

@@ -322,6 +322,12 @@ impl From<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
     }
 }
 
+impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
+    fn from(inner: lancedb::table::LsmWriteSpec) -> Self {
+        Self { inner }
+    }
+}
+
 #[pyclass(get_all, from_py_object)]
 #[derive(Clone, Debug)]
 pub struct AddColumnsResult {
@@ -1029,6 +1035,14 @@ impl Table {
         })
     }
 
+    pub fn get_lsm_write_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let spec = inner.get_lsm_write_spec().await.infer_error()?;
+            Ok(spec.map(LsmWriteSpec::from))
+        })
+    }
+
     pub fn close_lsm_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner_ref()?.clone();
         future_into_py(self_.py(), async move {