feat: refresh_column_async returns a job handle (#3939)

Mirrors create_index's dual surface: the blocking refresh_column keeps
returning {rows_filled, version}, and refresh_column_async returns the
same
Job handle create_index uses, running the refresh as an in-process task.
Invalid input is reported by the submitting call rather than by the job.

---

<sub>Stack created with <a
href="https://github.com/github/gh-stack">GitHub Stacks CLI</a> • <a
href="https://gh.io/stacks-feedback">Give Feedback 💬</a></sub>

docs/src/js/classes/Table.md

@@ -770,6 +770,39 @@ number of rows filled and the new version number of the table.
 
 ***
 
+### refreshColumnAsync()
+
+```ts
+abstract refreshColumnAsync(column): Promise<Job>
+```
+
+Like [Table#refreshColumn](Table.md#refreshcolumn), but returns a handle to the refresh
+job instead of blocking until it completes.
+
+The job may already be complete when returned; callers must not assume
+the column is filled until [Job.wait](Job.md#wait) resolves. Invalid input --
+an unknown column, or one that is not computed -- rejects here rather
+than failing the job. Local tables only.
+
+#### Parameters
+
+* **column**: `string`
+    The name of the computed column to fill.
+
+#### Returns
+
+`Promise`&lt;[`Job`](Job.md)&gt;
+
+#### Example
+
+```ts
+const job = await table.refreshColumnAsync("doubled");
+await job.wait();
+console.log(await job.status()); // "finished"
+```
+
+***
+
 ### restore()
 
 ```ts

nodejs/__test__/table.test.ts

@@ -3365,6 +3365,28 @@ describe("computed columns", () => {
     expect(rows.map((r) => r.doubled).sort()).toEqual([2, 4]);
   });
 
+  it("returns a job handle from refreshColumnAsync", async () => {
+    const db = await connect(tmpDir.name);
+    const table = await db.createTable("computed_job", [{ x: 1 }, { x: 2 }]);
+
+    await table.addColumns({
+      computed: [{ name: "doubled", valueSql: "x * 2" }],
+    });
+
+    const job = await table.refreshColumnAsync("doubled");
+    expect(job.id).toBeNull();
+    await job.wait();
+    expect(await job.status()).toBe("finished");
+
+    const rows = await table.query().toArray();
+    expect(rows.map((r) => r.doubled).sort()).toEqual([2, 4]);
+
+    // Bad input rejects at the call, not through the job.
+    await expect(table.refreshColumnAsync("x")).rejects.toThrow(
+      "not a computed column",
+    );
+  });
+
   it("fills rows added since the last refresh", async () => {
     const db = await connect(tmpDir.name);
     const table = await db.createTable("computed_append", [{ x: 1 }]);

nodejs/lancedb/table.ts

@@ -574,6 +574,24 @@ export abstract class Table {
    */
   abstract refreshColumn(column: string): Promise<RefreshColumnResult>;
 
+  /**
+   * Like {@link Table#refreshColumn}, but returns a handle to the refresh
+   * job instead of blocking until it completes.
+   *
+   * The job may already be complete when returned; callers must not assume
+   * the column is filled until {@link Job.wait} resolves. Invalid input --
+   * an unknown column, or one that is not computed -- rejects here rather
+   * than failing the job. Local tables only.
+   * @param {string} column The name of the computed column to fill.
+   * @example
+   * ```ts
+   * const job = await table.refreshColumnAsync("doubled");
+   * await job.wait();
+   * console.log(await job.status()); // "finished"
+   * ```
+   */
+  abstract refreshColumnAsync(column: string): Promise<Job>;
+
   /**
    * Alter the name or nullability of columns.
    * @param {ColumnAlteration[]} columnAlterations One or more alterations to
@@ -1179,6 +1197,10 @@ export class LocalTable extends Table {
     return await this.inner.refreshColumn(column);
   }
 
+  async refreshColumnAsync(column: string): Promise<Job> {
+    return await this.inner.refreshColumnAsync(column);
+  }
+
   async alterColumns(
     columnAlterations: ColumnAlteration[],
   ): Promise<AlterColumnsResult> {

nodejs/src/table.rs

@@ -371,6 +371,16 @@ impl Table {
         Ok(res.into())
     }
 
+    #[napi(catch_unwind)]
+    pub async fn refresh_column_async(&self, column: String) -> napi::Result<crate::job::Job> {
+        let job = self
+            .inner_ref()?
+            .refresh_column_async(column)
+            .await
+            .default_error()?;
+        Ok(crate::job::Job::new(job))
+    }
+
     #[napi(catch_unwind)]
     pub async fn add_columns_with_schema(
         &self,

python/python/lancedb/_lancedb.pyi

@@ -342,6 +342,7 @@ class Table:
         self, columns: list[tuple[str, str]]
     ) -> AddColumnsResult: ...
     async def refresh_column(self, column: str) -> RefreshColumnResult: ...
+    async def refresh_column_async(self, column: str) -> Job: ...
     async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...
     async def alter_columns(
         self, columns: list[dict[str, Any]]

python/python/lancedb/remote/table.py

@@ -973,6 +973,9 @@ class RemoteTable(Table):
     def refresh_column(self, column: str):
         raise NotImplementedError("computed columns are supported only on local tables")
 
+    def refresh_column_async(self, column: str) -> Job:
+        raise NotImplementedError("computed columns are supported only on local tables")
+
     def alter_columns(
         self, *alterations: Iterable[Dict[str, str]]
     ) -> AlterColumnsResult:

python/python/lancedb/table.py

@@ -2002,6 +2002,31 @@ class Table(ABC):
             version: the new version number of the table.
         """
 
+    @abstractmethod
+    def refresh_column_async(self, column: str) -> Job:
+        """
+        Like :meth:`refresh_column`, but returns a handle to the refresh job
+        instead of blocking until it completes.
+
+        The job may already be complete when returned; callers must not assume
+        the column is filled until :meth:`Job.wait` returns. Invalid input --
+        an unknown column, or one that is not computed -- raises here rather
+        than failing the job. Local tables only; LanceDB Cloud and Enterprise
+        raise ``NotImplementedError``.
+
+        Examples
+        --------
+        >>> import lancedb
+        >>> db = lancedb.connect("./.lancedb")
+        >>> table = db.create_table("computed_job_demo", [{"x": 1}, {"x": 2}])
+        >>> table.add_columns(computed={"doubled": "x * 2"})
+        AddColumnsResult(version=2)
+        >>> job = table.refresh_column_async("doubled")
+        >>> job.wait()
+        >>> job.status()
+        'finished'
+        """
+
     @abstractmethod
     def alter_columns(self, *alterations: Iterable[Dict[str, str]]):
         """
@@ -4020,6 +4045,13 @@ class LanceTable(Table):
         [`AsyncTable.refresh_column`][lancedb.AsyncTable.refresh_column]."""
         return LOOP.run(self._table.refresh_column(column))
 
+    def refresh_column_async(self, column: str) -> Job:
+        """Fill a computed column's unfilled rows, returning a handle to the
+        refresh job. See
+        [`Table.refresh_column_async`][lancedb.table.Table.refresh_column_async].
+        """
+        return Job(LOOP.run(self._table.refresh_column_async(column)))
+
     def alter_columns(
         self, *alterations: Iterable[Dict[str, str]]
     ) -> AlterColumnsResult:
@@ -6018,6 +6050,33 @@ class AsyncTable:
         """
         return await self._inner.refresh_column(column)
 
+    async def refresh_column_async(self, column: str) -> AsyncJob:
+        """
+        Like :meth:`refresh_column`, but returns a handle to the refresh job
+        instead of blocking until it completes.
+
+        The job may already be complete when returned; callers must not assume
+        the column is filled until :meth:`AsyncJob.wait` resolves. Invalid
+        input -- an unknown column, or one that is not computed -- raises here
+        rather than failing the job. Local tables only; LanceDB Cloud and
+        Enterprise raise ``NotImplementedError``.
+
+        Examples
+        --------
+        >>> import asyncio
+        >>> import lancedb
+        >>> async def refresh_in_background():
+        ...     db = await lancedb.connect_async("./.lancedb")
+        ...     table = await db.create_table("computed_job_async_demo", [{"x": 1}])
+        ...     await table.add_columns(computed={"doubled": "x * 2"})
+        ...     job = await table.refresh_column_async("doubled")
+        ...     await job.wait()
+        ...     return await job.status()
+        >>> asyncio.run(refresh_in_background())
+        'finished'
+        """
+        return AsyncJob(await self._inner.refresh_column_async(column))
+
     async def alter_columns(
         self, *alterations: Iterable[dict[str, Any]]
     ) -> AlterColumnsResult:

python/python/tests/test_table.py

@@ -3888,3 +3888,31 @@ async def test_computed_column_async(tmp_path):
     await table.refresh_column("tripled")
 
     assert (await table.to_arrow())["tripled"].to_pylist() == [9]
+
+
+def test_refresh_column_async_returns_job(tmp_path):
+    db = lancedb.connect(tmp_path)
+    table = db.create_table("computed_job", [{"x": 1}, {"x": 2}])
+    table.add_columns(computed={"doubled": "x * 2"})
+
+    job = table.refresh_column_async("doubled")
+    assert job.id is None  # in-process jobs have no server id
+    job.wait()
+    assert job.status() == "finished"
+    assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4]
+
+    # Bad input raises at the call, not through the job.
+    with pytest.raises(Exception, match="not a computed column"):
+        table.refresh_column_async("x")
+
+
+@pytest.mark.asyncio
+async def test_refresh_column_async_job_async_table(tmp_path):
+    db = await lancedb.connect_async(tmp_path)
+    table = await db.create_table("computed_job_async", [{"x": 3}])
+    await table.add_columns(computed={"tripled": "x * 3"})
+
+    job = await table.refresh_column_async("tripled")
+    await job.wait()
+    assert await job.status() == "finished"
+    assert (await table.to_arrow())["tripled"].to_pylist() == [9]

python/src/table.rs

@@ -1559,6 +1559,17 @@ impl Table {
         })
     }
 
+    pub fn refresh_column_async(
+        self_: PyRef<'_, Self>,
+        column: String,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let job = inner.refresh_column_async(column).await.infer_error()?;
+            Ok(crate::job::Job::new(job))
+        })
+    }
+
     pub fn add_columns_with_schema(
         self_: PyRef<'_, Self>,
         schema: PyArrowType<Schema>,

rust/lancedb/src/job.rs

@@ -141,7 +141,7 @@ impl SpawnedJob {
                 Ok(Err(err)) => Outcome::Failed(Arc::new(err)),
                 Err(err) if err.is_cancelled() => Outcome::Cancelled,
                 Err(err) => Outcome::Failed(Arc::new(Error::Runtime {
-                    message: format!("index job task failed: {err}"),
+                    message: format!("job task failed: {err}"),
                 })),
             };
             let _ = tx.send(Some(outcome));

rust/lancedb/src/table.rs

@@ -764,6 +764,13 @@ 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, returning a [`Job`] tracking
+    /// the operation.
+    async fn refresh_column_async(&self, _column: &str) -> Result<Job> {
+        Err(Error::NotSupported {
+            message: "computed columns are supported only on local tables".into(),
+        })
+    }
     /// Alter columns in the table.
     async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result<AlterColumnsResult>;
     /// Drop columns from the table.
@@ -1679,6 +1686,28 @@ impl Table {
         self.inner.refresh_column(column.as_ref()).await
     }
 
+    /// Like [`Table::refresh_column`], but returns a [`Job`] tracking the
+    /// operation instead of blocking until it completes.
+    ///
+    /// The job may already be complete when returned, and callers must not
+    /// assume the column is filled until [`Job::wait`] returns. Invalid input
+    /// -- an unknown column, or one that is not computed -- is reported by
+    /// this call rather than by the job. Local tables only: LanceDB Cloud and
+    /// Enterprise reject with `NotSupported`.
+    ///
+    /// ```
+    /// # use lancedb::Table;
+    /// # async fn refresh_in_background(table: &Table) -> Result<(), Box<dyn std::error::Error>> {
+    /// let job = table.refresh_column_async("doubled").await?;
+    /// println!("refresh running: {:?}", job.status().await?);
+    /// job.wait().await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn refresh_column_async(&self, column: impl AsRef<str>) -> Result<Job> {
+        self.inner.refresh_column_async(column.as_ref()).await
+    }
+
     /// Change a column's name or nullability.
     pub async fn alter_columns(
         &self,
@@ -3392,6 +3421,10 @@ impl BaseTable for NativeTable {
         Ok(result)
     }
 
+    async fn refresh_column_async(&self, column: &str) -> Result<Job> {
+        refresh::execute_refresh_column_async(self, column).await
+    }
+
     async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result<AlterColumnsResult> {
         let result = schema_evolution::execute_alter_columns(self, alterations).await?;
         self.bump_freshness();

rust/lancedb/src/table/refresh.rs

@@ -35,6 +35,7 @@ use serde::{Deserialize, Serialize};
 
 use super::computed_columns::{BoundExpression, ComputedColumnKind, computed_column_from_field};
 use super::{BaseTable, NativeTable};
+use crate::job::Job;
 use crate::{Error, Result};
 
 /// The result of refreshing a computed column.
@@ -115,6 +116,25 @@ pub(crate) async fn execute_refresh_column(
     })
 }
 
+/// Run the refresh as a [`Job`] in this process.
+pub(crate) async fn execute_refresh_column_async(table: &NativeTable, column: &str) -> Result<Job> {
+    // Validate before spawning so bad input is reported by this call rather
+    // than only by the job.
+    table.dataset.ensure_mutable()?;
+    ensure_no_lsm_write_spec(table).await?;
+    let dataset = table.dataset.get().await?;
+    declared_expression(&dataset, column)?;
+    drop(dataset);
+
+    let table = table.clone();
+    let column = column.to_string();
+    Ok(Job::spawned(tokio::spawn(async move {
+        execute_refresh_column(&table, &column).await?;
+        table.bump_freshness();
+        Ok(())
+    })))
+}
+
 /// Refuse to refresh under an LSM write spec.
 ///
 /// Refresh enumerates base fragments, and a write spec keeps visible rows in
@@ -606,6 +626,230 @@ mod tests {
         assert!(Arc::ptr_eq(&dataset.session(), &session));
     }
 
+    /// The async form's job settles with the fill visible, like
+    /// create_index's execute_async.
+    #[tokio::test]
+    async fn test_refresh_async_job_waits_for_the_fill() {
+        let table = table_with("refresh_async", vec![1, 2, 3]).await;
+        declare_doubled(&table).await.unwrap();
+
+        let job = table.refresh_column_async("doubled").await.unwrap();
+        assert!(job.id().is_none(), "in-process jobs have no server id");
+        job.wait().await.unwrap();
+        assert_eq!(job.status().await.unwrap(), "finished");
+        assert_eq!(
+            read(&table, "doubled").await,
+            vec![Some(2), Some(4), Some(6)]
+        );
+    }
+
+    /// Bad input is reported by the call, not by the job.
+    #[tokio::test]
+    async fn test_refresh_async_rejects_bad_input_before_spawning() {
+        let table = table_with("refresh_async_bad", vec![1, 2, 3]).await;
+
+        let err = table.refresh_column_async("x").await.unwrap_err();
+        assert!(matches!(err, Error::NotAComputedColumn { name } if name == "x"));
+
+        let err = table.refresh_column_async("nope").await.unwrap_err();
+        assert!(matches!(err, Error::ColumnNotFound { name } if name == "nope"));
+    }
+
+    #[tokio::test]
+    async fn test_refresh_async_job_reports_success_to_every_waiter() {
+        let table = table_with("refresh_async_waiters", vec![1, 2]).await;
+        declare_doubled(&table).await.unwrap();
+
+        let job = table.refresh_column_async("doubled").await.unwrap();
+        job.wait().await.unwrap();
+        // A second wait after completion observes the same outcome.
+        job.wait().await.unwrap();
+        assert_eq!(job.status().await.unwrap(), "finished");
+    }
+
+    #[tokio::test]
+    async fn test_refresh_rejects_a_plain_column() {
+        let table = table_with("refresh_plain", vec![1, 2, 3]).await;
+        let err = table.refresh_column("x").await.unwrap_err();
+        assert!(matches!(err, Error::NotAComputedColumn { name } if name == "x"));
+    }
+
+    #[tokio::test]
+    async fn test_refresh_rejects_an_unknown_column() {
+        let table = table_with("refresh_missing", vec![1, 2, 3]).await;
+        let err = table.refresh_column("nope").await.unwrap_err();
+        assert!(matches!(err, Error::ColumnNotFound { name } if name == "nope"));
+    }
+
+    /// The gate's reproducer: a poison value in a deleted row must not
+    /// abort filling the live rows, since nobody can read it.
+    #[tokio::test]
+    async fn test_a_deleted_rows_value_is_never_evaluated() {
+        let table = table_with("refresh_deleted_poison", vec![1, 0]).await;
+        table
+            .add_columns()
+            .computed("quotient", "10 / x")
+            .execute()
+            .await
+            .unwrap();
+        table.delete("x = 0").await.unwrap();
+
+        let result = table.refresh_column("quotient").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        assert_eq!(read(&table, "quotient").await, vec![Some(10)]);
+    }
+
+    /// The gate's reproducer: an already-filled row's value must not be
+    /// re-evaluated either -- its input may have mutated into one the
+    /// expression chokes on.
+    #[tokio::test]
+    async fn test_a_filled_rows_value_is_never_evaluated() {
+        let table = table_with("refresh_filled_poison", vec![1, 2]).await;
+        table
+            .add_columns()
+            .computed("quotient", "10 / x")
+            .execute()
+            .await
+            .unwrap();
+        table.refresh_column("quotient").await.unwrap();
+
+        table
+            .update()
+            .column("x", "0")
+            .only_if("x = 1")
+            .execute()
+            .await
+            .unwrap();
+        append(&table, vec![5]).await;
+
+        let result = table.refresh_column("quotient").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        assert_eq!(
+            read(&table, "quotient").await,
+            vec![Some(2), Some(5), Some(10)]
+        );
+    }
+
+    /// The gate's reproducer: the old internal projection alias is an
+    /// ordinary column name; a computed column may use it.
+    #[tokio::test]
+    async fn test_refresh_a_column_named_like_the_old_alias() {
+        let table = table_with("refresh_alias_name", vec![1, 2]).await;
+        table
+            .add_columns()
+            .computed("__lancedb_computed", "x * 2")
+            .execute()
+            .await
+            .unwrap();
+
+        let result = table.refresh_column("__lancedb_computed").await.unwrap();
+        assert_eq!(result.rows_filled, 2);
+        assert_eq!(
+            read(&table, "__lancedb_computed").await,
+            vec![Some(2), Some(4)]
+        );
+    }
+
+    /// The gate's reproducer: a late-gain fragment (filled, then one null row
+    /// compacted onto the end) fills without the old probe's buffering, which
+    /// this pins behaviorally; the memory bound is structural -- the fill
+    /// stream retains no batches at all.
+    #[tokio::test]
+    async fn test_refresh_fills_a_late_gain_fragment() {
+        let values: Vec<i32> = (0..20_000).collect();
+        let table = table_with("refresh_late_gain", values).await;
+        declare_doubled(&table).await.unwrap();
+        table.refresh_column("doubled").await.unwrap();
+
+        append(&table, vec![2_000_000]).await;
+        table
+            .optimize(crate::table::OptimizeAction::Compact {
+                options: crate::table::CompactionOptions::default(),
+                remap_options: None,
+            })
+            .await
+            .unwrap();
+
+        let result = table.refresh_column("doubled").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        let read_back = read(&table, "doubled").await;
+        assert_eq!(read_back.len(), 20_001);
+        assert_eq!(read_back.last().unwrap(), &Some(4_000_000));
+    }
+
+    /// The gate's reproducer: a nested input declares, refreshes, and guards
+    /// its root against invalidating schema changes.
+    #[tokio::test]
+    async fn test_a_nested_input_declares_and_refreshes() {
+        use arrow_array::{Int32Array, StructArray};
+        use arrow_schema::{DataType, Field, Fields};
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let age = Arc::new(Int32Array::from(vec![30, 40]));
+        let fields = Fields::from(vec![Field::new("age", DataType::Int32, true)]);
+        let metadata = StructArray::new(fields.clone(), vec![age as _], None);
+        let schema = Arc::new(arrow_schema::Schema::new(vec![Field::new(
+            "metadata",
+            DataType::Struct(fields),
+            true,
+        )]));
+        let batch =
+            arrow_array::RecordBatch::try_new(schema, vec![Arc::new(metadata) as _]).unwrap();
+        let table = conn
+            .create_table("refresh_nested", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        table
+            .add_columns()
+            .computed("next_age", "metadata.age + 1")
+            .execute()
+            .await
+            .unwrap();
+        let declaration =
+            &crate::table::computed_columns(table.schema().await.unwrap().as_ref())[0];
+        assert_eq!(declaration.inputs, vec!["metadata.age".to_string()]);
+
+        let result = table.refresh_column("next_age").await.unwrap();
+        assert_eq!(result.rows_filled, 2);
+        assert_eq!(read(&table, "next_age").await, vec![Some(31), Some(41)]);
+
+        // The dotted input guards its root.
+        let err = table.drop_columns(&["metadata"]).await.unwrap_err();
+        assert!(
+            matches!(&err, Error::InvalidInput { message } if message.contains("next_age")),
+            "{err:?}"
+        );
+
+        // Masking a struct input for a deleted row goes through the same
+        // nullif path as a primitive; a nested input plus deletions must not
+        // be the combination that breaks it.
+        table.delete("next_age = 31").await.unwrap();
+        append_struct_row(&table, 50).await;
+        let result = table.refresh_column("next_age").await.unwrap();
+        assert_eq!(result.rows_filled, 1);
+        assert_eq!(read(&table, "next_age").await, vec![Some(41), Some(51)]);
+    }
+
+    /// Append one `metadata: {age}` row to the nested-input table.
+    async fn append_struct_row(table: &Table, age: i32) {
+        use arrow_array::{Int32Array, StructArray};
+        use arrow_schema::{DataType, Field, Fields};
+
+        let ages = Arc::new(Int32Array::from(vec![age]));
+        let fields = Fields::from(vec![Field::new("age", DataType::Int32, true)]);
+        let metadata = StructArray::new(fields.clone(), vec![ages as _], None);
+        let schema = Arc::new(arrow_schema::Schema::new(vec![Field::new(
+            "metadata",
+            DataType::Struct(fields),
+            true,
+        )]));
+        let batch =
+            arrow_array::RecordBatch::try_new(schema, vec![Arc::new(metadata) as _]).unwrap();
+        table.add(batch).execute().await.unwrap();
+    }
+
     /// Both orders of declare+spec are refused at the source (see the
     /// schema_evolution tests); refresh's own check covers a dataset another
     /// writer left in that state.
@@ -639,6 +883,8 @@ mod tests {
             matches!(&err, Error::NotSupported { message } if message.contains("LSM")),
             "{err:?}"
         );
+        let err = table.refresh_column_async("doubled").await.unwrap_err();
+        assert!(matches!(err, Error::NotSupported { .. }));
     }
 
     /// After catch-up activation and unset, no spec remains but the catch-up