Bump version: 0.38.0-beta.1 → 0.38.0-beta.2

feat(python): expose LSM checkpoint and stats on sync RemoteTable (#3961 )
## Summary The sync `RemoteTable` carried `set_lsm_write_spec`, `unset_lsm_write_spec`, `get_lsm_write_spec`, and `close_lsm_writers`, but not `checkpoint_lsm`, `flush_lsm`, `compact_lsm`, or `get_lsm_stats`. That left the four LSM control methods reachable from `AsyncTable` only. They are also the four that *only* work against a remote table — `NativeTable` does not override the `BaseTable` defaults, so on a local table they return `NotSupported` (`rust/lancedb/src/table.rs:679-701`). The net effect for sync users: | | `checkpoint_lsm` / `get_lsm_stats` | |---|---| | `LanceTable` (sync, local) | present, but always `NotSupported` | | `RemoteTable` (sync, remote) | `AttributeError` — method absent | | `AsyncTable` (remote) | works | So there was no working sync path at all, despite the Rust `RemoteTable` implementing every one of these against real endpoints. ## Changes * Add `checkpoint_lsm`, `flush_lsm`, `compact_lsm`, and `get_lsm_stats` to `lancedb.remote.table.RemoteTable`, mirroring the delegation style of their neighbours. * Correct the docstrings on `set_lsm_write_spec` / `unset_lsm_write_spec`, which read `"""Not supported on LanceDB Cloud."""` although `rust/lancedb/src/remote/table.rs:2549-2601` implements both against `/v1/table/{}/set_lsm_write_spec/` and `/unset_lsm_write_spec/`. They appear to have been copy-pasted from `set_unenforced_primary_key` directly above. No Rust or PyO3 changes — the bindings and the `AsyncTable` methods already existed. The `Table` ABC is left alone, matching how the existing `*_lsm_write_spec` methods are declared on the concrete classes only. ## Tests Four new tests in `python/python/tests/test_remote_db.py`, against the existing mock HTTP server: * `test_get_lsm_stats_sync` — the server payload round-trips into the dict, and `include_generation_rows` defaults to `False` and is forwarded when set. * `test_get_lsm_stats_sync_returns_none_when_lsm_disabled` — a `{"lsm_stats": null}` envelope yields `None` rather than an error. * `test_flush_and_compact_lsm_sync` — both are one-shot POSTs answered `202` with no body. * `test_checkpoint_lsm_sync` — pins the binding to the endpoints it drives (`flush_lsm` then `get_lsm_stats`); the convergence loop itself is already covered in Rust. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-29 17:38:31 +00:00 · 2026-08-19 01:58:48 +00:00 · 2026-08-18 17:25:08 -05:00 · 2026-08-18 20:03:20 +08:00 · 2026-08-17 11:38:38 -07:00 · 2026-08-17 10:48:02 -07:00
63 changed files with 4778 additions and 227 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.37.1-beta.1"
+current_version = "0.38.0-beta.2"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -4,14 +4,14 @@ on:
  workflow_call:
    inputs:
      tag:
-        description: "Tag name from Lance. If omitted, the skill will use the latest Lance release that needs an update."
+        description: "Tag name from Lance (e.g. `v7.2.0-beta.1`). If omitted, the newest release is resolved automatically — stable releases are preferred over pre-releases — and the run is skipped if it is not newer than the version currently pinned in Cargo.toml."
        required: false
        default: ""
        type: string
  workflow_dispatch:
    inputs:
      tag:
-        description: "Tag name from Lance. Leave empty to use the latest Lance release that needs an update."
+        description: "Tag name from Lance (e.g. `v7.2.0-beta.1`). Leave empty to resolve the newest release automatically — stable releases are preferred over pre-releases — and skip the run if it is not newer than the version currently pinned in Cargo.toml."
        required: false
        default: ""
        type: string
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

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

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

 [[package]]
 name = "lance-arrow"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4913,7 +4911,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.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4927,7 +4925,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.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4936,8 +4934,8 @@ dependencies = [

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

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

 [[package]]
 name = "lance-datafusion"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5008,7 +5003,6 @@ dependencies = [
 "jsonb",
 "lance-arrow",
 "lance-core",
- "lance-datagen",
 "log",
 "pin-project",
 "prost",
@@ -5019,8 +5013,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5037,8 +5031,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5047,8 +5041,8 @@ dependencies = [

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

 [[package]]
 name = "lance-file"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5114,8 +5107,8 @@ dependencies = [

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

 [[package]]
 name = "lance-index-core"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5205,8 +5195,8 @@ dependencies = [

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

 [[package]]
 name = "lance-linalg"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 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.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5272,8 +5260,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace-impls"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5312,9 +5300,9 @@ dependencies = [

 [[package]]
 name = "lance-namespace-reqwest-client"
-version = "0.8.6"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba3f0a235e3ed5f8805205649ccc7d7d0f3df23ce1294242c9265ad488d7f19d"
+checksum = "0a030196da1c994b63a96a4f0bf5b0cfa459fe6dadc9e962320246ca328da22a"
 dependencies = [
 "reqwest 0.12.28",
 "serde",
@@ -5326,14 +5314,13 @@ dependencies = [

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

 [[package]]
 name = "lance-table"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5383,8 +5370,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5397,8 +5384,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "11.0.0-beta.7"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.7#e581c49338bc83baf1ea50c5e235bd702f3fbeea"
+version = "11.0.0-beta.13"
+source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.13#ee41152ceb9a78e5df4d2456fdbdb98542eb2059"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -5411,7 +5398,7 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.37.1-beta.1"
+version = "0.38.0-beta.1"
 dependencies = [
 "ahash",
 "anyhow",
@@ -5499,7 +5486,7 @@ dependencies = [

 [[package]]
 name = "lancedb-nodejs"
-version = "0.37.1-beta.1"
+version = "0.38.0-beta.1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5524,7 +5511,7 @@ dependencies = [

 [[package]]
 name = "lancedb-python"
-version = "0.37.1-beta.1"
+version = "0.38.0-beta.1"
 dependencies = [
 "arrow",
 "async-trait",
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=11.0.0-beta.7", default-features = false, "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=11.0.0-beta.7", default-features = false, "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=11.0.0-beta.7", default-features = false, "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=11.0.0-beta.7", "tag" = "v11.0.0-beta.7", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=11.0.0-beta.13", default-features = false, "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=11.0.0-beta.13", default-features = false, "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=11.0.0-beta.13", default-features = false, "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=11.0.0-beta.13", "tag" = "v11.0.0-beta.13", "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 }
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-core</artifactId>
-    <version>0.37.1-beta.1</version>
+    <version>0.38.0-beta.2</version>
 </dependency>
 ```

@@ -55,6 +55,38 @@ LanceNamespace namespaceClient = LanceDbNamespaceClientBuilder.newBuilder()
 | `region(String)` | AWS region (default: "us-east-1") | No |
 | `config(String, String)` | Additional configuration parameters | No |

+### Opening a Table with Vended Credentials
+
+When the catalog vends temporary object store credentials, open the table through the
+namespace client. The Lance dataset builder fetches the table location and storage options
+from the catalog and refreshes the credentials when they expire.
+
+```java
+import com.lancedb.LanceDbNamespaceClientBuilder;
+import org.lance.Dataset;
+import org.lance.namespace.LanceNamespace;
+
+import java.util.Arrays;
+
+LanceNamespace namespaceClient = LanceDbNamespaceClientBuilder.newBuilder()
+    .apiKey(System.getenv("LANCEDB_API_KEY"))
+    .database(System.getenv("LANCEDB_DATABASE"))
+    // Set the endpoint for a LanceDB Enterprise deployment.
+    // .endpoint("https://your-enterprise-endpoint")
+    .build();
+
+try (Dataset dataset = Dataset.open()
+        .namespaceClient(namespaceClient)
+        .tableId(Arrays.asList("my_namespace", "my_table"))
+        .build()) {
+    System.out.println("Rows: " + dataset.countRows());
+}
+```
+
+Do not call `describeTable()` and then open the returned location with `Dataset.open(uri)`.
+Opening through `namespaceClient()` is what applies the vended storage options and enables
+automatic credential refresh. No object store credentials need to be passed by the application.
+
 ## Metadata Operations

 ### Creating a Namespace Path
@@ -386,6 +386,29 @@ Drop an existing table.

 ***

+### dropTableAsync()
+
+```ts
+abstract dropTableAsync(name, namespacePath?): Promise<Job>
+```
+
+Start dropping a table and return its cleanup job.
+
+The table may become unavailable before its data files are removed. Wait
+on the returned job to know when cleanup has finished.
+
+#### Parameters
+
+* **name**: `string`
+
+* **namespacePath?**: `string`[]
+
+#### Returns
+
+`Promise`&lt;[`Job`](Job.md)&gt;
+
+***
+
 ### getJob()

 ```ts
@@ -69,14 +69,34 @@ abstract addColumns(newColumnTransforms): Promise<AddColumnsResult>

 Add new columns with defined values.

+The `{ computed }` form stores the expression rather than evaluating it
+now: the column is committed with no values, and rows get them from
+[Table#refreshColumn](Table.md#refreshcolumn). Declaring one therefore costs the same on a
+large table as on an empty one.
+
+A refresh does not revisit rows it has already filled, so mutating an
+input leaves the value computed at fill time; recomputing means dropping
+the column and declaring it again. While a declaration reads a column,
+that column cannot be renamed, retyped or dropped.
+
+On LanceDB Cloud and Enterprise the expression is planned by the
+server, and the refresh runs as a server job -- see
+[Table#refreshColumnAsync](Table.md#refreshcolumnasync).
+
 #### Parameters

-* **newColumnTransforms**: `Field`&lt;`any`&gt; \| `Field`&lt;`any`&gt;[] \| `Schema`&lt;`any`&gt; \| [`AddColumnsSql`](../interfaces/AddColumnsSql.md)[]
+* **newColumnTransforms**:
+    \| `Field`&lt;`any`&gt;
+    \| `Field`&lt;`any`&gt;[]
+    \| `Schema`&lt;`any`&gt;
+    \| [`AddColumnsSql`](../interfaces/AddColumnsSql.md)[]
+    \| `object`
    Either:
    - An array of objects with column names and SQL expressions to calculate values
    - A single Arrow Field defining one column with its data type (column will be initialized with null values)
    - An array of Arrow Fields defining columns with their data types (columns will be initialized with null values)
    - An Arrow Schema defining columns with their data types (columns will be initialized with null values)
+    - `{ computed }`, declaring columns defined by a SQL expression whose type and inputs are derived from it

 #### Returns

@@ -85,6 +105,13 @@ Add new columns with defined values.
 A promise that resolves to an object
 containing the new version number of the table after adding the columns.

+#### Example
+
+```ts
+await table.addColumns({ computed: [{ name: "doubled", valueSql: "x * 2" }] });
+const { rowsFilled } = await table.refreshColumn("doubled");
+```
+
 ***

 ### alterColumns()
@@ -718,6 +745,67 @@ for await (const batch of table.query()) {

 ***

+### refreshColumn()
+
+```ts
+abstract refreshColumn(column): Promise<RefreshColumnResult>
+```
+
+Fill the rows of a computed column that hold no value yet.
+
+Rows appended since the last refresh are filled by the next one; rows
+already filled are left as they are, so the call is idempotent and does
+not observe a mutated input. Local tables only: a remote refresh runs
+as a server job, through [Table#refreshColumnAsync](Table.md#refreshcolumnasync).
+
+#### Parameters
+
+* **column**: `string`
+    The name of the computed column to fill.
+
+#### Returns
+
+`Promise`&lt;[`RefreshColumnResult`](../interfaces/RefreshColumnResult.md)&gt;
+
+A promise that resolves to the
+number of rows filled and the new version number of the table.
+
+***
+
+### refreshColumnAsync()
+
+```ts
+abstract refreshColumnAsync(column): Promise<Job>
+```
+
+Like [Table#refreshColumn](Table.md#refreshcolumn), but returns a handle to the refresh
+job instead of blocking until it completes.
+
+The job may already be complete when returned; callers must not assume
+the column is filled until [Job.wait](Job.md#wait) resolves. Invalid input --
+an unknown column, or one that is not computed -- rejects here rather
+than failing the job. On local tables the job runs in-process; on
+LanceDB Cloud and Enterprise it is the server's backfill job.
+
+#### Parameters
+
+* **column**: `string`
+    The name of the computed column to fill.
+
+#### Returns
+
+`Promise`&lt;[`Job`](Job.md)&gt;
+
+#### Example
+
+```ts
+const job = await table.refreshColumnAsync("doubled");
+await job.wait();
+console.log(await job.status()); // "finished"
+```
+
+***
+
 ### restore()

 ```ts
@@ -105,6 +105,7 @@
 - [OptimizeOptions](interfaces/OptimizeOptions.md)
 - [OptimizeStats](interfaces/OptimizeStats.md)
 - [QueryExecutionOptions](interfaces/QueryExecutionOptions.md)
+- [RefreshColumnResult](interfaces/RefreshColumnResult.md)
 - [RemovalStats](interfaces/RemovalStats.md)
 - [RenameTableOptions](interfaces/RenameTableOptions.md)
 - [RestNamespaceConfig](interfaces/RestNamespaceConfig.md)
@@ -0,0 +1,23 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / RefreshColumnResult
+
+# Interface: RefreshColumnResult
+
+## Properties
+
+### rowsFilled
+
+```ts
+rowsFilled: number;
+```
+
+***
+
+### version
+
+```ts
+version: number;
+```
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.37.1-beta.1</version>
+      <version>0.38.0-beta.2</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

@@ -6,7 +6,7 @@

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.37.1-beta.1</version>
+    <version>0.38.0-beta.2</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -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.7</lance-core.version>
+        <lance-core.version>11.0.0-beta.13</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>
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.37.1-beta.1"
+version = "0.38.0-beta.2"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -89,6 +89,16 @@ describe("given a connection", () => {
    await db.createTable("test4", [{ id: 1 }, { id: 2 }]);
  });

+  it("should return a completed job when dropping a local table", async () => {
+    await db.createTable("async-drop", [{ id: 1 }]);
+
+    const job = await db.dropTableAsync("async-drop");
+    expect(job.id).toBeNull();
+    await expect(job.status()).resolves.toBe("finished");
+    await job.wait();
+    await expect(db.tableNames()).resolves.toEqual([]);
+  });
+
  it("should fail if creating table twice, unless overwrite is true", async () => {
    let tbl = await db.createTable("test", [{ id: 1 }, { id: 2 }]);
    await expect(tbl.countRows()).resolves.toBe(2);
@@ -3340,3 +3340,67 @@ describe("LSM merge insert", () => {
    await expect(table.query().useLsm(true).toArray()).rejects.toThrow();
  });
 });
+
+describe("computed columns", () => {
+  let tmpDir: tmp.DirResult;
+  beforeEach(() => {
+    tmpDir = tmp.dirSync({ unsafeCleanup: true });
+  });
+  afterEach(() => tmpDir.removeCallback());
+
+  it("declares a column and fills it on refresh", async () => {
+    const db = await connect(tmpDir.name);
+    const table = await db.createTable("computed", [{ x: 1 }, { x: 2 }]);
+
+    await table.addColumns({
+      computed: [{ name: "doubled", valueSql: "x * 2" }],
+    });
+    let rows = await table.query().toArray();
+    expect(rows.map((r) => r.doubled)).toEqual([null, null]);
+
+    const result = await table.refreshColumn("doubled");
+    expect(result.rowsFilled).toBe(2);
+
+    rows = await table.query().toArray();
+    expect(rows.map((r) => r.doubled).sort()).toEqual([2, 4]);
+  });
+
+  it("returns a job handle from refreshColumnAsync", async () => {
+    const db = await connect(tmpDir.name);
+    const table = await db.createTable("computed_job", [{ x: 1 }, { x: 2 }]);
+
+    await table.addColumns({
+      computed: [{ name: "doubled", valueSql: "x * 2" }],
+    });
+
+    const job = await table.refreshColumnAsync("doubled");
+    expect(job.id).toBeNull();
+    await job.wait();
+    expect(await job.status()).toBe("finished");
+
+    const rows = await table.query().toArray();
+    expect(rows.map((r) => r.doubled).sort()).toEqual([2, 4]);
+
+    // Bad input rejects at the call, not through the job.
+    await expect(table.refreshColumnAsync("x")).rejects.toThrow(
+      "not a computed column",
+    );
+  });
+
+  it("fills rows added since the last refresh", async () => {
+    const db = await connect(tmpDir.name);
+    const table = await db.createTable("computed_append", [{ x: 1 }]);
+
+    await table.addColumns({
+      computed: [{ name: "doubled", valueSql: "x * 2" }],
+    });
+    await table.refreshColumn("doubled");
+    await table.add([{ x: 5 }]);
+
+    const result = await table.refreshColumn("doubled");
+    expect(result.rowsFilled).toBe(1);
+
+    const rows = await table.query().toArray();
+    expect(rows.map((r) => r.doubled).sort()).toEqual([10, 2]);
+  });
+});
@@ -327,6 +327,14 @@ export abstract class Connection {
   */
  abstract dropTable(name: string, namespacePath?: string[]): Promise<void>;

+  /**
+   * Start dropping a table and return its cleanup job.
+   *
+   * The table may become unavailable before its data files are removed. Wait
+   * on the returned job to know when cleanup has finished.
+   */
+  abstract dropTableAsync(name: string, namespacePath?: string[]): Promise<Job>;
+
  /**
   * Drop all tables in the database.
   * @param {string[]} namespacePath The namespace path to drop tables from (defaults to root namespace).
@@ -705,6 +713,10 @@ export class LocalConnection extends Connection {
    return this.inner.dropTable(name, namespacePath ?? []);
  }

+  async dropTableAsync(name: string, namespacePath?: string[]): Promise<Job> {
+    return this.inner.dropTableAsync(name, namespacePath ?? []);
+  }
+
  async dropAllTables(namespacePath?: string[]): Promise<void> {
    return this.inner.dropAllTables(namespacePath ?? []);
  }
@@ -50,6 +50,7 @@ export {
  MergeResult,
  AddResult,
  AddColumnsResult,
+  RefreshColumnResult,
  AlterColumnsResult,
  UpdateFieldMetadataResult,
  DeleteResult,
@@ -33,6 +33,7 @@ import {
  Job,
  Branches as NativeBranches,
  OptimizeStats,
+  RefreshColumnResult,
  TableStatistics,
  Tags,
  UpdateFieldMetadataResult,
@@ -525,18 +526,75 @@ export abstract class Table {
  abstract vectorSearch(vector: IntoVector | MultiVector): VectorQuery;
  /**
   * Add new columns with defined values.
+   *
+   * The `{ computed }` form stores the expression rather than evaluating it
+   * now: the column is committed with no values, and rows get them from
+   * {@link Table#refreshColumn}. Declaring one therefore costs the same on a
+   * large table as on an empty one.
+   *
+   * A refresh does not revisit rows it has already filled, so mutating an
+   * input leaves the value computed at fill time; recomputing means dropping
+   * the column and declaring it again. While a declaration reads a column,
+   * that column cannot be renamed, retyped or dropped.
+   *
+   * On LanceDB Cloud and Enterprise the expression is planned by the
+   * server, and the refresh runs as a server job -- see
+   * {@link Table#refreshColumnAsync}.
   * @param {AddColumnsSql[] | Field | Field[] | Schema} newColumnTransforms Either:
   *   - An array of objects with column names and SQL expressions to calculate values
   *   - A single Arrow Field defining one column with its data type (column will be initialized with null values)
   *   - An array of Arrow Fields defining columns with their data types (columns will be initialized with null values)
   *   - An Arrow Schema defining columns with their data types (columns will be initialized with null values)
+   *   - `{ computed }`, declaring columns defined by a SQL expression whose type and inputs are derived from it
   * @returns {Promise<AddColumnsResult>} A promise that resolves to an object
   * containing the new version number of the table after adding the columns.
+   * @example
+   * ```ts
+   * await table.addColumns({ computed: [{ name: "doubled", valueSql: "x * 2" }] });
+   * const { rowsFilled } = await table.refreshColumn("doubled");
+   * ```
   */
  abstract addColumns(
-    newColumnTransforms: AddColumnsSql[] | Field | Field[] | Schema,
+    newColumnTransforms:
+      | AddColumnsSql[]
+      | Field
+      | Field[]
+      | Schema
+      | { computed: AddColumnsSql[] },
  ): Promise<AddColumnsResult>;

+  /**
+   * Fill the rows of a computed column that hold no value yet.
+   *
+   * Rows appended since the last refresh are filled by the next one; rows
+   * already filled are left as they are, so the call is idempotent and does
+   * not observe a mutated input. Local tables only: a remote refresh runs
+   * as a server job, through {@link Table#refreshColumnAsync}.
+   * @param {string} column The name of the computed column to fill.
+   * @returns {Promise<RefreshColumnResult>} A promise that resolves to the
+   * number of rows filled and the new version number of the table.
+   */
+  abstract refreshColumn(column: string): Promise<RefreshColumnResult>;
+
+  /**
+   * Like {@link Table#refreshColumn}, but returns a handle to the refresh
+   * job instead of blocking until it completes.
+   *
+   * The job may already be complete when returned; callers must not assume
+   * the column is filled until {@link Job.wait} resolves. Invalid input --
+   * an unknown column, or one that is not computed -- rejects here rather
+   * than failing the job. On local tables the job runs in-process; on
+   * LanceDB Cloud and Enterprise it is the server's backfill job.
+   * @param {string} column The name of the computed column to fill.
+   * @example
+   * ```ts
+   * const job = await table.refreshColumnAsync("doubled");
+   * await job.wait();
+   * console.log(await job.status()); // "finished"
+   * ```
+   */
+  abstract refreshColumnAsync(column: string): Promise<Job>;
+
  /**
   * Alter the name or nullability of columns.
   * @param {ColumnAlteration[]} columnAlterations One or more alterations to
@@ -1088,8 +1146,22 @@ export class LocalTable extends Table {
  // TODO: Support BatchUDF

  async addColumns(
-    newColumnTransforms: AddColumnsSql[] | Field | Field[] | Schema,
+    newColumnTransforms:
+      | AddColumnsSql[]
+      | Field
+      | Field[]
+      | Schema
+      | { computed: AddColumnsSql[] },
  ): Promise<AddColumnsResult> {
+    // Columns defined by an expression are declared, not materialized here.
+    if (
+      typeof newColumnTransforms === "object" &&
+      !Array.isArray(newColumnTransforms) &&
+      "computed" in newColumnTransforms
+    ) {
+      return await this.inner.addComputedColumns(newColumnTransforms.computed);
+    }
+
    // Handle single Field -> convert to array of Fields
    if (newColumnTransforms instanceof Field) {
      newColumnTransforms = [newColumnTransforms];
@@ -1124,6 +1196,14 @@ export class LocalTable extends Table {
    throw new Error("Invalid input type for addColumns");
  }

+  async refreshColumn(column: string): Promise<RefreshColumnResult> {
+    return await this.inner.refreshColumn(column);
+  }
+
+  async refreshColumnAsync(column: string): Promise<Job> {
+    return await this.inner.refreshColumnAsync(column);
+  }
+
  async alterColumns(
    columnAlterations: ColumnAlteration[],
  ): Promise<AlterColumnsResult> {
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.2",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.2",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.2",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.2",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.2",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.37.1-beta.1",
+  "version": "0.38.0-beta.2",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.37.1-beta.1",
+	"version": "0.38.0-beta.2",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -1,12 +1,12 @@
 {
  "name": "@lancedb/lancedb",
-  "version": "0.37.1-beta.1",
+  "version": "0.38.0-beta.1",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.37.1-beta.1",
+      "version": "0.38.0-beta.1",
      "cpu": [
        "x64",
        "arm64"
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.37.1-beta.1",
+  "version": "0.38.0-beta.2",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -334,6 +334,22 @@ impl Connection {
            .default_error()
    }

+    /// Start dropping a table and return its cleanup job.
+    #[napi(catch_unwind)]
+    pub async fn drop_table_async(
+        &self,
+        name: String,
+        namespace_path: Option<Vec<String>>,
+    ) -> napi::Result<crate::job::Job> {
+        let ns = namespace_path.unwrap_or_default();
+        let job = self
+            .get_inner()?
+            .drop_table_async(&name, &ns)
+            .await
+            .default_error()?;
+        Ok(crate::job::Job::new(job))
+    }
+
    #[napi(catch_unwind)]
    pub async fn drop_all_tables(&self, namespace_path: Option<Vec<String>>) -> napi::Result<()> {
        let ns = namespace_path.unwrap_or_default();
@@ -347,6 +347,40 @@ impl Table {
        Ok(res.into())
    }

+    #[napi(catch_unwind)]
+    pub async fn add_computed_columns(
+        &self,
+        columns: Vec<AddColumnsSql>,
+    ) -> napi::Result<AddColumnsResult> {
+        let table = self.inner_ref()?;
+        let mut builder = table.add_columns();
+        for column in columns {
+            builder = builder.computed(column.name, column.value_sql);
+        }
+        let res = builder.execute().await.default_error()?;
+        Ok(res.into())
+    }
+
+    #[napi(catch_unwind)]
+    pub async fn refresh_column(&self, column: String) -> napi::Result<RefreshColumnResult> {
+        let res = self
+            .inner_ref()?
+            .refresh_column(column)
+            .await
+            .default_error()?;
+        Ok(res.into())
+    }
+
+    #[napi(catch_unwind)]
+    pub async fn refresh_column_async(&self, column: String) -> napi::Result<crate::job::Job> {
+        let job = self
+            .inner_ref()?
+            .refresh_column_async(column)
+            .await
+            .default_error()?;
+        Ok(crate::job::Job::new(job))
+    }
+
    #[napi(catch_unwind)]
    pub async fn add_columns_with_schema(
        &self,
@@ -1196,6 +1230,21 @@ pub struct AddColumnsResult {
    pub version: i64,
 }

+#[napi(object)]
+pub struct RefreshColumnResult {
+    pub rows_filled: i64,
+    pub version: i64,
+}
+
+impl From<lancedb::table::RefreshColumnResult> for RefreshColumnResult {
+    fn from(value: lancedb::table::RefreshColumnResult) -> Self {
+        Self {
+            rows_filled: value.rows_filled as i64,
+            version: value.version as i64,
+        }
+    }
+}
+
 impl From<lancedb::table::AddColumnsResult> for AddColumnsResult {
    fn from(value: lancedb::table::AddColumnsResult) -> Self {
        Self {
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.37.1-beta.1"
+version = "0.38.0-beta.2"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -198,6 +198,9 @@ class Connection(object):
    async def drop_table(
        self, name: str, namespace_path: Optional[List[str]] = None
    ) -> None: ...
+    async def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job: ...
    async def drop_all_tables(
        self, namespace_path: Optional[List[str]] = None
    ) -> None: ...
@@ -335,6 +338,11 @@ class Table:
    ) -> list[FtsToken]: ...
    async def delete(self, filter: Union[str, PyExpr]) -> DeleteResult: ...
    async def add_columns(self, columns: list[tuple[str, str]]) -> AddColumnsResult: ...
+    async def add_computed_columns(
+        self, columns: list[tuple[str, str]]
+    ) -> AddColumnsResult: ...
+    async def refresh_column(self, column: str) -> RefreshColumnResult: ...
+    async def refresh_column_async(self, column: str) -> Job: ...
    async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...
    async def alter_columns(
        self, columns: list[dict[str, Any]]
@@ -680,6 +688,10 @@ class LsmWriteSpec:
 class AddColumnsResult:
    version: int

+class RefreshColumnResult:
+    rows_filled: int
+    version: int
+
 class AlterColumnsResult:
    version: int

@@ -524,6 +524,12 @@ class DBConnection(EnforceOverrides):
            namespace_path = []
        raise NotImplementedError

+    def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job:
+        """Start dropping a table and return its cleanup job."""
+        raise NotImplementedError
+
    def rename_table(
        self,
        cur_name: str,
@@ -1186,6 +1192,20 @@ class LanceDBConnection(DBConnection):
            )
        )

+    @override
+    def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job:
+        """Start dropping a table and return its cleanup job.
+
+        The table may become unavailable before its data files are removed.
+        Call :meth:`Job.wait` to wait for cleanup to finish.
+        """
+        if namespace_path is None:
+            namespace_path = []
+        job = LOOP.run(self._conn.drop_table_async(name, namespace_path=namespace_path))
+        return Job(job if isinstance(job, AsyncJob) else AsyncJob(job))
+
    @override
    def drop_all_tables(self, namespace_path: Optional[List[str]] = None):
        if namespace_path is None:
@@ -1963,6 +1983,23 @@ class AsyncConnection(object):
            if f"Table '{name}' was not found" not in str(e):
                raise e

+    async def drop_table_async(
+        self,
+        name: str,
+        *,
+        namespace_path: Optional[List[str]] = None,
+    ) -> AsyncJob:
+        """Start dropping a table and return its cleanup job.
+
+        The table may become unavailable before its data files are removed.
+        Await :meth:`AsyncJob.wait` to wait for cleanup to finish.
+        """
+        if namespace_path is None:
+            namespace_path = []
+        return AsyncJob(
+            await self._inner.drop_table_async(name, namespace_path=namespace_path)
+        )
+
    async def drop_all_tables(self, namespace_path: Optional[List[str]] = None):
        """Drop all tables from the database.

@@ -49,6 +49,7 @@ from lancedb._lancedb import (
 )
 from lancedb.background_loop import LOOP
 from lancedb.db import AsyncConnection, DBConnection
+from lancedb.job import AsyncJob, Job
 from lance_namespace import (
    LanceNamespace,
    connect as namespace_connect,
@@ -624,6 +625,18 @@ class LanceNamespaceDBConnection(DBConnection):
            namespace_path = []
        LOOP.run(self._inner.drop_table(name, namespace_path=namespace_path))

+    @override
+    def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job:
+        """Start dropping a table and return its cleanup job."""
+        if namespace_path is None:
+            namespace_path = []
+        job = LOOP.run(
+            self._inner.drop_table_async(name, namespace_path=namespace_path)
+        )
+        return Job(job if isinstance(job, AsyncJob) else AsyncJob(job))
+
    @override
    def rename_table(
        self,
@@ -1134,6 +1147,14 @@ class AsyncLanceNamespaceDBConnection:
            namespace_path = []
        await self._inner.drop_table(name, namespace_path=namespace_path)

+    async def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> AsyncJob:
+        """Start dropping a table and return its cleanup job."""
+        if namespace_path is None:
+            namespace_path = []
+        return await self._inner.drop_table_async(name, namespace_path=namespace_path)
+
    async def rename_table(
        self,
        cur_name: str,
@@ -2235,6 +2235,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
            reranker=self._reranker,
            limit=self._limit,
            with_row_ids=True,
+            offset=self._offset,
        )
        return self._finish_hybrid_results(results)

@@ -2256,6 +2257,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
        reranker,
        limit: int,
        with_row_ids: bool,
+        offset: Optional[int] = None,
    ) -> pa.Table:
        if norm == "rank":
            vector_results = LanceHybridQueryBuilder._rank(vector_results, "_distance")
@@ -2332,7 +2334,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
            score_i = results.column_names.index("_score")
            results = results.set_column(score_i, "_score", original_scores)

-        results = results.slice(length=limit)
+        results = results.slice(offset=offset or 0, length=limit)

        if not with_row_ids:
            results = results.drop(["_rowid"])
@@ -2679,8 +2681,12 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):

        # Apply common configurations
        if self._limit:
-            self._vector_query.limit(self._limit)
-            self._fts_query.limit(self._limit)
+            # The final offset/limit window is sliced out of the combined,
+            # reranked results, so each sub-query must fetch enough rows to
+            # cover the skipped prefix as well as the window itself.
+            sub_query_limit = self._limit + (self._offset or 0)
+            self._vector_query.limit(sub_query_limit)
+            self._fts_query.limit(sub_query_limit)
        if self._columns:
            self._vector_query.select(self._columns)
            self._fts_query.select(self._columns)
@@ -23,7 +23,7 @@ import pyarrow as pa

 from ..common import DATA
 from ..db import DBConnection, LOOP
-from ..job import Job
+from ..job import AsyncJob, Job

 if TYPE_CHECKING:
    from .._lancedb import JobDescription, JobInfo
@@ -663,6 +663,16 @@ class RemoteDBConnection(DBConnection):
            namespace_path = []
        LOOP.run(self._conn.drop_table(name, namespace_path=namespace_path))

+    @override
+    def drop_table_async(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Job:
+        """Start dropping a table and return its cleanup job."""
+        if namespace_path is None:
+            namespace_path = []
+        job = LOOP.run(self._conn.drop_table_async(name, namespace_path=namespace_path))
+        return Job(job if isinstance(job, AsyncJob) else AsyncJob(job))
+
    @override
    def rename_table(
        self,
@@ -958,8 +958,19 @@ class RemoteTable(Table):
    def count_rows(self, filter: Optional[str] = None) -> int:
        return LOOP.run(self._table.count_rows(filter))

-    def add_columns(self, transforms: Dict[str, str]) -> AddColumnsResult:
-        return LOOP.run(self._table.add_columns(transforms))
+    def add_columns(
+        self,
+        transforms: Dict[str, str] | None = None,
+        *,
+        computed: Dict[str, str] | None = None,
+    ) -> AddColumnsResult:
+        return LOOP.run(self._table.add_columns(transforms, computed=computed))
+
+    def refresh_column(self, column: str):
+        return LOOP.run(self._table.refresh_column(column))
+
+    def refresh_column_async(self, column: str) -> Job:
+        return Job(LOOP.run(self._table.refresh_column_async(column)))

    def alter_columns(
        self, *alterations: Iterable[Dict[str, str]]
@@ -979,17 +990,39 @@ class RemoteTable(Table):
        return LOOP.run(self._table.set_unenforced_primary_key(columns))

    def set_lsm_write_spec(self, spec: "LsmWriteSpec") -> None:
-        """Not supported on LanceDB Cloud."""
+        """Install an LsmWriteSpec."""
        return LOOP.run(self._table.set_lsm_write_spec(spec))

    def unset_lsm_write_spec(self) -> None:
-        """Not supported on LanceDB Cloud."""
+        """Remove the LsmWriteSpec."""
        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 checkpoint_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
+        return LOOP.run(self._table.checkpoint_lsm())
+
+    def flush_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.flush_lsm`][lancedb.AsyncTable.flush_lsm]."""
+        return LOOP.run(self._table.flush_lsm())
+
+    def compact_lsm(self) -> None:
+        """Synchronous version of
+        [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
+        return LOOP.run(self._table.compact_lsm())
+
+    def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
+        """Synchronous version of
+        [`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
+        return LOOP.run(
+            self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
+        )
+
    def close_lsm_writers(self) -> None:
        """No-op on LanceDB Cloud (no local shard writers)."""
        return LOOP.run(self._table.close_lsm_writers())
@@ -176,6 +176,7 @@ if TYPE_CHECKING:
        CompactionStats,
        Tag,
        AddColumnsResult,
+        RefreshColumnResult,
        AddResult,
        AlterColumnsResult,
        UpdateFieldMetadataResult,
@@ -1916,7 +1917,14 @@ class Table(ABC):

    @abstractmethod
    def add_columns(
-        self, transforms: Dict[str, str] | pa.Field | List[pa.Field] | pa.Schema
+        self,
+        transforms: Dict[str, str]
+        | pa.Field
+        | List[pa.Field]
+        | pa.Schema
+        | None = None,
+        *,
+        computed: Dict[str, str] | None = None,
    ):
        """
        Add new columns with defined values.
@@ -1930,11 +1938,95 @@ class Table(ABC):
            Alternatively, a pyarrow Field or Schema can be provided to add
            new columns with the specified data types. The new columns will
            be initialized with null values.
+        computed: Dict[str, str], optional
+            A map of column name to a SQL expression defining the column. The
+            column's type and inputs are derived from the expression, so no
+            data type is supplied.
+
+            Unlike ``transforms``, the expression is stored rather than
+            evaluated now: the column is committed with no values, and rows get
+            them from [`refresh_column`][lancedb.table.Table.refresh_column].
+            Declaring one therefore costs the same on a large table as on an
+            empty one.
+
+            A refresh does not revisit rows it has already filled, so mutating
+            an input leaves the value computed at fill time; recomputing means
+            dropping the column and declaring it again. While a declaration
+            reads a column, that column cannot be renamed, retyped or dropped.
+
+            On LanceDB Cloud and Enterprise the expression is planned by the
+            server, and the refresh runs as a server job -- see
+            [`refresh_column_async`][lancedb.table.Table.refresh_column_async].
+            Cannot be combined with ``transforms``.

        Returns
        -------
        AddColumnsResult
            version: the new version number of the table after adding columns.
+
+        Examples
+        --------
+        >>> import lancedb
+        >>> db = lancedb.connect("./.lancedb")
+        >>> table = db.create_table("computed_demo", [{"x": 1}, {"x": 2}])
+        >>> table.add_columns(computed={"doubled": "x * 2"})
+        AddColumnsResult(version=2)
+        >>> table.refresh_column("doubled")
+        RefreshColumnResult(rows_filled=2, version=3)
+        >>> table.to_arrow().sort_by("x").to_pandas()
+           x  doubled
+        0  1        2
+        1  2        4
+        """
+
+    @abstractmethod
+    def refresh_column(self, column: str) -> "RefreshColumnResult":
+        """
+        Fill the rows of a computed column that hold no value yet.
+
+        Declared with ``add_columns(computed=...)``, a column starts empty and
+        gets its values here. Rows appended since the last refresh are filled
+        by the next one; rows already filled are left as they are, so the call
+        is idempotent and does not observe a mutated input.
+
+        Local tables only: a remote refresh runs as a server job, through
+        [`refresh_column_async`][lancedb.table.Table.refresh_column_async].
+
+        Parameters
+        ----------
+        column: str
+            The name of the computed column to fill.
+
+        Returns
+        -------
+        RefreshColumnResult
+            rows_filled: the number of rows given a value.
+            version: the new version number of the table.
+        """
+
+    @abstractmethod
+    def refresh_column_async(self, column: str) -> Job:
+        """
+        Like :meth:`refresh_column`, but returns a handle to the refresh job
+        instead of blocking until it completes.
+
+        The job may already be complete when returned; callers must not assume
+        the column is filled until :meth:`Job.wait` returns. Invalid input --
+        an unknown column, or one that is not computed -- raises here rather
+        than failing the job. On local tables the job runs in-process; on
+        LanceDB Cloud and Enterprise it is the server's backfill job.
+
+        Examples
+        --------
+        >>> import lancedb
+        >>> db = lancedb.connect("./.lancedb")
+        >>> table = db.create_table("computed_job_demo", [{"x": 1}, {"x": 2}])
+        >>> table.add_columns(computed={"doubled": "x * 2"})
+        AddColumnsResult(version=2)
+        >>> job = table.refresh_column_async("doubled")
+        >>> job.wait()
+        >>> job.status()
+        'finished'
        """

    @abstractmethod
@@ -3939,9 +4031,28 @@ class LanceTable(Table):
        return LOOP.run(self._table.index_stats(index_name))

    def add_columns(
-        self, transforms: Dict[str, str] | pa.field | List[pa.field] | pa.Schema
+        self,
+        transforms: Dict[str, str]
+        | pa.field
+        | List[pa.field]
+        | pa.Schema
+        | None = None,
+        *,
+        computed: Dict[str, str] | None = None,
    ) -> AddColumnsResult:
-        return LOOP.run(self._table.add_columns(transforms))
+        return LOOP.run(self._table.add_columns(transforms, computed=computed))
+
+    def refresh_column(self, column: str) -> "RefreshColumnResult":
+        """Fill a computed column's unfilled rows. See
+        [`AsyncTable.refresh_column`][lancedb.AsyncTable.refresh_column]."""
+        return LOOP.run(self._table.refresh_column(column))
+
+    def refresh_column_async(self, column: str) -> Job:
+        """Fill a computed column's unfilled rows, returning a handle to the
+        refresh job. See
+        [`Table.refresh_column_async`][lancedb.table.Table.refresh_column_async].
+        """
+        return Job(LOOP.run(self._table.refresh_column_async(column)))

    def alter_columns(
        self, *alterations: Iterable[Dict[str, str]]
@@ -4735,7 +4846,7 @@ class AsyncTable:
        ``asyncio.wait_for`` for a wall-clock bound; abandoning it partway
        costs nothing.
        """
-        return await self._inner.checkpoint_lsm()
+        await self._inner.checkpoint_lsm()

    async def flush_lsm(self) -> None:
        """Seal every bucket's active memtable into L0.
@@ -4744,7 +4855,7 @@ class AsyncTable:
        `compact_lsm`. On a node that has not claimed this table, this claims
        it and replays its WAL log first.
        """
-        return await self._inner.flush_lsm()
+        await self._inner.flush_lsm()

    async def compact_lsm(self) -> None:
        """Trigger a background L0 to base compaction pass per bucket.
@@ -4753,7 +4864,7 @@ class AsyncTable:
        ``get_lsm_stats`` for progress, or use ``checkpoint_lsm`` to loop
        until the current L0 has reached base.
        """
-        return await self._inner.compact_lsm()
+        await self._inner.compact_lsm()

    async def get_lsm_stats(
        self, *, include_generation_rows: bool = False
@@ -5856,7 +5967,14 @@ class AsyncTable:
        return await self._inner.update(updates_sql, where)

    async def add_columns(
-        self, transforms: dict[str, str] | pa.field | List[pa.field] | pa.Schema
+        self,
+        transforms: dict[str, str]
+        | pa.field
+        | List[pa.field]
+        | pa.Schema
+        | None = None,
+        *,
+        computed: dict[str, str] | None = None,
    ) -> AddColumnsResult:
        """
        Add new columns with defined values.
@@ -5869,6 +5987,22 @@ class AsyncTable:
            each row in the table, and can reference existing columns.
            Alternatively, you can pass a pyarrow field or schema to add
            new columns with NULLs.
+        computed: Dict[str, str], optional
+            A map of column name to a SQL expression defining the column. The
+            column's type and inputs are derived from the expression.
+
+            Unlike ``transforms``, the expression is stored rather than
+            evaluated now: the column is committed with no values, and rows get
+            them from
+            [`refresh_column`][lancedb.table.AsyncTable.refresh_column].
+
+            A refresh does not revisit rows it has already filled, so mutating
+            an input leaves the value computed at fill time. While a
+            declaration reads a column, that column cannot be renamed, retyped
+            or dropped.
+
+            On LanceDB Cloud and Enterprise the expression is planned by
+            the server. Cannot be combined with ``transforms``.

        Returns
        -------
@@ -5882,11 +6016,71 @@ class AsyncTable:
            {isinstance(f, pa.Field) for f in transforms}
        ):
            transforms = pa.schema(transforms)
+        if computed:
+            if transforms:
+                raise ValueError(
+                    "add_columns cannot take both transforms and computed columns"
+                )
+            return await self._inner.add_computed_columns(list(computed.items()))
+        if transforms is None:
+            raise ValueError("add_columns requires transforms or computed columns")
        if isinstance(transforms, pa.Schema):
            return await self._inner.add_columns_with_schema(transforms)
        else:
            return await self._inner.add_columns(list(transforms.items()))

+    async def refresh_column(self, column: str) -> RefreshColumnResult:
+        """
+        Fill the rows of a computed column that hold no value yet.
+
+        Declared with ``add_columns(computed=...)``, a column starts empty and
+        gets its values here. Rows appended since the last refresh are filled
+        by the next one; rows already filled are left as they are, so the call
+        is idempotent and does not observe a mutated input.
+
+        Local tables only: a remote refresh runs as a server job, through
+        [`refresh_column_async`][lancedb.table.Table.refresh_column_async].
+
+        Parameters
+        ----------
+        column: str
+            The name of the computed column to fill.
+
+        Returns
+        -------
+        RefreshColumnResult
+            The number of rows filled and the new version of the table.
+        """
+        return await self._inner.refresh_column(column)
+
+    async def refresh_column_async(self, column: str) -> AsyncJob:
+        """
+        Like :meth:`refresh_column`, but returns a handle to the refresh job
+        instead of blocking until it completes.
+
+        The job may already be complete when returned; callers must not assume
+        the column is filled until :meth:`AsyncJob.wait` resolves. Invalid
+        input -- an unknown column, or one that is not computed -- raises here
+        rather than failing the job. On local tables the job runs
+        in-process; on LanceDB Cloud and Enterprise it is the server's
+        backfill job.
+
+        Examples
+        --------
+        >>> import asyncio
+        >>> import lancedb
+        >>> async def refresh_in_background():
+        ...     db = await lancedb.connect_async("./.lancedb")
+        ...     table = await db.create_table("computed_job_async_demo", [{"x": 1}])
+        ...     await table.add_columns(computed={"doubled": "x * 2"})
+        ...     job = await table.refresh_column_async("doubled")
+        ...     await job.wait()
+        ...     return await job.status()
+        >>> asyncio.run(refresh_in_background())
+        'finished'
+        """
+        return AsyncJob(await self._inner.refresh_column_async(column))
+
    async def alter_columns(
        self, *alterations: Iterable[dict[str, Any]]
    ) -> AlterColumnsResult:
@@ -755,8 +755,7 @@ def test_delete_table(tmp_db: lancedb.DBConnection):
    assert tmp_db.table_names() == []


-@pytest.mark.asyncio
-async def test_delete_table_async(tmp_db: lancedb.DBConnection):
+def test_drop_table_async(tmp_db: lancedb.DBConnection):
    data = pd.DataFrame(
        {
            "vector": [[3.1, 4.1], [5.9, 26.5]],
@@ -772,7 +771,10 @@ async def test_delete_table_async(tmp_db: lancedb.DBConnection):

    assert tmp_db.table_names() == ["test"]

-    tmp_db.drop_table("test")
+    job = tmp_db.drop_table_async("test")
+    assert job.id is None
+    assert job.status() == "finished"
+    job.wait()
    assert tmp_db.table_names() == []

    tmp_db.create_table("test", data=data)
@@ -781,6 +783,17 @@ async def test_delete_table_async(tmp_db: lancedb.DBConnection):
    tmp_db.drop_table("does_not_exist", ignore_missing=True)


+@pytest.mark.asyncio
+async def test_drop_table_async_connection(tmp_db_async: lancedb.AsyncConnection):
+    await tmp_db_async.create_table("test", data=pa.table({"id": [1, 2]}))
+
+    job = await tmp_db_async.drop_table_async("test")
+    assert job.id is None
+    assert await job.status() == "finished"
+    await job.wait()
+    assert await tmp_db_async.table_names() == []
+
+
 def test_drop_database(tmp_db: lancedb.DBConnection):
    data = pd.DataFrame(
        {
@@ -632,3 +632,101 @@ class TestExprBytesIntegration:
            .to_arrow()
        )
        assert result.num_rows == 2
+
+
+# ── datetime / timezone integration for lit() (issue #3262) ──────────────────
+
+
+class TestExprDatetimeTimezoneIntegration:
+    """Integration coverage for lit(datetime) against table timestamp columns.
+
+    PyArrow stores naive timestamps as UTC wall-clock microseconds. Python's
+    datetime.timestamp() treats naive values as *local* time, which used to
+    shift lit(naive) by the host UTC offset and break equality filters on
+    non-UTC machines. These cases lock the expected semantics.
+    """
+
+    def test_both_naive_match(self, tmp_path):
+        """Table naive + lit naive with the same wall clock must match."""
+        db = lancedb.connect(str(tmp_path / "naive"))
+        ts = datetime(2024, 7, 1, 10, 0, 0)
+        table = db.create_table(
+            "t", [{"id": 1, "ts": ts}, {"id": 2, "ts": datetime(2024, 7, 2, 10, 0, 0)}]
+        )
+        result = table.search().where(col("ts") == lit(ts)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_both_same_timezone_match(self, tmp_path):
+        """Table UTC + lit UTC for the same instant must match."""
+        db = lancedb.connect(str(tmp_path / "utc"))
+        ts = datetime(2024, 7, 1, 10, 0, 0, tzinfo=timezone.utc)
+        table = db.create_table(
+            "t",
+            pa.table(
+                {
+                    "id": [1, 2],
+                    "ts": pa.array(
+                        [ts, datetime(2024, 7, 2, 10, 0, 0, tzinfo=timezone.utc)],
+                        type=pa.timestamp("us", tz="UTC"),
+                    ),
+                }
+            ),
+        )
+        result = table.search().where(col("ts") == lit(ts)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_different_timezones_same_instant(self, tmp_path):
+        """UTC table row equals lit of the same instant in a different zone."""
+        db = lancedb.connect(str(tmp_path / "diff_tz"))
+        ts_utc = datetime(2024, 7, 1, 10, 0, 0, tzinfo=timezone.utc)
+        # Same instant as 06:00 in UTC-4
+        ts_est = datetime(2024, 7, 1, 6, 0, 0, tzinfo=timezone(timedelta(hours=-4)))
+        table = db.create_table(
+            "t",
+            pa.table(
+                {
+                    "id": [1],
+                    "ts": pa.array([ts_utc], type=pa.timestamp("us", tz="UTC")),
+                }
+            ),
+        )
+        result = table.search().where(col("ts") == lit(ts_est)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_table_tz_literal_naive(self, tmp_path):
+        """UTC table + naive lit uses wall-clock equality (10:00 == 10:00 UTC)."""
+        db = lancedb.connect(str(tmp_path / "tz_naive"))
+        ts_utc = datetime(2024, 7, 1, 10, 0, 0, tzinfo=timezone.utc)
+        ts_naive = datetime(2024, 7, 1, 10, 0, 0)
+        table = db.create_table(
+            "t",
+            pa.table(
+                {
+                    "id": [1],
+                    "ts": pa.array([ts_utc], type=pa.timestamp("us", tz="UTC")),
+                }
+            ),
+        )
+        result = table.search().where(col("ts") == lit(ts_naive)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_table_naive_literal_aware(self, tmp_path):
+        """Naive table + UTC lit with the same wall clock must match."""
+        db = lancedb.connect(str(tmp_path / "naive_aware"))
+        ts_naive = datetime(2024, 7, 1, 10, 0, 0)
+        ts_utc = datetime(2024, 7, 1, 10, 0, 0, tzinfo=timezone.utc)
+        table = db.create_table("t", [{"id": 1, "ts": ts_naive}])
+        result = table.search().where(col("ts") == lit(ts_utc)).to_list()
+        assert len(result) == 1
+        assert result[0]["id"] == 1
+
+    def test_naive_lit_sql_is_wall_clock_not_local_shifted(self):
+        """Regression: naive lit must not apply the host local UTC offset."""
+        ts = datetime(2024, 7, 1, 10, 0, 0)
+        sql = lit(ts).to_sql()
+        # Must encode 10:00 wall clock, not 10:00+local_offset.
+        assert "2024-07-01 10:00:00" in sql
@@ -203,6 +203,31 @@ async def test_async_hybrid_query_default_limit(table: AsyncTable):
    assert texts.count("a") == 1


+def test_hybrid_query_offset(sync_table: Table):
+    # The offset window of a hybrid query must be a suffix of the same query
+    # run without an offset -- it must not be silently ignored.
+    full = (
+        sync_table.search(query_type="hybrid")
+        .vector([0.0, 0.4])
+        .text("dog")
+        .limit(4)
+        .with_row_id(True)
+        .to_arrow()
+    )
+    assert len(full) == 4
+
+    offset_result = (
+        sync_table.search(query_type="hybrid")
+        .vector([0.0, 0.4])
+        .text("dog")
+        .offset(2)
+        .limit(2)
+        .with_row_id(True)
+        .to_arrow()
+    )
+    assert offset_result["_rowid"].to_pylist() == full["_rowid"].to_pylist()[2:]
+
+
 def test_hybrid_query_minimum_nprobes_zero_raises(sync_table: Table):
    # minimum_nprobes(0) must raise the same validation error a plain vector
    # query raises, not silently no-op because 0 is falsy.
@@ -1133,6 +1133,131 @@ def test_stats():
        assert res == stats


+@contextlib.contextmanager
+def lsm_test_table(lsm_handler):
+    """A remote table whose LSM routes are served by ``lsm_handler``.
+
+    ``lsm_handler(request, route)`` is called for ``/v1/table/test/<route>/``
+    where route is one of flush_lsm, compact_lsm, get_lsm_stats, and is
+    responsible for writing the response.
+    """
+    routes = ("flush_lsm", "compact_lsm", "get_lsm_stats")
+
+    def handler(request):
+        match = re.fullmatch(r"/v1/table/test/(\w+)/", request.path)
+        route = match.group(1) if match else None
+        if route in routes:
+            lsm_handler(request, route)
+        elif route == "describe":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b'{"version": 1, "schema": {"fields": []}}')
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        yield db.open_table("test")
+
+
+def read_json_body(request):
+    content_len = int(request.headers.get("Content-Length"))
+    return json.loads(request.rfile.read(content_len))
+
+
+def send_json(request, payload, status=200):
+    request.send_response(status)
+    request.send_header("Content-Type", "application/json")
+    request.end_headers()
+    request.wfile.write(json.dumps(payload).encode())
+
+
+def test_get_lsm_stats_sync():
+    """The sync wrapper round-trips the server payload into a dict."""
+    bucket = {
+        "shard_id": "b0",
+        "status": "Active",
+        "writer_epoch": 3,
+        "manifest_version": 12,
+        "current_generation": 6,
+        "replay_after_wal_entry_position": 40,
+        "wal_entry_position_last_seen": 42,
+        "generations": [{"generation": 5, "bytes": 1024, "rows": 7}],
+        "compacting": False,
+        "memtables": [
+            {
+                "generation": 6,
+                "rows": 2,
+                "bytes": 64,
+                "batches": 1,
+                "indexes": ["vec_idx"],
+            }
+        ],
+    }
+    seen_bodies = []
+
+    def lsm_handler(request, route):
+        assert route == "get_lsm_stats"
+        seen_bodies.append(read_json_body(request))
+        send_json(request, {"lsm_stats": {"buckets": [bucket]}})
+
+    with lsm_test_table(lsm_handler) as table:
+        assert table.get_lsm_stats() == {"buckets": [bucket]}
+        # Off by default, and forwarded when asked for.
+        assert seen_bodies == [{"include_generation_rows": False}]
+        table.get_lsm_stats(include_generation_rows=True)
+        assert seen_bodies[-1] == {"include_generation_rows": True}
+
+
+def test_get_lsm_stats_sync_returns_none_when_lsm_disabled():
+    """A null envelope means the LSM write path is not enabled, not an error."""
+
+    def lsm_handler(request, route):
+        send_json(request, {"lsm_stats": None})
+
+    with lsm_test_table(lsm_handler) as table:
+        assert table.get_lsm_stats() is None
+
+
+def test_flush_and_compact_lsm_sync():
+    """Both are one-shot POSTs answered 202 with no body."""
+    called = []
+
+    def lsm_handler(request, route):
+        called.append(route)
+        request.send_response(202)
+        request.end_headers()
+
+    with lsm_test_table(lsm_handler) as table:
+        assert table.flush_lsm() is None
+        assert table.compact_lsm() is None
+        assert called == ["flush_lsm", "compact_lsm"]
+
+
+def test_checkpoint_lsm_sync():
+    """Seal, read the watermark, and return once L0 holds nothing.
+
+    The convergence loop itself is covered in Rust; this pins the sync
+    binding to the endpoints it drives.
+    """
+    called = []
+
+    def lsm_handler(request, route):
+        called.append(route)
+        if route == "get_lsm_stats":
+            # An empty L0 yields no target watermark, so the loop is done
+            # after the seal without ever polling compaction.
+            send_json(request, {"lsm_stats": {"buckets": []}})
+        else:
+            request.send_response(202)
+            request.end_headers()
+
+    with lsm_test_table(lsm_handler) as table:
+        assert table.checkpoint_lsm() is None
+        assert called == ["flush_lsm", "get_lsm_stats"]
+
+
@contextlib.contextmanager
 def query_test_table(query_handler, *, server_version=Version("0.1.0")):
    def handler(request):
@@ -3854,3 +3854,65 @@ async def test_async_search_runs_embedding_on_dedicated_executor(
    assert all(name.startswith("lancedb-embedding") for name in captured_threads), (
        f"embedding ran off the dedicated executor: {captured_threads}"
    )
+
+
+def test_computed_column_declare_and_refresh(tmp_path):
+    db = lancedb.connect(tmp_path)
+    table = db.create_table("computed", [{"x": 1}, {"x": 2}])
+
+    table.add_columns(computed={"doubled": "x * 2"})
+    assert table.to_arrow()["doubled"].to_pylist() == [None, None]
+
+    result = table.refresh_column("doubled")
+    assert result.rows_filled == 2
+    assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4]
+
+    table.add([{"x": 5}])
+    assert table.refresh_column("doubled").rows_filled == 1
+    assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4, 10]
+
+
+def test_computed_column_rejects_transforms_and_computed_together(tmp_path):
+    db = lancedb.connect(tmp_path)
+    table = db.create_table("computed_mixed", [{"x": 1}])
+    with pytest.raises(ValueError):
+        table.add_columns({"a": "x + 1"}, computed={"b": "x * 2"})
+
+
+@pytest.mark.asyncio
+async def test_computed_column_async(tmp_path):
+    db = await lancedb.connect_async(tmp_path)
+    table = await db.create_table("computed_async", [{"x": 3}])
+
+    await table.add_columns(computed={"tripled": "x * 3"})
+    await table.refresh_column("tripled")
+
+    assert (await table.to_arrow())["tripled"].to_pylist() == [9]
+
+
+def test_refresh_column_async_returns_job(tmp_path):
+    db = lancedb.connect(tmp_path)
+    table = db.create_table("computed_job", [{"x": 1}, {"x": 2}])
+    table.add_columns(computed={"doubled": "x * 2"})
+
+    job = table.refresh_column_async("doubled")
+    assert job.id is None  # in-process jobs have no server id
+    job.wait()
+    assert job.status() == "finished"
+    assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4]
+
+    # Bad input raises at the call, not through the job.
+    with pytest.raises(Exception, match="not a computed column"):
+        table.refresh_column_async("x")
+
+
+@pytest.mark.asyncio
+async def test_refresh_column_async_job_async_table(tmp_path):
+    db = await lancedb.connect_async(tmp_path)
+    table = await db.create_table("computed_job_async", [{"x": 3}])
+    await table.add_columns(computed={"tripled": "x * 3"})
+
+    job = await table.refresh_column_async("tripled")
+    await job.wait()
+    assert await job.status() == "finished"
+    assert (await table.to_arrow())["tripled"].to_pylist() == [9]
@@ -346,6 +346,23 @@ impl Connection {
        })
    }

+    #[pyo3(signature = (name, namespace_path=None))]
+    pub fn drop_table_async(
+        self_: PyRef<'_, Self>,
+        name: String,
+        namespace_path: Option<Vec<String>>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        let ns_path = namespace_path.unwrap_or_default();
+        future_into_py(self_.py(), async move {
+            inner
+                .drop_table_async(name, &ns_path)
+                .await
+                .infer_error()
+                .map(crate::job::Job::new)
+        })
+    }
+
    #[pyo3(signature = (namespace_path=None,))]
    pub fn drop_all_tables(
        self_: PyRef<'_, Self>,
@@ -191,8 +191,27 @@ pub fn expr_lit(value: Bound<'_, PyAny>) -> PyResult<PyExpr> {
    }

    // datetime.datetime is a subclass of datetime.date, so it must be checked first.
+    //
+    // Python's datetime.timestamp() treats *naive* datetimes as local wall time.
+    // PyArrow (and therefore Lance table storage) encodes naive timestamps as
+    // UTC wall-clock microseconds. Using .timestamp() for naive values therefore
+    // shifts the literal by the local UTC offset on non-UTC machines, so
+    // `col("ts") == lit(naive_dt)` fails against a table that holds the same
+    // naive value. Fix: treat naive datetimes as UTC wall clock (match Arrow);
+    // keep aware datetimes on the real .timestamp() path (correct epoch).
    if let Ok(dt) = value.cast::<PyDateTime>() {
-        let ts: f64 = dt.call_method0("timestamp")?.extract()?;
+        let ts: f64 = if dt.getattr("tzinfo")?.is_none() {
+            // Force UTC interpretation of the naive wall clock.
+            let utc = pyo3::types::PyModule::import(value.py(), "datetime")?
+                .getattr("timezone")?
+                .getattr("utc")?;
+            let kwargs = pyo3::types::PyDict::new(value.py());
+            kwargs.set_item("tzinfo", utc)?;
+            let aware = dt.call_method("replace", (), Some(&kwargs))?;
+            aware.call_method0("timestamp")?.extract()?
+        } else {
+            dt.call_method0("timestamp")?.extract()?
+        };
        let micros = (ts * 1_000_000.0).round() as i64;
        return Ok(PyExpr(df_lit(ScalarValue::TimestampMicrosecond(
            Some(micros),
@@ -16,7 +16,8 @@ use query::{FTSQuery, HybridQuery, Query, VectorQuery};
 use session::Session;
 use table::{
    AddColumnsResult, AddResult, AlterColumnsResult, DeleteResult, DropColumnsResult, FtsToken,
-    LsmWriteSpec, MergeResult, PyBlobFile, Table, UpdateFieldMetadataResult, UpdateResult,
+    LsmWriteSpec, MergeResult, PyBlobFile, RefreshColumnResult, Table, UpdateFieldMetadataResult,
+    UpdateResult,
 };

 pub mod arrow;
@@ -57,6 +58,7 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_class::<VectorQuery>()?;
    m.add_class::<RecordBatchStream>()?;
    m.add_class::<AddColumnsResult>()?;
+    m.add_class::<RefreshColumnResult>()?;
    m.add_class::<AlterColumnsResult>()?;
    m.add_class::<UpdateFieldMetadataResult>()?;
    m.add_class::<AddResult>()?;
@@ -415,6 +415,32 @@ pub struct AddColumnsResult {
    pub version: u64,
 }

+#[pyclass(get_all, from_py_object)]
+#[derive(Clone, Debug)]
+pub struct RefreshColumnResult {
+    pub rows_filled: u64,
+    pub version: u64,
+}
+
+#[pymethods]
+impl RefreshColumnResult {
+    pub fn __repr__(&self) -> String {
+        format!(
+            "RefreshColumnResult(rows_filled={}, version={})",
+            self.rows_filled, self.version
+        )
+    }
+}
+
+impl From<lancedb::table::RefreshColumnResult> for RefreshColumnResult {
+    fn from(result: lancedb::table::RefreshColumnResult) -> Self {
+        Self {
+            rows_filled: result.rows_filled,
+            version: result.version,
+        }
+    }
+}
+
 #[pymethods]
 impl AddColumnsResult {
    pub fn __repr__(&self) -> String {
@@ -1510,6 +1536,40 @@ impl Table {
        })
    }

+    pub fn add_computed_columns(
+        self_: PyRef<'_, Self>,
+        columns: Vec<(String, String)>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let mut builder = inner.add_columns();
+            for (name, expression) in columns {
+                builder = builder.computed(name, expression);
+            }
+            let result = builder.execute().await.infer_error()?;
+            Ok(AddColumnsResult::from(result))
+        })
+    }
+
+    pub fn refresh_column(self_: PyRef<'_, Self>, column: String) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let result = inner.refresh_column(column).await.infer_error()?;
+            Ok(RefreshColumnResult::from(result))
+        })
+    }
+
+    pub fn refresh_column_async(
+        self_: PyRef<'_, Self>,
+        column: String,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let job = inner.refresh_column_async(column).await.infer_error()?;
+            Ok(crate::job::Job::new(job))
+        })
+    }
+
    pub fn add_columns_with_schema(
        self_: PyRef<'_, Self>,
        schema: PyArrowType<Schema>,
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.37.1-beta.1"
+version = "0.38.0-beta.2"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -565,6 +565,21 @@ impl Connection {
            .await
    }

+    /// Start dropping a table and return a handle to the cleanup job.
+    ///
+    /// The table may become unavailable before its physical data is removed.
+    /// Call [`crate::job::Job::wait`] to wait for cleanup to finish. Local
+    /// backends may complete the drop before returning the handle.
+    pub async fn drop_table_async(
+        &self,
+        name: impl AsRef<str>,
+        namespace_path: &[String],
+    ) -> Result<crate::job::Job> {
+        self.internal
+            .drop_table_async(name.as_ref(), namespace_path)
+            .await
+    }
+
    /// Drop the database
    ///
    /// This is the same as dropping all of the tables
@@ -323,6 +323,18 @@ pub trait Database:
    ) -> Result<()>;
    /// Drop a table in the database
    async fn drop_table(&self, name: &str, namespace_path: &[String]) -> Result<()>;
+    /// Start dropping a table and return a handle to the cleanup job.
+    ///
+    /// Backends without asynchronous cleanup complete the drop before
+    /// returning an already-finished job.
+    async fn drop_table_async(
+        &self,
+        name: &str,
+        namespace_path: &[String],
+    ) -> Result<crate::job::Job> {
+        self.drop_table(name, namespace_path).await?;
+        Ok(crate::job::Job::new_done())
+    }
    /// Drop all tables in the database
    async fn drop_all_tables(&self, namespace_path: &[String]) -> Result<()>;
    fn as_any(&self) -> &dyn std::any::Any;
@@ -1032,6 +1032,7 @@ impl Database for ListingDatabase {
        };

        Ok(ListTablesResponse {
+            context: None,
            tables: f,
            page_token: next_page_token,
        })
@@ -71,6 +71,14 @@ pub enum Error {
    IndexNotFound { name: String },
    #[snafu(display("Embedding function '{name}' was not found. : {reason}"))]
    EmbeddingFunctionNotFound { name: String, reason: String },
+    #[snafu(display("Column '{name}' was not found"))]
+    ColumnNotFound { name: String },
+    #[snafu(display("Column '{name}' already exists"))]
+    ColumnAlreadyExists { name: String },
+    #[snafu(display("Column '{name}' is not a computed column"))]
+    NotAComputedColumn { name: String },
+    #[snafu(display("Invalid expression for column '{column}': {message}"))]
+    InvalidExpression { column: String, message: String },

    #[snafu(display("Table '{name}' already exists"))]
    TableAlreadyExists { name: String },
@@ -141,7 +141,7 @@ impl SpawnedJob {
                Ok(Err(err)) => Outcome::Failed(Arc::new(err)),
                Err(err) if err.is_cancelled() => Outcome::Cancelled,
                Err(err) => Outcome::Failed(Arc::new(Error::Runtime {
-                    message: format!("index job task failed: {err}"),
+                    message: format!("job task failed: {err}"),
                })),
            };
            let _ = tx.send(Some(outcome));
@@ -19,6 +19,15 @@ const ARROW_FILE_CONTENT_TYPE: &str = "application/vnd.apache.arrow.file";
 #[cfg(test)]
 const JSON_CONTENT_TYPE: &str = "application/json";

+fn extract_job_id(body: &str) -> Option<String> {
+    serde_json::from_str::<serde_json::Value>(body)
+        .ok()?
+        .get("job_id")?
+        .as_str()
+        .filter(|job_id| !job_id.is_empty())
+        .map(str::to_string)
+}
+
 pub use client::{ClientConfig, HeaderProvider, RetryConfig, TimeoutConfig, TlsConfig};
 pub use db::{RemoteDatabaseOptions, RemoteDatabaseOptionsBuilder};
 pub use oauth::{OAuthConfig, OAuthFlow, OAuthHeaderProvider};
@@ -9,6 +9,7 @@ use http::StatusCode;
 use lance_io::object_store::StorageOptions;
 use lance_namespace_impls::{DynamicContextProvider, OperationInfo};
 use moka::future::Cache;
+use reqwest::Response;
 use reqwest::header::CONTENT_TYPE;

 use lance_namespace::models::{
@@ -23,15 +24,17 @@ use crate::database::{
    JobDescription, JobInfo, OpenTableRequest, ReadConsistency, TableNamesRequest,
 };
 use crate::error::Result;
+use crate::job::Job;
+use crate::remote::job::RemoteJob;
 use crate::remote::util::stream_as_body;
 use crate::table::BaseTable;

-use super::ARROW_STREAM_CONTENT_TYPE;
 use super::client::{
    ClientConfig, HeaderProvider, HttpSend, RequestResultExt, RestfulLanceDbClient, Sender,
 };
 use super::table::RemoteTable;
 use super::util::parse_server_version;
+use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id};

 // Request structure for the remote clone table API
 #[derive(serde::Serialize)]
@@ -326,6 +329,22 @@ impl RemoteDatabase {
    }
 }

+impl<S: HttpSend> RemoteDatabase<S> {
+    async fn submit_drop_table(
+        &self,
+        name: &str,
+        namespace_path: &[String],
+    ) -> Result<(String, Response)> {
+        let identifier = build_table_identifier(name, namespace_path, &self.client.id_delimiter);
+        let cache_key = build_cache_key(name, namespace_path);
+        let req = self.client.post(&format!("/v1/table/{}/drop/", identifier));
+        let (request_id, resp) = self.client.send(req).await?;
+        let resp = self.client.check_response(&request_id, resp).await?;
+        self.table_cache.remove(&cache_key).await;
+        Ok((request_id, resp))
+    }
+}
+
 #[cfg(all(test, feature = "remote"))]
 mod test_utils {
    use super::*;
@@ -894,13 +913,28 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
    }

    async fn drop_table(&self, name: &str, namespace_path: &[String]) -> Result<()> {
-        let identifier = build_table_identifier(name, namespace_path, &self.client.id_delimiter);
-        let cache_key = build_cache_key(name, namespace_path);
-        let req = self.client.post(&format!("/v1/table/{}/drop/", identifier));
-        let (request_id, resp) = self.client.send(req).await?;
-        self.client.check_response(&request_id, resp).await?;
-        self.table_cache.remove(&cache_key).await;
-        Ok(())
+        self.submit_drop_table(name, namespace_path)
+            .await
+            .map(|_| ())
+    }
+
+    async fn drop_table_async(&self, name: &str, namespace_path: &[String]) -> Result<Job> {
+        let (request_id, response) = self.submit_drop_table(name, namespace_path).await?;
+        let status = response.status();
+        let body = response.text().await.err_to_http(request_id.clone())?;
+        let job_id = extract_job_id(&body);
+        Ok(match job_id {
+            Some(job_id) => Job::new(Box::new(RemoteJob::new(self.client.clone(), job_id))),
+            None if status == StatusCode::ACCEPTED => {
+                return Err(Error::Http {
+                    source: "asynchronous drop-table response did not contain a valid job_id"
+                        .into(),
+                    request_id,
+                    status_code: Some(status),
+                });
+            }
+            None => Job::new_done(),
+        })
    }

    async fn drop_all_tables(&self, namespace_path: &[String]) -> Result<()> {
@@ -1492,6 +1526,67 @@ mod tests {
        // NOTE: the API will return 200 even if the table does not exist. So we shouldn't expect 404.
    }

+    #[tokio::test]
+    async fn test_drop_table_does_not_read_response_body() {
+        let conn = Connection::new_with_handler(|_| {
+            http::Response::builder()
+                .status(200)
+                .body(vec![0xff])
+                .unwrap()
+        });
+
+        conn.drop_table("table1", &[]).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_drop_table_async_returns_job() {
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::POST);
+            assert_eq!(request.url().path(), "/v1/table/table1/drop/");
+            http::Response::builder()
+                .status(202)
+                .body(r#"{"job_id":"drop-job-123"}"#)
+                .unwrap()
+        });
+
+        let job = conn.drop_table_async("table1", &[]).await.unwrap();
+        assert_eq!(job.id(), Some("drop-job-123"));
+    }
+
+    #[tokio::test]
+    async fn test_drop_table_async_old_server_returns_done_job() {
+        let conn = Connection::new_with_handler(|_| {
+            http::Response::builder().status(200).body("").unwrap()
+        });
+
+        let job = conn.drop_table_async("table1", &[]).await.unwrap();
+        assert_eq!(job.id(), None);
+        assert_eq!(job.status().await.unwrap(), "finished");
+    }
+
+    #[tokio::test]
+    async fn test_drop_table_async_rejects_accepted_response_without_job_id() {
+        let conn = Connection::new_with_handler(|_| {
+            http::Response::builder().status(202).body("{}").unwrap()
+        });
+
+        let error = conn.drop_table_async("table1", &[]).await.err().unwrap();
+        assert!(error.to_string().contains("valid job_id"));
+    }
+
+    #[tokio::test]
+    async fn test_drop_table_async_rejects_empty_job_id() {
+        let conn = Connection::new_with_handler(|_| {
+            http::Response::builder()
+                .status(202)
+                .body(r#"{"job_id":""}"#)
+                .unwrap()
+        });
+
+        let error = conn.drop_table_async("table1", &[]).await.err().unwrap();
+        assert!(error.to_string().contains("valid job_id"));
+    }
+
    #[tokio::test]
    async fn test_rename_table() {
        let conn = Connection::new_with_handler(|request| {
@@ -8,7 +8,7 @@ use self::insert::{RemoteWriteExec, WriteOp};
 use super::client::RequestResultExt;
 use super::client::{HttpSend, RestfulLanceDbClient, Sender};
 use super::db::ServerVersion;
-use super::{ARROW_FILE_CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE};
+use super::{ARROW_FILE_CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE, extract_job_id};
 use crate::blob::BlobFile;
 use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions};
 use crate::expr::expr_to_sql_string;
@@ -33,7 +33,9 @@ use crate::table::lsm_stats::GetLsmStatsResponse;
 use crate::table::merge::MergeFilter;
 use crate::table::query::create_multi_vector_plan;
 use crate::table::write_progress::FinishOnDrop;
-use crate::table::{AlterColumnsResult, FieldMetadataUpdate, UpdateFieldMetadataResult};
+use crate::table::{
+    AlterColumnsResult, FieldMetadataUpdate, RefreshColumnResult, UpdateFieldMetadataResult,
+};
 use crate::table::{AnyQuery, Filter, Predicate, PreprocessingOutput, TableStatistics};
 use crate::utils::background_cache::BackgroundCache;
 use crate::utils::{
@@ -140,6 +142,40 @@ impl FreshnessHeaders {
    }
 }

+/// A backfill job whose successful wait establishes a read-freshness
+/// baseline on the submitting handle, so a later read cannot be served
+/// from a cache older than the completed fill. A handle pinned by checkout
+/// at completion keeps its time-travel view instead.
+struct FreshnessJob<S: HttpSend> {
+    inner: RemoteJob<S>,
+    freshness: Arc<Mutex<FreshnessState>>,
+    version: Arc<RwLock<Option<u64>>>,
+}
+
+#[async_trait]
+impl<S: HttpSend> crate::job::JobHandle for FreshnessJob<S> {
+    fn id(&self) -> Option<&str> {
+        crate::job::JobHandle::id(&self.inner)
+    }
+
+    async fn status(&self) -> Result<String> {
+        crate::job::JobHandle::status(&self.inner).await
+    }
+
+    async fn wait(&self) -> Result<()> {
+        crate::job::JobHandle::wait(&self.inner).await?;
+        let version = self.version.read().await;
+        if version.is_none() {
+            self.freshness.lock().unwrap().checkout_baseline = Some(SystemTime::now());
+        }
+        Ok(())
+    }
+
+    async fn cancel(&self) -> Result<()> {
+        crate::job::JobHandle::cancel(&self.inner).await
+    }
+}
+
 fn compute_min_timestamp(
    state: &FreshnessState,
    interval: Option<Duration>,
@@ -274,10 +310,10 @@ pub struct RemoteTable<S: HttpSend = Sender> {
    identifier: String,
    server_version: ServerVersion,

-    version: RwLock<Option<u64>>,
+    version: Arc<RwLock<Option<u64>>>,
    location: RwLock<Option<String>>,
    schema_cache: BackgroundCache<SchemaRef, Error>,
-    freshness: Mutex<FreshnessState>,
+    freshness: Arc<Mutex<FreshnessState>>,
    /// The branch this handle is scoped to, or `None` for the main branch.
    /// Stamped onto every branch-accepting request so reads and writes resolve
    /// on the branch's own version chain rather than main's.
@@ -392,13 +428,7 @@ impl<S: HttpSend> RemoteTable<S> {
            .text()
            .await
            .ok()
-            .and_then(|body| serde_json::from_str::<serde_json::Value>(&body).ok())
-            .and_then(|value| {
-                value
-                    .get("job_id")
-                    .and_then(|id| id.as_str())
-                    .map(str::to_string)
-            });
+            .and_then(|body| extract_job_id(&body));

        if let Some(wait_timeout) = index.wait_timeout {
            let index_name = index.name.unwrap_or_else(|| format!("{}_idx", column));
@@ -421,10 +451,10 @@ impl<S: HttpSend> RemoteTable<S> {
            namespace,
            identifier,
            server_version,
-            version: RwLock::new(None),
+            version: Arc::new(RwLock::new(None)),
            location: RwLock::new(None),
            schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
-            freshness: Mutex::new(FreshnessState::default()),
+            freshness: Arc::new(Mutex::new(FreshnessState::default())),
            branch: None,
        }
    }
@@ -453,10 +483,10 @@ impl<S: HttpSend> RemoteTable<S> {
            namespace: self.namespace.clone(),
            identifier: self.identifier.clone(),
            server_version: self.server_version.clone(),
-            version: RwLock::new(None),
+            version: Arc::new(RwLock::new(None)),
            location: RwLock::new(None),
            schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
-            freshness: Mutex::new(FreshnessState::default()),
+            freshness: Arc::new(Mutex::new(FreshnessState::default())),
            branch,
        }
    }
@@ -1274,10 +1304,10 @@ mod test_utils {
                namespace: vec![],
                identifier: name,
                server_version: version.map(ServerVersion).unwrap_or_default(),
-                version: RwLock::new(None),
+                version: Arc::new(RwLock::new(None)),
                location: RwLock::new(None),
                schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
-                freshness: Mutex::new(FreshnessState::default()),
+                freshness: Arc::new(Mutex::new(FreshnessState::default())),
                branch: None,
            }
        }
@@ -1298,10 +1328,10 @@ mod test_utils {
                namespace: vec![],
                identifier: name,
                server_version: ServerVersion::default(),
-                version: RwLock::new(None),
+                version: Arc::new(RwLock::new(None)),
                location: RwLock::new(None),
                schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
-                freshness: Mutex::new(FreshnessState::default()),
+                freshness: Arc::new(Mutex::new(FreshnessState::default())),
                branch: None,
            }
        }
@@ -1331,10 +1361,10 @@ mod test_utils {
                namespace: vec![],
                identifier: name,
                server_version: version.map(ServerVersion).unwrap_or_default(),
-                version: RwLock::new(None),
+                version: Arc::new(RwLock::new(None)),
                location: RwLock::new(None),
                schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
-                freshness: Mutex::new(FreshnessState::default()),
+                freshness: Arc::new(Mutex::new(FreshnessState::default())),
                branch: None,
            }
        }
@@ -2714,6 +2744,86 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        }
    }

+    async fn add_computed_columns(&self, columns: &[(String, String)]) -> Result<AddColumnsResult> {
+        self.check_mutable().await?;
+        // The server plans the declaration: expression validation, type
+        // inference and the persisted binding all happen there.
+        let entries = columns
+            .iter()
+            .map(
+                |(name, expression)| lance_namespace::models::AddColumnsEntry {
+                    name: name.clone(),
+                    computed: Some(Some(expression.clone())),
+                    ..Default::default()
+                },
+            )
+            .collect::<Vec<_>>();
+        let mut body = serde_json::json!({ "new_columns": entries });
+        self.apply_branch_body(&mut body);
+        let request = self
+            .client
+            .post(&format!("/v1/table/{}/add_columns/", self.identifier))
+            .json(&body);
+        let (request_id, response) = self.send(request, true).await?;
+        let response = self.check_table_response(&request_id, response).await?;
+        let body = response.text().await.err_to_http(request_id.clone())?;
+
+        if body.trim().is_empty() {
+            // Backward compatible with old servers
+            return Ok(AddColumnsResult { version: 0 });
+        }
+
+        let result: AddColumnsResult = serde_json::from_str(&body).map_err(|e| Error::Http {
+            source: format!("Failed to parse add_columns response: {}", e).into(),
+            request_id,
+            status_code: None,
+        })?;
+
+        self.invalidate_schema_cache();
+        self.track_write_version(result.version);
+
+        Ok(result)
+    }
+
+    async fn refresh_column(&self, _column: &str) -> Result<RefreshColumnResult> {
+        // The server runs a refresh as a job and does not report a fill
+        // count, so the blocking form has no honest result to return.
+        Err(Error::NotSupported {
+            message: "a remote refresh runs as a server job; use refresh_column_async and \
+                      wait on the returned handle"
+                .into(),
+        })
+    }
+
+    async fn refresh_column_async(&self, column: &str) -> Result<Job> {
+        self.check_mutable().await?;
+        let mut body = serde_json::json!({ "column": column });
+        self.apply_branch_body(&mut body);
+        let request = self
+            .client
+            .post(&format!("/v1/table/{}/backfill_column", self.identifier))
+            .json(&body);
+        let (request_id, response) = self.send(request, true).await?;
+        let response = self.check_table_response(&request_id, response).await?;
+        let body = response.text().await.err_to_http(request_id.clone())?;
+
+        #[derive(serde::Deserialize)]
+        struct BackfillResponse {
+            job_id: String,
+        }
+        let response: BackfillResponse = serde_json::from_str(&body).map_err(|e| Error::Http {
+            source: format!("Failed to parse backfill_column response: {}", e).into(),
+            request_id,
+            status_code: None,
+        })?;
+
+        Ok(Job::new(Box::new(FreshnessJob {
+            inner: RemoteJob::new(self.client.clone(), response.job_id),
+            freshness: self.freshness.clone(),
+            version: self.version.clone(),
+        })))
+    }
+
    async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result<AlterColumnsResult> {
        self.check_mutable().await?;
        let body = alterations
@@ -6455,6 +6565,346 @@ mod tests {
        assert_eq!(result.version, if old_server { 0 } else { 43 });
    }

+    /// A declaration is sent as `{name, computed}` entries for the server to
+    /// plan; the client never types the expression itself.
+    #[tokio::test]
+    async fn test_add_computed_columns_sends_the_expression() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            assert_eq!(request.url().path(), "/v1/table/my_table/add_columns/");
+            let body = request.body().unwrap().as_bytes().unwrap();
+            let value: serde_json::Value = serde_json::from_slice(body).unwrap();
+            assert_eq!(
+                value["new_columns"],
+                serde_json::json!([{"name": "doubled", "computed": "x * 2"}])
+            );
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"version": 7}"#)
+                .unwrap()
+        });
+
+        let result = table
+            .add_columns()
+            .computed("doubled", "x * 2")
+            .execute()
+            .await
+            .unwrap();
+        assert_eq!(result.version, 7);
+    }
+
+    /// A remote refresh is a server job: the async form returns its handle,
+    /// and the blocking form refuses rather than invent a fill count.
+    #[tokio::test]
+    async fn test_refresh_column_async_submits_a_backfill_job() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            assert_eq!(request.url().path(), "/v1/table/my_table/backfill_column");
+            let body = request.body().unwrap().as_bytes().unwrap();
+            let value: serde_json::Value = serde_json::from_slice(body).unwrap();
+            assert_eq!(value["column"], "doubled");
+            http::Response::builder()
+                .status(202)
+                .body(r#"{"job_id": "j-42"}"#)
+                .unwrap()
+        });
+
+        let job = table.refresh_column_async("doubled").await.unwrap();
+        assert_eq!(job.id(), Some("j-42"));
+
+        let err = table.refresh_column("doubled").await.unwrap_err();
+        assert!(
+            matches!(&err, Error::NotSupported { message }
+                if message.contains("refresh_column_async")),
+            "{err:?}"
+        );
+    }
+
+    /// The gate's reproducer: after a successful wait, a same-handle read
+    /// must carry a freshness baseline so a stale server cache cannot serve
+    /// the pre-backfill snapshot.
+    #[tokio::test]
+    async fn test_backfill_wait_establishes_read_freshness() {
+        let saw_min_timestamp = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let saw = saw_min_timestamp.clone();
+        let table =
+            Table::new_with_handler("my_table", move |request| match request.url().path() {
+                "/v1/table/my_table/backfill_column" => http::Response::builder()
+                    .status(202)
+                    .body(r#"{"job_id": "j-7"}"#.to_string())
+                    .unwrap(),
+                "/v1/jobs/describe" => http::Response::builder()
+                    .status(200)
+                    .body(r#"{"job_id": "j-7", "job_state": "DONE"}"#.to_string())
+                    .unwrap(),
+                "/v1/table/my_table/count_rows/" => {
+                    saw.store(
+                        request.headers().contains_key("x-lancedb-min-timestamp"),
+                        std::sync::atomic::Ordering::SeqCst,
+                    );
+                    http::Response::builder()
+                        .status(200)
+                        .body("1".to_string())
+                        .unwrap()
+                }
+                path => panic!("unexpected request: {path}"),
+            });
+
+        let job = table.refresh_column_async("doubled").await.unwrap();
+        job.wait().await.unwrap();
+        table.count_rows(None).await.unwrap();
+        assert!(
+            saw_min_timestamp.load(std::sync::atomic::Ordering::SeqCst),
+            "read after wait carried no freshness baseline"
+        );
+    }
+
+    /// A checkout after submission wins over the completion fence: the
+    /// pinned view must not regain a timestamp floor from the job.
+    #[tokio::test]
+    async fn test_checkout_after_submit_beats_the_completion_fence() {
+        let saw_min_timestamp = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let saw = saw_min_timestamp.clone();
+        let table =
+            Table::new_with_handler("my_table", move |request| match request.url().path() {
+                "/v1/table/my_table/backfill_column" => http::Response::builder()
+                    .status(202)
+                    .body(r#"{"job_id": "j-8"}"#.to_string())
+                    .unwrap(),
+                "/v1/jobs/describe" => http::Response::builder()
+                    .status(200)
+                    .body(r#"{"job_id": "j-8", "job_state": "DONE"}"#.to_string())
+                    .unwrap(),
+                "/v1/table/my_table/describe/" => {
+                    let schema = Schema::new(vec![Field::new("x", DataType::Int32, true)]);
+                    http::Response::builder()
+                        .status(200)
+                        .body(describe_response(&schema))
+                        .unwrap()
+                }
+                "/v1/table/my_table/count_rows/" => {
+                    saw.store(
+                        request.headers().contains_key("x-lancedb-min-timestamp"),
+                        std::sync::atomic::Ordering::SeqCst,
+                    );
+                    http::Response::builder()
+                        .status(200)
+                        .body("1".to_string())
+                        .unwrap()
+                }
+                path => panic!("unexpected request: {path}"),
+            });
+
+        let job = table.refresh_column_async("doubled").await.unwrap();
+        table.checkout(3).await.unwrap();
+        job.wait().await.unwrap();
+        table.count_rows(None).await.unwrap();
+        assert!(
+            !saw_min_timestamp.load(std::sync::atomic::Ordering::SeqCst),
+            "completion fence overrode an explicit checkout"
+        );
+    }
+
+    /// Tag checkout resets freshness state wholesale; the fence must not
+    /// survive it.
+    #[tokio::test]
+    async fn test_tag_checkout_after_submit_beats_the_completion_fence() {
+        let saw_min_timestamp = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let saw = saw_min_timestamp.clone();
+        let table =
+            Table::new_with_handler("my_table", move |request| match request.url().path() {
+                "/v1/table/my_table/backfill_column" => http::Response::builder()
+                    .status(202)
+                    .body(r#"{"job_id": "j-9"}"#.to_string())
+                    .unwrap(),
+                "/v1/jobs/describe" => http::Response::builder()
+                    .status(200)
+                    .body(r#"{"job_id": "j-9", "job_state": "DONE"}"#.to_string())
+                    .unwrap(),
+                "/v1/table/my_table/tags/version/" => http::Response::builder()
+                    .status(200)
+                    .body(r#"{"version": 5}"#.to_string())
+                    .unwrap(),
+                "/v1/table/my_table/describe/" => {
+                    let schema = Schema::new(vec![Field::new("x", DataType::Int32, true)]);
+                    http::Response::builder()
+                        .status(200)
+                        .body(describe_response(&schema))
+                        .unwrap()
+                }
+                "/v1/table/my_table/count_rows/" => {
+                    saw.store(
+                        request.headers().contains_key("x-lancedb-min-timestamp"),
+                        std::sync::atomic::Ordering::SeqCst,
+                    );
+                    http::Response::builder()
+                        .status(200)
+                        .body("1".to_string())
+                        .unwrap()
+                }
+                path => panic!("unexpected request: {path}"),
+            });
+
+        let job = table.refresh_column_async("doubled").await.unwrap();
+        table.checkout_tag("v1").await.unwrap();
+        job.wait().await.unwrap();
+        table.count_rows(None).await.unwrap();
+        assert!(
+            !saw_min_timestamp.load(std::sync::atomic::Ordering::SeqCst),
+            "completion fence overrode a tag checkout"
+        );
+    }
+
+    /// A checkout landing while the submission request is in flight advances
+    /// the epoch past the token captured at submit.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_checkout_during_submission_beats_the_completion_fence() {
+        let saw_min_timestamp = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let saw = saw_min_timestamp.clone();
+        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+        let release_rx = Arc::new(std::sync::Mutex::new(release_rx));
+        let (arrived_tx, arrived_rx) = std::sync::mpsc::channel::<()>();
+        let arrived_tx = Arc::new(std::sync::Mutex::new(arrived_tx));
+        let table = Table::new_with_handler("my_table", move |request| {
+            match request.url().path() {
+                "/v1/table/my_table/backfill_column" => {
+                    // Signal arrival, then hold the response until the
+                    // test's checkout completes.
+                    arrived_tx.lock().unwrap().send(()).unwrap();
+                    release_rx
+                        .lock()
+                        .unwrap()
+                        .recv_timeout(std::time::Duration::from_secs(10))
+                        .unwrap();
+                    http::Response::builder()
+                        .status(202)
+                        .body(r#"{"job_id": "j-10"}"#.to_string())
+                        .unwrap()
+                }
+                "/v1/jobs/describe" => http::Response::builder()
+                    .status(200)
+                    .body(r#"{"job_id": "j-10", "job_state": "DONE"}"#.to_string())
+                    .unwrap(),
+                "/v1/table/my_table/describe/" => {
+                    let schema = Schema::new(vec![Field::new("x", DataType::Int32, true)]);
+                    http::Response::builder()
+                        .status(200)
+                        .body(describe_response(&schema))
+                        .unwrap()
+                }
+                "/v1/table/my_table/count_rows/" => {
+                    saw.store(
+                        request.headers().contains_key("x-lancedb-min-timestamp"),
+                        std::sync::atomic::Ordering::SeqCst,
+                    );
+                    http::Response::builder()
+                        .status(200)
+                        .body("1".to_string())
+                        .unwrap()
+                }
+                path => panic!("unexpected request: {path}"),
+            }
+        });
+
+        let submit = tokio::spawn({
+            let table = table.clone();
+            async move { table.refresh_column_async("doubled").await }
+        });
+        tokio::task::spawn_blocking(move || {
+            arrived_rx
+                .recv_timeout(std::time::Duration::from_secs(10))
+                .unwrap()
+        })
+        .await
+        .unwrap();
+        table.checkout(7).await.unwrap();
+        release_tx.send(()).unwrap();
+
+        let job = submit.await.unwrap().unwrap();
+        job.wait().await.unwrap();
+        table.count_rows(None).await.unwrap();
+        assert!(
+            !saw_min_timestamp.load(std::sync::atomic::Ordering::SeqCst),
+            "completion fence overrode a checkout that landed mid-submission"
+        );
+    }
+
+    /// checkout_latest keeps the handle on latest, so a completed backfill
+    /// must still establish its post-fill baseline -- strictly later than the
+    /// checkout's own, or a pre-fill cache could still serve.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_checkout_latest_during_submission_keeps_the_fence() {
+        let seen_min_timestamp = Arc::new(std::sync::Mutex::new(None::<String>));
+        let saw = seen_min_timestamp.clone();
+        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+        let release_rx = Arc::new(std::sync::Mutex::new(release_rx));
+        let (arrived_tx, arrived_rx) = std::sync::mpsc::channel::<()>();
+        let arrived_tx = Arc::new(std::sync::Mutex::new(arrived_tx));
+        let table =
+            Table::new_with_handler("my_table", move |request| match request.url().path() {
+                "/v1/table/my_table/backfill_column" => {
+                    arrived_tx.lock().unwrap().send(()).unwrap();
+                    release_rx
+                        .lock()
+                        .unwrap()
+                        .recv_timeout(std::time::Duration::from_secs(10))
+                        .unwrap();
+                    http::Response::builder()
+                        .status(202)
+                        .body(r#"{"job_id": "j-11"}"#.to_string())
+                        .unwrap()
+                }
+                "/v1/jobs/describe" => http::Response::builder()
+                    .status(200)
+                    .body(r#"{"job_id": "j-11", "job_state": "DONE"}"#.to_string())
+                    .unwrap(),
+                "/v1/table/my_table/count_rows/" => {
+                    *saw.lock().unwrap() = request
+                        .headers()
+                        .get("x-lancedb-min-timestamp")
+                        .map(|v| v.to_str().unwrap().to_string());
+                    http::Response::builder()
+                        .status(200)
+                        .body("1".to_string())
+                        .unwrap()
+                }
+                path => panic!("unexpected request: {path}"),
+            });
+
+        let submit = tokio::spawn({
+            let table = table.clone();
+            async move { table.refresh_column_async("doubled").await }
+        });
+        tokio::task::spawn_blocking(move || {
+            arrived_rx
+                .recv_timeout(std::time::Duration::from_secs(10))
+                .unwrap()
+        })
+        .await
+        .unwrap();
+        table.checkout_latest().await.unwrap();
+        let after_checkout = SystemTime::now();
+        // Real separation between the checkout baseline and completion.
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        release_tx.send(()).unwrap();
+
+        let job = submit.await.unwrap().unwrap();
+        job.wait().await.unwrap();
+        table.count_rows(None).await.unwrap();
+        let header = seen_min_timestamp
+            .lock()
+            .unwrap()
+            .clone()
+            .expect("no baseline");
+        let sent: SystemTime = chrono::DateTime::parse_from_rfc3339(&header)
+            .unwrap()
+            .into();
+        assert!(
+            sent > after_checkout,
+            "baseline {header} did not advance past the checkout"
+        );
+    }
+
    #[tokio::test]
    async fn test_prewarm_index() {
        let table = Table::new_with_handler("my_table", |request| {
@@ -68,6 +68,7 @@ pub mod add_columns;
 mod add_data;
 pub mod branch_merge;
 pub mod checkpoint;
+pub mod computed_columns;
 mod create_index;
 pub mod datafusion;
 pub(crate) mod dataset;
@@ -77,6 +78,7 @@ pub mod merge;
 pub mod optimize;
 mod primary_key;
 pub mod query;
+pub mod refresh;
 pub mod schema_evolution;
 pub mod update;
 pub mod write_progress;
@@ -90,6 +92,9 @@ pub use branch_merge::{
    MergeBranchResult, MergeBranchStatus, MergePreview, RowCountSummary,
 };
 pub use chrono::Duration;
+pub use computed_columns::{
+    ComputedColumn, ComputedColumnKind, computed_column_from_field, computed_columns,
+};
 pub use delete::DeleteResult;
 use futures::future::join_all;
 pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTags};
@@ -97,6 +102,7 @@ pub use lance::dataset::scanner::DatasetRecordBatchStream;
 pub use lance_index::optimize::OptimizeOptions;
 pub use lsm_stats::{BucketStats, GenerationStats, LsmStats, MemtableStats};
 pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats};
+pub use refresh::RefreshColumnResult;
 pub use schema_evolution::{
    AddColumnsResult, AlterColumnsResult, DropColumnsResult, FieldMetadataUpdate,
    UpdateFieldMetadataResult,
@@ -637,6 +643,15 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
            message: "set_lsm_write_spec is not supported on this table type".into(),
        })
    }
+    /// Switch this table to required index catch-up, one way.
+    ///
+    /// The default implementation returns `NotSupported`. Implementations
+    /// that support the MemWAL LSM write path must override this.
+    async fn require_mem_wal_index_catchup(&self) -> Result<()> {
+        Err(Error::NotSupported {
+            message: "require_mem_wal_index_catchup is not supported on this table type".into(),
+        })
+    }
    /// Remove the [`LsmWriteSpec`] from this table.
    ///
    /// This is a no-op if no spec is currently set.
@@ -732,6 +747,34 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
        transforms: NewColumnTransform,
        read_columns: Option<Vec<String>>,
    ) -> Result<AddColumnsResult>;
+    /// Declare computed columns, each defined by a SQL expression.
+    ///
+    /// Where the declaration is planned depends on the backend: a local table
+    /// validates and types the expression itself, a remote one sends the text
+    /// for the server to plan.
+    async fn add_computed_columns(
+        &self,
+        _columns: &[(String, String)],
+    ) -> Result<AddColumnsResult> {
+        Err(Error::NotSupported {
+            message: "computed columns are not supported on this table type".into(),
+        })
+    }
+    /// Fill a computed column's unfilled rows.
+    ///
+    /// The default returns `NotSupported`; Lance-backed tables override it.
+    async fn refresh_column(&self, _column: &str) -> Result<RefreshColumnResult> {
+        Err(Error::NotSupported {
+            message: "computed columns are supported only on local tables".into(),
+        })
+    }
+    /// Fill a computed column's unfilled rows, returning a [`Job`] tracking
+    /// the operation.
+    async fn refresh_column_async(&self, _column: &str) -> Result<Job> {
+        Err(Error::NotSupported {
+            message: "computed columns are supported only on local tables".into(),
+        })
+    }
    /// Alter columns in the table.
    async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result<AlterColumnsResult>;
    /// Drop columns from the table.
@@ -1624,6 +1667,53 @@ impl Table {
        AddColumnsBuilder::new(self.inner.clone())
    }

+    /// Fill the fragments of a computed column that hold no values yet.
+    ///
+    /// Declared with
+    /// [`AddColumnsBuilder::computed`](add_columns::AddColumnsBuilder::computed),
+    /// a column starts empty and gets its values here. Fragments appended
+    /// since the last refresh are filled by the next one; fragments already
+    /// filled are left as they are, so the call is idempotent and does not
+    /// observe a mutated input.
+    ///
+    /// Local tables only: a remote refresh runs as a server job, through
+    /// [`Table::refresh_column_async`].
+    ///
+    /// ```
+    /// # use lancedb::Table;
+    /// # async fn refresh(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
+    /// let result = table.refresh_column("doubled").await?;
+    /// println!("filled {} rows at version {}", result.rows_filled, result.version);
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn refresh_column(&self, column: impl AsRef<str>) -> Result<RefreshColumnResult> {
+        self.inner.refresh_column(column.as_ref()).await
+    }
+
+    /// Like [`Table::refresh_column`], but returns a [`Job`] tracking the
+    /// operation instead of blocking until it completes.
+    ///
+    /// The job may already be complete when returned, and callers must not
+    /// assume the column is filled until [`Job::wait`] returns. Invalid input
+    /// -- an unknown column, or one that is not computed -- is reported by
+    /// this call rather than by the job. On local tables the job runs as an
+    /// in-process task; on LanceDB Cloud and Enterprise it is the server's
+    /// backfill job.
+    ///
+    /// ```
+    /// # use lancedb::Table;
+    /// # async fn refresh_in_background(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
+    /// let job = table.refresh_column_async("doubled").await?;
+    /// println!("refresh running: {:?}", job.status().await?);
+    /// job.wait().await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn refresh_column_async(&self, column: impl AsRef<str>) -> Result<Job> {
+        self.inner.refresh_column_async(column.as_ref()).await
+    }
+
    /// Change a column's name or nullability.
    pub async fn alter_columns(
        &self,
@@ -1693,6 +1783,20 @@ impl Table {
        self.inner.set_lsm_write_spec(spec).await
    }

+    /// Switch this table to required index catch-up, one way.
+    ///
+    /// Separate from [`Self::set_lsm_write_spec`] on purpose: a table carrying
+    /// the bit retains its SSTables until an index records that it holds the
+    /// compacted rows, so turn it on only once something can repair coverage.
+    /// A writer that already holds the dataset can call the equivalent on
+    /// `DatasetMemWalExt` instead; this is the table-level entry point.
+    ///
+    /// Errors if no spec is set, or if the table already records SSTable
+    /// compaction progress from before this protocol.
+    pub async fn require_mem_wal_index_catchup(&self) -> Result<()> {
+        self.inner.require_mem_wal_index_catchup().await
+    }
+
    /// Remove the [`LsmWriteSpec`] from this table, reverting to the standard
    /// `merge_insert` write path.
    ///
@@ -2605,6 +2709,7 @@ impl NativeTable {
        namespace_client: Option<Arc<dyn LanceNamespace>>,
        pushdown_operations: HashSet<NamespaceClientPushdownOperation>,
    ) -> Result<Self> {
+        computed_columns::ensure_no_foreign_declarations(batches.arrow_schema().fields())?;
        // Default params uses format v1.
        let params = params.unwrap_or(WriteParams {
            ..Default::default()
@@ -3053,6 +3158,13 @@ impl BaseTable for NativeTable {
        let ds = self.dataset.get().await?;

        let table_schema = Schema::from(&ds.schema().clone());
+        computed_columns::ensure_not_written(
+            &table_schema,
+            add.data.schema().fields().iter().map(|f| f.name().as_str()),
+        )?;
+        if matches!(add.mode, AddDataMode::Overwrite) {
+            computed_columns::ensure_no_foreign_declarations(add.data.schema().fields())?;
+        }

        let num_partitions = if let Some(parallelism) = add.write_parallelism {
            parallelism
@@ -3213,6 +3325,11 @@ impl BaseTable for NativeTable {
        params: MergeInsertBuilder,
        new_data: Box<dyn RecordBatchReader + Send>,
    ) -> Result<MergeResult> {
+        let source_schema = arrow_array::RecordBatchReader::schema(&new_data);
+        computed_columns::ensure_not_written(
+            &Schema::from(self.dataset.get().await?.schema()),
+            source_schema.fields().iter().map(|f| f.name().as_str()),
+        )?;
        let result = merge::execute_merge_insert(self, params, new_data).await?;
        self.bump_freshness();
        Ok(result)
@@ -3226,6 +3343,10 @@ impl BaseTable for NativeTable {
        merge::lsm::set_lsm_write_spec(self, spec).await
    }

+    async fn require_mem_wal_index_catchup(&self) -> Result<()> {
+        merge::lsm::require_mem_wal_index_catchup(self).await
+    }
+
    async fn unset_lsm_write_spec(&self) -> Result<()> {
        merge::lsm::unset_lsm_write_spec(self).await
    }
@@ -3294,6 +3415,22 @@ impl BaseTable for NativeTable {
        Ok(result)
    }

+    async fn add_computed_columns(&self, columns: &[(String, String)]) -> Result<AddColumnsResult> {
+        let result = schema_evolution::execute_declare(self, columns).await?;
+        self.bump_freshness();
+        Ok(result)
+    }
+
+    async fn refresh_column(&self, column: &str) -> Result<RefreshColumnResult> {
+        let result = refresh::execute_refresh_column(self, column).await?;
+        self.bump_freshness();
+        Ok(result)
+    }
+
+    async fn refresh_column_async(&self, column: &str) -> Result<Job> {
+        refresh::execute_refresh_column_async(self, column).await
+    }
+
    async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result<AlterColumnsResult> {
        let result = schema_evolution::execute_alter_columns(self, alterations).await?;
        self.bump_freshness();
@@ -15,6 +15,7 @@ use crate::{Error, Result};
 pub struct AddColumnsBuilder {
    parent: Arc<dyn BaseTable>,
    transform: Option<NewColumnTransform>,
+    computed: Vec<(String, String)>,
    read_columns: Option<Vec<String>>,
 }

@@ -23,6 +24,7 @@ impl std::fmt::Debug for AddColumnsBuilder {
        f.debug_struct("AddColumnsBuilder")
            .field("parent", &self.parent)
            .field("has_transform", &self.transform.is_some())
+            .field("computed", &self.computed)
            .field("read_columns", &self.read_columns)
            .finish()
    }
@@ -33,19 +35,58 @@ impl AddColumnsBuilder {
        Self {
            parent,
            transform: None,
+            computed: Vec::new(),
            read_columns: None,
        }
    }

-    /// Set how the new columns' values are produced. Required.
+    /// Set how the new columns' values are produced.
    pub fn transform(mut self, transform: NewColumnTransform) -> Self {
        self.transform = Some(transform);
        self
    }

+    /// Add a column defined by `expression`, evaluated by a later refresh
+    /// rather than by this commit. Its type and inputs are derived from the
+    /// expression.
+    ///
+    /// The column is committed with no values, so declaring one costs the same
+    /// on an empty table as on a large one. Rows get values from
+    /// [`Table::refresh_column`](super::Table::refresh_column), which fills
+    /// every fragment that has none -- including fragments appended since the
+    /// last refresh.
+    ///
+    /// Refresh does not revisit a fragment it has filled, so mutating an input
+    /// leaves the value computed at fill time; recomputing means dropping the
+    /// column and declaring it again. An input cannot be renamed, retyped or
+    /// dropped while a declaration reads it, since the expression names it.
+    ///
+    /// On LanceDB Cloud and Enterprise the expression is planned by the
+    /// server, and the refresh runs as a server job -- see
+    /// [`Table::refresh_column_async`](super::Table::refresh_column_async).
+    ///
+    /// ```
+    /// # use lancedb::Table;
+    /// # async fn declare(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
+    /// table
+    ///     .add_columns()
+    ///     .computed("doubled", "x * 2")
+    ///     .execute()
+    ///     .await?;
+    /// let filled = table.refresh_column("doubled").await?;
+    /// println!("filled {} rows", filled.rows_filled);
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn computed(mut self, name: impl Into<String>, expression: impl Into<String>) -> Self {
+        self.computed.push((name.into(), expression.into()));
+        self
+    }
+
    /// Limit which existing columns a [`NewColumnTransform::BatchUDF`] mapper
-    /// receives. Every other transform determines what it reads, so setting
-    /// this alongside one is an error rather than a silent no-op.
+    /// receives. Every other transform, and a computed column, determines what
+    /// it reads, so setting this alongside one is an error rather than a silent
+    /// no-op.
    pub fn read_columns(mut self, columns: impl IntoIterator<Item = impl Into<String>>) -> Self {
        self.read_columns = Some(columns.into_iter().map(Into::into).collect());
        self
@@ -56,24 +97,42 @@ impl AddColumnsBuilder {
        let Self {
            parent,
            transform,
+            computed,
            read_columns,
        } = self;

-        let Some(transform) = transform else {
-            return Err(Error::InvalidInput {
-                message: "add_columns requires a transform".into(),
-            });
-        };
-
-        if read_columns.is_some() && !matches!(transform, NewColumnTransform::BatchUDF(_)) {
-            return Err(Error::InvalidInput {
-                message: "read_columns applies only to a BatchUDF transform; \
-                          every other transform determines what it reads"
+        match (transform, computed.is_empty()) {
+            (None, true) => Err(Error::InvalidInput {
+                message: "add_columns requires a transform or a computed column".into(),
+            }),
+            // The two commit through different transforms, so one call covering
+            // both would be two commits and could half-apply.
+            (Some(_), false) => Err(Error::InvalidInput {
+                message: "add_columns cannot mix a transform with computed columns; \
+                          they cannot be added atomically in one call"
                    .into(),
-            });
+            }),
+            (Some(transform), true) => {
+                if read_columns.is_some() && !matches!(transform, NewColumnTransform::BatchUDF(_)) {
+                    return Err(Error::InvalidInput {
+                        message: "read_columns applies only to a BatchUDF transform; \
+                                  every other transform determines what it reads"
+                            .into(),
+                    });
+                }
+                parent.add_columns(transform, read_columns).await
+            }
+            (None, false) => {
+                if read_columns.is_some() {
+                    return Err(Error::InvalidInput {
+                        message: "read_columns applies only to a BatchUDF transform; \
+                                  a computed column's inputs come from its expression"
+                            .into(),
+                    });
+                }
+                parent.add_computed_columns(&computed).await
+            }
        }
-
-        parent.add_columns(transform, read_columns).await
    }
 }

@@ -85,8 +144,8 @@ mod tests {
    use arrow_schema::{DataType, Field, Schema};
    use lance::dataset::{BatchUDF, NewColumnTransform};

-    use crate::Table;
    use crate::connect;
+    use crate::{Error, Table};

    async fn table_with_two_columns(name: &str) -> Table {
        let conn = connect("memory://").execute().await.unwrap();
@@ -98,10 +157,7 @@ mod tests {
    async fn test_requires_a_transform() {
        let table = table_with_two_columns("no_transform").await;
        let err = table.add_columns().execute().await.unwrap_err();
-        assert!(
-            err.to_string().contains("requires a transform"),
-            "got: {err}"
-        );
+        assert!(matches!(err, Error::InvalidInput { .. }));
    }

    #[tokio::test]
@@ -117,7 +173,7 @@ mod tests {
            .execute()
            .await
            .unwrap_err();
-        assert!(err.to_string().contains("BatchUDF"), "got: {err}");
+        assert!(matches!(err, Error::InvalidInput { .. }));

        let schema = table.schema().await.unwrap();
        assert!(
@@ -126,6 +182,47 @@ mod tests {
        );
    }

+    #[tokio::test]
+    async fn test_mixing_transform_and_computed_is_rejected() {
+        let table = table_with_two_columns("mixed_add").await;
+        let err = table
+            .add_columns()
+            .transform(NewColumnTransform::SqlExpressions(vec![(
+                "eager".into(),
+                "x * 2".into(),
+            )]))
+            .computed("lazy", "x * 3")
+            .execute()
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::InvalidInput { .. }));
+
+        let schema = table.schema().await.unwrap();
+        assert!(schema.field_with_name("eager").is_err());
+        assert!(schema.field_with_name("lazy").is_err());
+    }
+
+    #[tokio::test]
+    async fn test_read_columns_with_computed_is_rejected() {
+        let table = table_with_two_columns("read_cols_computed").await;
+        let err = table
+            .add_columns()
+            .computed("doubled", "x * 2")
+            .read_columns(["x"])
+            .execute()
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::InvalidInput { .. }));
+        assert!(
+            table
+                .schema()
+                .await
+                .unwrap()
+                .field_with_name("doubled")
+                .is_err()
+        );
+    }
+
    #[tokio::test]
    async fn test_read_columns_limits_what_a_batch_udf_sees() {
        let table = table_with_two_columns("read_cols_udf").await;
@@ -17,7 +17,7 @@ use datafusion_physical_plan::stream::RecordBatchStreamAdapter;
 use datafusion_physical_plan::{
    DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties,
 };
-use futures::TryStreamExt;
+use futures::StreamExt;
 use lance::Dataset;
 use lance::dataset::transaction::{Operation, Transaction};
 use lance::dataset::{CommitBuilder, InsertBuilder, WriteParams, WriteProgressFn};
@@ -194,12 +194,23 @@ impl ExecutionPlan for InsertExec {

        let output_bytes = MetricBuilder::new(&self.metrics).output_bytes(partition);
        let input_schema = input_stream.schema();
+        let declared: Vec<String> = crate::table::computed_columns::computed_columns(
+            &arrow_schema::Schema::from(self.dataset.schema()),
+        )
+        .into_iter()
+        .map(|declaration| declaration.name)
+        .collect();
        let input_stream: SendableRecordBatchStream =
            Box::pin(InstrumentedRecordBatchStreamAdapter::new(
                input_schema,
-                input_stream.map_ok(move |batch| {
+                input_stream.map(move |batch| {
+                    let batch = batch?;
+                    crate::table::computed_columns::ensure_batch_writes_no_computed_values(
+                        &declared, &batch,
+                    )
+                    .map_err(|e| datafusion::error::DataFusionError::External(Box::new(e)))?;
                    output_bytes.add(batch.get_array_memory_size());
-                    batch
+                    Ok(batch)
                }),
                partition,
                &self.metrics,
@@ -94,7 +94,16 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec)
        .await?
    };

+    table.checkout_latest().await?;
    let mut dataset = (*table.dataset.get().await?).clone();
+    let schema = arrow_schema::Schema::from(dataset.schema());
+    if !crate::table::computed_columns::computed_columns(&schema).is_empty() {
+        return Err(Error::NotSupported {
+            message: "an LSM write spec cannot be installed on a table with computed \
+                      columns: rows in un-compacted tiers are invisible to refresh"
+                .into(),
+        });
+    }
    let mut builder = dataset.initialize_mem_wal();
    let writer_config_defaults = match spec {
        LsmWriteSpec::Bucket {
@@ -183,6 +192,36 @@ fn index_name_list(indices: &[IndexConfig]) -> String {
    format!("[{}]", names.join(", "))
 }

+// =============================================================================
+// require_mem_wal_index_catchup
+// =============================================================================
+
+/// Switch this table to required index catch-up, one way.
+///
+/// Deliberately **not** part of installing the write spec. Until something can
+/// actually repair coverage, a table carrying the bit reports every index as
+/// not known to hold the compacted rows, so its SSTables are retained
+/// indefinitely -- and the WAL pod trims on the legacy rule meanwhile, leaving
+/// readers pointed at files that are gone. Turn this on only once remote
+/// maintenance owns the merge and the repair for the table.
+///
+/// Lance refuses the activation if the table already records SSTable
+/// compaction progress: those numbers predate this protocol and cannot be
+/// validated, so such a table must be drained rather than activated.
+#[allow(clippy::redundant_pub_crate)]
+pub(crate) async fn require_mem_wal_index_catchup(table: &NativeTable) -> Result<()> {
+    table.dataset.ensure_mutable()?;
+    let mut dataset = (*table.dataset.get().await?).clone();
+    if dataset.mem_wal_index_details().await?.is_none() {
+        return Err(Error::InvalidInput {
+            message: "require_mem_wal_index_catchup: no LSM write spec is set on this table".into(),
+        });
+    }
+    dataset.require_mem_wal_index_catchup().await?;
+    table.dataset.update(dataset);
+    Ok(())
+}
+
 // =============================================================================
 // unset_lsm_write_spec
 // =============================================================================
@@ -36,6 +36,7 @@ use lance::dataset::mem_wal::{
    DatasetMemWalExt, LsmScanner, ShardManifestStore, ShardSnapshot, ShardWriterConfig,
 };
 use lance_index::mem_wal::{MemWalIndexDetails, ShardManifest};
+use lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP;
 use uuid::Uuid;

 use super::NativeTable;
@@ -84,9 +85,8 @@ pub(super) async fn create_lsm_plan(
    let pk_columns = pk_columns(&ds_ref)?;
    // The base index an indexed arm relies on may lag compaction; resolve it so the
    // snapshot retains SSTables the index has not yet caught up to.
-    let arm_index = arm_maintained_index_name(&ds_ref, &query, &details).await?;
-    let (snapshots, in_memory) =
-        build_read_context(table, &ds_ref, &details, arm_index.as_deref()).await?;
+    let arm_indexes = arm_maintained_index_names(&ds_ref, &query, &details).await?;
+    let (snapshots, in_memory) = build_read_context(table, &ds_ref, &details, &arm_indexes).await?;

    let limit = query.base.limit;
    let offset = query.base.offset;
@@ -232,28 +232,44 @@ fn pk_columns(dataset: &Dataset) -> Result<Vec<String>> {
    Ok(pk)
 }

-/// Per-shard SSTable exclusion watermark: the generation at or below which SSTables
-/// are safe to drop for this arm. A generation is droppable only once it is
-/// compacted into the base table AND covered by `index_name`'s catch-up (for an
-/// indexed arm); a plain scan (`index_name == None`) uses the compaction watermark
-/// alone. Capping at the index catch-up keeps rows the base index has not yet
-/// indexed visible through their SSTable. First occurrence per shard mirrors Lance's
-/// `compacted_generation_for_shard`.
+/// Per-shard SSTable exclusion watermark: the generation at or below which
+/// SSTables are safe to drop for this query.
+///
+/// A generation is droppable only once it is compacted into the base table AND
+/// covered by the catch-up of every index the query relies on, so the watermark
+/// is the minimum across `index_names`. Gating on fewer than all of them would
+/// drop SSTables holding rows an uncounted index has not yet indexed, and that
+/// arm would silently return fewer rows.
+///
+/// See [`arm_maintained_index_names`] for which indexes are collected today: a
+/// vector search with a scalar prefilter is not yet among them.
+///
+/// An empty `index_names` (a plain scan) uses the compaction watermark alone.
+/// First occurrence per shard mirrors Lance's `compacted_generation_for_shard`.
 fn exclusion_watermarks(
    details: &MemWalIndexDetails,
-    index_name: Option<&str>,
+    index_names: &[String],
+    catchup_required: bool,
 ) -> HashMap<Uuid, u64> {
    let mut exclude: HashMap<Uuid, u64> = HashMap::new();
    for entry in &details.compacted_sstables {
        let mut watermark = entry.generation;
-        if let Some(name) = index_name
-            && let Some(caught_up) = details
+        for name in index_names {
+            match details
                .index_catchup
                .iter()
-                .find(|icp| icp.index_name == name)
+                .find(|icp| icp.index_name == *name)
                .and_then(|icp| icp.caught_up_generation_for_shard(&entry.shard_id))
-        {
-            watermark = watermark.min(caught_up);
+            {
+                Some(caught_up) => watermark = watermark.min(caught_up),
+                // No entry. On a table that requires catch-up this means the
+                // index is *not* known to hold these rows, and the base arm is
+                // index-only -- so every generation stays readable from its
+                // SSTable. Without the bit the field is not maintained at all,
+                // and absence carries no information.
+                None if catchup_required => watermark = 0,
+                None => {}
+            }
        }
        exclude.entry(entry.shard_id).or_insert(watermark);
    }
@@ -267,13 +283,26 @@ fn exclusion_watermarks(
 /// with a live cached `ShardWriter` (this session's in-flight writes) the
 /// writer's authoritative in-memory manifest and memtables override the
 /// on-disk view so a read sees data not yet flushed.
+/// Whether this table reads a missing `index_catchup` entry as "not caught up".
+///
+/// Both words must be set. A reader honouring the bit while a writer does not
+/// would retain SSTables the writer had already trimmed, and the reverse would
+/// serve rows from files the writer still expects to be excluded -- so a
+/// half-set manifest is treated as legacy, which is the conservative side.
+fn requires_index_catchup(dataset: &Dataset) -> bool {
+    let manifest = dataset.manifest();
+    manifest.reader_feature_flags & FLAG_MEM_WAL_INDEX_CATCHUP != 0
+        && manifest.writer_feature_flags & FLAG_MEM_WAL_INDEX_CATCHUP != 0
+}
+
 async fn build_read_context(
    table: &NativeTable,
    dataset: &Dataset,
    details: &MemWalIndexDetails,
-    index_name: Option<&str>,
+    index_names: &[String],
 ) -> Result<(Vec<ShardSnapshot>, HashMap<Uuid, InMemoryMemTables>)> {
-    let exclude = exclusion_watermarks(details, index_name);
+    let catchup_required = requires_index_catchup(dataset);
+    let exclude = exclusion_watermarks(details, index_names, catchup_required);

    let shard_ids = dataset.list_mem_wal_latest_shard_ids().await?;
    // Use the dataset's own object store (not `ObjectStore::from_uri`, which
@@ -487,19 +516,33 @@ async fn index_maintained(
    }))
 }

-/// The maintained base index the query's arm relies on (vector index for ANN, FTS
-/// index for full-text), used to gate SSTable compaction exclusion by index catch-up.
-/// `None` for a plain scan or when no maintained index covers the searched column.
-async fn arm_maintained_index_name(
+/// Every maintained base index this query relies on, used to gate SSTable
+/// exclusion by index catch-up.
+///
+/// Returns a list because the watermark must be the lowest across every index a
+/// query relies on. Today it never holds more than one: `reject_unsupported`
+/// refuses hybrid search, so the vector and full-text arms are mutually
+/// exclusive.
+///
+/// The case that is genuinely multi-index -- a vector search with a scalar or
+/// bitmap prefilter -- is **not collected yet**. Identifying those needs the
+/// planner's chosen indexes, not the columns the filter names, and no Lance API
+/// exposes them. Until it does, such a query is gated on its vector index alone.
+///
+/// Empty for a plain scan, or when no maintained index covers the searched
+/// column.
+async fn arm_maintained_index_names(
    dataset: &Dataset,
    query: &VectorQueryRequest,
    details: &MemWalIndexDetails,
-) -> Result<Option<String>> {
+) -> Result<Vec<String>> {
    use lance::index::DatasetIndexExt;
-    // Resolve the arm's searched column, the index-detail type it relies on, and a
+
+    // Each arm's searched column, the index-detail type it relies on, and a
    // label for diagnostics — catch-up is taken from the vector/FTS index
    // specifically, not a BTree on the same column.
-    let (column, type_url_suffix, arm) = if !query.query_vector.is_empty() {
+    let mut arms: Vec<(String, &str, &str)> = Vec::new();
+    if !query.query_vector.is_empty() {
        let arrow_schema = ArrowSchema::from(dataset.schema());
        let column = match &query.column {
            Some(column) => column.clone(),
@@ -508,31 +551,43 @@ async fn arm_maintained_index_name(
                default_vector_column(&arrow_schema, dim)?
            }
        };
-        (column, "VectorIndexDetails", "vector")
-    } else if let Some(fts) = &query.base.full_text_search {
-        match fts.columns().into_iter().next() {
-            Some(column) => (column, "InvertedIndexDetails", "full-text"),
-            None => return Ok(None),
-        }
-    } else {
-        return Ok(None);
-    };
-    let Some(field) = dataset.schema().field(&column) else {
-        return Ok(None);
-    };
+        arms.push((column, "VectorIndexDetails", "vector"));
+    }
+    if let Some(fts) = &query.base.full_text_search
+        && let Some(column) = fts.columns().into_iter().next()
+    {
+        arms.push((column, "InvertedIndexDetails", "full-text"));
+    }
+    if arms.is_empty() {
+        return Ok(Vec::new());
+    }
+
    let indices = dataset.load_indices().await?;
-    let segment_names: Vec<String> = indices
-        .iter()
-        .filter(|idx| {
-            idx.fields.contains(&field.id)
-                && idx
-                    .index_details
-                    .as_ref()
-                    .is_some_and(|d| d.type_url.ends_with(type_url_suffix))
-        })
-        .map(|idx| idx.name.clone())
-        .collect();
-    resolve_single_index(segment_names, &details.maintained_indexes, arm, &column)
+    let mut names = Vec::with_capacity(arms.len());
+    for (column, type_url_suffix, arm) in arms {
+        let Some(field) = dataset.schema().field(&column) else {
+            continue;
+        };
+        let segment_names: Vec<String> = indices
+            .iter()
+            .filter(|idx| {
+                idx.fields.contains(&field.id)
+                    && idx
+                        .index_details
+                        .as_ref()
+                        .is_some_and(|d| d.type_url.ends_with(type_url_suffix))
+            })
+            .map(|idx| idx.name.clone())
+            .collect();
+        if let Some(name) =
+            resolve_single_index(segment_names, &details.maintained_indexes, arm, &column)?
+        {
+            names.push(name);
+        }
+    }
+    names.sort();
+    names.dedup();
+    Ok(names)
 }

 /// Resolve the single logical index from the names of its matching physical
@@ -734,22 +789,117 @@ mod tests {
        };

        // Plain scan: drop every compacted generation (through 5).
-        assert_eq!(exclusion_watermarks(&details, None).get(&shard), Some(&5));
+        assert_eq!(
+            exclusion_watermarks(&details, &[], false).get(&shard),
+            Some(&5)
+        );

        // FTS arm with a lagging index: exclusion is capped at the index catch-up
        // (2), so SSTable generations 3..=5 are retained until the index covers
        // them — otherwise those documents would silently vanish from FTS results.
        assert_eq!(
-            exclusion_watermarks(&details, Some("fts_idx")).get(&shard),
+            exclusion_watermarks(&details, &["fts_idx".to_string()], false).get(&shard),
            Some(&2)
        );

        // A caught-up index — or one untracked in index_catchup — falls back to the
        // compaction watermark.
        assert_eq!(
-            exclusion_watermarks(&details, Some("caught_up_idx")).get(&shard),
+            exclusion_watermarks(&details, &["caught_up_idx".to_string()], false).get(&shard),
            Some(&5)
        );
+
+        // The same missing entry, once the table requires catch-up: absence now
+        // means "not known to hold these rows", so nothing may be excluded and
+        // every generation stays readable from its SSTable. This is the whole
+        // point of the protocol -- an indexed query against a table whose index
+        // has not caught up must not silently lose rows.
+        assert_eq!(
+            exclusion_watermarks(&details, &["untracked_idx".to_string()], true).get(&shard),
+            Some(&0)
+        );
+
+        // A tracked index is unaffected by the mode: the recorded position is
+        // information either way, and it still caps the exclusion.
+        assert_eq!(
+            exclusion_watermarks(&details, &["fts_idx".to_string()], true).get(&shard),
+            Some(&2)
+        );
+
+        // One missing entry is enough to hold everything back, even alongside an
+        // index that has caught up.
+        let mixed = vec!["fts_idx".to_string(), "untracked_idx".to_string()];
+        assert_eq!(
+            exclusion_watermarks(&details, &mixed, true).get(&shard),
+            Some(&0)
+        );
+    }
+
+    /// A hybrid search reads a vector and a full-text index, and either may lag.
+    /// Retaining to the lower of the two is what keeps both arms complete;
+    /// gating on one alone would drop SSTables the other has not indexed.
+    #[test]
+    fn exclusion_watermark_takes_the_minimum_across_every_index_used() {
+        let shard = Uuid::from_u128(1);
+        let details = MemWalIndexDetails {
+            compacted_sstables: vec![CompactedSsTable::new(shard, 9)],
+            index_catchup: vec![
+                IndexCatchupProgress::new(
+                    "vec_idx".to_string(),
+                    vec![CompactedSsTable::new(shard, 7)],
+                ),
+                IndexCatchupProgress::new(
+                    "fts_idx".to_string(),
+                    vec![CompactedSsTable::new(shard, 4)],
+                ),
+            ],
+            maintained_indexes: vec!["vec_idx".to_string(), "fts_idx".to_string()],
+            ..Default::default()
+        };
+
+        // Each index alone stops at its own catch-up.
+        assert_eq!(
+            exclusion_watermarks(&details, &["vec_idx".to_string()], false).get(&shard),
+            Some(&7)
+        );
+        assert_eq!(
+            exclusion_watermarks(&details, &["fts_idx".to_string()], false).get(&shard),
+            Some(&4)
+        );
+
+        // Used together, the lower one governs regardless of order.
+        let both = ["vec_idx".to_string(), "fts_idx".to_string()];
+        assert_eq!(
+            exclusion_watermarks(&details, &both, false).get(&shard),
+            Some(&4)
+        );
+        let reversed = ["fts_idx".to_string(), "vec_idx".to_string()];
+        assert_eq!(
+            exclusion_watermarks(&details, &reversed, false).get(&shard),
+            Some(&4)
+        );
+    }
+
+    /// An index with no catch-up entry contributes no cap today, so a lagging
+    /// sibling must still govern rather than being widened by the untracked one.
+    #[test]
+    fn an_untracked_index_does_not_widen_a_lagging_sibling() {
+        let shard = Uuid::from_u128(1);
+        let details = MemWalIndexDetails {
+            compacted_sstables: vec![CompactedSsTable::new(shard, 9)],
+            index_catchup: vec![IndexCatchupProgress::new(
+                "fts_idx".to_string(),
+                vec![CompactedSsTable::new(shard, 4)],
+            )],
+            maintained_indexes: vec!["fts_idx".to_string(), "untracked_idx".to_string()],
+            ..Default::default()
+        };
+
+        let both = ["fts_idx".to_string(), "untracked_idx".to_string()];
+        assert_eq!(
+            exclusion_watermarks(&details, &both, false).get(&shard),
+            Some(&4)
+        );
    }

    #[test]
@@ -0,0 +1,954 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+//! Filling computed columns.
+//!
+//! A row without a value gets one; a row that has one keeps it. Refresh is
+//! therefore idempotent and does not observe input mutation -- once a row is
+//! filled, changing what the expression reads leaves the stored result alone.
+//!
+//! Two passes per fragment. The first scans only the unfilled live rows and
+//! evaluates the expression over them, which yields the exact fill count and
+//! decides whether the fragment is staged at all -- a fragment where nothing
+//! would change stages nothing, which is what lets an expression yielding
+//! null settle instead of restaging forever. The second streams the
+//! fragment's physical rows into `write_column` a batch at a time, so peak
+//! memory is bounded by a scan batch. The expression is evaluated by this
+//! module, never through a projection alias, and only over rows being
+//! filled: every other row -- deleted, or already holding a value -- has its
+//! inputs masked to null first, so a poison value in a row nobody is filling
+//! cannot fail the refresh.
+
+use std::sync::Arc;
+
+use arrow_array::{ArrayRef, BooleanArray, RecordBatch, RecordBatchOptions};
+use arrow_schema::Schema as ArrowSchema;
+use datafusion_expr::ColumnarValue;
+use futures::{Stream, StreamExt, TryStreamExt};
+use lance::Dataset;
+use lance::dataset::WriteDestination;
+use lance::dataset::fragment::FileFragment;
+use lance::dataset::transaction::Operation;
+use lance_core::ROW_ID;
+use lance_core::datatypes::Schema as LanceSchema;
+use serde::{Deserialize, Serialize};
+
+use super::computed_columns::{BoundExpression, ComputedColumnKind, computed_column_from_field};
+use super::{BaseTable, NativeTable};
+use crate::job::Job;
+use crate::{Error, Result};
+
+/// The result of refreshing a computed column.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
+pub struct RefreshColumnResult {
+    /// Rows that had a value computed.
+    #[serde(default)]
+    pub rows_filled: u64,
+    /// The commit version associated with the operation.
+    #[serde(default)]
+    pub version: u64,
+}
+
+/// Internal implementation of the refresh logic.
+pub(crate) async fn execute_refresh_column(
+    table: &NativeTable,
+    column: &str,
+) -> Result<RefreshColumnResult> {
+    table.dataset.ensure_mutable()?;
+    ensure_no_lsm_write_spec(table).await?;
+    let dataset = table.dataset.get().await?;
+
+    let expression = declared_expression(&dataset, column)?;
+    let schema = Arc::new(ArrowSchema::from(dataset.schema()));
+    let bound = Arc::new(super::computed_columns::bind(schema, column, &expression)?);
+    let field = dataset
+        .schema()
+        .field(column)
+        .ok_or_else(|| Error::ColumnNotFound {
+            name: column.to_string(),
+        })?;
+    // The dataset's own field, so the identity write_column checks against the
+    // manifest holds by construction.
+    let column_schema = LanceSchema {
+        fields: vec![field.clone()],
+        metadata: Default::default(),
+    };
+
+    let mut rows_filled = 0u64;
+    let mut replacements = Vec::new();
+    for fragment in dataset.get_fragments() {
+        let gained = count_fragment_gains(&dataset, &fragment, &bound, column).await?;
+        if gained == 0 {
+            continue;
+        }
+        rows_filled += gained;
+        let values = fill_stream(&dataset, &fragment, bound.clone(), column).await?;
+        replacements.push(fragment.write_column(values, &column_schema).await?);
+    }
+
+    if replacements.is_empty() {
+        return Ok(RefreshColumnResult {
+            rows_filled: 0,
+            version: dataset.version().version,
+        });
+    }
+
+    let read_version = dataset.version().version;
+    // The dataset's own session, so registrations and caches survive the
+    // commit being installed on the handle.
+    let session = dataset.session();
+    let new_dataset = Dataset::commit(
+        WriteDestination::Dataset(dataset.clone()),
+        Operation::DataReplacement { replacements },
+        Some(read_version),
+        None,
+        None,
+        session,
+        false,
+    )
+    .await?;
+
+    let version = new_dataset.version().version;
+    table.dataset.update(new_dataset);
+    Ok(RefreshColumnResult {
+        rows_filled,
+        version,
+    })
+}
+
+/// Run the refresh as a [`Job`] in this process.
+pub(crate) async fn execute_refresh_column_async(table: &NativeTable, column: &str) -> Result<Job> {
+    // Validate before spawning so bad input is reported by this call rather
+    // than only by the job.
+    table.dataset.ensure_mutable()?;
+    ensure_no_lsm_write_spec(table).await?;
+    let dataset = table.dataset.get().await?;
+    declared_expression(&dataset, column)?;
+    drop(dataset);
+
+    let table = table.clone();
+    let column = column.to_string();
+    Ok(Job::spawned(tokio::spawn(async move {
+        execute_refresh_column(&table, &column).await?;
+        table.bump_freshness();
+        Ok(())
+    })))
+}
+
+/// Refuse to refresh under an LSM write spec.
+///
+/// Refresh enumerates base fragments, and a write spec keeps visible rows in
+/// un-compacted MemWAL tiers it cannot reach -- success would silently omit
+/// readable rows.
+async fn ensure_no_lsm_write_spec(table: &NativeTable) -> Result<()> {
+    // The catch-up flag outlives unset and marks retained SSTable rows.
+    let catchup = table.dataset.get().await?.manifest().reader_feature_flags
+        & lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP
+        != 0;
+    if catchup || table.get_lsm_write_spec().await?.is_some() {
+        return Err(Error::NotSupported {
+            message: "refresh_column is not supported on a table with an LSM write \
+                      spec: rows in un-compacted tiers are invisible to refresh"
+                .into(),
+        });
+    }
+    Ok(())
+}
+
+/// The SQL expression `column` is declared with.
+fn declared_expression(dataset: &Dataset, column: &str) -> Result<String> {
+    let schema = ArrowSchema::from(dataset.schema());
+    let field = schema
+        .field_with_name(column)
+        .map_err(|_| Error::ColumnNotFound {
+            name: column.to_string(),
+        })?;
+    let declaration =
+        computed_column_from_field(field).ok_or_else(|| Error::NotAComputedColumn {
+            name: column.to_string(),
+        })?;
+    match declaration.kind {
+        ComputedColumnKind::Sql { expression } => Ok(expression),
+        ComputedColumnKind::Unrecognized { kind } => Err(Error::NotSupported {
+            message: format!(
+                "computed column '{column}' is defined by '{kind}', which this version of \
+                 lancedb cannot evaluate"
+            ),
+        }),
+    }
+}
+
+/// Quote `name` as a lance SQL identifier.
+///
+/// Lance's dialect delimits with backticks, so a double-quoted name would
+/// parse as a string literal rather than a column.
+fn quote_identifier(name: &str) -> String {
+    format!("`{}`", name.replace('`', "``"))
+}
+
+/// Assemble the batch evaluation runs against: the bound roots, in read-schema
+/// order. Built by name so scan-side column order never matters.
+fn evaluation_batch(
+    batch: &RecordBatch,
+    bound: &BoundExpression,
+    mask_out: Option<&BooleanArray>,
+) -> lance_core::Result<RecordBatch> {
+    let mut columns = Vec::with_capacity(bound.roots.len());
+    for name in &bound.roots {
+        let column = batch.column_by_name(name).ok_or_else(|| {
+            lance_core::Error::invalid_input(format!(
+                "refreshing a computed column read no {name} column"
+            ))
+        })?;
+        // Rows outside the mask must not reach the expression: a value in a
+        // deleted or already-filled row can be one it would choke on.
+        columns.push(match mask_out {
+            Some(mask) => arrow::compute::nullif(column, mask)?,
+            None => column.clone(),
+        });
+    }
+    Ok(RecordBatch::try_new_with_options(
+        bound.read_schema.clone(),
+        columns,
+        &RecordBatchOptions::new().with_row_count(Some(batch.num_rows())),
+    )?)
+}
+
+/// Evaluate the expression over `batch`, materializing a constant result to
+/// the batch's length.
+fn evaluate(bound: &BoundExpression, batch: &RecordBatch) -> lance_core::Result<ArrayRef> {
+    let value = bound
+        .physical
+        .evaluate(batch)
+        .map_err(lance_core::Error::from)?;
+    match value {
+        ColumnarValue::Array(array) => Ok(array),
+        scalar => scalar
+            .into_array(batch.num_rows())
+            .map_err(lance_core::Error::from),
+    }
+}
+
+/// How many rows of one fragment would gain a value.
+///
+/// Scans only the unfilled live rows -- deleted rows never reach the
+/// expression here, the filter having already excluded them -- and counts the
+/// non-null results. Exact, so it is both the staging decision and the
+/// fragment's contribution to `rows_filled`.
+async fn count_fragment_gains(
+    dataset: &Dataset,
+    fragment: &FileFragment,
+    bound: &BoundExpression,
+    column: &str,
+) -> Result<u64> {
+    let mut scanner = dataset.scan();
+    scanner
+        .with_fragments(vec![fragment.metadata().clone()])
+        .with_row_id()
+        .filter(&format!("{} IS NULL", quote_identifier(column)))?
+        .project(&bound.roots)?;
+
+    let mut gained = 0u64;
+    let mut batches = scanner.try_into_stream().await?;
+    while let Some(batch) = batches.try_next().await? {
+        let evaluated = evaluate(bound, &evaluation_batch(&batch, bound, None)?)?;
+        gained += (batch.num_rows() - evaluated.null_count()) as u64;
+    }
+    Ok(gained)
+}
+
+/// Stream one fragment's column in physical order, filling the unfilled live
+/// rows and keeping every other value.
+///
+/// Deleted rows are carried through so the values line up positionally with
+/// the fragment's data files; they are never read back, but the column file
+/// has to cover them.
+async fn fill_stream(
+    dataset: &Dataset,
+    fragment: &FileFragment,
+    bound: Arc<BoundExpression>,
+    column: &str,
+) -> Result<impl Stream<Item = lance_core::Result<RecordBatch>> + Send + use<>> {
+    let mut projection: Vec<String> = bound.roots.clone();
+    projection.push(column.to_string());
+    let mut scanner = dataset.scan();
+    scanner
+        .with_fragments(vec![fragment.metadata().clone()])
+        .with_row_id()
+        .include_deleted_rows()
+        .project(&projection)?;
+
+    let projected = Arc::new(ArrowSchema::new(vec![
+        ArrowSchema::from(dataset.schema())
+            .field_with_name(column)
+            .map_err(|_| Error::ColumnNotFound {
+                name: column.to_string(),
+            })?
+            .clone(),
+    ]));
+
+    let column = column.to_string();
+    let batches = scanner.try_into_stream().await?;
+    Ok(batches.map(move |batch| {
+        let batch = batch?;
+        let missing = |name: &str| {
+            lance_core::Error::invalid_input(format!(
+                "refreshing a computed column read no {name} column"
+            ))
+        };
+        let existing = batch
+            .column_by_name(&column)
+            .ok_or_else(|| missing(&column))?;
+        let row_ids = batch
+            .column_by_name(ROW_ID)
+            .ok_or_else(|| missing(ROW_ID))?;
+
+        // Only an unfilled live row gains a value; a deleted row has a null
+        // row id and keeps its (null) slot.
+        let unfilled = arrow::compute::is_null(existing.as_ref())?;
+        let live = arrow::compute::is_not_null(row_ids.as_ref())?;
+        let fill = arrow::compute::and(&unfilled, &live)?;
+        let keep = arrow::compute::not(&fill)?;
+
+        let computed = evaluate(&bound, &evaluation_batch(&batch, &bound, Some(&keep))?)?;
+        let merged = arrow_select::zip::zip(&fill, &computed, existing)?;
+        Ok(RecordBatch::try_new(projected.clone(), vec![merged])?)
+    }))
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use arrow_array::{Int32Array, record_batch};
+    use futures::TryStreamExt;
+
+    use crate::connect;
+    use crate::query::{ExecutableQuery, QueryBase, Select};
+    use crate::{Error, Result, Table};
+
+    async fn table_with(name: &str, values: Vec<i32>) -> Table {
+        let conn = connect("memory://").execute().await.unwrap();
+        let batch = record_batch!(("x", Int32, values)).unwrap();
+        conn.create_table(name, batch).execute().await.unwrap()
+    }
+
+    async fn declare_doubled(table: &Table) -> Result<u64> {
+        Ok(table
+            .add_columns()
+            .computed("doubled", "x * 2")
+            .execute()
+            .await?
+            .version)
+    }
+
+    async fn read(table: &Table, column: &str) -> Vec<Option<i32>> {
+        let batches = table
+            .query()
+            .select(Select::columns(&[column]))
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let mut values: Vec<Option<i32>> = batches
+            .iter()
+            .flat_map(|batch| {
+                batch[column]
+                    .as_any()
+                    .downcast_ref::<Int32Array>()
+                    .unwrap()
+                    .iter()
+                    .collect::<Vec<_>>()
+            })
+            .collect();
+        values.sort();
+        values
+    }
+
+    async fn append(table: &Table, values: Vec<i32>) {
+        let batch = record_batch!(("x", Int32, values)).unwrap();
+        table.add(batch).execute().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_refresh_fills_a_declared_column() {
+        let table = table_with("refresh_fills", vec![1, 2, 3]).await;
+        let declared = declare_doubled(&table).await.unwrap();
+        assert_eq!(read(&table, "doubled").await, vec![None, None, None]);
+
+        let result = table.refresh_column("doubled").await.unwrap();
+        assert!(result.version > declared);
+        assert_eq!(result.rows_filled, 3);
+        assert_eq!(
+            read(&table, "doubled").await,
+            vec![Some(2), Some(4), Some(6)]
+        );
+    }
+
+    /// Values written after the last refresh must be reachable by another one.
+    #[tokio::test]
+    async fn test_refresh_fills_rows_appended_since_the_last_refresh() {
+        let table = table_with("refresh_appended", vec![1, 2]).await;
+        declare_doubled(&table).await.unwrap();
+        table.refresh_column("doubled").await.unwrap();
+
+        append(&table, vec![5, 6]).await;
+        assert_eq!(
+            read(&table, "doubled").await,
+            vec![None, None, Some(2), Some(4)]
+        );
+
+        let result = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(result.rows_filled, 2);
+        assert_eq!(
+            read(&table, "doubled").await,
+            vec![Some(2), Some(4), Some(10), Some(12)]
+        );
+    }
+
+    #[tokio::test]
+    async fn test_refresh_with_nothing_to_fill() {
+        let table = table_with("refresh_noop", vec![1, 2, 3]).await;
+        declare_doubled(&table).await.unwrap();
+        table.refresh_column("doubled").await.unwrap();
+
+        let again = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(again.rows_filled, 0);
+        assert_eq!(
+            read(&table, "doubled").await,
+            vec![Some(2), Some(4), Some(6)]
+        );
+    }
+
+    /// A row is filled only by gaining a value, so an expression yielding null
+    /// settles at once instead of re-selecting the same rows forever. Nothing
+    /// is staged, so the version does not move either.
+    #[tokio::test]
+    async fn test_refresh_converges_on_a_null_result() {
+        let table = table_with("refresh_null_result", vec![1, 2, 3]).await;
+        let declared = table
+            .add_columns()
+            .computed("maybe", "nullif(x, x)")
+            .execute()
+            .await
+            .unwrap()
+            .version;
+
+        let first = table.refresh_column("maybe").await.unwrap();
+        assert_eq!(first.rows_filled, 0);
+        assert_eq!(first.version, declared);
+        assert_eq!(read(&table, "maybe").await, vec![None, None, None]);
+
+        let again = table.refresh_column("maybe").await.unwrap();
+        assert_eq!(again.rows_filled, 0);
+        assert_eq!(again.version, declared);
+    }
+
+    /// The contract's boundary: a filled fragment is not revisited, so
+    /// mutating an input leaves the value computed at fill time.
+    #[tokio::test]
+    async fn test_refresh_does_not_observe_input_mutation() {
+        let table = table_with("refresh_mutation", vec![1]).await;
+        declare_doubled(&table).await.unwrap();
+        table.refresh_column("doubled").await.unwrap();
+        assert_eq!(read(&table, "doubled").await, vec![Some(2)]);
+
+        table.update().column("x", "3").execute().await.unwrap();
+
+        let again = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(again.rows_filled, 0);
+        assert_eq!(read(&table, "doubled").await, vec![Some(2)]);
+    }
+
+    /// A row rewrite before the first refresh materializes the declared
+    /// column as null behind a covering data file. Those rows are still
+    /// unfilled and a later refresh has to reach them.
+    #[tokio::test]
+    async fn test_update_before_the_first_refresh() {
+        let table = table_with("refresh_update_first", vec![1]).await;
+        declare_doubled(&table).await.unwrap();
+
+        table.update().column("x", "3").execute().await.unwrap();
+
+        let result = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        assert_eq!(read(&table, "doubled").await, vec![Some(6)]);
+    }
+
+    /// The contract holds row by row, not fragment by fragment: revisiting a
+    /// fragment to fill one row must not recompute a filled row sitting beside
+    /// it, even where the input behind it has since changed.
+    #[tokio::test]
+    async fn test_refresh_does_not_recompute_a_filled_row_beside_an_unfilled_one() {
+        let table = table_with("refresh_mixed", vec![1, 2]).await;
+        declare_doubled(&table).await.unwrap();
+        table.refresh_column("doubled").await.unwrap();
+
+        append(&table, vec![5]).await;
+        table
+            .update()
+            .column("x", "100")
+            .only_if("x = 1")
+            .execute()
+            .await
+            .unwrap();
+        table
+            .optimize(crate::table::OptimizeAction::Compact {
+                options: crate::table::CompactionOptions::default(),
+                remap_options: None,
+            })
+            .await
+            .unwrap();
+
+        let result = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        // 2 is the mutated row keeping the value it was filled with, not 200.
+        assert_eq!(
+            read(&table, "doubled").await,
+            vec![Some(2), Some(4), Some(10)]
+        );
+    }
+
+    /// Filling a fragment must not disturb the values it already holds, which
+    /// is what makes a compaction-mixed fragment safe to revisit.
+    #[tokio::test]
+    async fn test_refresh_preserves_already_filled_rows() {
+        let table = table_with("refresh_preserves", vec![1, 2]).await;
+        declare_doubled(&table).await.unwrap();
+        table.refresh_column("doubled").await.unwrap();
+
+        append(&table, vec![5]).await;
+        table
+            .optimize(crate::table::OptimizeAction::Compact {
+                options: crate::table::CompactionOptions::default(),
+                remap_options: None,
+            })
+            .await
+            .unwrap();
+
+        let result = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        assert_eq!(
+            read(&table, "doubled").await,
+            vec![Some(2), Some(4), Some(10)]
+        );
+    }
+
+    #[tokio::test]
+    async fn test_refresh_leaves_deleted_rows_alone() {
+        let table = table_with("refresh_deleted", vec![1, 2, 3, 4]).await;
+        declare_doubled(&table).await.unwrap();
+        table.delete("x = 2").await.unwrap();
+
+        let result = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(result.rows_filled, 3);
+        assert_eq!(
+            read(&table, "doubled").await,
+            vec![Some(2), Some(6), Some(8)]
+        );
+    }
+
+    #[tokio::test]
+    async fn test_refresh_a_constant_expression() {
+        let table = table_with("refresh_constant", vec![1, 2, 3]).await;
+        table
+            .add_columns()
+            .computed("answer", "42")
+            .execute()
+            .await
+            .unwrap();
+
+        let result = table.refresh_column("answer").await.unwrap();
+        assert_eq!(result.rows_filled, 3);
+    }
+
+    /// A name needing quotes reaches the evaluator intact: it is carried as a
+    /// projection alias, never spliced into SQL text.
+    #[tokio::test]
+    async fn test_refresh_a_column_whose_name_needs_quoting() {
+        let table = table_with("refresh_quoted", vec![1, 2, 3]).await;
+        table
+            .add_columns()
+            .computed("double value", "x * 2")
+            .execute()
+            .await
+            .unwrap();
+
+        let result = table.refresh_column("double value").await.unwrap();
+        assert_eq!(result.rows_filled, 3);
+        assert_eq!(
+            read(&table, "double value").await,
+            vec![Some(2), Some(4), Some(6)]
+        );
+    }
+
+    /// A fragment spanning several scan batches exercises the streamed fill:
+    /// the probe buffers only until the first gained value and the rest flows
+    /// through write_column a batch at a time.
+    #[tokio::test]
+    async fn test_refresh_streams_a_multi_batch_fragment() {
+        let values: Vec<i32> = (0..20_000).collect();
+        let table = table_with("refresh_multi_batch", values.clone()).await;
+        declare_doubled(&table).await.unwrap();
+
+        let result = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(result.rows_filled, 20_000);
+
+        let read_back = read(&table, "doubled").await;
+        assert_eq!(read_back.len(), 20_000);
+        let mut expected: Vec<Option<i32>> = values.iter().map(|v| Some(v * 2)).collect();
+        expected.sort();
+        assert_eq!(read_back, expected);
+    }
+
+    /// The gate's reproducer: the commit must reuse the configured session,
+    /// or registrations and caches vanish from the handle after a refresh.
+    #[tokio::test]
+    async fn test_refresh_preserves_the_configured_session() {
+        let session = Arc::new(lance::session::Session::default());
+        let conn = crate::connect("memory://")
+            .session(session.clone())
+            .execute()
+            .await
+            .unwrap();
+        let batch = record_batch!(("x", Int32, [1, 2])).unwrap();
+        let table = conn
+            .create_table("session_kept", batch)
+            .execute()
+            .await
+            .unwrap();
+        declare_doubled(&table).await.unwrap();
+        table.refresh_column("doubled").await.unwrap();
+
+        let dataset = table.as_native().unwrap().dataset.get().await.unwrap();
+        assert!(Arc::ptr_eq(&dataset.session(), &session));
+    }
+
+    /// The async form's job settles with the fill visible, like
+    /// create_index's execute_async.
+    #[tokio::test]
+    async fn test_refresh_async_job_waits_for_the_fill() {
+        let table = table_with("refresh_async", vec![1, 2, 3]).await;
+        declare_doubled(&table).await.unwrap();
+
+        let job = table.refresh_column_async("doubled").await.unwrap();
+        assert!(job.id().is_none(), "in-process jobs have no server id");
+        job.wait().await.unwrap();
+        assert_eq!(job.status().await.unwrap(), "finished");
+        assert_eq!(
+            read(&table, "doubled").await,
+            vec![Some(2), Some(4), Some(6)]
+        );
+    }
+
+    /// Bad input is reported by the call, not by the job.
+    #[tokio::test]
+    async fn test_refresh_async_rejects_bad_input_before_spawning() {
+        let table = table_with("refresh_async_bad", vec![1, 2, 3]).await;
+
+        let err = table.refresh_column_async("x").await.unwrap_err();
+        assert!(matches!(err, Error::NotAComputedColumn { name } if name == "x"));
+
+        let err = table.refresh_column_async("nope").await.unwrap_err();
+        assert!(matches!(err, Error::ColumnNotFound { name } if name == "nope"));
+    }
+
+    #[tokio::test]
+    async fn test_refresh_async_job_reports_success_to_every_waiter() {
+        let table = table_with("refresh_async_waiters", vec![1, 2]).await;
+        declare_doubled(&table).await.unwrap();
+
+        let job = table.refresh_column_async("doubled").await.unwrap();
+        job.wait().await.unwrap();
+        // A second wait after completion observes the same outcome.
+        job.wait().await.unwrap();
+        assert_eq!(job.status().await.unwrap(), "finished");
+    }
+
+    #[tokio::test]
+    async fn test_refresh_rejects_a_plain_column() {
+        let table = table_with("refresh_plain", vec![1, 2, 3]).await;
+        let err = table.refresh_column("x").await.unwrap_err();
+        assert!(matches!(err, Error::NotAComputedColumn { name } if name == "x"));
+    }
+
+    #[tokio::test]
+    async fn test_refresh_rejects_an_unknown_column() {
+        let table = table_with("refresh_missing", vec![1, 2, 3]).await;
+        let err = table.refresh_column("nope").await.unwrap_err();
+        assert!(matches!(err, Error::ColumnNotFound { name } if name == "nope"));
+    }
+
+    /// The gate's reproducer: a poison value in a deleted row must not
+    /// abort filling the live rows, since nobody can read it.
+    #[tokio::test]
+    async fn test_a_deleted_rows_value_is_never_evaluated() {
+        let table = table_with("refresh_deleted_poison", vec![1, 0]).await;
+        table
+            .add_columns()
+            .computed("quotient", "10 / x")
+            .execute()
+            .await
+            .unwrap();
+        table.delete("x = 0").await.unwrap();
+
+        let result = table.refresh_column("quotient").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        assert_eq!(read(&table, "quotient").await, vec![Some(10)]);
+    }
+
+    /// The gate's reproducer: an already-filled row's value must not be
+    /// re-evaluated either -- its input may have mutated into one the
+    /// expression chokes on.
+    #[tokio::test]
+    async fn test_a_filled_rows_value_is_never_evaluated() {
+        let table = table_with("refresh_filled_poison", vec![1, 2]).await;
+        table
+            .add_columns()
+            .computed("quotient", "10 / x")
+            .execute()
+            .await
+            .unwrap();
+        table.refresh_column("quotient").await.unwrap();
+
+        table
+            .update()
+            .column("x", "0")
+            .only_if("x = 1")
+            .execute()
+            .await
+            .unwrap();
+        append(&table, vec![5]).await;
+
+        let result = table.refresh_column("quotient").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        assert_eq!(
+            read(&table, "quotient").await,
+            vec![Some(2), Some(5), Some(10)]
+        );
+    }
+
+    /// The gate's reproducer: the old internal projection alias is an
+    /// ordinary column name; a computed column may use it.
+    #[tokio::test]
+    async fn test_refresh_a_column_named_like_the_old_alias() {
+        let table = table_with("refresh_alias_name", vec![1, 2]).await;
+        table
+            .add_columns()
+            .computed("__lancedb_computed", "x * 2")
+            .execute()
+            .await
+            .unwrap();
+
+        let result = table.refresh_column("__lancedb_computed").await.unwrap();
+        assert_eq!(result.rows_filled, 2);
+        assert_eq!(
+            read(&table, "__lancedb_computed").await,
+            vec![Some(2), Some(4)]
+        );
+    }
+
+    /// The gate's reproducer: a late-gain fragment (filled, then one null row
+    /// compacted onto the end) fills without the old probe's buffering, which
+    /// this pins behaviorally; the memory bound is structural -- the fill
+    /// stream retains no batches at all.
+    #[tokio::test]
+    async fn test_refresh_fills_a_late_gain_fragment() {
+        let values: Vec<i32> = (0..20_000).collect();
+        let table = table_with("refresh_late_gain", values).await;
+        declare_doubled(&table).await.unwrap();
+        table.refresh_column("doubled").await.unwrap();
+
+        append(&table, vec![2_000_000]).await;
+        table
+            .optimize(crate::table::OptimizeAction::Compact {
+                options: crate::table::CompactionOptions::default(),
+                remap_options: None,
+            })
+            .await
+            .unwrap();
+
+        let result = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        let read_back = read(&table, "doubled").await;
+        assert_eq!(read_back.len(), 20_001);
+        assert_eq!(read_back.last().unwrap(), &Some(4_000_000));
+    }
+
+    /// The gate's reproducer: a nested input declares, refreshes, and guards
+    /// its root against invalidating schema changes.
+    #[tokio::test]
+    async fn test_a_nested_input_declares_and_refreshes() {
+        use arrow_array::{Int32Array, StructArray};
+        use arrow_schema::{DataType, Field, Fields};
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let age = Arc::new(Int32Array::from(vec![30, 40]));
+        let fields = Fields::from(vec![Field::new("age", DataType::Int32, true)]);
+        let metadata = StructArray::new(fields.clone(), vec![age as _], None);
+        let schema = Arc::new(arrow_schema::Schema::new(vec![Field::new(
+            "metadata",
+            DataType::Struct(fields),
+            true,
+        )]));
+        let batch =
+            arrow_array::RecordBatch::try_new(schema, vec![Arc::new(metadata) as _]).unwrap();
+        let table = conn
+            .create_table("refresh_nested", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        table
+            .add_columns()
+            .computed("next_age", "metadata.age + 1")
+            .execute()
+            .await
+            .unwrap();
+        let declaration =
+            &crate::table::computed_columns(table.schema().await.unwrap().as_ref())[0];
+        assert_eq!(declaration.inputs, vec!["metadata.age".to_string()]);
+
+        let result = table.refresh_column("next_age").await.unwrap();
+        assert_eq!(result.rows_filled, 2);
+        assert_eq!(read(&table, "next_age").await, vec![Some(31), Some(41)]);
+
+        // The dotted input guards its root.
+        let err = table.drop_columns(&["metadata"]).await.unwrap_err();
+        assert!(
+            matches!(&err, Error::InvalidInput { message } if message.contains("next_age")),
+            "{err:?}"
+        );
+
+        // Masking a struct input for a deleted row goes through the same
+        // nullif path as a primitive; a nested input plus deletions must not
+        // be the combination that breaks it.
+        table.delete("next_age = 31").await.unwrap();
+        append_struct_row(&table, 50).await;
+        let result = table.refresh_column("next_age").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        assert_eq!(read(&table, "next_age").await, vec![Some(41), Some(51)]);
+    }
+
+    /// Append one `metadata: {age}` row to the nested-input table.
+    async fn append_struct_row(table: &Table, age: i32) {
+        use arrow_array::{Int32Array, StructArray};
+        use arrow_schema::{DataType, Field, Fields};
+
+        let ages = Arc::new(Int32Array::from(vec![age]));
+        let fields = Fields::from(vec![Field::new("age", DataType::Int32, true)]);
+        let metadata = StructArray::new(fields.clone(), vec![ages as _], None);
+        let schema = Arc::new(arrow_schema::Schema::new(vec![Field::new(
+            "metadata",
+            DataType::Struct(fields),
+            true,
+        )]));
+        let batch =
+            arrow_array::RecordBatch::try_new(schema, vec![Arc::new(metadata) as _]).unwrap();
+        table.add(batch).execute().await.unwrap();
+    }
+
+    /// Both orders of declare+spec are refused at the source (see the
+    /// schema_evolution tests); refresh's own check covers a dataset another
+    /// writer left in that state.
+    #[tokio::test]
+    async fn test_refresh_refuses_a_foreign_lsm_state() {
+        use crate::table::LsmWriteSpec;
+
+        let tmp_dir = tempfile::tempdir().unwrap();
+        let conn = connect(tmp_dir.path().to_str().unwrap())
+            .execute()
+            .await
+            .unwrap();
+        let schema = Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+            "x",
+            arrow_schema::DataType::Int32,
+            false,
+        )]));
+        let batch =
+            arrow_array::RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1]))])
+                .unwrap();
+        let table = conn.create_table("lsm", batch).execute().await.unwrap();
+        table.set_unenforced_primary_key(["x"]).await.unwrap();
+        table
+            .set_lsm_write_spec(LsmWriteSpec::unsharded())
+            .await
+            .unwrap();
+        super::super::computed_columns::add_foreign_kind(&table, "doubled", "sql").await;
+
+        let err = table.refresh_column("doubled").await.unwrap_err();
+        assert!(
+            matches!(&err, Error::NotSupported { message } if message.contains("LSM")),
+            "{err:?}"
+        );
+        let err = table.refresh_column_async("doubled").await.unwrap_err();
+        assert!(matches!(err, Error::NotSupported { .. }));
+    }
+
+    /// After catch-up activation and unset, no spec remains but the catch-up
+    /// flag still marks retained SSTable rows; refresh refuses on the flag.
+    #[tokio::test]
+    async fn test_refresh_refuses_retained_catchup_state() {
+        use crate::table::LsmWriteSpec;
+
+        let tmp_dir = tempfile::tempdir().unwrap();
+        let conn = connect(tmp_dir.path().to_str().unwrap())
+            .execute()
+            .await
+            .unwrap();
+        let schema = Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+            "x",
+            arrow_schema::DataType::Int32,
+            false,
+        )]));
+        let batch = arrow_array::RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from(vec![1]))],
+        )
+        .unwrap();
+        let table = conn
+            .create_table("catchup", batch.clone())
+            .execute()
+            .await
+            .unwrap();
+        table.set_unenforced_primary_key(["x"]).await.unwrap();
+        table
+            .set_lsm_write_spec(LsmWriteSpec::unsharded())
+            .await
+            .unwrap();
+        table.require_mem_wal_index_catchup().await.unwrap();
+        let mut merge = table.merge_insert(&["x"]);
+        merge
+            .when_matched_update_all(None)
+            .when_not_matched_insert_all()
+            .use_lsm(true);
+        merge
+            .execute(Box::new(arrow_array::RecordBatchIterator::new(
+                vec![Ok(batch)],
+                schema,
+            )))
+            .await
+            .unwrap();
+        table.unset_lsm_write_spec().await.unwrap();
+        super::super::computed_columns::add_foreign_kind(&table, "doubled", "sql").await;
+
+        let err = table.refresh_column("doubled").await.unwrap_err();
+        assert!(
+            matches!(&err, Error::NotSupported { message } if message.contains("LSM")),
+            "{err:?}"
+        );
+    }
+
+    /// A declaration of a kind this version cannot evaluate is refused by
+    /// name, rather than mistaken for a plain column or fed to the SQL path.
+    #[tokio::test]
+    async fn test_refresh_rejects_a_kind_it_cannot_evaluate() {
+        let table = table_with("refresh_foreign", vec![1, 2, 3]).await;
+        super::super::computed_columns::add_foreign_kind(&table, "embedding", "udf").await;
+
+        let err = table.refresh_column("embedding").await.unwrap_err();
+        assert!(matches!(err, Error::NotSupported { message } if message.contains("udf")));
+    }
+}
@@ -8,12 +8,14 @@
 //! - [`alter_columns`](execute_alter_columns): Rename columns, change types, or modify nullability
 //! - [`drop_columns`](execute_drop_columns): Remove columns from the table

+use arrow_schema::Schema as ArrowSchema;
 use lance::dataset::{ColumnAlteration, NewColumnTransform};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;

-use super::NativeTable;
-use crate::Result;
+use super::computed_columns;
+use super::{BaseTable, NativeTable};
+use crate::{Error, Result};

 /// The result of an add columns operation.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
@@ -98,6 +100,48 @@ pub(crate) async fn execute_add_columns(
    table: &NativeTable,
    transforms: NewColumnTransform,
    read_columns: Option<Vec<String>>,
+) -> Result<AddColumnsResult> {
+    // Declarations are admitted only through [`execute_declare`].
+    match &transforms {
+        NewColumnTransform::AllNulls(schema) => {
+            computed_columns::ensure_no_foreign_declarations(schema.fields())?
+        }
+        NewColumnTransform::BatchUDF(udf) => {
+            computed_columns::ensure_no_foreign_declarations(udf.output_schema.fields())?
+        }
+        _ => {}
+    }
+    commit_add_columns(table, transforms, read_columns).await
+}
+
+/// Declare validated computed columns. The only admission path for
+/// declaration metadata.
+pub(crate) async fn execute_declare(
+    table: &NativeTable,
+    columns: &[(String, String)],
+) -> Result<AddColumnsResult> {
+    // An LSM write spec keeps visible rows in tiers refresh cannot reach;
+    // checked against latest committed state, not this handle's snapshot.
+    // The catch-up flag outlives unset and marks retained SSTable rows.
+    table.checkout_latest().await?;
+    let catchup = table.dataset.get().await?.manifest().reader_feature_flags
+        & lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP
+        != 0;
+    if catchup || table.get_lsm_write_spec().await?.is_some() {
+        return Err(Error::NotSupported {
+            message: "computed columns are not supported on a table with an LSM write \
+                      spec: rows in un-compacted tiers are invisible to refresh"
+                .into(),
+        });
+    }
+    let transform = computed_columns::declare(table.schema().await?, columns)?;
+    commit_add_columns(table, transform, None).await
+}
+
+pub(crate) async fn commit_add_columns(
+    table: &NativeTable,
+    transforms: NewColumnTransform,
+    read_columns: Option<Vec<String>>,
 ) -> Result<AddColumnsResult> {
    table.dataset.ensure_mutable()?;
    let mut dataset = (*table.dataset.get().await?).clone();
@@ -116,6 +160,21 @@ pub(crate) async fn execute_alter_columns(
 ) -> Result<AlterColumnsResult> {
    table.dataset.ensure_mutable()?;
    let mut dataset = (*table.dataset.get().await?).clone();
+    // Nullability is not part of what an expression resolves against, so only
+    // a rename or a retype can invalidate a binding.
+    let schema = std::sync::Arc::new(ArrowSchema::from(dataset.schema()));
+    let rebinding = alterations
+        .iter()
+        .filter(|alteration| alteration.rename.is_some() || alteration.data_type.is_some())
+        .map(|alteration| alteration.path.as_str())
+        .collect::<Vec<_>>();
+    computed_columns::ensure_not_an_input(&schema, &rebinding)?;
+    let retyped = alterations
+        .iter()
+        .filter(|alteration| alteration.data_type.is_some())
+        .map(|alteration| alteration.path.as_str())
+        .collect::<Vec<_>>();
+    computed_columns::ensure_not_retyped(schema.as_ref(), &retyped)?;
    dataset.alter_columns(alterations).await?;
    let version = dataset.version().version;
    table.dataset.update(dataset);
@@ -131,6 +190,10 @@ pub(crate) async fn execute_drop_columns(
 ) -> Result<DropColumnsResult> {
    table.dataset.ensure_mutable()?;
    let mut dataset = (*table.dataset.get().await?).clone();
+    computed_columns::ensure_not_an_input(
+        &std::sync::Arc::new(ArrowSchema::from(dataset.schema())),
+        columns,
+    )?;
    dataset.drop_columns(columns).await?;
    let version = dataset.version().version;
    table.dataset.update(dataset);
@@ -147,6 +210,44 @@ pub(crate) async fn execute_update_field_metadata(
    table.dataset.ensure_mutable()?;
    let mut dataset = (*table.dataset.get().await?).clone();

+    // A declaration is validated as a whole at declare time; editing its keys
+    // here would bypass that, fabricate one on a plain column, or move a
+    // binding out from under a refresh. A replace on a declared column would
+    // silently erase it.
+    let schema = ArrowSchema::from(dataset.schema());
+    let declared: Vec<String> = computed_columns::computed_columns(&schema)
+        .into_iter()
+        .map(|declaration| declaration.name)
+        .collect();
+    for update in updates {
+        if update
+            .metadata
+            .keys()
+            .any(|key| computed_columns::is_declaration_key(key))
+        {
+            return Err(Error::InvalidInput {
+                message: format!(
+                    "metadata keys of a computed-column declaration cannot be edited \
+                     (path '{}'); drop the column and declare it again",
+                    update.path
+                ),
+            });
+        }
+        if update.replace
+            && declared
+                .iter()
+                .any(|name| name == computed_columns::root(&update.path))
+        {
+            return Err(Error::InvalidInput {
+                message: format!(
+                    "replacing all metadata of computed column '{}' would erase its \
+                     declaration; drop the column and declare it again",
+                    update.path
+                ),
+            });
+        }
+    }
+
    let mut builder = dataset.update_field_metadata();
    for update in updates {
        let entries = update.metadata.iter().map(|(k, v)| (k.clone(), v.clone()));
@@ -82,6 +82,10 @@ pub(crate) async fn execute_update(

    // 1. Snapshot the current dataset
    let dataset = table.dataset.get().await?;
+    super::computed_columns::ensure_not_written(
+        &arrow_schema::Schema::from(dataset.schema()),
+        update.columns.iter().map(|(name, _)| name.as_str()),
+    )?;

    // 2. Initialize the Lance Core builder
    let mut builder = LanceUpdateBuilder::new(dataset);
Author	SHA1	Message	Date
Lance Release	d3a7d27d36	Bump version: 0.38.0-beta.1 → 0.38.0-beta.2	2026-08-19 01:58:48 +00:00
Dan Rammer	cdebea118d	feat(python): expose LSM checkpoint and stats on sync RemoteTable (#3961 ) ## Summary The sync `RemoteTable` carried `set_lsm_write_spec`, `unset_lsm_write_spec`, `get_lsm_write_spec`, and `close_lsm_writers`, but not `checkpoint_lsm`, `flush_lsm`, `compact_lsm`, or `get_lsm_stats`. That left the four LSM control methods reachable from `AsyncTable` only. They are also the four that only work against a remote table — `NativeTable` does not override the `BaseTable` defaults, so on a local table they return `NotSupported` (`rust/lancedb/src/table.rs:679-701`). The net effect for sync users: \| \| `checkpoint_lsm` / `get_lsm_stats` \| \|---\|---\| \| `LanceTable` (sync, local) \| present, but always `NotSupported` \| \| `RemoteTable` (sync, remote) \| `AttributeError` — method absent \| \| `AsyncTable` (remote) \| works \| So there was no working sync path at all, despite the Rust `RemoteTable` implementing every one of these against real endpoints. ## Changes * Add `checkpoint_lsm`, `flush_lsm`, `compact_lsm`, and `get_lsm_stats` to `lancedb.remote.table.RemoteTable`, mirroring the delegation style of their neighbours. * Correct the docstrings on `set_lsm_write_spec` / `unset_lsm_write_spec`, which read `"""Not supported on LanceDB Cloud."""` although `rust/lancedb/src/remote/table.rs:2549-2601` implements both against `/v1/table/{}/set_lsm_write_spec/` and `/unset_lsm_write_spec/`. They appear to have been copy-pasted from `set_unenforced_primary_key` directly above. No Rust or PyO3 changes — the bindings and the `AsyncTable` methods already existed. The `Table` ABC is left alone, matching how the existing `_lsm_write_spec` methods are declared on the concrete classes only. ## Tests Four new tests in `python/python/tests/test_remote_db.py`, against the existing mock HTTP server: `test_get_lsm_stats_sync` — the server payload round-trips into the dict, and `include_generation_rows` defaults to `False` and is forwarded when set. * `test_get_lsm_stats_sync_returns_none_when_lsm_disabled` — a `{"lsm_stats": null}` envelope yields `None` rather than an error. * `test_flush_and_compact_lsm_sync` — both are one-shot POSTs answered `202` with no body. * `test_checkpoint_lsm_sync` — pins the binding to the endpoints it drives (`flush_lsm` then `get_lsm_stats`); the convergence loop itself is already covered in Rust. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>	2026-08-18 17:25:08 -05:00
Xuanwo	76942306b7	docs(java): add vended credentials example (#3958 ) ## Context Java users opening catalog-backed tables with vended credentials currently lack a documented workflow. Opening the catalog-returned URI directly drops the namespace-provided storage options and automatic credential refresh. Document the namespace-backed `Dataset.open()` path so temporary object store credentials are applied and refreshed transparently.	2026-08-18 20:03:20 +08:00
Adityaj0	d742b174c4	fix: hybrid search silently ignores .offset() (#3769 ) ## Summary `LanceHybridQueryBuilder` (sync hybrid search, `table.search(query_type="hybrid")`) silently ignored `.offset()`. `self._offset` was never forwarded to the vector/FTS sub-queries and never applied when slicing the final combined/reranked result, so `.offset(N)` behaved identically to `.offset(0)` — no error, just wrong pagination. Fixes #3765 ## Changes - `_create_query_builders()`: each sub-query now fetches `limit + offset` rows so there's enough data to slice the correct window out of after combining/reranking. - `_combine_hybrid_results()` / `to_arrow()`: the final table is sliced with `offset=self._offset` instead of always starting at 0. ## Test plan - [x] New regression test `test_hybrid_query_offset` in `python/python/tests/test_hybrid_query.py` - [x] `uv run --extra tests pytest python/tests/test_hybrid_query.py -vv` — 13 passed - [x] `uv run --extra dev ruff format` / `ruff check` — clean Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com> Co-authored-by: Will Jones <willjones127@gmail.com>	2026-08-17 11:38:38 -07:00
Igor Ganapolsky	a075aa62f8	fix(python): treat naive lit(datetime) as UTC wall clock (#3262 ) (#3775 ) ## Summary Fixes naive `lit(datetime)` equality filters against table timestamp columns on non-UTC hosts, and adds the integration matrix from #3262. ## Failure (before) On a machine in US Eastern (UTC−4 / EDT), with PyPI `lancedb==0.36.0`: ```python from datetime import datetime import lancedb from lancedb.expr import col, lit db = lancedb.connect("memory://") ts = datetime(2024, 7, 1, 10, 0, 0) # naive table = db.create_table("t", [{"id": 1, "ts": ts}]) rows = table.search().where(col("ts") == lit(ts)).to_list() # actual: [] (0 rows) # expected: 1 row ``` ### Root cause In `python/src/expr.rs`, `expr_lit` converted every `datetime` via Python's `.timestamp()`: - naive `.timestamp()` = local wall → UTC epoch (shifted by host offset) - PyArrow naive storage = UTC wall-clock microseconds (no local shift) So `lit(naive)` became `CAST('2024-07-01 14:00:00' AS TIMESTAMP)` on EDT while the table held `10:00:00`. ## After Naive datetimes are interpreted as UTC wall clock (`replace(tzinfo=timezone.utc).timestamp()`), matching Arrow storage. Aware datetimes still use `.timestamp()` (correct epoch). Same repro on this branch: 1 matching row. ## Tests Added `TestExprDatetimeTimezoneIntegration` covering: \| Case \| Result \| \|------\|--------\| \| both naive \| match \| \| both same TZ (UTC) \| match \| \| different TZs, same instant \| match \| \| table TZ + naive lit \| match (wall clock) \| \| table naive + aware lit \| match \| \| naive lit SQL is wall clock, not local-shifted \| asserts `10:00:00` in SQL \| ### Verification ```bash cd python maturin develop pytest python/tests/test_expr.py -v ``` 102 passed (full `test_expr.py`, including the 6 new cases). Closes #3262 --------- Co-authored-by: Will Jones <willjones127@gmail.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>	2026-08-17 10:48:02 -07:00
Lance Release	040a4120c8	Bump version: 0.38.0-beta.0 → 0.38.0-beta.1	2026-08-17 16:56:54 +00:00
Wyatt Alt	928c3dde2d	feat: computed columns on remote tables (#3941 ) LanceDB Cloud and Enterprise support computed columns through the REST API, so declaration dispatches per backend: local tables plan the expression themselves, remote ones send {name, computed} entries for the server to plan. A remote refresh is the server's backfill job -- refresh_column_async submits it and returns a handle whose successful wait establishes a read-freshness baseline on the submitting handle, unless a checkout has pinned the handle by the time the job completes; the blocking form refuses rather than invent a fill count the server does not report. Declaration entries are built from the namespace client's AddColumnsEntry model (lance-namespace 0.11.0, via the lance beta.13 pin), so the payload shape is compile-checked against the published contract. --- <sub>Stack created with <a href="https://github.com/github/gh-stack">GitHub Stacks CLI</a> • <a href="https://gh.io/stacks-feedback">Give Feedback 💬</a></sub>	2026-08-14 17:21:55 -07:00
LanceDB Robot	980818df26	chore: update lance dependency to v11.0.0-beta.13 (#3947 ) Updates the Lance Rust workspace dependencies and Java lance-core dependency to [v11.0.0-beta.13](https://github.com/lance-format/lance/releases/tag/v11.0.0-beta.13). Adds the required `ListTablesResponse.context` compatibility field and validates the workspace with Clippy warnings denied.	2026-08-14 16:20:39 -07:00
Wyatt Alt	c429863122	feat: refresh_column_async returns a job handle (#3939 ) Mirrors create_index's dual surface: the blocking refresh_column keeps returning {rows_filled, version}, and refresh_column_async returns the same Job handle create_index uses, running the refresh as an in-process task. Invalid input is reported by the submitting call rather than by the job. --- <sub>Stack created with <a href="https://github.com/github/gh-stack">GitHub Stacks CLI</a> • <a href="https://gh.io/stacks-feedback">Give Feedback 💬</a></sub>	2026-08-14 16:05:04 -07:00
Wyatt Alt	fc0d917d32	feat: refresh computed columns (#3938 ) table.refresh_column("doubled") fills the rows of a declared column that hold no value, in two passes per fragment: the first scans only the unfilled live rows to count exact gains and decide staging, the second streams the fragment's physical rows into a standalone column file published in one DataReplacement -- committed under the dataset's own session -- so peak memory is bounded by a scan batch. A row that holds a value keeps it; deleted and already-filled rows never reach the expression, so a poison value in them cannot fail the refresh. Refresh refuses under an LSM write spec, including the mem-wal catch-up flag that outlives unset and marks retained SSTable rows. --- <sub>Stack created with <a href="https://github.com/github/gh-stack">GitHub Stacks CLI</a> • <a href="https://gh.io/stacks-feedback">Give Feedback 💬</a></sub>	2026-08-14 14:43:41 -07:00
Wyatt Alt	def869bb78	feat: declare computed columns by SQL expression (#3937 ) add_columns().computed("doubled", "x * 2") stores the expression in field metadata and commits the column empty; a later refresh fills it. Type and inputs are derived from the expression. The declaration stays authoritative for its lifetime: writes that would give the column a value (append, update, merge, SQL insert), schema changes that would break the stored expression or reshape its output, metadata edits, volatile expressions, declaration metadata arriving through any path but the validated declare call, and LSM write specs in either order against latest committed state are all refused. The LSM check also refuses on the mem-wal catch-up feature flag, which outlives unset and marks retained SSTable rows. Simultaneous declare/install interleavings conflict at commit via lance's mem-wal rule (lance#8539). Local tables only. --- <sub>Stack created with <a href="https://github.com/github/gh-stack">GitHub Stacks CLI</a> • <a href="https://gh.io/stacks-feedback">Give Feedback 💬</a></sub>	2026-08-14 14:17:41 -07:00
LanceDB Robot	9e4d8bd1c7	chore: update lance dependency to v11.0.0-beta.11 (#3946 ) Updates the Rust workspace Lance crates and Java lance-core dependency to v11.0.0-beta.11. No compatibility fixes were required; formatting and full-workspace clippy validation pass. Lance tag: https://github.com/lance-format/lance/releases/tag/v11.0.0-beta.11	2026-08-14 08:31:58 -07:00
XY Zhan	4148dfef72	feat(lsm): require recorded index catch-up, as an explicit activation (#3911 ) > Stacked on #3780. Blocked only on #3922 (`lance` → `v11.0.0-beta.6`), so CI > stays red until that lands. ## Missing coverage must mean "not known to be covered" #3780 caps the SSTable exclusion watermark at an index's recorded catch-up when there is one, and silently ignores the case where there is none. On a table that requires catch-up, an absent entry means the index is not known to hold the compacted rows — and the LSM base arm reads base through the index (`fast_search`, no brute-force tail), so dropping that SSTable loses those rows for that query. ```rust Some(caught_up) => watermark = watermark.min(caught_up), None if catchup_required => watermark = 0, // retain everything None => {} ``` `catchup_required` reads the manifest feature bit directly, and requires both words: a half-set manifest is treated as legacy, which is the conservative side. Without the bit the field is not maintained at all, so absence carries no information and behaviour is unchanged. ## Activation, as a table-level entry point `Table::require_mem_wal_index_catchup()` performs the one-way switch, separate from `set_lsm_write_spec`: a table carrying the bit retains every generation until something records catch-up, so it has to follow the deployment of whatever repairs coverage, not the creation of the table. This is a convenience, not the only path — a writer holding the dataset calls the equivalent on `DatasetMemWalExt`, which is what the WAL pod does. Lance enforces the preconditions either way: the MemWAL index must exist, and the table must not already carry `compacted_sstables` from before this protocol, since those numbers cannot be validated. ## Still correct after the Lance rework lance-format/lance#8481 replaced the transmitted `IndexCatchupAdvance` with a position derived at commit time from the version a transaction read. That changed how a writer earns coverage; it did not change what a reader may conclude from its absence. The rule here, and the field it reads, are unchanged. ## Tests Existing `exclusion_watermarks` unit tests carry the new argument. Coverage against a real dataset follows once #3922 lands and this can build.	2026-08-14 09:32:02 -04:00
LanceDB Robot	0ac70a8b9f	chore: update lance dependency to v11.0.0-beta.10 (#3944 ) Updates the Rust workspace Lance dependencies and Java lance-core dependency to v11.0.0-beta.10. No compatibility fixes were required; workspace clippy with all features and Rust formatting pass. Lance tag: https://github.com/lance-format/lance/releases/tag/v11.0.0-beta.10	2026-08-14 18:46:46 +08:00
Lance Release	91c5f344d2	Bump version: 0.37.1-beta.1 → 0.38.0-beta.0	2026-08-14 01:09:50 +00:00
Jack Ye	ffd35c1a8f	feat: add asynchronous drop table API (#3936 ) ## Summary - add `drop_table_async` and return a job handle while preserving `drop_table` - consume remote 202 responses with cleanup job IDs and retain older-server compatibility - expose the API through Python and TypeScript connection wrappers	2026-08-13 18:05:44 -07:00
Wyatt Alt	790d0c684c	docs(ci): clarify tag input on codex-update-lance-dependency (#3924 ) Say what resolving "latest" actually does: pick the newest release, preferring stable over pre-release, and skip the run if it is not newer than the version pinned in Cargo.toml. Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>	2026-08-13 11:26:58 -07:00
XY Zhan	251f194696	refactor(lsm): gate SSTable exclusion on every index a query relies on (#3780 ) `exclusion_watermarks` resolved a single index and capped SSTable exclusion at that index's catch-up watermark. It now takes every index the query relies on and retains to the lowest of them, and the resolver collects arms together rather than returning at the first match. This is groundwork, not a fix for a reachable bug: `reject_unsupported` refuses hybrid search, so the vector and full-text arms are mutually exclusive and the list never holds more than one entry today. The generalisation is what the remaining work below plugs into. Unchanged: a plain scan uses the compaction watermark alone, an index with no catch-up entry contributes no cap, and a caught-up index falls back to the compaction watermark. Taking a minimum over more indexes can only lower a watermark, so the failure direction is "read an SSTable unnecessarily", never "miss rows". ## Tests Three in `lsm`: the existing lagging-index test updated for the new signature; `exclusion_watermark_takes_the_minimum_across_every_index_used` (two indexes at 7 and 4 against compaction at 9 — each alone stops at its own watermark, together the lower governs, order-independent); and `an_untracked_index_does_not_widen_a_lagging_sibling`. `cargo test -p lancedb --lib` — 45 lsm tests, 484 in the crate. `cargo fmt --check` clean. ## Follow-ups This crate pins lance to a released tag, so anything needing unreleased Lance symbols waits for a bump. 1. Select legacy versus strict semantics from the feature bit. On a table with `FLAG_MEM_WAL_INDEX_CATCHUP` set, a missing entry must mean "not caught up" and retain the SSTables, instead of leaving the compaction watermark unchanged. Needs the bit from lance-format/lance#8263. This must land before any table is activated — otherwise the bit is set while queries still read permissively. 2. Collect scalar and bitmap-family prefilter indexes. The genuinely multi-index query is a vector search with a scalar prefilter, and it is gated on the vector index alone today. Identifying the others needs the planner's chosen indexes, not the columns the filter names, so it needs a Lance-side helper. 3. Verify a retained SSTable can actually answer. Both base and SSTable arms use `fast_search`; a source without a compatible index contributes nothing, so retention alone does not guarantee its rows are returned. Needs a flat-search fallback or an explicit error in Lance's `LsmScanner`. 4. Planner-level integration tests. Current tests exercise the watermark arithmetic directly. End-to-end coverage over real queries — prefilter forms, legacy versus activated, missing index and missing shard entries — depends on 1–3.	2026-08-13 13:23:37 -04:00
LanceDB Robot	4b7325bd74	chore: update lance dependency to v11.0.0-beta.8 (#3928 ) Updates the Rust workspace and Java lance-core dependency to Lance v11.0.0-beta.8, with refreshed Cargo lockfile metadata. No compatibility fixes were required. Lance tag: https://github.com/lance-format/lance/releases/tag/v11.0.0-beta.8	2026-08-14 00:00:18 +08:00