From c429863122489cc19a47aabc187fad1f37ef9bfc Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Fri, 14 Aug 2026 16:05:04 -0700 Subject: [PATCH] feat: refresh_column_async returns a job handle (#3939) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- Stack created with GitHub Stacks CLI • Give Feedback 💬 --- docs/src/js/classes/Table.md | 33 ++++ nodejs/__test__/table.test.ts | 22 +++ nodejs/lancedb/table.ts | 22 +++ nodejs/src/table.rs | 10 ++ python/python/lancedb/_lancedb.pyi | 1 + python/python/lancedb/remote/table.py | 3 + python/python/lancedb/table.py | 59 ++++++ python/python/tests/test_table.py | 28 +++ python/src/table.rs | 11 ++ rust/lancedb/src/job.rs | 2 +- rust/lancedb/src/table.rs | 33 ++++ rust/lancedb/src/table/refresh.rs | 246 ++++++++++++++++++++++++++ 12 files changed, 469 insertions(+), 1 deletion(-) diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index 278559cc4..712c15ad0 100644 --- a/docs/src/js/classes/Table.md +++ b/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 +``` + +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`<[`Job`](Job.md)> + +#### Example + +```ts +const job = await table.refreshColumnAsync("doubled"); +await job.wait(); +console.log(await job.status()); // "finished" +``` + +*** + ### restore() ```ts diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts index bc495d24b..5396a251a 100644 --- a/nodejs/__test__/table.test.ts +++ b/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 }]); diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index 5b8d00076..4469e41a0 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -574,6 +574,24 @@ export abstract class Table { */ abstract refreshColumn(column: string): Promise; + /** + * 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; + /** * 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 { + return await this.inner.refreshColumnAsync(column); + } + async alterColumns( columnAlterations: ColumnAlteration[], ): Promise { diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index 40ed7d9f0..4c45be668 100644 --- a/nodejs/src/table.rs +++ b/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 { + 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, diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 96bbecad8..22878fd85 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/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]] diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py index 5bd446775..b1bc5bded 100644 --- a/python/python/lancedb/remote/table.py +++ b/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: diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index db25c4ebc..9c5925cb7 100644 --- a/python/python/lancedb/table.py +++ b/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: diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index ddc1f450f..bb011f8c0 100644 --- a/python/python/tests/test_table.py +++ b/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] diff --git a/python/src/table.rs b/python/src/table.rs index a4c3c307a..35ee92dc4 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -1559,6 +1559,17 @@ impl Table { }) } + pub fn refresh_column_async( + self_: PyRef<'_, Self>, + column: String, + ) -> PyResult> { + 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, diff --git a/rust/lancedb/src/job.rs b/rust/lancedb/src/job.rs index 789ce8312..d77dd6974 100644 --- a/rust/lancedb/src/job.rs +++ b/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)); diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index bfa060638..093d63438 100644 --- a/rust/lancedb/src/table.rs +++ b/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 { + 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; /// 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> { + /// 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) -> Result { + 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 { + refresh::execute_refresh_column_async(self, column).await + } + async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result { let result = schema_evolution::execute_alter_columns(self, alterations).await?; self.bump_freshness(); diff --git a/rust/lancedb/src/table/refresh.rs b/rust/lancedb/src/table/refresh.rs index ab4f8e452..edc78387e 100644 --- a/rust/lancedb/src/table/refresh.rs +++ b/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 { + // 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 = (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