feat(remote): support set_unenforced_primary_key

`RemoteTable::set_unenforced_primary_key` returned `NotSupported`, so the
call failed against LanceDB Cloud and enterprise from every SDK -- Python
and TypeScript both forward to it. That also blocked sharded LSM writes,
since the server rejects bucket/identity sharding on a table that declares
no unenforced primary key.

No server-side support was missing. The unenforced primary key is Lance
schema field metadata, and the existing `update_field_metadata` endpoint
writes exactly that, so the remote table now installs the key through it.
The commit layer behind that endpoint installs the position, enforces
immutability and runs `verify_primary_key()` -- the same code a native
table reaches, so both paths agree on semantics and not just on messages.

The request validation and the metadata edit move into shared helpers so
the native and remote paths cannot drift. Native behaviour is unchanged.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011xV8EebEZDmDDV8hVBm6MD

Cargo.lock

@@ -5553,7 +5553,6 @@ dependencies = [
  "serde_json",
  "serde_with",
  "serial_test",
- "sha2 0.10.9",
  "snafu 0.8.9",
  "tempfile",
  "test-log",
@@ -7704,9 +7703,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 +7732,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",

docs/src/js/classes/Table.md

@@ -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
 

nodejs/lancedb/table.ts

@@ -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.

python/python/lancedb/remote/table.py

@@ -1008,7 +1008,7 @@ class RemoteTable(Table):
         return LOOP.run(self._table.drop_columns(columns))
 
     def set_unenforced_primary_key(self, columns: Union[str, Iterable[str]]) -> None:
-        """Not supported on LanceDB Cloud."""
+        """Set the unenforced primary key for this table to a single column."""
         return LOOP.run(self._table.set_unenforced_primary_key(columns))
 
     def set_lsm_write_spec(self, spec: "LsmWriteSpec") -> None:

python/python/lancedb/table.py

@@ -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].
@@ -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)))
@@ -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].

python/python/tests/test_table.py

@@ -4183,14 +4183,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 +4208,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]

rust/lancedb/Cargo.toml

@@ -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"

rust/lancedb/src/materialized_view/refresh.rs

@@ -1124,9 +1124,8 @@ struct RowScope {
 
 /// 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.
+/// those fields and neither adding nor removing rows. 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.
@@ -1177,19 +1176,6 @@ async fn only_computed_rewrites_since(view_ds: &Dataset, recorded: u64) -> Resul
                                 .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 {
@@ -3373,46 +3359,6 @@ mod tests {
         );
     }
 
-    /// 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
@@ -3578,9 +3524,9 @@ mod tests {
     }
 
     /// 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.
+    /// commits a data replacement; 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};

rust/lancedb/src/remote/table.rs

@@ -32,6 +32,7 @@ use crate::table::Tags;
 use crate::table::UpdateResult;
 use crate::table::lsm_stats::GetLsmStatsResponse;
 use crate::table::merge::MergeFilter;
+use crate::table::primary_key;
 use crate::table::query::create_multi_vector_plan;
 use crate::table::write_progress::FinishOnDrop;
 use crate::table::{
@@ -68,6 +69,7 @@ use lance::arrow::json::{JsonDataType, JsonSchema};
 use lance::dataset::refs::TagContents;
 use lance::dataset::scanner::DatasetRecordBatchStream;
 use lance::dataset::{ColumnAlteration, NewColumnTransform, Version};
+use lance_core::datatypes::Schema as LanceSchema;
 use lance_datafusion::exec::{OneShotExec, execute_plan};
 use reqwest::{RequestBuilder, Response};
 use serde::{Deserialize, Serialize};
@@ -2992,10 +2994,27 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
         }
     }
 
-    async fn set_unenforced_primary_key(&self, _columns: &[&str]) -> Result<()> {
-        Err(Error::NotSupported {
-            message: "set_unenforced_primary_key is not supported on LanceDB cloud.".into(),
-        })
+    /// The unenforced primary key is Lance schema field metadata, so this
+    /// installs it through the `update_field_metadata` endpoint. The commit
+    /// layer behind that endpoint is what actually installs and validates the
+    /// key, exactly as on a native table; the checks here only fail fast with
+    /// the same messages a native table gives.
+    async fn set_unenforced_primary_key(&self, columns: &[&str]) -> Result<()> {
+        self.check_mutable().await?;
+
+        let arrow_schema = self.schema().await?;
+        let schema = LanceSchema::try_from(arrow_schema.as_ref()).map_err(|e| Error::Schema {
+            message: format!("Invalid schema: {}", e),
+        })?;
+        primary_key::validate(&schema, columns)?;
+
+        self.update_field_metadata(&[FieldMetadataUpdate {
+            path: columns[0].to_string(),
+            metadata: primary_key::install_edit(),
+            replace: false,
+        }])
+        .await?;
+        Ok(())
     }
 
     async fn flush_lsm(&self) -> Result<()> {
@@ -11832,6 +11851,120 @@ mod tests {
         assert_eq!(result.version, 7);
     }
 
+    /// The unenforced primary key is field metadata, so the remote table
+    /// installs it through the `update_field_metadata` endpoint.
+    #[tokio::test]
+    async fn test_set_unenforced_primary_key() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            match request.url().path() {
+                "/v1/table/my_table/describe/" => {
+                    let schema = Schema::new(vec![
+                        Field::new("id", DataType::Int64, false),
+                        Field::new("name", DataType::Utf8, true),
+                    ]);
+                    http::Response::builder()
+                        .status(200)
+                        .body(describe_response(&schema))
+                        .unwrap()
+                }
+                "/v1/table/my_table/update_field_metadata/" => {
+                    let body = request_body_json(&request);
+                    assert_eq!(body["updates"].as_array().unwrap().len(), 1);
+                    let update = &body["updates"][0];
+                    assert_eq!(update["path"], "id");
+                    assert_eq!(update["replace"], json!(false));
+                    assert_eq!(
+                        update["metadata"]["lance-schema:unenforced-primary-key:position"],
+                        "1"
+                    );
+                    assert_eq!(
+                        update["metadata"]["lance-schema:unenforced-primary-key"],
+                        json!(null)
+                    );
+                    http::Response::builder()
+                        .status(200)
+                        .body(r#"{"version": 3, "fields": {}}"#.to_string())
+                        .unwrap()
+                }
+                path => panic!("Unexpected path: {}", path),
+            }
+        });
+
+        table.set_unenforced_primary_key(["id"]).await.unwrap();
+    }
+
+    /// Requests the native table rejects are rejected here too, before any
+    /// write reaches the server.
+    #[tokio::test]
+    async fn test_set_unenforced_primary_key_rejects_invalid_requests() {
+        let table = Table::new_with_handler("my_table", |request| match request.url().path() {
+            "/v1/table/my_table/describe/" => {
+                let schema = Schema::new(vec![
+                    Field::new("id", DataType::Int64, false),
+                    Field::new("score", DataType::Float32, true),
+                ]);
+                http::Response::builder()
+                    .status(200)
+                    .body(describe_response(&schema))
+                    .unwrap()
+            }
+            path => panic!("Unexpected path: {}", path),
+        });
+
+        for columns in [
+            vec![],
+            vec!["id", "score"],
+            vec!["nonexistent"],
+            vec!["score"],
+        ] {
+            let err = table
+                .set_unenforced_primary_key(columns.clone())
+                .await
+                .unwrap_err();
+            assert!(
+                matches!(err, Error::InvalidInput { .. }),
+                "unexpected error for {:?}: {:?}",
+                columns,
+                err
+            );
+        }
+    }
+
+    /// The key is immutable once set, and the schema the server already
+    /// reports is enough to say so.
+    #[tokio::test]
+    async fn test_set_unenforced_primary_key_already_set() {
+        let table = Table::new_with_handler("my_table", |request| match request.url().path() {
+            "/v1/table/my_table/describe/" => {
+                let schema = Schema::new(vec![
+                    Field::new("id", DataType::Int64, false).with_metadata(HashMap::from([(
+                        "lance-schema:unenforced-primary-key:position".to_string(),
+                        "1".to_string(),
+                    )])),
+                    Field::new("name", DataType::Utf8, false),
+                ]);
+                http::Response::builder()
+                    .status(200)
+                    .body(describe_response(&schema))
+                    .unwrap()
+            }
+            path => panic!("Unexpected path: {}", path),
+        });
+
+        for column in ["name", "id"] {
+            let err = table
+                .set_unenforced_primary_key([column])
+                .await
+                .unwrap_err();
+            assert!(
+                err.to_string().contains("already set"),
+                "unexpected error: {:?}",
+                err
+            );
+        }
+    }
+
     // ----- Branch support -----
 
     /// Parse a request's in-memory JSON body. Only valid for JSON-body ops

rust/lancedb/src/table.rs

@@ -75,11 +75,10 @@ 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;
-mod primary_key;
+pub(crate) mod primary_key;
 pub mod query;
 pub mod refresh;
 pub mod schema_evolution;
@@ -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,
@@ -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`].

rust/lancedb/src/table/add_columns.rs

@@ -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

rust/lancedb/src/table/computed_columns.rs

@@ -71,22 +71,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 +139,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 +163,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()),
     ])
 }
 
@@ -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>,
@@ -1489,9 +1470,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.

rust/lancedb/src/table/primary_key.rs

@@ -11,24 +11,26 @@
 //! Only a single-column primary key is supported, and the key cannot be
 //! changed once set.
 
+use std::collections::HashMap;
+
 use arrow_schema::DataType;
-use lance_core::datatypes::{LANCE_UNENFORCED_PRIMARY_KEY, LANCE_UNENFORCED_PRIMARY_KEY_POSITION};
+use lance_core::datatypes::{
+    Field as LanceField, LANCE_UNENFORCED_PRIMARY_KEY, LANCE_UNENFORCED_PRIMARY_KEY_POSITION,
+    Schema as LanceSchema,
+};
 
 use crate::error::{Error, Result};
 use crate::table::NativeTable;
 
-/// Set the unenforced primary key on `table` to the single column in `columns`.
+/// Validate a `set_unenforced_primary_key` request against `schema`, returning
+/// the field the key would be installed on.
 ///
-/// Fails if `columns` is not exactly one column (compound primary keys are not
-/// supported), if the column does not exist or has an unsupported dtype, or if
-/// the table already has an unenforced primary key (changing the primary key
-/// is not supported).
-pub(super) async fn set_unenforced_primary_key(
-    table: &NativeTable,
-    columns: &[&str],
-) -> Result<()> {
-    table.dataset.ensure_mutable()?;
-
+/// Shared by [`NativeTable`] and the remote table so both reject the same
+/// requests with the same messages. Fails if `columns` is not exactly one
+/// column (compound primary keys are not supported), if the column does not
+/// exist or has an unsupported dtype, or if the table already has an
+/// unenforced primary key (changing the primary key is not supported).
+pub fn validate<'a>(schema: &'a LanceSchema, columns: &[&str]) -> Result<&'a LanceField> {
     if columns.is_empty() {
         return Err(Error::InvalidInput {
             message: "set_unenforced_primary_key: a column is required".into(),
@@ -44,43 +46,71 @@ pub(super) async fn set_unenforced_primary_key(
     }
     let column = columns[0];
 
+    // The primary key is immutable once set. The Lance commit layer is the
+    // source of truth for this (it also covers the concurrent-writer race);
+    // this check just fails fast with a clear message.
+    if !schema.unenforced_primary_key().is_empty() {
+        return Err(Error::InvalidInput {
+            message: "set_unenforced_primary_key: an unenforced primary key is already set on this table; changing it is not supported".into(),
+        });
+    }
+
+    let field = schema.field(column).ok_or_else(|| Error::InvalidInput {
+        message: format!(
+            "set_unenforced_primary_key: column '{}' not found on table",
+            column
+        ),
+    })?;
+    if !is_supported_pk_dtype(&field.data_type()) {
+        return Err(Error::InvalidInput {
+            message: format!(
+                "set_unenforced_primary_key: column '{}' has dtype {:?} which is not supported as a primary key. Supported: Int32, Int64, Utf8, LargeUtf8, Binary, LargeBinary, FixedSizeBinary",
+                column,
+                field.data_type()
+            ),
+        });
+    }
+    Ok(field)
+}
+
+/// The field metadata edit that installs the primary key on a field: keys to
+/// set (`Some`) or delete (`None`).
+///
+/// Position metadata is 1-indexed; `Schema::unenforced_primary_key` treats
+/// position 0 as a legacy "no specific position" fallback, so the legacy
+/// boolean key is cleared and only the position governs.
+pub fn install_edit() -> HashMap<String, Option<String>> {
+    HashMap::from([
+        (LANCE_UNENFORCED_PRIMARY_KEY.to_string(), None),
+        (
+            LANCE_UNENFORCED_PRIMARY_KEY_POSITION.to_string(),
+            Some("1".to_string()),
+        ),
+    ])
+}
+
+/// Set the unenforced primary key on `table` to the single column in `columns`.
+pub(super) async fn set_unenforced_primary_key(
+    table: &NativeTable,
+    columns: &[&str],
+) -> Result<()> {
+    table.dataset.ensure_mutable()?;
+
     let updates = {
         let dataset = table.dataset.get().await?;
-        let schema = dataset.schema();
+        let field = validate(dataset.schema(), columns)?;
 
-        // The primary key is immutable once set. The Lance commit layer is the
-        // source of truth for this (it also covers the concurrent-writer race);
-        // this check just fails fast with a clear message.
-        if !schema.unenforced_primary_key().is_empty() {
-            return Err(Error::InvalidInput {
-                message: "set_unenforced_primary_key: an unenforced primary key is already set on this table; changing it is not supported".into(),
-            });
-        }
-
-        let field = schema.field(column).ok_or_else(|| Error::InvalidInput {
-            message: format!(
-                "set_unenforced_primary_key: column '{}' not found on table",
-                column
-            ),
-        })?;
-        if !is_supported_pk_dtype(&field.data_type()) {
-            return Err(Error::InvalidInput {
-                message: format!(
-                    "set_unenforced_primary_key: column '{}' has dtype {:?} which is not supported as a primary key. Supported: Int32, Int64, Utf8, LargeUtf8, Binary, LargeBinary, FixedSizeBinary",
-                    column,
-                    field.data_type()
-                ),
-            });
-        }
-
-        // Position metadata is 1-indexed; `Schema::unenforced_primary_key`
-        // treats position 0 as a legacy "no specific position" fallback.
         let mut metadata = field.metadata.clone();
-        metadata.remove(LANCE_UNENFORCED_PRIMARY_KEY);
-        metadata.insert(
-            LANCE_UNENFORCED_PRIMARY_KEY_POSITION.to_string(),
-            "1".to_string(),
-        );
+        for (key, value) in install_edit() {
+            match value {
+                Some(value) => {
+                    metadata.insert(key, value);
+                }
+                None => {
+                    metadata.remove(&key);
+                }
+            }
+        }
         vec![(field_id_to_u32(field.id, &field.name)?, metadata)]
     };
 

rust/lancedb/src/table/refresh.rs

@@ -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)]