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

2026-09-11 07:42:26 +00:00 · 2026-09-05 23:25:29 +00:00
70 changed files with 402 additions and 4003 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.39.0-beta.6"
+current_version = "0.39.0-beta.2"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -1,20 +0,0 @@
-name: Typo checker
-on:
-  push:
-    branches:
-      - main
-  pull_request:
-
-permissions:
-  contents: read
-
-jobs:
-  run:
-    name: Spell Check with Typos
-    runs-on: ubuntu-latest
-    steps:
-      - name: Check out code
-        uses: actions/checkout@v6
-
-      - name: Check spelling of the entire repository
-        uses: crate-ci/typos@6802cc60d4e7f78b9d5454f6cf3935c042d5e1e3 # v1.26.0
@@ -10,10 +10,6 @@ repos:
    rev: v0.9.9
    hooks:
      - id: ruff
-  - repo: https://github.com/crate-ci/typos
-    rev: v1.26.0
-    hooks:
-      - id: typos
  # - repo: https://github.com/RobertCraigie/pyright-python
  #   rev: v1.1.395
  #   hooks:
@@ -1,19 +0,0 @@
-[default]
-extend-ignore-re = ["(?Rm)^.*(#|//)\\s*spellchecker:disable-line$"]
-
-[default.extend-words]
-# Azure Kubernetes Service, mentioned in rust/lancedb/src/remote/oauth.rs.
-AKS = "AKS"
-# RabitQ is the name of a vector quantization algorithm, not a typo of "Rabbit".
-Rabit = "Rabit"
-# `VarBuilder::from_mmaped_safetensors` is the real (if oddly-spelled) name of
-# the candle-core API we call in rust/lancedb/src/embeddings/sentence_transformers.rs.
-mmaped = "mmaped"
-# `WriteableBuffer` is the real name of a type from Python's `_typeshed` stubs,
-# used in python/python/lancedb/_blob.py.
-Writeable = "Writeable"
-
-[files]
-extend-exclude = [
-    "*_THIRD_PARTY_LICENSES.*",
-]
@@ -3526,8 +3526,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-array",
 "rand 0.9.5",
@@ -4886,8 +4886,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

 [[package]]
 name = "lance"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4959,8 +4959,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4982,7 +4982,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4996,7 +4996,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5005,8 +5005,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -5016,8 +5016,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5054,8 +5054,8 @@ dependencies = [

 [[package]]
 name = "lance-datafusion"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5085,8 +5085,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5103,8 +5103,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5113,8 +5113,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5147,8 +5147,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5179,8 +5179,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5244,8 +5244,8 @@ dependencies = [

 [[package]]
 name = "lance-index-core"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5267,8 +5267,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5308,8 +5308,8 @@ dependencies = [

 [[package]]
 name = "lance-linalg"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5323,23 +5323,21 @@ dependencies = [

 [[package]]
 name = "lance-namespace"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow",
 "async-trait",
 "bytes",
 "lance-core",
 "lance-namespace-reqwest-client",
- "serde",
- "serde_json",
 "snafu 0.9.0",
 ]

 [[package]]
 name = "lance-namespace-impls"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5378,9 +5376,9 @@ dependencies = [

 [[package]]
 name = "lance-namespace-reqwest-client"
-version = "0.12.0"
+version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d8d23e54b1634d5bbb434f8dd33dc3c05f6e58d876a9a27b3b4aef58ddbe11af"
+checksum = "1d06b1fbb5d41f93bc652b61e2872af92e8a6c5f6b4ce8839a8ecfa05365d359"
 dependencies = [
 "reqwest 0.12.28",
 "serde",
@@ -5392,8 +5390,8 @@ dependencies = [

 [[package]]
 name = "lance-select"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5407,8 +5405,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5448,8 +5446,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5462,8 +5460,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "12.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.16#f7df098f5860cf9b3ac5339e7a641430cfbc6351"
+version = "12.0.0-beta.11"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.11#4a0e26895729feb86d0cb9c09d551bfd619c6472"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -5476,7 +5474,7 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.39.0-beta.6"
+version = "0.39.0-beta.1"
 dependencies = [
 "ahash",
 "anyhow",
@@ -5553,7 +5551,6 @@ dependencies = [
 "serde_json",
 "serde_with",
 "serial_test",
- "sha2 0.10.9",
 "snafu 0.8.9",
 "tempfile",
 "test-log",
@@ -5568,7 +5565,7 @@ dependencies = [

 [[package]]
 name = "lancedb-nodejs"
-version = "0.39.0-beta.6"
+version = "0.39.0-beta.1"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5593,7 +5590,7 @@ dependencies = [

 [[package]]
 name = "lancedb-python"
-version = "0.39.0-beta.6"
+version = "0.39.0-beta.1"
 dependencies = [
 "arrow",
 "async-trait",
@@ -7704,9 +7701,9 @@ dependencies = [

 [[package]]
 name = "prost"
-version = "0.14.4"
+version = "0.14.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1"
+checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568"
 dependencies = [
 "bytes",
 "prost-derive",
@@ -7733,9 +7730,9 @@ dependencies = [

 [[package]]
 name = "prost-derive"
-version = "0.14.4"
+version = "0.14.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
+checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b"
 dependencies = [
 "anyhow",
 "itertools 0.14.0",
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=12.0.0-beta.16", default-features = false, "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=12.0.0-beta.16", "tag" = "v12.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=12.0.0-beta.11", default-features = false, "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=12.0.0-beta.11", default-features = false, "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=12.0.0-beta.11", default-features = false, "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=12.0.0-beta.11", "tag" = "v12.0.0-beta.11", "git" = "https://github.com/lance-format/lance.git" }
 lancedb = { path = "rust/lancedb", default-features = false }
 ahash = "0.8"
 # Note that this one does not include pyarrow
@@ -155,7 +155,7 @@ paths:
                vector:
                  type: FixedSizeList
                  description: |
-                    The targeted vector to search for. Required.
+                    The targetted vector to search for. Required.
                vector_column:
                  type: string
                  description: |
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-core</artifactId>
-    <version>0.39.0-beta.6</version>
+    <version>0.39.0-beta.2</version>
 </dependency>
 ```

@@ -141,7 +141,7 @@ Currently this causes multiple copies of the row to be created
 but that behavior is subject to change.

 An optional condition may be specified.  If it is, then only
-matched rows that satisfy the condition will be updated.  Any
+matched rows that satisfy the condtion will be updated.  Any
 rows that do not satisfy the condition will be left as they
 are.  Failing to satisfy the condition does not cause a
 "matched row" to become a "not matched" row.
@@ -74,10 +74,10 @@ 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 also recomputes the rows whose inputs changed since they were
-computed, so a mutated input is reflected by the next refresh. While a
-declaration reads a column, that column cannot be renamed, retyped or
-dropped.
+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
@@ -854,10 +854,10 @@ 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, and
-rows whose inputs changed since they were computed are recomputed;
-everything else is left as it is. Local tables only: a remote refresh
-runs as a server job, through [Table#refreshColumnAsync](Table.md#refreshcolumnasync).
+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

@@ -1266,7 +1266,7 @@ value is 0")
 Note: if your condition is something like "some_id_column == 7" and
 you are updating many rows (with different ids) then you will get
 better performance with a single [`merge_insert`] call instead of
-repeatedly calling this method.
+repeatedly calilng this method.

 ##### Parameters

@@ -118,7 +118,7 @@ Number of sub-vectors of PQ.
 This value controls how much the vector is compressed during the quantization step.
 The more sub vectors there are the less the vector is compressed.  The default is
 the dimension of the vector divided by 16.  If the dimension is not evenly divisible
-by 16 we use the dimension divided by 8.
+by 16 we use the dimension divded by 8.

 The above two cases are highly preferred.  Having 8 or 16 values per subvector allows
 us to use efficient SIMD instructions.
@@ -16,7 +16,7 @@ optional config: Index;

 Advanced index configuration

-This option allows you to specify a specific index to create and also
+This option allows you to specify a specfic index to create and also
 allows you to pass in configuration for training the index.

 See the static methods on Index for details on the various index types.
@@ -112,7 +112,7 @@ Number of sub-vectors of PQ.
 This value controls how much the vector is compressed during the quantization step.
 The more sub vectors there are the less the vector is compressed.  The default is
 the dimension of the vector divided by 16.  If the dimension is not evenly divisible
-by 16 we use the dimension divided by 8.
+by 16 we use the dimension divded by 8.

 The above two cases are highly preferred.  Having 8 or 16 values per subvector allows
 us to use efficient SIMD instructions.
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.39.0-beta.6</version>
+      <version>0.39.0-beta.2</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.39.0-beta.6</version>
+    <version>0.39.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>12.0.0-beta.16</lance-core.version>
+        <lance-core.version>12.0.0-beta.11</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.39.0-beta.6"
+version = "0.39.0-beta.2"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -281,7 +281,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
        numIndices: 0,
        numRows: 3,
        // Full on-disk size of the two data files, footers and metadata included.
-        totalBytes: 550,
+        totalBytes: 684,
      });

      // Index files count toward totalBytes too (only deletion files and
@@ -289,7 +289,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      await table.createIndex("id", { config: Index.btree() });
      const statsWithIndex = await table.stats();
      expect(statsWithIndex.numIndices).toBe(1);
-      expect(statsWithIndex.totalBytes).toBeGreaterThan(550);
+      expect(statsWithIndex.totalBytes).toBeGreaterThan(684);
    });

    it("should overwrite data if asked", async () => {
@@ -3252,7 +3252,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      const db = await connect(tmpDir.name);
      const data = [
        { text: "fa", vector: [0.1, 0.2, 0.3] },
-        { text: "fo", vector: [0.4, 0.5, 0.6] }, // spellchecker:disable-line
+        { text: "fo", vector: [0.4, 0.5, 0.6] },
        { text: "fob", vector: [0.4, 0.5, 0.6] },
        { text: "focus", vector: [0.4, 0.5, 0.6] },
        { text: "foo", vector: [0.4, 0.5, 0.6] },
@@ -3277,7 +3277,7 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      const resultSet = new Set(fuzzyResults.map((r) => r.text));
      expect(resultSet.has("foo")).toBe(true);
      expect(resultSet.has("fob")).toBe(true);
-      expect(resultSet.has("fo")).toBe(true); // spellchecker:disable-line
+      expect(resultSet.has("fo")).toBe(true);
      expect(resultSet.has("food")).toBe(true);

      const prefixResults = await table
@@ -600,7 +600,7 @@ function makeVector(
  }
  if (values.length === 0) {
    throw Error(
-      "makeVector requires at least one value or the type must be specified",
+      "makeVector requires at least one value or the type must be specfied",
    );
  }
  const sampleValue = values.find((val) => val !== null && val !== undefined);
@@ -858,7 +858,7 @@ async function applyEmbeddings<T>(
 * customized by the `embeddingDataType` property of the embedding function.
 *
 * If a schema is provided in `makeTableOptions` then it should include the
- * embedding columns.  If no schema is provided then embedding columns will
+ * embedding columns.  If no schema is provded then embedding columns will
 * be placed at the end of the table, after all of the input columns.
 */
 export async function convertToTable(
@@ -26,7 +26,7 @@ export interface IvfPqOptions {
   * This value controls how much the vector is compressed during the quantization step.
   * The more sub vectors there are the less the vector is compressed.  The default is
   * the dimension of the vector divided by 16.  If the dimension is not evenly divisible
-   * by 16 we use the dimension divided by 8.
+   * by 16 we use the dimension divded by 8.
   *
   * The above two cases are highly preferred.  Having 8 or 16 values per subvector allows
   * us to use efficient SIMD instructions.
@@ -228,7 +228,7 @@ export interface HnswPqOptions {
   * This value controls how much the vector is compressed during the quantization step.
   * The more sub vectors there are the less the vector is compressed.  The default is
   * the dimension of the vector divided by 16.  If the dimension is not evenly divisible
-   * by 16 we use the dimension divided by 8.
+   * by 16 we use the dimension divded by 8.
   *
   * The above two cases are highly preferred.  Having 8 or 16 values per subvector allows
   * us to use efficient SIMD instructions.
@@ -825,7 +825,7 @@ export interface IndexOptions {
  /**
   * Advanced index configuration
   *
-   * This option allows you to specify a specific index to create and also
+   * This option allows you to specify a specfic index to create and also
   * allows you to pass in configuration for training the index.
   *
   * See the static methods on Index for details on the various index types.
@@ -27,7 +27,7 @@ export class MergeInsertBuilder {
   * but that behavior is subject to change.
   *
   * An optional condition may be specified.  If it is, then only
-   * matched rows that satisfy the condition will be updated.  Any
+   * matched rows that satisfy the condtion will be updated.  Any
   * rows that do not satisfy the condition will be left as they
   * are.  Failing to satisfy the condition does not cause a
   * "matched row" to become a "not matched" row.
@@ -3,7 +3,7 @@

 // The utilities in this file help sanitize data from the user's arrow
 // library into the types expected by vectordb's arrow library.  Node
-// generally allows for multiple versions of the same library (and sometimes
+// generally allows for mulitple versions of the same library (and sometimes
 // even multiple copies of the same version) to be installed at the same
 // time.  However, arrow-js uses instanceof which expected that the input
 // comes from the exact same library instance.  This is not always the case
@@ -313,7 +313,7 @@ export abstract class Table {
   * Note: if your condition is something like "some_id_column == 7" and
   * you are updating many rows (with different ids) then you will get
   * better performance with a single [`merge_insert`] call instead of
-   * repeatedly calling this method.
+   * repeatedly calilng this method.
   * @param {Map<string, string> | Record<string, string>} updates - the
   * columns to update
   * @returns {Promise<UpdateResult>} A promise that resolves to an object
@@ -542,10 +542,10 @@ export abstract class Table {
   * {@link Table#refreshColumn}. Declaring one therefore costs the same on a
   * large table as on an empty one.
   *
-   * A refresh also recomputes the rows whose inputs changed since they were
-   * computed, so a mutated input is reflected by the next refresh. While a
-   * declaration reads a column, that column cannot be renamed, retyped or
-   * dropped.
+   * 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
@@ -576,10 +576,10 @@ export abstract class Table {
  /**
   * Fill the rows of a computed column that hold no value yet.
   *
-   * Rows appended since the last refresh are filled by the next one, and
-   * rows whose inputs changed since they were computed are recomputed;
-   * everything else is left as it is. Local tables only: a remote refresh
-   * runs as a server job, through {@link Table#refreshColumnAsync}.
+   * 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.
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.39.0-beta.6",
+	"version": "0.39.0-beta.2",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.39.0-beta.6",
+	"version": "0.39.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.39.0-beta.6",
+	"version": "0.39.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.39.0-beta.6",
+	"version": "0.39.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.39.0-beta.6",
+	"version": "0.39.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.39.0-beta.6",
+  "version": "0.39.0-beta.2",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.39.0-beta.6",
+	"version": "0.39.0-beta.2",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.39.0-beta.6",
+  "version": "0.39.0-beta.2",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -41,7 +41,7 @@ importers:
        version: 3.7.0(@emnapi/core@1.10.0)(@emnapi/runtime@1.11.3)(@types/node@22.7.4)
      '@opentelemetry/sdk-metrics':
        specifier: ^2.10.0
-        version: 2.11.0(@opentelemetry/api@1.9.1)
+        version: 2.10.0(@opentelemetry/api@1.9.1)
      '@types/axios':
        specifier: ^0.14.0
        version: 0.14.4
@@ -80,7 +80,7 @@ importers:
        version: 0.2.7
      ts-jest:
        specifier: ^29.1.2
-        version: 29.4.12(@babel/core@7.29.7)(@jest/transform@29.7.0)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.29.7))(jest-util@29.7.0)(jest@29.7.0(@types/node@22.7.4))(typescript@5.5.4)
+        version: 29.4.9(@babel/core@7.29.7)(@jest/transform@29.7.0)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.29.7))(jest-util@29.7.0)(jest@29.7.0(@types/node@22.7.4))(typescript@5.5.4)
      typedoc:
        specifier: 0.26.4
        version: 0.26.4(typescript@5.5.4)
@@ -1394,20 +1394,20 @@ packages:
    resolution: {integrity: sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==}
    engines: {node: '>=8.0.0'}

-  '@opentelemetry/core@2.11.0':
-    resolution: {integrity: sha512-7YP44XH0tV6+Mb54x2YGf84i7yi+31MBZlE8JwvozkxyTvXbSp10X7cI7YE49ChJ3shMJoBmCJF3+1QFBJctGA==}
+  '@opentelemetry/core@2.10.0':
+    resolution: {integrity: sha512-/wNZ8twnEQQA4HoHu22+vcsdru6pWPWxW+7w+FlxT6Id7PE/WIbZmVKkte+PF72e0F2dnImFeHD2syyE1Mw6MQ==}
    engines: {node: ^18.19.0 || >=20.6.0}
    peerDependencies:
      '@opentelemetry/api': '>=1.0.0 <1.10.0'

-  '@opentelemetry/resources@2.11.0':
-    resolution: {integrity: sha512-Ie7+8q8MDF4FAEQCKVMTx3ReUvxiIAgIiiW3c9JdmP8+HMcDy20puT+AHjexnExgnbvBxjQ9fjkFDWrikJ2jQA==}
+  '@opentelemetry/resources@2.10.0':
+    resolution: {integrity: sha512-q6MMm2zhggzsHVNbabYwut+a6nbuQQe3URUoxaojM/8K1IBfwwPzvxIjNi2/lI1TFe+fMHMW9MWhrtDLEXEnkA==}
    engines: {node: ^18.19.0 || >=20.6.0}
    peerDependencies:
      '@opentelemetry/api': '>=1.3.0 <1.10.0'

-  '@opentelemetry/sdk-metrics@2.11.0':
-    resolution: {integrity: sha512-7GXXcObyHyDUUSG+L+kJoquty01bzm7ivE7+SSgXXJcHuPzGviptxwARmI2c+bnnxjexGQbJnyNlN8HxBP/Y7A==}
+  '@opentelemetry/sdk-metrics@2.10.0':
+    resolution: {integrity: sha512-t6r1VSvXNtSDnPXU1FbZeetJb7yyovHmgu0wRSoftxtE0g2rSNhQZQUy69sRUCL+iioJpX8SN/S6wq6ZtvLySQ==}
    engines: {node: ^18.19.0 || >=20.6.0}
    peerDependencies:
      '@opentelemetry/api': '>=1.9.0 <1.10.0'
@@ -1480,7 +1480,6 @@ packages:
  '@smithy/core@3.24.1':
    resolution: {integrity: sha512-3mT7o4qQyUWttYnVK3A0Z/u3Xha3E81tXn32Tz6vjZiUXhBrkEivpw1hBYfh84iFF9CSzkBU9Y1DJ3Q6RQ231g==}
    engines: {node: '>=18.0.0'}
-    deprecated: Deprecated due to bug in browser bundling instructions https://github.com/smithy-lang/smithy-typescript/issues/2025

  '@smithy/credential-provider-imds@4.3.1':
    resolution: {integrity: sha512-0S/acwHnqX4WrjXzhdiDRxsG2s9SC0cpPIK9nZ1R6UOHd+j7uL28+4bHu22urbLk2TVw3fkp6na/+fkUt/pLNQ==}
@@ -3239,8 +3238,8 @@ packages:
    peerDependencies:
      typescript: '>=4.2.0'

-  ts-jest@29.4.12:
-    resolution: {integrity: sha512-Ov6ClY53Fflh6BGAnY2DlTq1hYDrTycz2PVTXBWFW2CU+9zrEqAp9fWdGXl42EXO5RLSFAcAZ2JFKbP+zBTFfw==}
+  ts-jest@29.4.9:
+    resolution: {integrity: sha512-LTb9496gYPMCqjeDLdPrKuXtncudeV1yRZnF4Wo5l3SFi0RYEnYRNgMrFIdg+FHvfzjCyQk1cLncWVqiSX+EvQ==}
    engines: {node: ^14.15.0 || ^16.10.0 || ^18.0.0 || >=20.0.0}
    hasBin: true
    peerDependencies:
@@ -5111,22 +5110,22 @@ snapshots:

  '@opentelemetry/api@1.9.1': {}

-  '@opentelemetry/core@2.11.0(@opentelemetry/api@1.9.1)':
+  '@opentelemetry/core@2.10.0(@opentelemetry/api@1.9.1)':
    dependencies:
      '@opentelemetry/api': 1.9.1
      '@opentelemetry/semantic-conventions': 1.43.0

-  '@opentelemetry/resources@2.11.0(@opentelemetry/api@1.9.1)':
+  '@opentelemetry/resources@2.10.0(@opentelemetry/api@1.9.1)':
    dependencies:
      '@opentelemetry/api': 1.9.1
-      '@opentelemetry/core': 2.11.0(@opentelemetry/api@1.9.1)
+      '@opentelemetry/core': 2.10.0(@opentelemetry/api@1.9.1)
      '@opentelemetry/semantic-conventions': 1.43.0

-  '@opentelemetry/sdk-metrics@2.11.0(@opentelemetry/api@1.9.1)':
+  '@opentelemetry/sdk-metrics@2.10.0(@opentelemetry/api@1.9.1)':
    dependencies:
      '@opentelemetry/api': 1.9.1
-      '@opentelemetry/core': 2.11.0(@opentelemetry/api@1.9.1)
-      '@opentelemetry/resources': 2.11.0(@opentelemetry/api@1.9.1)
+      '@opentelemetry/core': 2.10.0(@opentelemetry/api@1.9.1)
+      '@opentelemetry/resources': 2.10.0(@opentelemetry/api@1.9.1)

  '@opentelemetry/semantic-conventions@1.43.0': {}

@@ -5575,7 +5574,7 @@ snapshots:
      globby: 11.1.0
      is-glob: 4.0.3
      minimatch: 9.0.9
-      semver: 7.8.5
+      semver: 7.8.0
      ts-api-utils: 1.4.3(typescript@5.5.4)
    optionalDependencies:
      typescript: 5.5.4
@@ -6427,7 +6426,7 @@ snapshots:
      '@babel/parser': 7.29.3
      '@istanbuljs/schema': 0.1.6
      istanbul-lib-coverage: 3.2.2
-      semver: 7.8.5
+      semver: 7.8.0
    transitivePeerDependencies:
      - supports-color

@@ -6706,7 +6705,7 @@ snapshots:
      jest-util: 29.7.0
      natural-compare: 1.4.0
      pretty-format: 29.7.0
-      semver: 7.8.5
+      semver: 7.8.0
    transitivePeerDependencies:
      - supports-color

@@ -6829,7 +6828,7 @@ snapshots:

  make-dir@4.0.0:
    dependencies:
-      semver: 7.8.5
+      semver: 7.8.0

  make-error@1.3.6: {}

@@ -7163,7 +7162,8 @@ snapshots:

  semver@7.8.0: {}

-  semver@7.8.5: {}
+  semver@7.8.5:
+    optional: true

  sharp@0.35.4(@types/node@22.7.4):
    dependencies:
@@ -7327,7 +7327,7 @@ snapshots:
    dependencies:
      typescript: 5.5.4

-  ts-jest@29.4.12(@babel/core@7.29.7)(@jest/transform@29.7.0)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.29.7))(jest-util@29.7.0)(jest@29.7.0(@types/node@22.7.4))(typescript@5.5.4):
+  ts-jest@29.4.9(@babel/core@7.29.7)(@jest/transform@29.7.0)(@jest/types@29.6.3)(babel-jest@29.7.0(@babel/core@7.29.7))(jest-util@29.7.0)(jest@29.7.0(@types/node@22.7.4))(typescript@5.5.4):
    dependencies:
      bs-logger: 0.2.6
      fast-json-stable-stringify: 2.1.0
@@ -7336,7 +7336,7 @@ snapshots:
      json5: 2.2.3
      lodash.memoize: 4.1.2
      make-error: 1.3.6
-      semver: 7.8.5
+      semver: 7.8.0
      type-fest: 4.41.0
      typescript: 5.5.4
      yargs-parser: 21.1.1
@@ -127,7 +127,7 @@ impl Job {
 }

 /// Serialise Arrow batches as a single IPC stream for the TypeScript layer.
-fn batches_to_ipc_buffer(batches: &[RecordBatch]) -> napi::Result<Buffer> {
+pub(crate) fn batches_to_ipc_buffer(batches: &[RecordBatch]) -> napi::Result<Buffer> {
    let Some(first) = batches.first() else {
        return Ok(Buffer::from(Vec::<u8>::new()));
    };
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.39.0-beta.6"
+version = "0.39.0-beta.2"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -21,7 +21,7 @@ class GteEmbeddings(TextEmbeddingFunction):
    An embedding function that uses GTE-LARGE MLX format(for Apple silicon devices only)
    as well as the standard cpu/gpu version from: https://huggingface.co/thenlper/gte-large.

-    For Apple users, you will need the mlx package installed, which can be done with:
+    For Apple users, you will need the mlx package insalled, which can be done with:
        pip install mlx

    Parameters
@@ -60,7 +60,7 @@ class InstructorEmbeddingFunction(TextEmbeddingFunction):

    import lancedb
    from lancedb.pydantic import LanceModel, Vector
-    from lancedb.embeddings import get_registry, InstructorEmbeddingFunction
+    from lancedb.embeddings import get_registry, InstuctorEmbeddingFunction

    instructor = get_registry().get("instructor").create(
        source_instruction="represent the document for retrieval",
@@ -751,7 +751,7 @@ class IvfPq:
        This value controls how much the vector is compressed during the
        quantization step.  The more sub vectors there are the less the vector is
        compressed.  The default is the dimension of the vector divided by 16.  If
-        the dimension is not evenly divisible by 16 we use the dimension divided by
+        the dimension is not evenly divisible by 16 we use the dimension divded by
        8.

        The above two cases are highly preferred.  Having 8 or 16 values per
@@ -78,10 +78,6 @@ if TYPE_CHECKING:
 T = TypeVar("T", bound="LanceModel")
 AnalyzePlanDistributedMetrics = Literal["aggregate", "per_worker", "full"]

-# Number of rows a hybrid query returns when no limit was set on it. This
-# mirrors the default the Rust query builder applies to its sub-queries.
-DEFAULT_HYBRID_LIMIT = 10
-

@runtime_checkable
 class _LanceScanner(Protocol):
@@ -863,7 +859,7 @@ class Query(pydantic.BaseModel):
        return query

    # This tells pydantic to allow custom types (needed for the `vector` query since
-    # pa.Array wouldn't be allowed otherwise)
+    # pa.Array wouln't be allowed otherwise)
    model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)


@@ -3897,54 +3893,14 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):

        return self

-    def _create_child_queries(
-        self,
-    ) -> Tuple["AsyncFTSQuery", "AsyncVectorQuery", int, int]:
-        """Build the sub-queries that make up this hybrid query.
-
-        Execution, `explain_plan` and `analyze_plan` all go through here so that
-        the plans that are reported are the plans that actually run.
-
-        Returns the two sub-queries along with the effective limit and offset of
-        the hybrid query itself.
-        """
-        fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table)
-        vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table)
-
-        fts_req = fts_query._inner.to_query_request()
-        vec_req = vec_query._inner.to_query_request()
-
-        # Only one of the two sub-queries carries the limit when it was never
-        # set explicitly: nearest_to()/nearest_to_text() build the sibling query
-        # from scratch, and that is where the default gets filled in. Which one
-        # that is depends on the order the hybrid query was built in, so look at
-        # both rather than at a single side.
-        limit = fts_req.limit if fts_req.limit is not None else vec_req.limit
-        if limit is None:
-            limit = DEFAULT_HYBRID_LIMIT
-        offset = fts_req.offset or vec_req.offset or 0
-
-        fts_query.with_row_id()
-        vec_query.with_row_id()
-
-        # offset() pushes the offset down into both sub-queries, which would make
-        # each of them skip its own first `offset` rows. The window has to be
-        # taken out of the combined, reranked results instead, so fetch the
-        # skipped prefix here too and slice it off afterwards.
-        fts_query.limit(limit + offset)
-        vec_query.limit(limit + offset)
-        fts_query.offset(0)
-        vec_query.offset(0)
-
-        return fts_query, vec_query, limit, offset
-
    async def to_batches(
        self,
        *,
        max_batch_length: Optional[int] = None,
        timeout: Optional[timedelta] = None,
    ) -> AsyncRecordBatchReader:
-        fts_query, vec_query, limit, offset = self._create_child_queries()
+        fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table)
+        vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table)

        req = fts_query._inner.to_query_request()
        blob_auto_row_id = False
@@ -3964,6 +3920,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        self._blob_auto_row_id = blob_auto_row_id
        self._blob_paths = blob_paths

+        fts_query.with_row_id()
+        vec_query.with_row_id()
+
        fts_results, vector_results = await asyncio.gather(
            fts_query.to_arrow(timeout=timeout),
            vec_query.to_arrow(timeout=timeout),
@@ -3975,9 +3934,8 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
            norm=self._norm,
            fts_query=fts_query.get_query(),
            reranker=self._reranker,
-            limit=limit,
+            limit=self._inner.get_limit(),
            with_row_ids=True,
-            offset=offset,
        )
        if (
            not self._user_requested_row_id()
@@ -4006,14 +3964,14 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        ...     print(plan)
        >>> asyncio.run(doctest_example()) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
        RRFReranker(K=60)
-            ProjectionExec: expr=[vector@0 as vector, text@3 as text, _distance@2 as _distance, _rowid@1 as _rowid]
+            ProjectionExec: expr=[vector@0 as vector, text@3 as text, _distance@2 as _distance]
              LanceRead: uri=..., projection=[text], source=stream(_rowid)
                GlobalLimitExec: skip=0, fetch=10
                  FilterExec: _distance@2 IS NOT NULL
                    SortExec: TopK(fetch=10), expr=[_distance@2 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false]
                      KNNVectorDistance: metric=l2
                        LanceRead: uri=..., projection=[vector], ...
-            ProjectionExec: expr=[vector@2 as vector, text@3 as text, _score@1 as _score, _rowid@0 as _rowid]
+            ProjectionExec: expr=[vector@2 as vector, text@3 as text, _score@1 as _score]
              LanceRead: uri=..., projection=[vector, text], source=stream(_rowid)
                GlobalLimitExec: skip=0, fetch=10
                  MatchQuery: column=text, query=[hello]
@@ -4028,9 +3986,8 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        plan : str
        """  # noqa: E501

-        fts_query, vec_query, _, _ = self._create_child_queries()
-        vector_plan = await vec_query.explain_plan(verbose)
-        fts_plan = await fts_query.explain_plan(verbose)
+        vector_plan = await self._inner.to_vector_query().explain_plan(verbose)
+        fts_plan = await self._inner.to_fts_query().explain_plan(verbose)
        # Indent sub-plans under the reranker
        indented_vector = "\n".join("  " + line for line in vector_plan.splitlines())
        indented_fts = "\n".join("  " + line for line in fts_plan.splitlines())
@@ -4057,12 +4014,14 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        -------
        plan : str
        """
-        fts_query, vec_query, _, _ = self._create_child_queries()
-
        results = ["Vector Search Query:"]
-        results.append(await vec_query.analyze_plan(distributed_metrics))
+        results.append(
+            await self._inner.to_vector_query().analyze_plan(distributed_metrics)
+        )
        results.append("FTS Search Query:")
-        results.append(await fts_query.analyze_plan(distributed_metrics))
+        results.append(
+            await self._inner.to_fts_query().analyze_plan(distributed_metrics)
+        )

        return "\n".join(results)

@@ -720,7 +720,7 @@ class RemoteTable(Table):
        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
-            The targeted vector to search for.
+            The targetted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image
@@ -175,7 +175,7 @@ class Reranker(ABC):
            if the results haven't been executed yet or the results in arrow format.
        query : str or None,
            The input query. Some rerankers might not need the query to rerank.
-            In that case, it can be set to None explicitly. This is intended to
+            In that case, it can be set to None explicitly. This is inteded to
            be handled by the reranker implementations.
        deduplicate : bool, optional
            Whether to deduplicate the results based on the `_rowid` column,
@@ -1619,7 +1619,7 @@ class Table(ABC):
        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
-            The targeted vector to search for.
+            The targetted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image
@@ -2188,10 +2188,10 @@ class Table(ABC):
            Declaring one therefore costs the same on a large table as on an
            empty one.

-            A refresh also recomputes the rows whose inputs changed since they
-            were computed, so a mutated input is reflected by the next refresh.
-            While a declaration reads a column, that column cannot be renamed,
-            retyped or dropped.
+            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
@@ -2211,7 +2211,7 @@ class Table(ABC):
        >>> table.add_columns(computed={"doubled": "x * 2"})
        AddColumnsResult(version=2)
        >>> table.refresh_column("doubled")
-        RefreshColumnResult(rows_filled=2, version=4)
+        RefreshColumnResult(rows_filled=2, version=3)
        >>> table.to_arrow().sort_by("x").to_pandas()
           x  doubled
        0  1        2
@@ -2225,8 +2225,8 @@ class Table(ABC):

        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, and rows whose inputs changed since they were computed
-        are recomputed; everything else is left as it is.
+        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].
@@ -3841,7 +3841,7 @@ class LanceTable(Table):
        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
-            The targeted vector to search for.
+            The targetted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image
@@ -4318,14 +4318,13 @@ class LanceTable(Table):
        return LOOP.run(self._table.add_columns(transforms, computed=computed))

    def refresh_column(self, column: str) -> "RefreshColumnResult":
-        """Fill a computed column's unfilled rows and recompute those whose
-        inputs changed. See
+        """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[RefreshColumnJobResult]:
-        """Fill a computed column's unfilled rows and recompute those whose
-        inputs changed, returning a handle to the refresh job. See
+        """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)))
@@ -5639,7 +5638,7 @@ class AsyncTable:
        if fill_value is None:
            fill_value = 0.0

-        # _sanitize_data is an old code path, but we will use it until the
+        # _santitize_data is an old code path, but we will use it until the
        # new code path is ready.
        if mode == "overwrite":
            # For overwrite, apply the same preprocessing as create_table
@@ -5815,7 +5814,7 @@ class AsyncTable:
        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
-            The targeted vector to search for.
+            The targetted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image
@@ -6313,10 +6312,10 @@ class AsyncTable:
            them from
            [`refresh_column`][lancedb.table.AsyncTable.refresh_column].

-            A refresh also recomputes the rows whose inputs changed since they
-            were computed, so a mutated input is reflected by the next refresh.
-            While a declaration reads a column, that column cannot be renamed,
-            retyped or dropped.
+            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``.
@@ -6378,8 +6377,8 @@ class AsyncTable:

        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, and rows whose inputs changed since they were computed
-        are recomputed; everything else is left as it is.
+        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].
@@ -297,10 +297,7 @@ def test_blob_v2_projection_sources_use_typed_column_name():


 def _legacy_v1_table(name):
-    # Legacy v1 blob columns are only writable at file version <= 2.1.
-    db = lancedb.connect(
-        "memory:///", storage_options={"new_table_data_storage_version": "2.1"}
-    )
+    db = lancedb.connect("memory:///")
    schema = pa.schema(
        [
            pa.field("id", pa.int64()),
@@ -327,8 +327,8 @@ def test_embedding_function_with_pandas(tmp_path):
        ) -> List[np.array]:
            return [np.random.randn(self.ndims()).tolist() for _ in range(len(texts))]

-    registry = get_registry()
-    func = registry.get("mock-embedding").create()
+    registery = get_registry()
+    func = registery.get("mock-embedding").create()

    class TestSchema(LanceModel):
        text: str = func.SourceField()
@@ -394,9 +394,9 @@ def test_multiple_embeddings_for_pandas(tmp_path):
        ) -> List[np.array]:
            return [np.random.randn(self.ndims()).tolist() for _ in range(len(texts))]

-    registry = get_registry()
-    func1 = registry.get("mock-embedding").create()
-    func2 = registry.get("mock-embedding2").create()
+    registery = get_registry()
+    func1 = registery.get("mock-embedding").create()
+    func2 = registery.get("mock-embedding2").create()

    class TestSchema(LanceModel):
        text: str = func1.SourceField()
@@ -1011,13 +1011,8 @@ def test_fts_ngram(mem_db: DBConnection):
    assert set(r["text"] for r in results) == {"lance database", "lance is cool"}

    results = (
-        table.search(
-            "nce",  # spellchecker:disable-line
-            query_type="fts",
-        )
-        .limit(10)
-        .to_list()
-    )
+        table.search("nce", query_type="fts").limit(10).to_list()
+    )  # spellchecker:disable-line
    assert len(results) == 2
    assert set(r["text"] for r in results) == {"lance database", "lance is cool"}

@@ -1039,13 +1034,8 @@ def test_fts_ngram(mem_db: DBConnection):
    assert set(r["text"] for r in results) == {"lance database", "lance is cool"}

    results = (
-        table.search(
-            "nce",  # spellchecker:disable-line
-            query_type="fts",
-        )
-        .limit(10)
-        .to_list()
-    )
+        table.search("nce", query_type="fts").limit(10).to_list()
+    )  # spellchecker:disable-line
    assert len(results) == 0

    results = table.search("la", query_type="fts").limit(10).to_list()
@@ -203,93 +203,6 @@ async def test_async_hybrid_query_default_limit(table: AsyncTable):
    assert texts.count("a") == 1


-@pytest.mark.asyncio
-async def test_async_hybrid_query_offset(table: AsyncTable):
-    # The offset window of a hybrid query must be a suffix of the same query
-    # run without an offset. Skipping the first rows of each sub-query instead
-    # of the first rows of the fused result silently changes which rows land in
-    # the window.
-    full = await (
-        table.query()
-        .nearest_to([0.0, 0.4])
-        .nearest_to_text("dog")
-        .limit(4)
-        .with_row_id()
-        .to_arrow()
-    )
-    assert len(full) == 4
-
-    second_page = await (
-        table.query()
-        .nearest_to([0.0, 0.4])
-        .nearest_to_text("dog")
-        .offset(2)
-        .limit(2)
-        .with_row_id()
-        .to_arrow()
-    )
-    assert second_page["_rowid"].to_pylist() == full["_rowid"].to_pylist()[2:]
-
-    first_page = await (
-        table.query()
-        .nearest_to([0.0, 0.4])
-        .nearest_to_text("dog")
-        .limit(2)
-        .with_row_id()
-        .to_arrow()
-    )
-    # Paging through the result must visit every row exactly once: no row
-    # repeated from the previous page and none dropped between the two.
-    paged = first_page["_rowid"].to_pylist() + second_page["_rowid"].to_pylist()
-    assert sorted(paged) == sorted(full["_rowid"].to_pylist())
-
-
-@pytest.mark.asyncio
-async def test_async_hybrid_query_fts_first_default_limit(table: AsyncTable):
-    # nearest_to() and nearest_to_text() build their new sibling sub-query from
-    # scratch, and that is the sub-query the default limit ends up on. So the
-    # side that carries the limit depends on the order the hybrid query was
-    # built in, and looking at only one side loses the limit for half the ways
-    # a hybrid query can be written. Without a limit the combined results are
-    # not truncated at all and the whole union of both candidate lists is
-    # returned.
-    await table.add([{"text": "dog", "vector": [50.0 + i, 50.0]} for i in range(10)])
-
-    result = await (
-        table.query().nearest_to_text("dog").nearest_to([0.1, 0.1]).to_arrow()
-    )
-    assert len(result) == 10
-
-    offset_result = await (
-        table.query().nearest_to_text("dog").nearest_to([0.1, 0.1]).offset(2).to_arrow()
-    )
-    assert len(offset_result) == 10
-
-
-@pytest.mark.asyncio
-async def test_async_hybrid_query_explain_plan_matches_execution(table: AsyncTable):
-    # Paging rewrites the sub-queries: each one fetches limit + offset rows with
-    # no offset of its own, and the window is sliced out after fusion. The plans
-    # have to be built from those rewritten sub-queries, otherwise explain_plan
-    # and analyze_plan describe a query that is never run.
-    query = (
-        table.query().nearest_to([0.0, 0.4]).nearest_to_text("dog").offset(2).limit(2)
-    )
-    await query.to_arrow()
-
-    plan = await query.explain_plan()
-    assert [
-        line.strip() for line in plan.splitlines() if "GlobalLimitExec" in line
-    ] == [
-        "GlobalLimitExec: skip=0, fetch=4",
-        "GlobalLimitExec: skip=0, fetch=4",
-    ]
-
-    analyzed = await query.analyze_plan()
-    assert analyzed.count("skip=0, fetch=4") == 2
-    assert "skip=2" not in analyzed
-
-
 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.
@@ -193,13 +193,7 @@ class TestNamespaceConnection:
            ),
        )

-        # Legacy v1 blob columns are only writable at file version <= 2.1.
-        table = db.create_table(
-            "blob_table",
-            data,
-            namespace_path=["test_ns"],
-            storage_options={"new_table_data_storage_version": "2.1"},
-        )
+        table = db.create_table("blob_table", data, namespace_path=["test_ns"])
        df = table.to_pandas(blob_mode="lazy").sort_values("id")

        blob = df["blob"].iloc[0]
@@ -40,10 +40,6 @@ from utils import exception_output
 from importlib.util import find_spec


-# Legacy v1 blob columns are only writable at file version <= 2.1.
-LEGACY_BLOB_STORAGE_OPTIONS = {"new_table_data_storage_version": "2.1"}
-
-
 def _blob_query_data():
    return pa.table(
        {
@@ -123,17 +119,13 @@ def _assert_blob_bytes_projection(df):

 def _blob_query_table(db, name, blob_schema):
    if blob_schema == "v1":
-        return db.create_table(
-            name, _blob_query_data(), storage_options=LEGACY_BLOB_STORAGE_OPTIONS
-        )
+        return db.create_table(name, _blob_query_data())
    return _create_blob_v2_query_table(db, name)


 async def _blob_query_table_async(db, name, blob_schema):
    if blob_schema == "v1":
-        return await db.create_table(
-            name, _blob_query_data(), storage_options=LEGACY_BLOB_STORAGE_OPTIONS
-        )
+        return await db.create_table(name, _blob_query_data())
    return await _create_blob_v2_query_table_async(db, name)


@@ -283,9 +275,7 @@ async def test_query_to_pandas_kwargs(table, table_async):
 def test_plain_scan_query_to_pandas_blob_modes(tmp_db, blob_mode):
    pytest.importorskip("lance")
    table = tmp_db.create_table(
-        f"test_query_to_pandas_blob_{blob_mode}",
-        _blob_query_data(),
-        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
+        f"test_query_to_pandas_blob_{blob_mode}", _blob_query_data()
    )

    df = (
@@ -332,9 +322,7 @@ def test_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow(
 ):
    pytest.importorskip("lance")
    table = tmp_db.create_table(
-        "test_query_to_pandas_blob_no_arrow_collect",
-        _blob_query_data(),
-        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
+        "test_query_to_pandas_blob_no_arrow_collect", _blob_query_data()
    )
    query = table.search().where("id = 1").select(["id", "blob"])

@@ -359,9 +347,7 @@ def test_plain_scan_query_to_pandas_blob_descriptions_flatten_uses_scanner(
 ):
    pytest.importorskip("lance")
    table = tmp_db.create_table(
-        "test_query_to_pandas_blob_desc_flatten",
-        _blob_query_data(),
-        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
+        "test_query_to_pandas_blob_desc_flatten", _blob_query_data()
    )
    query = table.search().where("id = 1").select(["id", "blob"])

@@ -379,11 +365,7 @@ def test_plain_scan_query_to_pandas_blob_descriptions_flatten_uses_scanner(
 def test_plain_scan_query_to_pandas_scanner_state(tmp_db):
    pytest.importorskip("lance")
    data = _blob_query_data()
-    table = tmp_db.create_table(
-        "test_query_to_pandas_scanner_state",
-        data.slice(0, 2),
-        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
-    )
+    table = tmp_db.create_table("test_query_to_pandas_scanner_state", data.slice(0, 2))
    table.add(data.slice(2, 2))

    fragments = table.to_lance().get_fragments()
@@ -418,9 +400,7 @@ def test_plain_scan_query_to_pandas_scanner_state(tmp_db):
 async def test_async_plain_scan_query_to_pandas_blob_projection(tmp_db_async):
    pytest.importorskip("lance")
    table = await tmp_db_async.create_table(
-        "test_async_query_to_pandas_blob_projection",
-        _blob_query_data(),
-        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
+        "test_async_query_to_pandas_blob_projection", _blob_query_data()
    )

    lazy_df = await (
@@ -472,9 +452,7 @@ async def test_async_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow
 ):
    pytest.importorskip("lance")
    table = await tmp_db_async.create_table(
-        "test_async_query_to_pandas_blob_no_arrow_collect",
-        _blob_query_data(),
-        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
+        "test_async_query_to_pandas_blob_no_arrow_collect", _blob_query_data()
    )
    query = table.query().where("id = 1").select(["id", "blob"])

@@ -496,11 +474,7 @@ async def test_async_plain_scan_query_to_pandas_blob_mode_does_not_collect_arrow

 def test_vector_query_to_pandas_blob_mode_requires_native_path(tmp_db):
    pytest.importorskip("lance")
-    table = tmp_db.create_table(
-        "test_vector_query_blob_mode",
-        _blob_query_data(),
-        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
-    )
+    table = tmp_db.create_table("test_vector_query_blob_mode", _blob_query_data())

    with pytest.raises(RuntimeError, match="Lance native pandas conversion"):
        table.search([1.0, 0.0]).select(["blob", "vector"]).limit(1).to_pandas(
@@ -511,9 +485,7 @@ def test_vector_query_to_pandas_blob_mode_requires_native_path(tmp_db):
 def test_vector_query_to_pandas_blob_descriptions_requires_plain_scan(tmp_db):
    pytest.importorskip("lance")
    table = tmp_db.create_table(
-        "test_vector_query_blob_descriptions",
-        _blob_query_data(),
-        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
+        "test_vector_query_blob_descriptions", _blob_query_data()
    )

    with pytest.raises(RuntimeError, match="plain scan query"):
@@ -81,7 +81,7 @@ def get_test_table(tmp_path):
        "but his son was mortal",
        "there hasn't been a good battlefield game since 2142",
        "I wish they would make another one",
-        "campaigns are not as good as they used to be",
+        "campains are not as good as they used to be",
        "Multiplayer and open world games have destroyed the single player experience",
        "Maybe the future is console games",
        "I don't know",
@@ -64,23 +64,15 @@ async def _blob_v2_table_async(db: AsyncConnection, name: str):
    return table


-# Legacy v1 blob columns are only writable at file version <= 2.1.
-LEGACY_BLOB_STORAGE_OPTIONS = {"new_table_data_storage_version": "2.1"}
-
-
 def _blob_table(db: DBConnection, name: str, blob_schema: str):
    if blob_schema == "v1":
-        return db.create_table(
-            name, data=_blob_test_data(), storage_options=LEGACY_BLOB_STORAGE_OPTIONS
-        )
+        return db.create_table(name, data=_blob_test_data())
    return _blob_v2_table(db, name)


 async def _blob_table_async(db: AsyncConnection, name: str, blob_schema: str):
    if blob_schema == "v1":
-        return await db.create_table(
-            name, data=_blob_test_data(), storage_options=LEGACY_BLOB_STORAGE_OPTIONS
-        )
+        return await db.create_table(name, data=_blob_test_data())
    return await _blob_v2_table_async(db, name)


@@ -155,11 +147,7 @@ def test_table_to_pandas_invalid_blob_mode_non_blob_table(tmp_db: DBConnection):
@pytest.mark.parametrize("blob_mode", ["lazy", "bytes", "descriptions"])
 def test_table_to_pandas_blob_modes(tmp_db: DBConnection, blob_mode):
    pytest.importorskip("lance")
-    table = tmp_db.create_table(
-        f"test_to_pandas_blob_{blob_mode}",
-        _blob_test_data(),
-        storage_options=LEGACY_BLOB_STORAGE_OPTIONS,
-    )
+    table = tmp_db.create_table(f"test_to_pandas_blob_{blob_mode}", _blob_test_data())

    df = table.to_pandas(blob_mode=blob_mode)

@@ -3354,7 +3342,7 @@ def test_empty_query(mem_db: DBConnection):
    # None is the same as default
    df = table.search().select(["id"]).limit(None).to_arrow()
    assert df.num_rows == 100
-    # invalid limist is the same as None, which is the same as default
+    # invalid limist is the same as None, wihch is the same as default
    df = table.search().select(["id"]).limit(-1).to_arrow()
    assert df.num_rows == 100
    # valid limit should work
@@ -3971,7 +3959,7 @@ def test_stats(mem_db: DBConnection):
    print(f"{stats=}")
    assert stats == {
        # Full on-disk size of the data file, footer and metadata included.
-        "total_bytes": 637,
+        "total_bytes": 633,
        "num_rows": 2,
        "num_indices": 0,
        "fragment_stats": {
@@ -4183,14 +4171,13 @@ def test_refresh_column_async_returns_job(tmp_path):
    assert result.rows_failed == 0
    assert result.rows_remaining == 0
    assert result.source_version == 2
-    # The fill lands at 3; the stamp recording its inputs is published at 4.
-    assert result.published_version == 4
+    assert result.published_version == 3
    assert job.status() == "finished"
    assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4]

    no_op = table.refresh_column_async("doubled").wait()
    assert no_op.rows_assigned == 0
-    assert no_op.source_version == 4
+    assert no_op.source_version == 3
    assert no_op.published_version is None

    # Bad input raises at the call, not through the job.
@@ -4209,6 +4196,6 @@ async def test_refresh_column_async_job_async_table(tmp_path):
    assert isinstance(result, lancedb.RefreshColumnResult)
    assert result.rows_assigned == 1
    assert result.source_version == 2
-    assert result.published_version == 4
+    assert result.published_version == 3
    assert await job.status() == "finished"
    assert (await table.to_arrow())["tripled"].to_pylist() == [9]
@@ -334,7 +334,7 @@ pub struct PyQueryRequest {
    pub column: Option<String>,
    pub query_vector: Option<PyQueryVectors>,
    pub minimum_nprobes: Option<usize>,
-    // None means user did not set it and default should be used (currently 20)
+    // None means user did not set it and default shoud be used (currenty 20)
    // Some(0) means user set it to None and there is no limit
    pub maximum_nprobes: Option<usize>,
    pub lower_bound: Option<f32>,
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.39.0-beta.6"
+version = "0.39.0-beta.2"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -95,14 +95,13 @@ candle-transformers = { version = "0.9.1", optional = true }
 candle-nn = { version = "0.9.1", optional = true }
 tokenizers = { version = "0.19.1", optional = true }
 semver = { workspace = true }
-roaring = "0.11.4"
-sha2 = "0.10"

 [dev-dependencies]
 anyhow = "1"
 lance-testing = { workspace = true }
 tempfile = { workspace = true }
 random_word = { version = "0.4.3", features = ["en"] }
+roaring = "0.11.4"
 tokio = { workspace = true, features = ["io-util", "macros", "net", "test-util"] }
 uuid = { workspace = true }
 walkdir = "2"
@@ -163,7 +163,7 @@ pub struct PolarsDataFrameRecordBatchReader {
 impl PolarsDataFrameRecordBatchReader {
    /// Creates a new `PolarsDataFrameRecordBatchReader` from a given Polars DataFrame.
    /// If the input dataframe does not have aligned chunks, this function undergoes
-    /// the costly operation of reallocating each series as a single contiguous chunk.
+    /// the costly operation of reallocating each series as a single contigous chunk.
    pub fn new(mut df: DataFrame) -> Result<Self> {
        df.align_chunks();
        let arrow_schema =
@@ -532,11 +532,10 @@ mod tests {
    fn storage_version_bumps_to_v2_2() {
        let mut params = WriteParams::default();
        ensure_blob_storage_version(&blob_schema(), &mut params);
-        let resolved = params
-            .data_storage_version
-            .unwrap_or(LanceFileVersion::Stable)
-            .resolve();
-        assert_eq!(resolved, ConcreteFileVersion::V2_2);
+        assert_eq!(
+            params.data_storage_version.unwrap().resolve(),
+            ConcreteFileVersion::V2_2
+        );
        assert!(!params.enable_stable_row_ids);
    }

@@ -548,11 +547,10 @@ mod tests {
        };
        ensure_blob_storage_version(&blob_schema(), &mut params);
        assert!(params.enable_stable_row_ids);
-        let resolved = params
-            .data_storage_version
-            .unwrap_or(LanceFileVersion::Stable)
-            .resolve();
-        assert_eq!(resolved, ConcreteFileVersion::V2_2);
+        assert_eq!(
+            params.data_storage_version.unwrap().resolve(),
+            ConcreteFileVersion::V2_2
+        );
    }

    #[test]
@@ -827,7 +827,7 @@ impl Connection {
 pub struct ConnectRequest {
    /// Database URI
    ///
-    /// ### Accepted URI formats
+    /// ### Accpeted URI formats
    ///
    /// - `/path/to/database` - local database on file system.
    /// - `s3://bucket/path/to/database` or `gs://bucket/path/to/database` - database on cloud object store
@@ -512,7 +512,7 @@ impl ListingDatabase {
                // iter thru the query params and extract the commit store param
                let mut engine = None;
                let mut mirrored_store = None;
-                let mut filtered_queries = vec![];
+                let mut filtered_querys = vec![];

                // WARNING: specifying engine is NOT a publicly supported feature in lancedb yet
                // THE API WILL CHANGE
@@ -528,13 +528,13 @@ impl ListingDatabase {
                        mirrored_store = Some(value.to_string());
                    } else {
                        // to owned so we can modify the url
-                        filtered_queries.push((key.to_string(), value.to_string()));
+                        filtered_querys.push((key.to_string(), value.to_string()));
                    }
                }

                // Filter out the commit store query param -- it's a lancedb param
                url.query_pairs_mut().clear();
-                url.query_pairs_mut().extend_pairs(filtered_queries);
+                url.query_pairs_mut().extend_pairs(filtered_querys);
                // Take a copy of the query string so we can propagate it to lance.
                // `query_pairs_mut()` leaves the URL with `Some("")` even when no
                // pairs survive (or none existed in the first place), so an empty
@@ -896,11 +896,11 @@ impl Database for ListingDatabase {
    }

    async fn read_consistency(&self) -> Result<ReadConsistency> {
-        if let Some(interval) = self.read_consistency_interval {
-            if interval.is_zero() {
+        if let Some(read_consistency_inverval) = self.read_consistency_interval {
+            if read_consistency_inverval.is_zero() {
                Ok(ReadConsistency::Strong)
            } else {
-                Ok(ReadConsistency::Eventual(interval))
+                Ok(ReadConsistency::Eventual(read_consistency_inverval))
            }
        } else {
            Ok(ReadConsistency::Manual)
@@ -3043,15 +3043,15 @@ mod tests {
    /// across platforms — see the `file://` test below).
    fn capture_query_like_connect(input_uri: &str) -> Option<String> {
        let mut url = url::Url::parse(input_uri).unwrap();
-        let mut filtered_queries = Vec::new();
+        let mut filtered_querys = Vec::new();
        for (key, value) in url.query_pairs() {
            if key == ENGINE || key == MIRRORED_STORE {
                continue;
            }
-            filtered_queries.push((key.to_string(), value.to_string()));
+            filtered_querys.push((key.to_string(), value.to_string()));
        }
        url.query_pairs_mut().clear();
-        url.query_pairs_mut().extend_pairs(filtered_queries);
+        url.query_pairs_mut().extend_pairs(filtered_querys);
        url.query().filter(|q| !q.is_empty()).map(|s| s.to_string())
    }

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

 //! Namespace-based database implementation that delegates table management to lance-namespace

-use lance_datafusion::utils::StreamingWriteSource;
 use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, Mutex};

@@ -251,11 +250,11 @@ impl Database for LanceNamespaceDatabase {
    }

    async fn read_consistency(&self) -> Result<ReadConsistency> {
-        if let Some(interval) = self.read_consistency_interval {
-            if interval.is_zero() {
+        if let Some(read_consistency_inverval) = self.read_consistency_interval {
+            if read_consistency_inverval.is_zero() {
                Ok(ReadConsistency::Strong)
            } else {
-                Ok(ReadConsistency::Eventual(interval))
+                Ok(ReadConsistency::Eventual(read_consistency_inverval))
            }
        } else {
            Ok(ReadConsistency::Manual)
@@ -305,10 +304,6 @@ impl Database for LanceNamespaceDatabase {
    }

    async fn create_table(&self, request: DbCreateTableRequest) -> Result<Arc<dyn BaseTable>> {
-        // Refuse a bad declaration before the namespace records a table.
-        crate::table::computed_columns::ensure_declarations_are_planned(
-            &request.data.arrow_schema(),
-        )?;
        let mut table_id = request.namespace_path.clone();
        table_id.push(request.name.clone());
        let mut existing_table = None;
@@ -125,7 +125,7 @@ macro_rules! impl_pq_params_setter {
        /// This value controls how much the vector is compressed during the quantization step.
        /// The more sub vectors there are the less the vector is compressed.  The default is
        /// the dimension of the vector divided by 16.  If the dimension is not evenly divisible
-        /// by 16 we use the dimension divided by 8.
+        /// by 16 we use the dimension divded by 8.
        ///
        /// The above two cases are highly preferred.  Having 8 or 16 values per subvector allows
        /// us to use efficient SIMD instructions.
@@ -24,8 +24,8 @@ use std::time::{SystemTime, UNIX_EPOCH};

 use arrow_array::cast::AsArray;
 use arrow_array::types::UInt64Type;
-use arrow_array::{RecordBatch, UInt64Array, new_null_array};
-use arrow_schema::{FieldRef, Schema as ArrowSchema, SchemaRef};
+use arrow_array::{RecordBatch, UInt64Array};
+use arrow_schema::{Schema as ArrowSchema, SchemaRef};
 use datafusion::common::ScalarValue;
 use datafusion::error::DataFusionError;
 use datafusion::physical_plan::SendableRecordBatchStream;
@@ -34,7 +34,7 @@ use datafusion::prelude::{col, lit};
 use futures::{StreamExt, TryStreamExt};
 use lance::Dataset;
 use lance::dataset::mem_wal::DatasetMemWalExt;
-use lance::dataset::transaction::{Operation, Transaction, UpdateMode};
+use lance::dataset::transaction::{Operation, Transaction};
 use lance::dataset::write::delete::DeleteBuilder;
 use lance::dataset::write::merge_insert::inserted_rows::{
    KeyExistenceFilter, KeyExistenceFilterBuilder, KeyValue,
@@ -51,9 +51,6 @@ use super::{
    definition_to_metadata,
 };
 use crate::database::OpenTableRequest;
-use crate::table::computed_columns::{
-    computed_column_from_field, computed_columns, ensure_declarations_are_planned,
-};
 use crate::table::{NativeTable, NativeTableExt, Table};
 use crate::{Error, Result};

@@ -170,52 +167,30 @@ pub(crate) async fn execute_refresh(
        .map(|p| (p.output.clone(), p.expression.clone()))
        .collect();
    validate_inputs(&source_ds, definition)?;
-    let (replanned, planned_fields, _renames) = super::plan(
+    let (replanned, mut planned_fields, _renames) = super::plan(
        source_schema,
        &definition.source_table,
        &definition.source_namespace,
-        Some(&projections),
+        &projections,
        definition.filter.as_deref(),
        definition.limit,
    )?;
-    let mut planned_fields = planned_fields;
    planned_fields.push(arrow_schema::Field::new(
        SOURCE_ROW_ID_COLUMN,
        arrow_schema::DataType::UInt64,
        false,
    ));
-    // A computed column is not planned from the source: refresh writes it
-    // NULL and its declaration's owner fills it. Its declaration must still
-    // be complete, and it must be able to hold NULL.
    let physical = ArrowSchema::from(view_ds.schema());
-    let mut computed = computed_columns(&physical).into_iter().map(|c| c.name);
-    if let Some(name) = computed.by_ref().find(|name| {
-        physical
-            .field_with_name(name)
-            .is_ok_and(|f| !f.is_nullable())
-    }) {
-        return Err(Error::Schema {
-            message: format!(
-                "computed column '{name}' of view '{}' cannot hold NULL; recreate the view",
-                view.name()
-            ),
-        });
-    }
-    ensure_declarations_are_planned(&physical)?;
-    let physical_planned: Vec<&FieldRef> = physical
+    let planned_shape: Vec<_> = planned_fields
+        .iter()
+        .map(|f| (f.name().clone(), f.data_type().clone(), f.is_nullable()))
+        .collect();
+    let physical_shape: Vec<_> = physical
        .fields()
        .iter()
-        .filter(|f| computed_column_from_field(f).is_none())
+        .map(|f| (f.name().clone(), f.data_type().clone(), f.is_nullable()))
        .collect();
-    // A projected column that became nullable at the source still fits the
-    // view's nullable field; the reverse would not.
-    let matches = planned_fields.len() == physical_planned.len()
-        && planned_fields.iter().zip(&physical_planned).all(|(e, p)| {
-            e.name() == p.name()
-                && e.data_type() == p.data_type()
-                && (p.is_nullable() || !e.is_nullable())
-        });
-    if !matches {
+    if planned_shape != physical_shape {
        return Err(Error::Schema {
            message: format!(
                "the stored definition of view '{}' does not produce this \
@@ -254,18 +229,11 @@ pub(crate) async fn execute_refresh(
        .get(SOURCE_VERSION_TS_META_KEY)
        .and_then(|raw| raw.parse().ok());
    // The watermark speaks only for the view state its refresh left behind;
-    // any other commit on the view since then is drift, except a fill of its
-    // computed columns, which rewrites nothing refresh certifies.
-    let recorded_view_version = metadata
+    // any other commit on the view since then is drift.
+    let view_intact = metadata
        .get(VIEW_VERSION_META_KEY)
-        .and_then(|raw| raw.parse::<u64>().ok());
-    let view_intact = match recorded_view_version {
-        Some(recorded) if recorded == view_ds.version().version => true,
-        Some(recorded) if recorded < view_ds.version().version => {
-            only_computed_rewrites_since(&view_ds, recorded).await?
-        }
-        _ => false,
-    };
+        .and_then(|raw| raw.parse::<u64>().ok())
+        == Some(view_ds.version().version);

    if !full && watermark == Some(source_version) && view_intact && recorded_ts == Some(source_ts) {
        return Ok(RefreshMaterializedViewResult {
@@ -1122,83 +1090,6 @@ struct RowScope {
    limit: Option<u64>,
 }

-/// Whether every commit on the view after `recorded` is a fill of its
-/// computed columns: a column rewrite or data replacement touching only
-/// those fields and neither adding nor removing rows, or the freshness
-/// stamp a fill leaves on them. A version whose transaction cannot be read
-/// is not proven, so it counts as drift.
-async fn only_computed_rewrites_since(view_ds: &Dataset, recorded: u64) -> Result<bool> {
-    // A fill may write any field under a computed column, so the whole
-    // subtree counts, not only the root.
-    let physical = ArrowSchema::from(view_ds.schema());
-    fn subtree(field: &lance_core::datatypes::Field, ids: &mut Vec<u32>) {
-        ids.push(field.id as u32);
-        for child in &field.children {
-            subtree(child, ids);
-        }
-    }
-    let mut computed_fields = Vec::new();
-    for column in computed_columns(&physical) {
-        if let Some(field) = view_ds.schema().field(&column.name) {
-            subtree(field, &mut computed_fields);
-        }
-    }
-    if computed_fields.is_empty() {
-        return Ok(false);
-    }
-    for version in recorded + 1..=view_ds.version().version {
-        let Some(transaction) = view_ds.read_transaction_by_version(version).await? else {
-            return Ok(false);
-        };
-        let fill = match &transaction.operation {
-            Operation::Update {
-                removed_fragment_ids,
-                new_fragments,
-                fields_modified,
-                update_mode: Some(UpdateMode::RewriteColumns),
-                ..
-            } => {
-                removed_fragment_ids.is_empty()
-                    && new_fragments.is_empty()
-                    && !fields_modified.is_empty()
-                    && fields_modified
-                        .iter()
-                        .all(|field| computed_fields.contains(field))
-            }
-            // What `refresh_column` commits for a SQL declaration.
-            Operation::DataReplacement { replacements } => {
-                !replacements.is_empty()
-                    && replacements.iter().all(|group| {
-                        !group.1.fields.is_empty()
-                            && group
-                                .1
-                                .fields
-                                .iter()
-                                .all(|field| computed_fields.contains(&(*field as u32)))
-                    })
-            }
-            // The stamp `refresh_column` writes after its fill (see
-            // `table::freshness`): field metadata on computed columns, no data.
-            Operation::UpdateConfig {
-                config_updates: None,
-                table_metadata_updates: None,
-                schema_metadata_updates: None,
-                field_metadata_updates,
-            } => {
-                !field_metadata_updates.is_empty()
-                    && field_metadata_updates
-                        .keys()
-                        .all(|field| computed_fields.contains(&(*field as u32)))
-            }
-            _ => false,
-        };
-        if !fill {
-            return Ok(false);
-        }
-    }
-    Ok(true)
-}
-
 async fn compute_stream(
    source: &Dataset,
    definition: &MaterializedViewDefinition,
@@ -1267,10 +1158,6 @@ async fn compute_stream(
        let batch = batch.map_err(|e| DataFusionError::External(Box::new(e)))?;
        let mut columns = Vec::with_capacity(out_schema.fields().len());
        for field in out_schema.fields() {
-            if computed_column_from_field(field).is_some() {
-                columns.push(new_null_array(field.data_type(), batch.num_rows()));
-                continue;
-            }
            let name = if field.name() == SOURCE_ROW_ID_COLUMN {
                ROW_ID
            } else {
@@ -2881,7 +2768,7 @@ mod tests {
        let (conn, source) = db_with_source(vec![1]).await;
        let prepared = crate::materialized_view::prepare_declaration(
            &source,
-            Some(&[("x".into(), "x".into()), ("twice".into(), "x * 2".into())]),
+            &[("x".into(), "x".into()), ("twice".into(), "x * 2".into())],
            None,
            None,
        )
@@ -3245,464 +3132,4 @@ mod tests {
        let err = view.refresh().execute().await.unwrap_err();
        assert!(err.to_string().contains("source table 'src'"), "{err}");
    }
-
-    /// A view with a computed column, declared over `people` and refreshed.
-    async fn refreshed_computed_view(conn: &Connection) -> MaterializedView {
-        use crate::materialized_view::tests::{computed_field, people, test_binding};
-        let source = people(conn).await;
-        let view = crate::materialized_view::prepare_declaration(
-            &source,
-            Some(&[
-                ("id".to_string(), "id".to_string()),
-                ("name".to_string(), "name".to_string()),
-            ]),
-            None,
-            None,
-        )
-        .await
-        .unwrap()
-        .with_computed_columns(
-            vec![(2, computed_field("emb", "fb_1", "name"))],
-            &[test_binding("fb_1", "name", "emb")],
-        )
-        .unwrap()
-        .create("v")
-        .await
-        .unwrap();
-        let result = view.refresh().execute().await.unwrap();
-        assert_eq!(result.mode, RefreshMode::Rebuild);
-        view
-    }
-
-    async fn unfilled(view: &MaterializedView) -> usize {
-        view.table()
-            .count_rows(Some("emb IS NULL".to_string()))
-            .await
-            .unwrap()
-    }
-
-    async fn append_people(conn: &Connection, ids: Vec<i32>, names: Vec<&str>) {
-        let batch = record_batch!(("id", Int32, ids), ("name", Utf8, names)).unwrap();
-        conn.open_table("people")
-            .execute()
-            .await
-            .unwrap()
-            .add(batch)
-            .execute()
-            .await
-            .unwrap();
-    }
-
-    /// Commit the fill job's shape on the view: a column rewrite of
-    /// `fields`, touching no rows. The data is left as it is; what matters
-    /// here is how the next refresh classifies the commit.
-    async fn commit_column_rewrite(view: &MaterializedView, fields: &[&str]) {
-        let native = view.table().as_native().unwrap();
-        native.dataset.reload().await.unwrap();
-        let dataset = native.dataset.get().await.unwrap().as_ref().clone();
-        let fields_modified = fields
-            .iter()
-            .map(|name| dataset.schema().field(name).unwrap().id as u32)
-            .collect();
-        let updated_fragments = dataset
-            .get_fragments()
-            .iter()
-            .map(|fragment| fragment.metadata().clone())
-            .collect();
-        let operation = Operation::Update {
-            removed_fragment_ids: Vec::new(),
-            updated_fragments,
-            new_fragments: Vec::new(),
-            fields_modified,
-            compacted_sstables: Vec::new(),
-            fields_for_preserving_frag_bitmap: Vec::new(),
-            update_mode: Some(UpdateMode::RewriteColumns),
-            inserted_rows_filter: None,
-            updated_fragment_offsets: None,
-        };
-        let read_version = dataset.version().version;
-        CommitBuilder::new(WriteDestination::Dataset(Arc::new(dataset)))
-            .execute(Transaction::new(read_version, operation, None))
-            .await
-            .unwrap();
-    }
-
-    /// Refresh never computes a computed column: every row it writes, on a
-    /// rebuild, an append and a rewrite, carries NULL there, and the
-    /// declaration survives all three.
-    #[tokio::test]
-    async fn test_computed_columns_are_written_null_and_kept() {
-        let conn = connect("memory://").execute().await.unwrap();
-        let view = refreshed_computed_view(&conn).await;
-        assert_eq!(unfilled(&view).await, 3);
-
-        append_people(&conn, vec![4, 5], vec!["d", "e"]).await;
-        let result = view.refresh().execute().await.unwrap();
-        assert_eq!(result.mode, RefreshMode::Incremental);
-        assert_eq!(unfilled(&view).await, 5);
-
-        conn.open_table("people")
-            .execute()
-            .await
-            .unwrap()
-            .update()
-            .column("name", "'z'")
-            .only_if("id = 1")
-            .execute()
-            .await
-            .unwrap();
-        view.refresh().execute().await.unwrap();
-        assert_eq!(unfilled(&view).await, 5);
-        assert_eq!(read(view.table(), "id").await, vec![1, 2, 3, 4, 5]);
-
-        let schema = view.table().schema().await.unwrap();
-        assert!(
-            crate::table::computed_columns::function_bindings(&schema)
-                .unwrap()
-                .iter()
-                .any(|b| b.binding_id() == "fb_1"),
-            "the binding envelope was lost"
-        );
-        assert!(
-            computed_column_from_field(schema.field_with_name("emb").unwrap()).is_some(),
-            "the declaration was lost"
-        );
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::NoOp
-        );
-    }
-
-    /// Field metadata on `field` only, the commit shape of the freshness
-    /// stamp `refresh_column` leaves after its fill.
-    async fn commit_field_metadata(view: &MaterializedView, field: &str, key: &str) {
-        let native = view.table().as_native().unwrap();
-        native.dataset.reload().await.unwrap();
-        let mut dataset = native.dataset.get().await.unwrap().as_ref().clone();
-        dataset
-            .update_field_metadata()
-            .update(field, [(key.to_string(), "{}".to_string())])
-            .unwrap()
-            .await
-            .unwrap();
-    }
-
-    /// The stamp is metadata on the computed column and rewrites nothing
-    /// refresh certifies, so it is not drift; the same commit shape on a
-    /// projected column is, like any other write to it.
-    #[tokio::test]
-    async fn test_a_freshness_stamp_is_not_drift() {
-        let conn = connect("memory://").execute().await.unwrap();
-        let view = refreshed_computed_view(&conn).await;
-
-        commit_field_metadata(
-            &view,
-            "emb",
-            crate::table::computed_columns::SOURCE_SIGNATURE_META_KEY,
-        )
-        .await;
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::NoOp
-        );
-
-        commit_field_metadata(&view, "id", "probe").await;
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::Rebuild
-        );
-    }
-
-    /// The fill job's commit rewrites only computed columns. It is the one
-    /// commit on a view that is not drift: the next refresh carries on from
-    /// its watermark instead of rebuilding, which would null what the fill
-    /// just wrote.
-    #[tokio::test]
-    async fn test_a_computed_column_fill_is_not_drift() {
-        let conn = connect("memory://").execute().await.unwrap();
-        let view = refreshed_computed_view(&conn).await;
-
-        commit_column_rewrite(&view, &["emb"]).await;
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::NoOp
-        );
-
-        commit_column_rewrite(&view, &["emb"]).await;
-        append_people(&conn, vec![4], vec!["d"]).await;
-        let result = view.refresh().execute().await.unwrap();
-        assert_eq!(result.mode, RefreshMode::Incremental);
-        assert_eq!(result.rows_written, 1);
-        assert_eq!(read(view.table(), "id").await, vec![1, 2, 3, 4]);
-    }
-
-    /// A column rewrite that reaches a projected column is drift like any
-    /// other write: refresh certifies those columns and must recompute them.
-    #[tokio::test]
-    async fn test_a_rewrite_of_a_projected_column_is_drift() {
-        let conn = connect("memory://").execute().await.unwrap();
-        let view = refreshed_computed_view(&conn).await;
-
-        commit_column_rewrite(&view, &["emb", "name"]).await;
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::Rebuild
-        );
-    }
-
-    /// The declaration contract is checked before any refresh mutation: a
-    /// missing binding envelope and a column that lost its declaration both
-    /// fail closed.
-    #[tokio::test]
-    async fn test_a_broken_declaration_is_refused_before_refresh() {
-        let conn = connect("memory://").execute().await.unwrap();
-        let view = refreshed_computed_view(&conn).await;
-        let native = view.table().as_native().unwrap();
-        let mut dataset = native.dataset.get().await.unwrap().as_ref().clone();
-        dataset
-            .update_schema_metadata(vec![(
-                crate::table::computed_columns::FUNCTION_BINDINGS_META_KEY.to_string(),
-                None,
-            )])
-            .await
-            .unwrap();
-        let err = view.refresh().execute().await.unwrap_err().to_string();
-        assert!(err.contains("references missing binding 'fb_1'"), "{err}");
-
-        let conn = connect("memory://").execute().await.unwrap();
-        let view = refreshed_computed_view(&conn).await;
-        let native = view.table().as_native().unwrap();
-        let mut dataset = native.dataset.get().await.unwrap().as_ref().clone();
-        dataset
-            .replace_field_metadata(vec![(
-                dataset.schema().field("emb").unwrap().id as u32,
-                HashMap::new(),
-            )])
-            .await
-            .unwrap();
-        let err = view.refresh().execute().await.unwrap_err().to_string();
-        assert!(err.contains("does not match binding 'fb_1'"), "{err}");
-    }
-
-    /// An input the view does not project is materialized on every refresh
-    /// path, before the provenance column, with the source's values.
-    #[tokio::test]
-    async fn test_internal_inputs_are_materialized_and_refreshed() {
-        use crate::materialized_view::tests::{computed_field, strict_people, test_binding};
-        let conn = connect("memory://").execute().await.unwrap();
-        let source = strict_people(&conn).await;
-        let mut prepared = crate::materialized_view::prepare_declaration(
-            &source,
-            Some(&[("id".to_string(), "id".to_string())]),
-            None,
-            None,
-        )
-        .await
-        .unwrap();
-        let input = prepared.input_column("name").unwrap();
-        let view = prepared
-            .with_computed_columns(
-                vec![(1, computed_field("emb", "fb_1", &input))],
-                &[test_binding("fb_1", &input, "emb")],
-            )
-            .unwrap()
-            .create("v")
-            .await
-            .unwrap();
-        let names: Vec<String> = view
-            .table()
-            .schema()
-            .await
-            .unwrap()
-            .fields()
-            .iter()
-            .map(|f| f.name().clone())
-            .collect();
-        assert_eq!(names, ["id", "emb", "__input_name", SOURCE_ROW_ID_COLUMN]);
-
-        let unfilled_inputs = || async {
-            view.table()
-                .count_rows(Some("__input_name IS NULL".to_string()))
-                .await
-                .unwrap()
-        };
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::Rebuild
-        );
-        assert_eq!(view.table().count_rows(None).await.unwrap(), 3);
-        assert_eq!(unfilled_inputs().await, 0);
-
-        let more = arrow_array::RecordBatch::try_new(
-            source.schema().await.unwrap(),
-            vec![
-                Arc::new(Int32Array::from(vec![4])),
-                Arc::new(arrow_array::StringArray::from(vec!["d"])),
-            ],
-        )
-        .unwrap();
-        source.add(more).execute().await.unwrap();
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::Incremental
-        );
-        assert_eq!(unfilled_inputs().await, 0);
-        assert_eq!(
-            view.table()
-                .count_rows(Some("__input_name = 'd'".to_string()))
-                .await
-                .unwrap(),
-            1
-        );
-
-        source
-            .update()
-            .column("name", "'z'")
-            .only_if("id = 1")
-            .execute()
-            .await
-            .unwrap();
-        view.refresh().execute().await.unwrap();
-        assert_eq!(
-            view.table()
-                .count_rows(Some("__input_name = 'z'".to_string()))
-                .await
-                .unwrap(),
-            1
-        );
-        assert_eq!(
-            unfilled(&view).await,
-            4,
-            "rewritten and new rows are unfilled"
-        );
-    }
-
-    /// A SQL declaration is filled by `refresh_column` on the view, which
-    /// commits a data replacement and then its freshness stamp; the next
-    /// refresh continues from its watermark and keeps what the fill wrote,
-    /// and only rows the view added since come back unfilled.
-    #[tokio::test]
-    async fn test_a_sql_fill_is_not_drift() {
-        use crate::materialized_view::tests::{people, sql_field};
-        let conn = connect("memory://").execute().await.unwrap();
-        let source = people(&conn).await;
-        let view = crate::materialized_view::prepare_declaration(
-            &source,
-            Some(&[("id".to_string(), "id".to_string())]),
-            None,
-            None,
-        )
-        .await
-        .unwrap()
-        .with_computed_columns(
-            vec![(
-                1,
-                sql_field("next", arrow_schema::DataType::Int32, "id + 1", r#"["id"]"#),
-            )],
-            &[],
-        )
-        .unwrap()
-        .create("v")
-        .await
-        .unwrap();
-        let filled = || async {
-            view.table()
-                .count_rows(Some("next = id + 1".to_string()))
-                .await
-                .unwrap()
-        };
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::Rebuild
-        );
-        assert_eq!(
-            view.table()
-                .refresh_column("next")
-                .await
-                .unwrap()
-                .rows_filled,
-            3
-        );
-        assert_eq!(filled().await, 3);
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::NoOp
-        );
-        assert_eq!(filled().await, 3);
-
-        append_people(&conn, vec![4], vec!["d"]).await;
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::Incremental
-        );
-        assert_eq!(filled().await, 3);
-        assert_eq!(
-            view.table()
-                .refresh_column("next")
-                .await
-                .unwrap()
-                .rows_filled,
-            1
-        );
-        assert_eq!(filled().await, 4);
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::NoOp
-        );
-    }
-
-    /// A fill of a nested computed column writes its child fields; that is
-    /// still a fill, not drift.
-    #[tokio::test]
-    async fn test_a_nested_sql_fill_is_not_drift() {
-        use crate::materialized_view::tests::{people, sql_field};
-        let conn = connect("memory://").execute().await.unwrap();
-        let source = people(&conn).await;
-        let payload = sql_field(
-            "payload",
-            arrow_schema::DataType::Struct(
-                vec![arrow_schema::Field::new(
-                    "value",
-                    arrow_schema::DataType::Utf8,
-                    true,
-                )]
-                .into(),
-            ),
-            "named_struct('value', name)",
-            r#"["name"]"#,
-        );
-        let view = crate::materialized_view::prepare_declaration(
-            &source,
-            Some(&[("name".to_string(), "name".to_string())]),
-            None,
-            None,
-        )
-        .await
-        .unwrap()
-        .with_computed_columns(vec![(1, payload)], &[])
-        .unwrap()
-        .create("v")
-        .await
-        .unwrap();
-        view.refresh().execute().await.unwrap();
-        assert_eq!(
-            view.table()
-                .refresh_column("payload")
-                .await
-                .unwrap()
-                .rows_filled,
-            3
-        );
-        assert_eq!(
-            view.refresh().execute().await.unwrap().mode,
-            RefreshMode::NoOp
-        );
-        assert_eq!(
-            view.table()
-                .count_rows(Some("payload.value = name".to_string()))
-                .await
-                .unwrap(),
-            3
-        );
-    }
 }
@@ -1299,7 +1299,7 @@ impl VectorQuery {
    /// This can be useful when there is a narrow filter to allow these queries to
    /// spend more time searching and avoid potential false negatives.
    ///
-    /// Set to None to search all partitions, if needed, to satisfy the limit
+    /// Set to None to search all partitions, if needed, to satsify the limit
    pub fn maximum_nprobes(mut self, maximum_nprobes: Option<usize>) -> Result<Self> {
        if let Some(maximum_nprobes) = maximum_nprobes {
            if maximum_nprobes == 0 {
@@ -8021,7 +8021,7 @@ mod tests {
            match request.url().path() {
                "/v1/table/my_table/backfill_column" => http::Response::builder()
                    .status(202)
-                    .body(br#"{"job_id": "j-42"}"#.to_vec())
+                    .body(r#"{"job_id": "j-42"}"#.as_bytes().to_vec())
                    .unwrap(),
                "/v1/jobs/describe" => http::Response::builder()
                    .status(200)
@@ -75,7 +75,6 @@ mod create_index;
 pub mod datafusion;
 pub(crate) mod dataset;
 pub mod delete;
-pub mod freshness;
 pub mod lsm_stats;
 pub mod merge;
 pub mod optimize;
@@ -241,7 +240,7 @@ enum BadVectorHandling {
    /// An error is returned
    #[default]
    Error,
-    /// The offending row is dropped
+    /// The offending row is droppped
    Drop,
    /// The invalid/missing items are replaced by fill_value
    Fill(f32),
@@ -779,8 +778,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
            message: "Function columns are supported only on LanceDB Cloud and Enterprise".into(),
        })
    }
-    /// Fill a computed column's unfilled rows and recompute those whose
-    /// inputs changed.
+    /// 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> {
@@ -788,8 +786,8 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
            message: "computed columns are supported only on local tables".into(),
        })
    }
-    /// Fill a computed column's unfilled rows and recompute those whose
-    /// inputs changed, returning a [`Job`] tracking the operation.
+    /// Fill a computed column's unfilled rows, returning a [`Job`] tracking
+    /// the operation.
    async fn refresh_column_async(
        &self,
        _column: &str,
@@ -1328,7 +1326,7 @@ impl Table {
    /// Note: if your condition is something like "some_id_column == 7" and
    /// you are updating many rows (with different ids) then you will get
    /// better performance with a single [`merge_insert`] call instead of
-    /// repeatedly calling this method.
+    /// repeatedly calilng this method.
    pub fn update(&self) -> UpdateBuilder {
        UpdateBuilder::new(self.inner.clone())
    }
@@ -1751,10 +1749,9 @@ impl Table {
    /// 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, and fragments whose
-    /// inputs changed since they were computed are recomputed (see
-    /// [`freshness`](crate::table::freshness)); everything else is left as
-    /// it is.
+    /// 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`].
@@ -2807,7 +2804,7 @@ impl NativeTable {
        namespace_client: Option<Arc<dyn LanceNamespace>>,
        pushdown_operations: HashSet<NamespaceClientPushdownOperation>,
    ) -> Result<Self> {
-        let batches = computed_columns::admit_create_source(batches)?;
+        computed_columns::ensure_no_foreign_declarations(batches.arrow_schema().fields())?;
        // Default params uses format v1.
        let params = params.unwrap_or(WriteParams {
            ..Default::default()
@@ -2907,7 +2904,6 @@ impl NativeTable {
        pushdown_operations: HashSet<NamespaceClientPushdownOperation>,
        session: Option<Arc<lance::session::Session>>,
    ) -> Result<Self> {
-        let batches = computed_columns::admit_create_source(batches)?;
        // Build table_id from namespace + name for the storage options provider
        let mut table_id = namespace.clone();
        table_id.push(name.to_string());
@@ -5681,7 +5677,7 @@ mod tests {
            TableStatistics {
                num_rows: 250,
                num_indices: 0,
-                total_bytes: 8969,
+                total_bytes: 8925,
                fragment_stats: FragmentStatistics {
                    num_fragments: 11,
                    num_small_fragments: 11,
@@ -60,11 +60,10 @@ impl AddColumnsBuilder {
    /// every fragment that has none -- including fragments appended since the
    /// last refresh.
    ///
-    /// A refresh also recomputes the rows of a fragment whose inputs changed
-    /// since it was computed (see [`freshness`](super::freshness)), so a
-    /// mutated input is reflected by the next refresh. An input cannot be
-    /// renamed, retyped or dropped while a declaration reads it, since the
-    /// expression names it.
+    /// 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
@@ -21,7 +21,6 @@
 //! [`computed_columns`] and [`computed_column_from_field`] read declarations
 //! back off a schema.

-use futures::StreamExt;
 use std::collections::{BTreeSet, HashMap, HashSet};
 use std::sync::Arc;

@@ -71,22 +70,6 @@ pub const FUNCTION_BINDINGS_META_KEY: &str = "lancedb::function_bindings";
 /// Version of the schema-level Function binding envelope.
 pub const FUNCTION_BINDINGS_VERSION: u32 = 1;

-/// Field metadata key holding `{fragment id -> input signature}` as JSON,
-/// recorded by the refresh that last computed each fragment. Outside the
-/// declaration namespace on purpose: a declaration is immutable through
-/// metadata edits, this is rewritten by every refresh. Seeded empty at
-/// declaration, so a column is tracked from birth; a column without it was
-/// declared before signatures existed.
-pub const SOURCE_SIGNATURE_META_KEY: &str = "computed_refresh.source_signature";
-
-/// Field metadata key holding the definition digest a column was last
-/// computed under. A change to it makes every row stale.
-pub const DEFINITION_VERSION_META_KEY: &str = "computed_refresh.definition_version";
-
-/// Field metadata key holding the table version the signature map describes:
-/// where a refresh starts following compactions to carry freshness forward.
-pub const RECORDED_AT_VERSION_META_KEY: &str = "computed_refresh.recorded_at_version";
-
 /// Value of [`KIND_META_KEY`] for a column defined by a SQL expression.
 pub const SQL_KIND: &str = "sql";

@@ -155,7 +138,6 @@ fn computed_column_metadata(expression: &str, inputs: &[String]) -> HashMap<Stri
            INPUTS_META_KEY.to_string(),
            serde_json::to_string(inputs).unwrap_or_else(|_| "[]".to_string()),
        ),
-        (SOURCE_SIGNATURE_META_KEY.to_string(), "{}".to_string()),
    ])
 }

@@ -180,7 +162,6 @@ pub fn function_computed_column_metadata(
            INPUTS_META_KEY.to_string(),
            serde_json::to_string(inputs).unwrap_or_else(|_| "[]".to_string()),
        ),
-        (SOURCE_SIGNATURE_META_KEY.to_string(), "{}".to_string()),
    ])
 }

@@ -778,33 +759,14 @@ fn function_output_field(name: &str, nullable: bool, raw: &str) -> Result<JsonAr
    Ok(field)
 }

-/// Whether two fields describe the same Function output.
-///
-/// `compare_identity` covers the field's own name and nullability. Struct
-/// children carry both as part of the declaration and compare with it on. List
-/// children do not: Lance rewrites a list item's name and nullability when it
-/// writes, so a stored `fixed_size_list<item: float not null>` comes back as
-/// `fixed_size_list<item: float>` and never matches the declaration again.
-/// Comparing those by type alone keeps this agreeing with the server, which
-/// draws the same distinction and is what accepted the column when it was
-/// declared.
-fn function_output_field_matches(
-    expected: &ArrowField,
-    actual: &ArrowField,
-    compare_identity: bool,
-) -> bool {
-    if compare_identity
-        && (expected.name() != actual.name() || expected.is_nullable() != actual.is_nullable())
-    {
-        return false;
-    }
-    match (expected.is_blob_v2(), actual.is_blob_v2()) {
-        (false, false) => function_output_type_matches(expected.data_type(), actual.data_type()),
-        (true, true) => {
+fn function_output_field_matches(expected: &ArrowField, actual: &ArrowField) -> bool {
+    expected.name() == actual.name()
+        && expected.is_nullable() == actual.is_nullable()
+        && if expected.is_blob_v2() {
            has_supported_blob_v2_layout(expected) && has_supported_blob_v2_layout(actual)
+        } else {
+            function_output_type_matches(expected.data_type(), actual.data_type())
        }
-        _ => false,
-    }
 }

 fn function_output_type_matches(expected: &DataType, actual: &DataType) -> bool {
@@ -817,19 +779,33 @@ fn function_output_type_matches(expected: &DataType, actual: &DataType) -> bool
                && expected
                    .iter()
                    .zip(actual)
-                    .all(|(expected, actual)| function_output_field_matches(expected, actual, true))
+                    .all(|(expected, actual)| function_output_field_matches(expected, actual))
        }
        (DataType::List(expected), DataType::List(actual))
        | (DataType::LargeList(expected), DataType::LargeList(actual)) => {
-            function_output_field_matches(expected, actual, false)
+            function_output_field_matches(expected, actual)
        }
        (
            DataType::FixedSizeList(expected, expected_size),
            DataType::FixedSizeList(actual, actual_size),
-        ) => expected_size == actual_size && function_output_field_matches(expected, actual, false),
+        ) => expected_size == actual_size && function_output_field_matches(expected, actual),
        (DataType::Map(expected, expected_sorted), DataType::Map(actual, actual_sorted)) => {
-            expected_sorted == actual_sorted
-                && function_output_field_matches(expected, actual, true)
+            expected_sorted == actual_sorted && function_output_field_matches(expected, actual)
+        }
+        _ => false,
+    }
+}
+
+fn function_output_type_has_blob(data_type: &DataType) -> bool {
+    match data_type {
+        DataType::Struct(fields) => fields
+            .iter()
+            .any(|field| field.is_blob_v2() || function_output_type_has_blob(field.data_type())),
+        DataType::List(field)
+        | DataType::LargeList(field)
+        | DataType::FixedSizeList(field, _)
+        | DataType::Map(field, _) => {
+            field.is_blob_v2() || function_output_type_has_blob(field.data_type())
        }
        _ => false,
    }
@@ -915,13 +891,16 @@ fn ensure_binding_matches_schema(schema: &ArrowSchema, binding: &FunctionBinding
                binding.binding_id()
            )));
        }
-        let type_matches = if output.arrow_type == FUNCTION_BLOB_V2_TYPE {
-            has_supported_blob_v2_layout(field)
+        let (type_matches, has_semantic_blob) = if output.arrow_type == FUNCTION_BLOB_V2_TYPE {
+            (has_supported_blob_v2_layout(field), true)
        } else {
            let expected_type = parse_output_arrow_type(&output.arrow_type)?;
            let expected_type = lance_namespace::schema::convert_json_arrow_type(&expected_type)
                .map_err(|e| invalid_function(format!("invalid Function output type: {e}")))?;
-            function_output_type_matches(&expected_type, field.data_type())
+            (
+                function_output_type_matches(&expected_type, field.data_type()),
+                function_output_type_has_blob(&expected_type),
+            )
        };
        if !type_matches {
            return Err(invalid_function(format!(
@@ -952,16 +931,19 @@ fn ensure_binding_matches_schema(schema: &ArrowSchema, binding: &FunctionBinding
                binding.binding_id()
            )));
        }
-        // Rebuild from the declaration rather than from the stored field. The
-        // stored field carries Lance's write-time normalization, which would
-        // never round-trip back to the schema the binding recorded -- the same
-        // reason list children compare by type above. Whether the column on
-        // disk still matches is settled by that comparison, not here.
-        output_fields.push(function_output_field(
-            field.name(),
-            true,
-            &output.arrow_type,
-        )?);
+        if has_semantic_blob {
+            output_fields.push(function_output_field(
+                field.name(),
+                true,
+                &output.arrow_type,
+            )?);
+        } else {
+            let json = lance_namespace::schema::arrow_schema_to_json(&ArrowSchema::new(vec![
+                ArrowField::new(field.name().clone(), field.data_type().clone(), true),
+            ]))
+            .map_err(|e| invalid_function(format!("invalid Function output schema: {e}")))?;
+            output_fields.push(json.fields.into_iter().next().unwrap());
+        }
    }
    if let Some(assignment) = binding.assignment() {
        if binding
@@ -1313,8 +1295,7 @@ pub(crate) fn ensure_not_an_input(schema: &SchemaRef, paths: &[&str]) -> Result<
 }

 /// Reject a write that supplies values for a computed column directly:
-/// only refresh materializes one, and only refresh decides what it
-/// recomputes.
+/// only refresh materializes one, and refresh never revisits a filled row.
 pub(crate) fn ensure_not_written<'a>(
    schema: &ArrowSchema,
    written: impl IntoIterator<Item = &'a str>,
@@ -1358,106 +1339,6 @@ pub(crate) fn ensure_batch_writes_no_computed_values(
    Ok(())
 }

-/// Validate every computed-column declaration `schema` carries against the
-/// schema itself: every field with declaration metadata is a complete
-/// declaration, a SQL declaration re-plans to the field it declares, a
-/// Function declaration satisfies the binding contract, and no declaration
-/// reads another computed column. What passes here is what `refresh_column`
-/// can execute.
-pub(crate) fn ensure_declarations_are_planned(schema: &ArrowSchema) -> Result<()> {
-    let invalid = |message: String| Error::InvalidInput { message };
-    // A field with any declaration key is a declaration; a partial one is
-    // not "no declaration", it is a broken one.
-    for field in schema.fields() {
-        if field.metadata().keys().any(|k| is_declaration_key(k))
-            && computed_column_from_field(field).is_none()
-        {
-            return Err(invalid(format!(
-                "field '{}' carries an incomplete computed-column declaration",
-                field.name()
-            )));
-        }
-    }
-    let declared: HashSet<String> = computed_columns(schema)
-        .into_iter()
-        .map(|c| c.name)
-        .collect();
-    for column in computed_columns(schema) {
-        let field = schema.field_with_name(&column.name)?;
-        if !field.is_nullable() {
-            return Err(invalid(format!(
-                "computed column '{}' must be nullable until a refresh fills it",
-                column.name
-            )));
-        }
-        match &column.kind {
-            ComputedColumnKind::Sql { expression } => {
-                let others: Vec<ArrowField> = schema
-                    .fields()
-                    .iter()
-                    .filter(|f| f.name() != &column.name)
-                    .map(|f| f.as_ref().clone())
-                    .collect();
-                let bound = bind(Arc::new(ArrowSchema::new(others)), &column.name, expression)?;
-                if let Some(input) = bound.roots.iter().find(|r| declared.contains(*r)) {
-                    return Err(invalid(format!(
-                        "computed column '{}' reads computed column '{input}'",
-                        column.name
-                    )));
-                }
-                if &bound.data_type != field.data_type() {
-                    return Err(invalid(format!(
-                        "computed column '{}' is declared as {} but its expression yields {}",
-                        column.name,
-                        field.data_type(),
-                        bound.data_type
-                    )));
-                }
-                let mut declared_inputs = column.inputs.clone();
-                declared_inputs.sort();
-                if declared_inputs != bound.inputs {
-                    return Err(invalid(format!(
-                        "computed column '{}' declares inputs {:?} but its expression reads {:?}",
-                        column.name, declared_inputs, bound.inputs
-                    )));
-                }
-            }
-            ComputedColumnKind::Function { binding_id, .. } => {
-                // The binding validator resolves each input's leaf; the
-                // no-computed-input rule is about the root it hangs from.
-                let bindings = function_bindings(schema)?;
-                let Some(binding) = bindings.iter().find(|b| b.binding_id() == binding_id) else {
-                    continue; // reported by the binding validator below
-                };
-                // Roots come from the canonical path parser: a quoted
-                // top-level name may itself contain a dot.
-                if let Some(input) = binding
-                    .inputs()
-                    .iter()
-                    .filter_map(|input| resolve_field_path(schema, &input.field_path).ok())
-                    .map(|resolved| resolved.root.name().as_str())
-                    .find(|r| declared.contains(*r))
-                {
-                    return Err(invalid(format!(
-                        "computed column '{}' reads computed column '{input}'",
-                        column.name
-                    )));
-                }
-            }
-            ComputedColumnKind::Unrecognized { kind } => {
-                return Err(Error::NotSupported {
-                    message: format!(
-                        "computed column '{}' is defined by '{kind}', which this version \
-                         of lancedb cannot fill",
-                        column.name
-                    ),
-                });
-            }
-        }
-    }
-    ensure_supported_function_metadata(schema)
-}
-
 /// Reject fields carrying declaration metadata that did not come through
 /// [`plan`]. One authority for creation, overwrite and raw transforms.
 pub(crate) fn ensure_no_foreign_declarations<'a>(
@@ -1489,9 +1370,7 @@ fn ensure_no_foreign_declaration(field: &ArrowField) -> Result<()> {
 /// kind, the expression, the inputs -- would bypass that validation or move
 /// a binding out from under a refresh. Drop the column and declare it again.
 pub(crate) fn is_declaration_key(key: &str) -> bool {
-    key == COMPUTED_COLUMN_META_KEY
-        || key.starts_with("computed_column.")
-        || key.starts_with("computed_refresh.")
+    key == COMPUTED_COLUMN_META_KEY || key.starts_with("computed_column.")
 }

 /// Reject retyping a computed column itself.
@@ -1918,54 +1797,6 @@ pub(super) async fn add_foreign_kind(table: &crate::Table, name: &str, kind: &st
    .unwrap();
 }

-/// Admit a table's initial data: every declaration it carries is validated,
-/// and the stream refuses any batch with values in a computed column, whose
-/// values come from refresh alone. One boundary for every way a table is
-/// created.
-pub(crate) fn admit_create_source<S: lance_datafusion::utils::StreamingWriteSource>(
-    batches: S,
-) -> Result<UnfilledDeclarations<S>> {
-    let schema = batches.arrow_schema();
-    ensure_declarations_are_planned(&schema)?;
-    let declared = computed_columns(&schema)
-        .into_iter()
-        .map(|c| c.name)
-        .collect();
-    Ok(UnfilledDeclarations {
-        inner: batches,
-        declared,
-    })
-}
-
-/// A write source whose computed columns must arrive unfilled.
-pub(crate) struct UnfilledDeclarations<S> {
-    inner: S,
-    declared: Vec<String>,
-}
-
-impl<S: lance_datafusion::utils::StreamingWriteSource> lance_datafusion::utils::StreamingWriteSource
-    for UnfilledDeclarations<S>
-{
-    fn arrow_schema(&self) -> SchemaRef {
-        self.inner.arrow_schema()
-    }
-
-    fn into_stream(self) -> datafusion_physical_plan::SendableRecordBatchStream {
-        if self.declared.is_empty() {
-            return self.inner.into_stream();
-        }
-        let schema = self.inner.arrow_schema();
-        let declared = self.declared;
-        let stream = self.inner.into_stream().map(move |batch| {
-            let batch = batch?;
-            ensure_batch_writes_no_computed_values(&declared, &batch)
-                .map_err(|e| datafusion_common::DataFusionError::External(Box::new(e)))?;
-            Ok(batch)
-        });
-        Box::pin(datafusion_physical_plan::stream::RecordBatchStreamAdapter::new(schema, stream))
-    }
-}
-
 #[cfg(test)]
 mod tests {
    /// The gate's reproducer: the validator applies the same schema-level
@@ -1984,69 +1815,6 @@ mod tests {
        assert!(super::validate_declarations(schema, &declarations).is_err());
    }

-    #[test]
-    fn list_children_match_by_type_but_struct_children_by_identity() {
-        use arrow_schema::Field as F;
-
-        // Lance rewrites a list item's name and nullability on write, so the
-        // stored field is no longer identical to what was declared. Comparing
-        // those by type keeps a table with a vector output usable.
-        let declared =
-            DataType::FixedSizeList(Arc::new(F::new("item", DataType::Float32, false)), 4);
-        let stored = DataType::FixedSizeList(Arc::new(F::new("item", DataType::Float32, true)), 4);
-        assert!(super::function_output_type_matches(&declared, &stored));
-
-        let renamed =
-            DataType::FixedSizeList(Arc::new(F::new("element", DataType::Float32, true)), 4);
-        assert!(super::function_output_type_matches(&declared, &renamed));
-
-        // The dimension is still part of the declaration.
-        let resized = DataType::FixedSizeList(Arc::new(F::new("item", DataType::Float32, true)), 8);
-        assert!(!super::function_output_type_matches(&declared, &resized));
-
-        // Struct children keep comparing by name and nullability.
-        let struct_declared =
-            DataType::Struct(vec![F::new("changed", DataType::Boolean, false)].into());
-        let struct_nullable =
-            DataType::Struct(vec![F::new("changed", DataType::Boolean, true)].into());
-        let struct_renamed =
-            DataType::Struct(vec![F::new("altered", DataType::Boolean, false)].into());
-        assert!(super::function_output_type_matches(
-            &struct_declared,
-            &struct_declared
-        ));
-        assert!(!super::function_output_type_matches(
-            &struct_declared,
-            &struct_nullable
-        ));
-        assert!(!super::function_output_type_matches(
-            &struct_declared,
-            &struct_renamed
-        ));
-
-        // A list nested inside a struct gets the list rule.
-        let nested_declared = DataType::Struct(
-            vec![F::new(
-                "tokens",
-                DataType::List(Arc::new(F::new("item", DataType::Utf8, false))),
-                true,
-            )]
-            .into(),
-        );
-        let nested_stored = DataType::Struct(
-            vec![F::new(
-                "tokens",
-                DataType::List(Arc::new(F::new("item", DataType::Utf8, true))),
-                true,
-            )]
-            .into(),
-        );
-        assert!(super::function_output_type_matches(
-            &nested_declared,
-            &nested_stored
-        ));
-    }
-
    #[test]
    fn output_arrow_type_grammar_matches_the_shared_golden() {
        let golden: serde_json::Value = serde_json::from_str(include_str!(
@@ -2816,8 +2584,6 @@ mod tests {
        );
    }

-    /// A create carries a declaration only if it re-plans completely; this
-    /// one lacks its inputs and is refused before its forged value matters.
    #[tokio::test]
    async fn test_create_table_cannot_inject_a_declaration() {
        let conn = connect("memory://").execute().await.unwrap();
@@ -2845,7 +2611,7 @@ mod tests {
            .await
            .unwrap_err();
        assert!(
-            matches!(&err, Error::InvalidInput { message } if message.contains("computed column 'doubled'")),
+            matches!(&err, Error::InvalidInput { message } if message.contains("computed()")),
            "{err:?}"
        );
    }
@@ -3523,70 +3289,6 @@ mod tests {
        assert!(output_schema.field(0).is_blob_v2());
    }

-    #[test]
-    fn binding_accepts_a_lance_normalized_list_child() {
-        // The whole guard, not just the type helper: this also reaches the
-        // output-schema comparison at the end of ensure_binding_matches_schema,
-        // which used to rebuild the schema from the stored field and so failed
-        // on exactly the same normalization.
-        let input = ArrowField::new("value", DataType::Int64, false);
-        let application = FunctionApplication::from_json(
-            &serde_json::json!({
-                "function": {"name": "embed", "version": "fv_embed"},
-                "inputs": [{
-                    "parameter": "value",
-                    "kind": "column",
-                    "value": {"path": "value"}
-                }],
-                "output": {
-                    "kind": "scalar",
-                    "arrow_type": "fixed_size_list<float32, 4>",
-                    "nullable": false
-                }
-            })
-            .to_string(),
-        )
-        .unwrap();
-        let plan = plan_function_application(
-            &ArrowSchema::new(vec![input.clone()]),
-            &application,
-            Some("embedding"),
-        )
-        .unwrap();
-        let binding = binding_from_plan(&plan);
-
-        // The declaration says the item is non-nullable; Lance rewrites it to
-        // nullable on write, so this is what the column looks like on disk.
-        let stored = DataType::FixedSizeList(
-            Arc::new(ArrowField::new("item", DataType::Float32, true)),
-            4,
-        );
-        let output = ArrowField::new("embedding", stored, true).with_metadata(
-            function_computed_column_metadata(binding.binding_id(), 0, &["value".into()]),
-        );
-
-        ensure_binding_matches_schema(&ArrowSchema::new(vec![input.clone(), output]), &binding)
-            .unwrap();
-
-        // A different element type is still a mismatch.
-        let wrong = ArrowField::new(
-            "embedding",
-            DataType::FixedSizeList(
-                Arc::new(ArrowField::new("item", DataType::Float64, true)),
-                4,
-            ),
-            true,
-        )
-        .with_metadata(function_computed_column_metadata(
-            binding.binding_id(),
-            0,
-            &["value".into()],
-        ));
-        assert!(
-            ensure_binding_matches_schema(&ArrowSchema::new(vec![input, wrong]), &binding).is_err()
-        );
-    }
-
    #[test]
    fn test_blob_scalar_binding_accepts_full_logical_layout() {
        let input = crate::blob("image", false);
@@ -5,17 +5,12 @@
 //!
 //! [`super::cast::cast_to_table_schema`] calls [`coerce_blob_expr`].

-use std::fmt;
-use std::hash::{Hash, Hasher};
 use std::sync::Arc;

-use arrow_array::{Array, BooleanArray, RecordBatch};
-use arrow_schema::{DataType, Field, FieldRef, Fields, Schema};
-use arrow_select::nullif::nullif;
+use arrow_schema::{DataType, Field, FieldRef, Fields};
 use datafusion::functions::core::{get_field, named_struct};
 use datafusion_common::ScalarValue;
 use datafusion_common::config::ConfigOptions;
-use datafusion_expr::ColumnarValue;
 use datafusion_physical_expr::ScalarFunctionExpr;
 use datafusion_physical_expr::expressions::{CastExpr, Literal};
 use datafusion_physical_plan::PhysicalExpr;
@@ -138,102 +133,16 @@ pub(super) fn coerce_blob_expr(
        ns_args.push(value);
    }

-    let built: Arc<dyn PhysicalExpr> = Arc::new(ScalarFunctionExpr::new(
+    let expr: Arc<dyn PhysicalExpr> = Arc::new(ScalarFunctionExpr::new(
        &format!("named_struct({})", table_field.name()),
        named_struct(),
        ns_args,
        table_field.clone(),
        config.clone(),
    ));
-
-    // `named_struct` always yields a valid struct, so a null input would land
-    // as a row that set neither `data` nor `uri` -- not an absent blob but a
-    // malformed one, which Lance rejects on write.
-    let expr: Arc<dyn PhysicalExpr> = Arc::new(AbsentBlobIsNull {
-        source: input_expr,
-        built,
-        field: table_field.clone(),
-    });
    Ok((expr, table_field.clone()))
 }

-/// Carries the source column's nullity onto the struct built for it.
-///
-/// This is its own expression rather than a `CASE` because the projection
-/// takes its output field from `return_field`, and the generic implementation
-/// rebuilds a bare field -- which would drop the `lance.blob.v2` extension
-/// metadata and stop the column being recognised as a blob at all.
-#[derive(Debug, Clone)]
-struct AbsentBlobIsNull {
-    source: Arc<dyn PhysicalExpr>,
-    built: Arc<dyn PhysicalExpr>,
-    field: FieldRef,
-}
-
-impl fmt::Display for AbsentBlobIsNull {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "absent_blob_is_null({}, {})", self.source, self.built)
-    }
-}
-
-impl PartialEq for AbsentBlobIsNull {
-    fn eq(&self, other: &Self) -> bool {
-        self.source.eq(&other.source) && self.built.eq(&other.built) && self.field == other.field
-    }
-}
-
-impl Eq for AbsentBlobIsNull {}
-
-impl Hash for AbsentBlobIsNull {
-    fn hash<H: Hasher>(&self, state: &mut H) {
-        self.source.hash(state);
-        self.built.hash(state);
-        self.field.hash(state);
-    }
-}
-
-impl PhysicalExpr for AbsentBlobIsNull {
-    fn return_field(&self, _input_schema: &Schema) -> datafusion_common::Result<FieldRef> {
-        Ok(self.field.clone())
-    }
-
-    fn nullable(&self, _input_schema: &Schema) -> datafusion_common::Result<bool> {
-        Ok(true)
-    }
-
-    fn evaluate(&self, batch: &RecordBatch) -> datafusion_common::Result<ColumnarValue> {
-        let rows = batch.num_rows();
-        let built = self.built.evaluate(batch)?.into_array(rows)?;
-        let source = self.source.evaluate(batch)?.into_array(rows)?;
-        let Some(nulls) = source.logical_nulls() else {
-            return Ok(ColumnarValue::Array(built));
-        };
-        // `nullif` nulls the rows the mask marks true, which is where the
-        // source had no value.
-        let absent = BooleanArray::new(!nulls.inner(), None);
-        Ok(ColumnarValue::Array(nullif(built.as_ref(), &absent)?))
-    }
-
-    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
-        vec![&self.source, &self.built]
-    }
-
-    fn with_new_children(
-        self: Arc<Self>,
-        children: Vec<Arc<dyn PhysicalExpr>>,
-    ) -> datafusion_common::Result<Arc<dyn PhysicalExpr>> {
-        Ok(Arc::new(Self {
-            source: children[0].clone(),
-            built: children[1].clone(),
-            field: self.field.clone(),
-        }))
-    }
-
-    fn fmt_sql(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "{self}")
-    }
-}
-
 enum BlobInputShape<'a> {
    Bytes,
    String,
@@ -404,11 +313,6 @@ mod tests {
        let data = image.column_by_name("data").unwrap();
        assert!(!data.is_null(0));
        assert!(data.is_null(1));
-        // The row itself has to be null, not merely a struct whose children
-        // are. A present-but-empty struct set neither `data` nor `uri`, which
-        // Lance rejects as malformed rather than reading as an absent blob.
-        assert!(!image.is_null(0));
-        assert!(image.is_null(1));
    }

    #[tokio::test]
@@ -52,7 +52,7 @@ enum ConsistencyMode {
    /// refresh_window = min(3s, TTL/4)
    ///
    /// | t < TTL - refresh_window | t < TTL                           | t >= TTL            |
-    /// |  Return value            | Background refresh & return value |  synchronous refresh |
+    /// |  Return value            | Background refresh & return value |  syncronous refresh |
    Eventual(BackgroundCache<Arc<Dataset>, Error>),
 }

@@ -103,7 +103,7 @@ impl MergeInsertBuilder {
    /// but that behavior is subject to change.
    ///
    /// An optional condition may be specified.  If it is, then only
-    /// matched rows that satisfy the condition will be updated.  Any
+    /// matched rows that satisfy the condtion will be updated.  Any
    /// rows that do not satisfy the condition will be left as they
    /// are.  Failing to satisfy the condition does not cause a
    /// "matched row" to become a "not matched" row.
@@ -904,7 +904,7 @@ fn unsharded_shard_id() -> Uuid {

 /// Build a [`ShardWriterConfig`] from the persisted `writer_config_defaults`.
 ///
-/// Unknown or unparsable keys are ignored; absent keys keep the
+/// Unknown or unparseable keys are ignored; absent keys keep the
 /// [`ShardWriterConfig`] default. The shard id is set by `mem_wal_writer`.
 fn shard_writer_config_from_defaults(defaults: &HashMap<String, String>) -> ShardWriterConfig {
    let mut config = ShardWriterConfig::default().with_shard_spec_id(SHARDING_SPEC_ID);
@@ -3,9 +3,9 @@

 //! Filling computed columns.
 //!
-//! A row without a value gets one; a row that has one keeps it unless its
-//! fragment's inputs moved since it was computed, which `freshness` decides
-//! from the manifest and stamps after every fill.
+//! 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.
 //!
 //! A column's computed inputs are filled first -- the dependency graph is
 //! walked once, each reachable column filled once in dependency order, each
@@ -31,7 +31,6 @@

 use std::collections::HashSet;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};

 use arrow_array::{
    Array, ArrayRef, BooleanArray, LargeBinaryArray, RecordBatch, RecordBatchOptions, StructArray,
@@ -49,7 +48,6 @@ use lance_core::datatypes::{BlobHandling, Schema as LanceSchema};
 use serde::{Deserialize, Serialize};

 use super::computed_columns::{BoundExpression, ComputedColumnKind, computed_column_from_field};
-use super::freshness::{self, SignatureMap, StalenessPlan};
 use super::{BaseTable, NativeTable};
 use crate::job::Job;
 use crate::{Error, Result};
@@ -112,81 +110,28 @@ async fn execute_refresh_column_with_source(
    };
    let output_is_blob = field.is_blob_v2();

-    // Which fragments the null filter cannot speak for: their inputs moved
-    // since they were computed, or the definition did. Decided once, from
-    // the manifest the values are read from.
-    let inputs = freshness::fields_for_paths(dataset.schema(), &bound.inputs)?;
-    let definition = freshness::definition_version(&expression);
-    let staleness = freshness::staleness_against(&dataset, column, &definition, &inputs).await?;
-
    let mut rows_filled = 0u64;
    let mut replacements = Vec::new();
-    // Fragments this refresh computed in full, signed at the version read.
-    let mut computed = SignatureMap::new();
    for fragment in dataset.get_fragments() {
-        let fragment_id = u32::try_from(fragment.id()).map_err(|_| Error::Runtime {
-            message: format!("fragment id {} does not fit a signature map", fragment.id()),
-        })?;
-        // A recompute rewrites every live row, so it is staged without the
-        // probe and counted as it fills; a null fill probes first, since a
-        // fragment with nothing to gain is not worth a write.
-        let recompute = staleness.is_dirty(fragment_id);
-        let whole = recompute || {
-            let (gained, unfilled) =
-                count_fragment_gains(&dataset, &fragment, &bound, column).await?;
-            if gained == 0 {
-                continue;
-            }
-            rows_filled += gained;
-            unfilled == u64::try_from(fragment.count_rows(None).await?).unwrap_or(u64::MAX)
-        };
-        if whole {
-            computed.insert(
-                fragment_id,
-                freshness::fragment_input_signature(fragment.metadata(), &inputs)?,
-            );
+        let gained = count_fragment_gains(&dataset, &fragment, &bound, column).await?;
+        if gained == 0 {
+            continue;
        }
-        let gained = Arc::new(AtomicU64::new(0));
-        let values = fill_stream(
-            &dataset,
-            &fragment,
-            bound.clone(),
-            column,
-            output_is_blob,
-            recompute,
-            gained.clone(),
-        )
-        .await?;
+        rows_filled += gained;
+        let values =
+            fill_stream(&dataset, &fragment, bound.clone(), column, output_is_blob).await?;
        replacements.push(fragment.write_columns(values, &column_schema).await?);
-        if recompute {
-            rows_filled += gained.load(Ordering::Relaxed);
-        }
    }

    let source_version = dataset.version().version;
    if replacements.is_empty() {
-        // Nothing to fill; the stamp may still have something to record -- a
-        // column not yet enrolled, or fragments a compaction carried.
-        let mut latest = (*dataset).clone();
-        let stamped = record(
-            &mut latest,
-            (&dataset, &staleness),
-            column,
-            &definition,
-            &inputs,
-            computed,
-        )
-        .await;
-        if stamped.is_some() {
-            table.dataset.update(latest);
-        }
        return Ok(RefreshExecution {
            result: RefreshColumnResult {
                rows_filled: 0,
-                version: stamped.unwrap_or(source_version),
+                version: source_version,
            },
            source_version,
-            published_version: stamped,
+            published_version: None,
        });
    }

@@ -204,17 +149,7 @@ async fn execute_refresh_column_with_source(
    )
    .await?;

-    let mut new_dataset = new_dataset;
-    let version = record(
-        &mut new_dataset,
-        (&dataset, &staleness),
-        column,
-        &definition,
-        &inputs,
-        computed,
-    )
-    .await
-    .unwrap_or(new_dataset.version().version);
+    let version = new_dataset.version().version;
    table.dataset.update(new_dataset);
    Ok(RefreshExecution {
        result: RefreshColumnResult {
@@ -226,31 +161,6 @@ async fn execute_refresh_column_with_source(
    })
 }

-/// Stamp the input state the refresh computed from (see
-/// [`freshness::record_freshness`]); the version the stamp landed at, which
-/// is the last one the refresh wrote. Never fails the refresh: the values
-/// are committed, and a missing stamp only costs a recompute next time.
-async fn record(
-    latest: &mut Dataset,
-    pinned: (&Dataset, &StalenessPlan),
-    column: &str,
-    definition: &str,
-    inputs: &freshness::InputFields,
-    computed: SignatureMap,
-) -> Option<u64> {
-    match freshness::record_freshness(latest, Some(pinned), column, definition, inputs, computed)
-        .await
-    {
-        Ok(record) => record.version,
-        Err(error) => {
-            log::warn!(
-                "could not record the input state computed column '{column}' was refreshed from ({error}); its fragments will recompute on the next refresh"
-            );
-            None
-        }
-    }
-}
-
 /// Refuse while a computed input still has rows a refresh of it would fill:
 /// read now, its placeholder null would be evaluated as a value and kept.
 async fn ensure_inputs_filled(
@@ -278,9 +188,7 @@ async fn ensure_inputs_filled(
        let input_bound = super::computed_columns::bind(schema.clone(), input, expression)?;
        let mut unfilled = 0u64;
        for fragment in dataset.get_fragments() {
-            unfilled += count_fragment_gains(dataset, &fragment, &input_bound, input)
-                .await?
-                .0;
+            unfilled += count_fragment_gains(dataset, &fragment, &input_bound, input).await?;
        }
        if unfilled > 0 {
            return Err(Error::InvalidInput {
@@ -528,35 +436,32 @@ fn blob_array_from_binary(
 /// 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`. Returns the gains and the rows
-/// scanned.
+/// fragment's contribution to `rows_filled`.
 async fn count_fragment_gains(
    dataset: &Dataset,
    fragment: &FileFragment,
    bound: &BoundExpression,
    column: &str,
-) -> Result<(u64, u64)> {
+) -> Result<u64> {
    let mut scanner = dataset.scan();
    scanner
        .with_fragments(vec![fragment.metadata().clone()])
        .with_row_id()
-        .project(&bound.roots)?
-        .filter(&format!("{} IS NULL", quote_identifier(column)))?;
+        .filter(&format!("{} IS NULL", quote_identifier(column)))?
+        .project(&bound.roots)?;
    configure_blob_inputs(&mut scanner, dataset.schema(), bound, None)?;

    let mut gained = 0u64;
-    let mut considered = 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;
-        considered += batch.num_rows() as u64;
    }
-    Ok((gained, considered))
+    Ok(gained)
 }

 /// Stream one fragment's column in physical order, filling the unfilled live
-/// rows -- every live row, for a recompute -- and keeping every other value.
+/// 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
@@ -567,8 +472,6 @@ async fn fill_stream(
    bound: Arc<BoundExpression>,
    column: &str,
    output_is_blob: bool,
-    recompute: bool,
-    gained: Arc<AtomicU64>,
 ) -> Result<impl Stream<Item = lance_core::Result<RecordBatch>> + Send + use<>> {
    let mut projection: Vec<String> = bound.roots.clone();
    projection.push(column.to_string());
@@ -618,20 +521,14 @@ async fn fill_stream(
            .column_by_name(ROW_ID)
            .ok_or_else(|| missing(ROW_ID))?;

-        // Only an unfilled live row gains a value, or every live row under a
-        // recompute; a deleted row has a null row id and keeps its (null) slot.
+        // 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 = if recompute {
-            live
-        } else {
-            let unfilled = arrow::compute::is_null(existing.as_ref())?;
-            arrow::compute::and(&unfilled, &live)?
-        };
+        let fill = arrow::compute::and(&unfilled, &live)?;
        let keep = arrow::compute::not(&fill)?;

        let computed = evaluate(&bound, &evaluation_batch(&batch, &bound, Some(&keep))?)?;
-        let values = arrow::compute::and(&fill, &arrow::compute::is_not_null(&computed)?)?;
-        gained.fetch_add(values.true_count() as u64, Ordering::Relaxed);
        let merged = arrow_select::zip::zip(&fill, &computed, existing)?;
        let merged = if output_is_blob {
            blob_array_from_binary(&merged, projected.field(0))?
@@ -840,8 +737,7 @@ mod tests {
            .await
            .unwrap();
        assert_eq!(no_op.rows_assigned, 0);
-        // The fill, then the stamp recording what it computed from.
-        assert_eq!(no_op.source_version, 4);
+        assert_eq!(no_op.source_version, 3);
        assert_eq!(no_op.published_version, None);
    }

@@ -881,8 +777,8 @@ mod tests {
    }

    /// 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: the
-    /// second refresh finds the fragment signed and moves nothing.
+    /// 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;
@@ -896,145 +792,28 @@ mod tests {

        let first = table.refresh_column("maybe").await.unwrap();
        assert_eq!(first.rows_filled, 0);
-        assert!(first.version > declared);
+        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, first.version);
+        assert_eq!(again.version, declared);
    }

-    /// A filled row whose input moved is recomputed: the update rewrites
-    /// the row into a fragment the stamp never signed, and only that one.
+    /// 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_recomputes_a_row_whose_input_moved() {
-        let table = table_with("refresh_mutation", vec![1, 2]).await;
+    async fn test_refresh_does_not_observe_input_mutation() {
+        let table = table_with("refresh_mutation", vec![1]).await;
        declare_doubled(&table).await.unwrap();
-        append(&table, vec![5]).await;
        table.refresh_column("doubled").await.unwrap();
-        assert_eq!(
-            read(&table, "doubled").await,
-            vec![Some(2), Some(4), Some(10)]
-        );
+        assert_eq!(read(&table, "doubled").await, vec![Some(2)]);

-        table
-            .update()
-            .column("x", "7")
-            .only_if("x = 5")
-            .execute()
-            .await
-            .unwrap();
-
-        let again = table.refresh_column("doubled").await.unwrap();
-        assert_eq!(again.rows_filled, 1);
-        assert_eq!(
-            read(&table, "doubled").await,
-            vec![Some(2), Some(4), Some(14)]
-        );
-
-        let settled = table.refresh_column("doubled").await.unwrap();
-        assert_eq!(settled.rows_filled, 0);
-        assert_eq!(settled.version, again.version);
-    }
-
-    /// A deleted row is never computed and the rows that stay keep their
-    /// values: a delete recomputes nothing and stamps nothing.
-    #[tokio::test]
-    async fn test_a_delete_recomputes_nothing() {
-        let table = table_with("refresh_delete", vec![1, 2, 3]).await;
-        declare_doubled(&table).await.unwrap();
-        let filled = table.refresh_column("doubled").await.unwrap();
-
-        table.delete("x = 2").await.unwrap();
-        let deleted = table.version().await.unwrap();
+        table.update().column("x", "3").execute().await.unwrap();

        let again = table.refresh_column("doubled").await.unwrap();
        assert_eq!(again.rows_filled, 0);
-        assert_eq!(again.version, deleted);
-        assert!(deleted > filled.version);
-        assert_eq!(read(&table, "doubled").await, vec![Some(2), Some(6)]);
-    }
-
-    /// Compaction copies inputs unchanged, so a fragment it builds from
-    /// signed ones is fresh: the refresh recomputes nothing and only records
-    /// the new fragment.
-    #[tokio::test]
-    async fn test_a_compaction_of_signed_fragments_recomputes_nothing() {
-        let table = table_with("refresh_compact_signed", vec![1, 2]).await;
-        declare_doubled(&table).await.unwrap();
-        append(&table, vec![5]).await;
-        table.refresh_column("doubled").await.unwrap();
-
-        table
-            .optimize(crate::table::OptimizeAction::Compact {
-                options: crate::table::CompactionOptions::default(),
-                remap_options: None,
-            })
-            .await
-            .unwrap();
-        let compacted = table.version().await.unwrap();
-
-        let carried = table.refresh_column("doubled").await.unwrap();
-        assert_eq!(carried.rows_filled, 0);
-        assert_eq!(carried.version, compacted + 1);
-        let settled = table.refresh_column("doubled").await.unwrap();
-        assert_eq!(settled.version, carried.version);
-        assert_eq!(
-            read(&table, "doubled").await,
-            vec![Some(2), Some(4), Some(10)]
-        );
-    }
-
-    /// A column declared before signatures existed has no map. Its first
-    /// refresh keeps the null-fill contract and enrolls what it read from;
-    /// from then on a moved input is recomputed like any other.
-    #[tokio::test]
-    async fn test_an_unsigned_column_is_enrolled_by_its_first_refresh() {
-        let table = table_with("refresh_legacy", vec![1, 2]).await;
-        declare_doubled(&table).await.unwrap();
-        table.refresh_column("doubled").await.unwrap();
-        table
-            .update()
-            .column("x", "3")
-            .only_if("x = 1")
-            .execute()
-            .await
-            .unwrap();
-
-        let native = table.as_native().unwrap();
-        let mut dataset = (*native.dataset.get().await.unwrap()).clone();
-        let declaration = dataset
-            .schema()
-            .field("doubled")
-            .unwrap()
-            .metadata
-            .iter()
-            .filter(|(key, _)| !key.starts_with("computed_refresh."))
-            .map(|(key, value)| (key.clone(), value.clone()))
-            .collect::<Vec<_>>();
-        dataset
-            .update_field_metadata()
-            .replace("doubled", declaration)
-            .unwrap()
-            .await
-            .unwrap();
-        table.checkout_latest().await.unwrap();
-
-        // Null-fill only: the moved row keeps the value it was filled with.
-        let enrolled = table.refresh_column("doubled").await.unwrap();
-        assert_eq!(enrolled.rows_filled, 0);
-        assert_eq!(read(&table, "doubled").await, vec![Some(2), Some(4)]);
-
-        table
-            .update()
-            .column("x", "5")
-            .only_if("x = 3")
-            .execute()
-            .await
-            .unwrap();
-        let again = table.refresh_column("doubled").await.unwrap();
-        assert_eq!(again.rows_filled, 1);
-        assert_eq!(read(&table, "doubled").await, vec![Some(4), Some(10)]);
+        assert_eq!(read(&table, "doubled").await, vec![Some(2)]);
    }

    /// A row rewrite before the first refresh materializes the declared
@@ -1052,11 +831,11 @@ mod tests {
        assert_eq!(read(&table, "doubled").await, vec![Some(6)]);
    }

-    /// A fragment compacted out of one the stamp never signed cannot vouch
-    /// for any of its rows: every live row is recomputed, the moved one
-    /// included.
+    /// 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_a_compaction_of_an_unsigned_fragment_recomputes_it() {
+    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();
@@ -1078,70 +857,18 @@ mod tests {
            .unwrap();

        let result = table.refresh_column("doubled").await.unwrap();
-        assert_eq!(result.rows_filled, 3);
-        assert_eq!(
-            read(&table, "doubled").await,
-            vec![Some(4), Some(10), Some(200)]
-        );
-    }
-
-    /// The gate's reproducer: a raw lance append may carry a value for the
-    /// computed column. Compaction cannot certify it, so the product is
-    /// recomputed and the supplied value replaced.
-    #[tokio::test]
-    async fn test_raw_append_values_are_not_trusted_after_compaction() {
-        use arrow_array::RecordBatchIterator;
-        use lance::Dataset;
-        use lance::dataset::{WriteMode, WriteParams};
-
-        let dir = tempfile::tempdir().unwrap();
-        let conn = connect(dir.path().to_str().unwrap())
-            .execute()
-            .await
-            .unwrap();
-        let batch = record_batch!(("x", Int32, [1, 2])).unwrap();
-        let table = conn
-            .create_table("raw_append", batch)
-            .execute()
-            .await
-            .unwrap();
-        declare_doubled(&table).await.unwrap();
-        table.refresh_column("doubled").await.unwrap();
-
-        let batch = record_batch!(("x", Int32, [5]), ("doubled", Int32, [Some(999_i32)])).unwrap();
-        let schema = batch.schema();
-        let uri = table.uri().await.unwrap();
-        Dataset::write(
-            RecordBatchIterator::new(vec![Ok(batch)], schema),
-            &uri,
-            Some(WriteParams {
-                mode: WriteMode::Append,
-                ..Default::default()
-            }),
-        )
-        .await
-        .unwrap();
-        table.checkout_latest().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, 3);
+        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)]
        );
    }

-    /// An appended fragment holds no values, so compacting it into a signed
-    /// one leaves the product fresh: only the appended rows are filled.
+    /// 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_a_compaction_with_an_appended_fragment_fills_only_its_rows() {
+    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();
@@ -1268,8 +995,7 @@ mod tests {
        assert_eq!(result.rows_failed, 0);
        assert_eq!(result.rows_remaining, 0);
        assert_eq!(result.source_version, 2);
-        // The fill lands at 3; the stamp recording its inputs is published at 4.
-        assert_eq!(result.published_version, Some(4));
+        assert_eq!(result.published_version, Some(3));
        assert_eq!(job.status().await.unwrap(), "finished");
        assert_eq!(
            read(&table, "doubled").await,
@@ -1333,32 +1059,31 @@ mod tests {
        assert_eq!(read(&table, "quotient").await, vec![Some(10)]);
    }

-    /// A filled row whose input moved is re-evaluated, and a row whose
-    /// input did not move is not: the untouched fragment is never read, so
-    /// its poison input is never reached.
+    /// 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_only_a_moved_rows_value_is_re_evaluated() {
-        let table = table_with("refresh_filled_poison", vec![1, 0]).await;
+    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 / coalesce(nullif(x, 0), 1)")
+            .computed("quotient", "10 / x")
            .execute()
            .await
            .unwrap();
        table.refresh_column("quotient").await.unwrap();
-        assert_eq!(read(&table, "quotient").await, vec![Some(10), Some(10)]);

-        append(&table, vec![5]).await;
        table
            .update()
-            .column("x", "2")
+            .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, 2);
+        assert_eq!(result.rows_filled, 1);
        assert_eq!(
            read(&table, "quotient").await,
            vec![Some(2), Some(5), Some(10)]
@@ -19,7 +19,6 @@ use lancedb::{
    connect, connect_namespace,
    database::listing::{
        ListingDatabaseOptions, NewTableConfig, OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS,
-        OPT_NEW_TABLE_STORAGE_VERSION,
    },
    query::{ExecutableQuery, QueryBase},
    table::{AddDataMode, CompactionOptions, OptimizeAction, OptimizeStats, WriteOptions},
@@ -147,10 +146,7 @@ async fn non_blob_table_keeps_default_format_and_row_id_setting() -> Result<()>
    let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
    let table = db.create_empty_table("t", schema).execute().await?;

-    assert_eq!(
-        storage_format_version(&table).await,
-        LanceFileVersion::Stable.resolve()
-    );
+    assert!(!supports_blob_v2(storage_format_version(&table).await));
    assert!(!uses_stable_row_ids(&table).await);
    Ok(())
 }
@@ -813,11 +809,7 @@ async fn fetch_blobs_rejects_unknown_column() -> Result<()> {
 #[tokio::test]
 async fn fetch_blobs_rejects_legacy_v1_blob_column() -> Result<()> {
    let tmp = tempdir().unwrap();
-    // Legacy v1 blob columns are only writable at file version <= 2.1.
-    let db = connect(tmp.path().to_str().unwrap())
-        .storage_options([(OPT_NEW_TABLE_STORAGE_VERSION, "2.1")])
-        .execute()
-        .await?;
+    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
    let legacy = Field::new("image", DataType::LargeBinary, true).with_metadata(
        std::collections::HashMap::from([("lance-encoding:blob".to_string(), "true".to_string())]),
    );