feat: let a computed-column batch read its own earlier declarations

`add_columns().computed()` accepted several columns in one call but bound
each against the table's schema as it stood before the call, so `a` and
`b = a + 1` had to be two commits. A server staging declarations behind
other schema work has no atomic way to do that, and a caller reading the
builder's plural signature reasonably expects the batch to be one.

Each accepted column now joins the schema the next one resolves against,
so the batch is planned and committed as one. Order is the dependency
order; reading ahead is still an unknown column. `validate_declarations`
exposes the schema-level checks -- the Function-binding guard and the
planning -- without a commit, for callers that must reject before earlier
work in the same request lands; LSM state is table state and stays a
commit-time check.

Refresh order matters for a dependent column: `b = coalesce(a, 0)`
refreshed before `a` would bake zeros from `a`'s placeholder null, and the
fill-once contract keeps them. Refresh now refuses, naming the input,
while a computed input still has rows a refresh of it would fill -- the
same probe refresh already uses to detect a no-op. Otherwise it is one
snapshot and one commit, as before; a concurrent append is not in the
commit and waits for the next refresh. Refreshing dependencies on the
caller's behalf was considered and rejected: it is not how materialized
views or our own backfill scheduler behave, and it needs multi-commit
fencing that an explicit per-row fill marker would make unnecessary.

Cargo.lock

@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
 [[package]]
 name = "fsst"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-array",
  "rand 0.9.5",
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
 
 [[package]]
 name = "lance"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arc-swap",
  "arrow",
@@ -4888,8 +4888,8 @@ dependencies = [
 
 [[package]]
 name = "lance-arrow"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4911,7 +4911,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.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4925,7 +4925,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.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -4934,8 +4934,8 @@ dependencies = [
 
 [[package]]
 name = "lance-bitpacking"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrayref",
  "crunchy",
@@ -4945,8 +4945,8 @@ dependencies = [
 
 [[package]]
 name = "lance-core"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4983,8 +4983,8 @@ dependencies = [
 
 [[package]]
 name = "lance-datafusion"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5013,8 +5013,8 @@ dependencies = [
 
 [[package]]
 name = "lance-datagen"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5031,8 +5031,8 @@ dependencies = [
 
 [[package]]
 name = "lance-derive"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -5041,8 +5041,8 @@ dependencies = [
 
 [[package]]
 name = "lance-encoding"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -5075,8 +5075,8 @@ dependencies = [
 
 [[package]]
 name = "lance-file"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -5107,8 +5107,8 @@ dependencies = [
 
 [[package]]
 name = "lance-index"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arc-swap",
  "arrow",
@@ -5172,8 +5172,8 @@ dependencies = [
 
 [[package]]
 name = "lance-index-core"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -5195,8 +5195,8 @@ dependencies = [
 
 [[package]]
 name = "lance-io"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5236,8 +5236,8 @@ dependencies = [
 
 [[package]]
 name = "lance-linalg"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -5251,8 +5251,8 @@ dependencies = [
 
 [[package]]
 name = "lance-namespace"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow",
  "async-trait",
@@ -5264,8 +5264,8 @@ dependencies = [
 
 [[package]]
 name = "lance-namespace-impls"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow",
  "arrow-ipc",
@@ -5318,8 +5318,8 @@ dependencies = [
 
 [[package]]
 name = "lance-select"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5333,8 +5333,8 @@ dependencies = [
 
 [[package]]
 name = "lance-table"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5374,8 +5374,8 @@ dependencies = [
 
 [[package]]
 name = "lance-testing"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -5388,8 +5388,8 @@ dependencies = [
 
 [[package]]
 name = "lance-tokenizer"
-version = "12.0.0-beta.1"
+version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
  "frostem",
  "icu_segmenter",

Cargo.toml

@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"
 
 [workspace.dependencies]
-lance = { "version" = "=12.0.0-beta.1", default-features = false, "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=12.0.0-beta.1", default-features = false, "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=12.0.0-beta.1", default-features = false, "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "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

java/pom.xml

@@ -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.1</lance-core.version>
+        <lance-core.version>12.0.0-beta.2</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>

rust/lancedb/src/table/computed_columns.rs

@@ -22,7 +22,7 @@
 use std::collections::{BTreeSet, HashMap};
 use std::sync::Arc;
 
-use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema, SchemaRef};
+use arrow_schema::{DataType, Field as ArrowField, Fields, Schema as ArrowSchema, SchemaRef};
 use datafusion_common::tree_node::TreeNode;
 use datafusion_physical_plan::PhysicalExpr;
 use lance::dataset::NewColumnTransform;
@@ -1273,6 +1273,11 @@ pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result<
 /// refresh time: that the expression parses, that every column it reads
 /// exists, and that the target name is free. A declaration that survives this
 /// is one a refresh can always act on.
+///
+/// Each accepted column joins the schema the next one resolves against, so a
+/// batch may declare `a` and then `b = a + 1` in one commit. Refresh order
+/// then matters, and refresh enforces it: `b` is refused while `a` still has
+/// unfilled rows.
 pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Vec<ArrowField>> {
     if columns.is_empty() {
         return Err(Error::InvalidInput {
@@ -1280,11 +1285,11 @@ pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Ve
         });
     }
 
+    let mut schema = schema;
     let mut fields = Vec::with_capacity(columns.len());
-    let mut declared: Vec<&str> = Vec::with_capacity(columns.len());
 
     for (name, expression) in columns {
-        if schema.field_with_name(name).is_ok() || declared.contains(&name.as_str()) {
+        if schema.field_with_name(name).is_ok() {
             return Err(Error::ColumnAlreadyExists { name: name.clone() });
         }
 
@@ -1292,16 +1297,50 @@ pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Ve
 
         // Declared columns start entirely null, so nullability is a property
         // of the declaration rather than of what the expression yields.
-        fields.push(
+        let field = ArrowField::new(name, bound.data_type, true)
-            ArrowField::new(name, bound.data_type, true)
+            .with_metadata(computed_column_metadata(expression, &bound.inputs));
-                .with_metadata(computed_column_metadata(expression, &bound.inputs)),
+        schema = Arc::new(ArrowSchema::new_with_metadata(
-        );
+            schema
-        declared.push(name);
+                .fields()
+                .iter()
+                .cloned()
+                .chain(std::iter::once(Arc::new(field.clone())))
+                .collect::<Fields>(),
+            schema.metadata().clone(),
+        ));
+        fields.push(field);
     }
 
     Ok(fields)
 }
 
+/// Run the schema-level checks of
+/// [`AddColumnsBuilder::computed`](super::AddColumnsBuilder::computed) against
+/// `schema` without committing: the Function-binding guard and the planning of
+/// every declaration. For callers that stage declarations behind other work
+/// and need those rejections before any of it lands.
+///
+/// Only the schema is consulted. Declaring also refuses a table with an LSM
+/// write spec or retained SSTables; that is table state, checked at commit.
+///
+/// ```
+/// # use std::sync::Arc;
+/// # use arrow_schema::{DataType, Field, Schema};
+/// use lancedb::table::computed_columns::validate_declarations;
+///
+/// let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
+/// let declarations = vec![
+///     ("a".to_string(), "x + 1".to_string()),
+///     ("b".to_string(), "a * 2".to_string()),
+/// ];
+/// assert!(validate_declarations(schema.clone(), &declarations).is_ok());
+/// assert!(validate_declarations(schema, &[("c".into(), "random()".into())]).is_err());
+/// ```
+pub fn validate_declarations(schema: SchemaRef, columns: &[(String, String)]) -> Result<()> {
+    ensure_no_function_bindings_for_mutation(schema.as_ref(), "schema evolution")?;
+    plan(schema, columns).map(drop)
+}
+
 /// Build the transform that declares `columns` against `schema`.
 ///
 /// An all-null column is how a binding with no values yet is carried into a
@@ -1340,6 +1379,22 @@ pub(super) async fn add_foreign_kind(table: &crate::Table, name: &str, kind: &st
 
 #[cfg(test)]
 mod tests {
+    /// The gate's reproducer: the validator applies the same schema-level
+    /// guard declaring does, so a staging caller is refused before it commits
+    /// anything else.
+    #[test]
+    fn test_validate_declarations_matches_schema_admission_barriers() {
+        let schema = Arc::new(ArrowSchema::new_with_metadata(
+            vec![ArrowField::new("x", DataType::Int32, true)],
+            HashMap::from([(
+                FUNCTION_BINDINGS_META_KEY.to_string(),
+                "not valid binding metadata".to_string(),
+            )]),
+        ));
+        let declarations = vec![("a".to_string(), "x + 1".to_string())];
+        assert!(super::validate_declarations(schema, &declarations).is_err());
+    }
+
     #[test]
     fn output_arrow_type_grammar_matches_the_shared_golden() {
         let golden: serde_json::Value = serde_json::from_str(include_str!(
@@ -1582,6 +1637,40 @@ mod tests {
         assert!(declared(&table).await.is_empty());
     }
 
+    /// A batch may build on itself: one commit, and the later entry's inputs
+    /// name the earlier one.
+    #[tokio::test]
+    async fn test_a_declaration_may_read_one_declared_before_it() {
+        let table = table_with_ints("chain").await;
+        let before = table.version().await.unwrap();
+        add_computed(
+            &table,
+            &[("a".into(), "x + 1".into()), ("b".into(), "a * 2".into())],
+        )
+        .await
+        .unwrap();
+        assert_eq!(table.version().await.unwrap(), before + 1);
+        let declared = declared(&table).await;
+        assert_eq!(declared[1].name, "b");
+        assert_eq!(declared[1].inputs, vec!["a".to_string()]);
+
+        // Order is the dependency order; reading ahead is still unknown.
+        let err = add_computed(
+            &table,
+            &[("c".into(), "d + 1".into()), ("d".into(), "x + 1".into())],
+        )
+        .await
+        .unwrap_err();
+        assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "c"));
+        assert!(
+            validate_declarations(
+                table.schema().await.unwrap(),
+                &[("e".into(), "random()".into())]
+            )
+            .is_err()
+        );
+    }
+
     /// A column added by an ordinary transform is materialized, not bound, so
     /// it carries no declaration to report.
     #[tokio::test]

rust/lancedb/src/table/refresh.rs

@@ -7,6 +7,16 @@
 //! 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
+//! fill its own commit. Every fill in the pass, the requested column's
+//! included, covers only the fragments of the snapshot the pass started
+//! from: a commit may rebase over a concurrent append, and the fragment that
+//! admits carries placeholder nulls no earlier fill covered, so it waits for
+//! a later refresh rather than being read as values. Two concurrent fills of
+//! one input collide on its field in lance's conflict check, so a dependent
+//! fill can only commit over inputs that were durable when it read them.
+//!
 //! Two passes per fragment. The first scans only the unfilled live rows and
 //! evaluates the expression over them, which yields the exact fill count and
 //! decides whether the fragment is staged at all -- a fragment where nothing
@@ -41,7 +51,8 @@ use crate::{Error, Result};
 /// The result of refreshing a computed column.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
 pub struct RefreshColumnResult {
-    /// Rows that had a value computed.
+    /// Rows that had a value computed, in the requested column only; inputs
+    /// filled on its behalf are not counted.
     #[serde(default)]
     pub rows_filled: u64,
     /// The commit version associated with the operation.
@@ -52,6 +63,7 @@ pub struct RefreshColumnResult {
 struct RefreshExecution {
     result: RefreshColumnResult,
     source_version: u64,
+    published_version: Option<u64>,
 }
 
 /// Internal implementation of the refresh logic.
@@ -74,7 +86,12 @@ async fn execute_refresh_column_with_source(
 
     let expression = declared_expression(&dataset, column)?;
     let schema = Arc::new(ArrowSchema::from(dataset.schema()));
-    let bound = Arc::new(super::computed_columns::bind(schema, column, &expression)?);
+    let bound = Arc::new(super::computed_columns::bind(
+        schema.clone(),
+        column,
+        &expression,
+    )?);
+    ensure_inputs_filled(&dataset, &schema, column, &bound).await?;
     let field = dataset
         .schema()
         .field(column)
@@ -100,25 +117,25 @@ async fn execute_refresh_column_with_source(
         replacements.push(fragment.write_columns(values, &column_schema).await?);
     }
 
+    let source_version = dataset.version().version;
     if replacements.is_empty() {
-        let source_version = dataset.version().version;
         return Ok(RefreshExecution {
             result: RefreshColumnResult {
                 rows_filled: 0,
                 version: source_version,
             },
             source_version,
+            published_version: None,
         });
     }
 
-    let read_version = dataset.version().version;
     // The dataset's own session, so registrations and caches survive the
     // commit being installed on the handle.
     let session = dataset.session();
     let new_dataset = Dataset::commit(
         WriteDestination::Dataset(dataset.clone()),
         Operation::DataReplacement { replacements },
-        Some(read_version),
+        Some(source_version),
         None,
         None,
         session,
@@ -133,10 +150,52 @@ async fn execute_refresh_column_with_source(
             rows_filled,
             version,
         },
-        source_version: read_version,
+        source_version,
+        published_version: Some(version),
     })
 }
 
+/// 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(
+    dataset: &Dataset,
+    schema: &Arc<ArrowSchema>,
+    column: &str,
+    bound: &BoundExpression,
+) -> Result<()> {
+    for input in &bound.roots {
+        let Some(declaration) = schema
+            .field_with_name(input)
+            .ok()
+            .and_then(computed_column_from_field)
+        else {
+            continue;
+        };
+        let ComputedColumnKind::Sql { expression } = &declaration.kind else {
+            return Err(Error::NotSupported {
+                message: format!(
+                    "computed column '{column}' reads '{input}', whose fill state this \
+                     refresh cannot check; refresh '{input}' first"
+                ),
+            });
+        };
+        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?;
+        }
+        if unfilled > 0 {
+            return Err(Error::InvalidInput {
+                message: format!(
+                    "computed column '{column}' reads '{input}', which has {unfilled} unfilled \
+                     rows; refresh '{input}' first"
+                ),
+            });
+        }
+    }
+    Ok(())
+}
+
 /// Run the refresh as a [`Job`] in this process.
 pub(crate) async fn execute_refresh_column_async(
     table: &NativeTable,
@@ -160,8 +219,7 @@ pub(crate) async fn execute_refresh_column_async(
             rows_failed: 0,
             rows_remaining: 0,
             source_version: execution.source_version,
-            published_version: (execution.result.rows_filled > 0)
+            published_version: execution.published_version,
-                .then_some(execution.result.version),
         })
     })))
 }
@@ -384,7 +442,8 @@ mod tests {
             .version)
     }
 
-    async fn read(table: &Table, column: &str) -> Vec<Option<i32>> {
+    async fn read(table: &Table, column: &str) -> Vec<Option<i64>> {
+        use arrow_array::{Array, Int64Array};
         let batches = table
             .query()
             .select(Select::columns(&[column]))
@@ -394,15 +453,19 @@ mod tests {
             .try_collect::<Vec<_>>()
             .await
             .unwrap();
-        let mut values: Vec<Option<i32>> = batches
+        let mut values: Vec<Option<i64>> = batches
             .iter()
             .flat_map(|batch| {
-                batch[column]
+                let array = &batch[column];
-                    .as_any()
+                match array.as_any().downcast_ref::<Int32Array>() {
-                    .downcast_ref::<Int32Array>()
+                    Some(ints) => ints.iter().map(|v| v.map(i64::from)).collect::<Vec<_>>(),
-                    .unwrap()
+                    None => array
-                    .iter()
+                        .as_any()
-                    .collect::<Vec<_>>()
+                        .downcast_ref::<Int64Array>()
+                        .unwrap()
+                        .iter()
+                        .collect::<Vec<_>>(),
+                }
             })
             .collect();
         values.sort();
@@ -414,6 +477,98 @@ mod tests {
         table.add(batch).execute().await.unwrap();
     }
 
+    /// The gate's reproducer: `b = coalesce(a, 0)` refreshed before `a`
+    /// must not bake zeros from `a`'s placeholder null. It is refused, and
+    /// names the input, until `a` is filled -- after every append too.
+    #[tokio::test]
+    async fn test_dependent_refresh_refuses_an_unfilled_input() {
+        let table = table_with("dependent_refresh_order", vec![1, 2, 3]).await;
+        table
+            .add_columns()
+            .computed("a", "x + 1")
+            .computed("b", "coalesce(a, 0)")
+            .execute()
+            .await
+            .unwrap();
+
+        let err = table.refresh_column("b").await.unwrap_err();
+        assert!(
+            matches!(&err, Error::InvalidInput { message } if message.contains("refresh 'a' first")),
+            "{err}"
+        );
+        assert_eq!(read(&table, "b").await, vec![None, None, None]);
+
+        assert_eq!(table.refresh_column("a").await.unwrap().rows_filled, 3);
+        assert_eq!(table.refresh_column("b").await.unwrap().rows_filled, 3);
+        assert_eq!(read(&table, "b").await, vec![Some(2), Some(3), Some(4)]);
+
+        append(&table, vec![10]).await;
+        assert!(table.refresh_column("b").await.is_err());
+        table.refresh_column("a").await.unwrap();
+        assert_eq!(table.refresh_column("b").await.unwrap().rows_filled, 1);
+        assert_eq!(
+            table.count_rows(Some("b = 0".to_string())).await.unwrap(),
+            0
+        );
+    }
+
+    /// Names that need quoting, and a nested input, survive the trip through
+    /// declaration metadata and the dependency check: the recorded inputs
+    /// are matched by name, never re-parsed as SQL.
+    #[tokio::test]
+    async fn test_dependent_refresh_handles_awkward_column_names() {
+        use arrow_array::{Int32Array, StructArray};
+        use arrow_schema::{DataType, Field, Fields};
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let age_fields = Fields::from(vec![Field::new("age", DataType::Int32, true)]);
+        let meta = StructArray::new(
+            age_fields.clone(),
+            vec![Arc::new(Int32Array::from(vec![10, 20])) as _],
+            None,
+        );
+        let schema = Arc::new(arrow_schema::Schema::new(vec![
+            Field::new("camelCase", DataType::Int32, true),
+            Field::new("with-hyphen", DataType::Int32, true),
+            Field::new("meta", DataType::Struct(age_fields), true),
+        ]));
+        let batch = arrow_array::RecordBatch::try_new(
+            schema,
+            vec![
+                Arc::new(Int32Array::from(vec![1, 2])) as _,
+                Arc::new(Int32Array::from(vec![100, 200])) as _,
+                Arc::new(meta) as _,
+            ],
+        )
+        .unwrap();
+        let table = conn
+            .create_table("awkward_names", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        table
+            .add_columns()
+            .computed("y", "`camelCase` * 2")
+            .computed("z", "coalesce(y, 0) + `with-hyphen` + meta.age")
+            .execute()
+            .await
+            .unwrap();
+        let z = crate::table::computed_columns::computed_columns(
+            table.schema().await.unwrap().as_ref(),
+        )
+        .into_iter()
+        .find(|c| c.name == "z")
+        .unwrap();
+        assert_eq!(z.inputs, vec!["meta.age", "with-hyphen", "y"]);
+
+        let err = table.refresh_column("z").await.unwrap_err();
+        assert!(err.to_string().contains("refresh 'y' first"), "{err}");
+        assert_eq!(table.refresh_column("y").await.unwrap().rows_filled, 2);
+        assert_eq!(table.refresh_column("z").await.unwrap().rows_filled, 2);
+        assert_eq!(read(&table, "z").await, vec![Some(112), Some(224)]);
+    }
+
     #[tokio::test]
     async fn test_refresh_fills_a_declared_column() {
         let table = table_with("refresh_fills", vec![1, 2, 3]).await;
@@ -651,7 +806,8 @@ mod tests {
 
         let read_back = read(&table, "doubled").await;
         assert_eq!(read_back.len(), 20_000);
-        let mut expected: Vec<Option<i32>> = values.iter().map(|v| Some(v * 2)).collect();
+        let mut expected: Vec<Option<i64>> =
+            values.iter().map(|v| Some(i64::from(v * 2))).collect();
         expected.sort();
         assert_eq!(read_back, expected);
     }