diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index f87fd3d13..4c7613959 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -335,6 +335,10 @@ class Table: ) -> list[FtsToken]: ... async def delete(self, filter: Union[str, PyExpr]) -> DeleteResult: ... async def add_columns(self, columns: list[tuple[str, str]]) -> AddColumnsResult: ... + async def add_computed_columns( + self, columns: list[tuple[str, str]] + ) -> AddColumnsResult: ... + async def refresh_column(self, column: str) -> RefreshColumnResult: ... async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ... async def alter_columns( self, columns: list[dict[str, Any]] @@ -680,6 +684,10 @@ class LsmWriteSpec: class AddColumnsResult: version: int +class RefreshColumnResult: + rows_filled: int + version: int + class AlterColumnsResult: version: int diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py index acc2f4c9d..82da3779f 100644 --- a/python/python/lancedb/remote/table.py +++ b/python/python/lancedb/remote/table.py @@ -958,8 +958,16 @@ class RemoteTable(Table): def count_rows(self, filter: Optional[str] = None) -> int: return LOOP.run(self._table.count_rows(filter)) - def add_columns(self, transforms: Dict[str, str]) -> AddColumnsResult: - return LOOP.run(self._table.add_columns(transforms)) + def add_columns( + self, + transforms: Dict[str, str] | None = None, + *, + computed: Dict[str, str] | None = None, + ) -> AddColumnsResult: + return LOOP.run(self._table.add_columns(transforms, computed=computed)) + + def refresh_column(self, column: str): + return LOOP.run(self._table.refresh_column(column)) def alter_columns( self, *alterations: Iterable[Dict[str, str]] diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index c566fc532..3ca245e55 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -176,6 +176,7 @@ if TYPE_CHECKING: CompactionStats, Tag, AddColumnsResult, + RefreshColumnResult, AddResult, AlterColumnsResult, UpdateFieldMetadataResult, @@ -1916,7 +1917,14 @@ class Table(ABC): @abstractmethod def add_columns( - self, transforms: Dict[str, str] | pa.Field | List[pa.Field] | pa.Schema + self, + transforms: Dict[str, str] + | pa.Field + | List[pa.Field] + | pa.Schema + | None = None, + *, + computed: Dict[str, str] | None = None, ): """ Add new columns with defined values. @@ -3939,9 +3947,21 @@ class LanceTable(Table): return LOOP.run(self._table.index_stats(index_name)) def add_columns( - self, transforms: Dict[str, str] | pa.field | List[pa.field] | pa.Schema + self, + transforms: Dict[str, str] + | pa.field + | List[pa.field] + | pa.Schema + | None = None, + *, + computed: Dict[str, str] | None = None, ) -> AddColumnsResult: - return LOOP.run(self._table.add_columns(transforms)) + return LOOP.run(self._table.add_columns(transforms, computed=computed)) + + def refresh_column(self, column: str) -> "RefreshColumnResult": + """Fill a computed column's unfilled rows. See + [`AsyncTable.refresh_column`][lancedb.AsyncTable.refresh_column].""" + return LOOP.run(self._table.refresh_column(column)) def alter_columns( self, *alterations: Iterable[Dict[str, str]] @@ -5856,7 +5876,14 @@ class AsyncTable: return await self._inner.update(updates_sql, where) async def add_columns( - self, transforms: dict[str, str] | pa.field | List[pa.field] | pa.Schema + self, + transforms: dict[str, str] + | pa.field + | List[pa.field] + | pa.Schema + | None = None, + *, + computed: dict[str, str] | None = None, ) -> AddColumnsResult: """ Add new columns with defined values. @@ -5882,11 +5909,35 @@ class AsyncTable: {isinstance(f, pa.Field) for f in transforms} ): transforms = pa.schema(transforms) + if computed: + if transforms: + raise ValueError( + "add_columns cannot take both transforms and computed columns" + ) + return await self._inner.add_computed_columns(list(computed.items())) + if transforms is None: + raise ValueError("add_columns requires transforms or computed columns") if isinstance(transforms, pa.Schema): return await self._inner.add_columns_with_schema(transforms) else: return await self._inner.add_columns(list(transforms.items())) + async def refresh_column(self, column: str) -> RefreshColumnResult: + """ + Compute and store values for a computed column's unfilled rows. + + Parameters + ---------- + column: str + The name of the computed column to fill. + + Returns + ------- + RefreshColumnResult + The number of rows filled and the new version of the table. + """ + return await self._inner.refresh_column(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 2a069c712..ddc1f450f 100644 --- a/python/python/tests/test_table.py +++ b/python/python/tests/test_table.py @@ -3854,3 +3854,37 @@ async def test_async_search_runs_embedding_on_dedicated_executor( assert all(name.startswith("lancedb-embedding") for name in captured_threads), ( f"embedding ran off the dedicated executor: {captured_threads}" ) + + +def test_computed_column_declare_and_refresh(tmp_path): + db = lancedb.connect(tmp_path) + table = db.create_table("computed", [{"x": 1}, {"x": 2}]) + + table.add_columns(computed={"doubled": "x * 2"}) + assert table.to_arrow()["doubled"].to_pylist() == [None, None] + + result = table.refresh_column("doubled") + assert result.rows_filled == 2 + assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4] + + table.add([{"x": 5}]) + assert table.refresh_column("doubled").rows_filled == 1 + assert sorted(table.to_arrow()["doubled"].to_pylist()) == [2, 4, 10] + + +def test_computed_column_rejects_transforms_and_computed_together(tmp_path): + db = lancedb.connect(tmp_path) + table = db.create_table("computed_mixed", [{"x": 1}]) + with pytest.raises(ValueError): + table.add_columns({"a": "x + 1"}, computed={"b": "x * 2"}) + + +@pytest.mark.asyncio +async def test_computed_column_async(tmp_path): + db = await lancedb.connect_async(tmp_path) + table = await db.create_table("computed_async", [{"x": 3}]) + + await table.add_columns(computed={"tripled": "x * 3"}) + await table.refresh_column("tripled") + + assert (await table.to_arrow())["tripled"].to_pylist() == [9] diff --git a/python/src/lib.rs b/python/src/lib.rs index 6b0c0cf97..a19bf172d 100644 --- a/python/src/lib.rs +++ b/python/src/lib.rs @@ -16,7 +16,8 @@ use query::{FTSQuery, HybridQuery, Query, VectorQuery}; use session::Session; use table::{ AddColumnsResult, AddResult, AlterColumnsResult, DeleteResult, DropColumnsResult, FtsToken, - LsmWriteSpec, MergeResult, PyBlobFile, Table, UpdateFieldMetadataResult, UpdateResult, + LsmWriteSpec, MergeResult, PyBlobFile, RefreshColumnResult, Table, UpdateFieldMetadataResult, + UpdateResult, }; pub mod arrow; @@ -57,6 +58,7 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/python/src/table.rs b/python/src/table.rs index cae6b5d9a..a4c3c307a 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -415,6 +415,32 @@ pub struct AddColumnsResult { pub version: u64, } +#[pyclass(get_all, from_py_object)] +#[derive(Clone, Debug)] +pub struct RefreshColumnResult { + pub rows_filled: u64, + pub version: u64, +} + +#[pymethods] +impl RefreshColumnResult { + pub fn __repr__(&self) -> String { + format!( + "RefreshColumnResult(rows_filled={}, version={})", + self.rows_filled, self.version + ) + } +} + +impl From for RefreshColumnResult { + fn from(result: lancedb::table::RefreshColumnResult) -> Self { + Self { + rows_filled: result.rows_filled, + version: result.version, + } + } +} + #[pymethods] impl AddColumnsResult { pub fn __repr__(&self) -> String { @@ -1510,6 +1536,29 @@ impl Table { }) } + pub fn add_computed_columns( + self_: PyRef<'_, Self>, + columns: Vec<(String, String)>, + ) -> PyResult> { + let inner = self_.inner_ref()?.clone(); + future_into_py(self_.py(), async move { + let mut builder = inner.add_columns(); + for (name, expression) in columns { + builder = builder.computed(name, expression); + } + let result = builder.execute().await.infer_error()?; + Ok(AddColumnsResult::from(result)) + }) + } + + pub fn refresh_column(self_: PyRef<'_, Self>, column: String) -> PyResult> { + let inner = self_.inner_ref()?.clone(); + future_into_py(self_.py(), async move { + let result = inner.refresh_column(column).await.infer_error()?; + Ok(RefreshColumnResult::from(result)) + }) + } + pub fn add_columns_with_schema( self_: PyRef<'_, Self>, schema: PyArrowType,