diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index 3fa3b08db..97bdea628 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -69,14 +69,33 @@ abstract addColumns(newColumnTransforms): Promise Add new columns with defined values. +The `{ computed }` form stores the expression rather than evaluating it +now: the column is committed with no values, and a later refresh fills +the rows. Declaring one therefore costs the same on a large table as on +an empty one. + +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. + +Computed columns are local-only: LanceDB Cloud and Enterprise reject a +declaration. + #### Parameters -* **newColumnTransforms**: `Field`<`any`> \| `Field`<`any`>[] \| `Schema`<`any`> \| [`AddColumnsSql`](../interfaces/AddColumnsSql.md)[] +* **newColumnTransforms**: + \| `Field`<`any`> + \| `Field`<`any`>[] + \| `Schema`<`any`> + \| [`AddColumnsSql`](../interfaces/AddColumnsSql.md)[] + \| `object` Either: - An array of objects with column names and SQL expressions to calculate values - A single Arrow Field defining one column with its data type (column will be initialized with null values) - An array of Arrow Fields defining columns with their data types (columns will be initialized with null values) - An Arrow Schema defining columns with their data types (columns will be initialized with null values) + - `{ computed }`, declaring columns defined by a SQL expression whose type and inputs are derived from it #### Returns @@ -85,6 +104,12 @@ Add new columns with defined values. A promise that resolves to an object containing the new version number of the table after adding the columns. +#### Example + +```ts +await table.addColumns({ computed: [{ name: "doubled", valueSql: "x * 2" }] }); +``` + *** ### alterColumns() diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts index d263d9cab..5ff18da3e 100644 --- a/nodejs/__test__/table.test.ts +++ b/nodejs/__test__/table.test.ts @@ -3340,3 +3340,22 @@ describe("LSM merge insert", () => { await expect(table.query().useLsm(true).toArray()).rejects.toThrow(); }); }); + +describe("computed columns", () => { + let tmpDir: tmp.DirResult; + beforeEach(() => { + tmpDir = tmp.dirSync({ unsafeCleanup: true }); + }); + afterEach(() => tmpDir.removeCallback()); + + it("declares a column with no values", async () => { + const db = await connect(tmpDir.name); + const table = await db.createTable("computed", [{ x: 1 }, { x: 2 }]); + + await table.addColumns({ + computed: [{ name: "doubled", valueSql: "x * 2" }], + }); + const rows = await table.query().toArray(); + expect(rows.map((r) => r.doubled)).toEqual([null, null]); + }); +}); diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index 04705475b..6234b8fbf 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -525,16 +525,39 @@ export abstract class Table { abstract vectorSearch(vector: IntoVector | MultiVector): VectorQuery; /** * Add new columns with defined values. + * + * The `{ computed }` form stores the expression rather than evaluating it + * now: the column is committed with no values, and a later refresh fills + * the rows. Declaring one therefore costs the same on a large table as on + * an empty one. + * + * 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. + * + * Computed columns are local-only: LanceDB Cloud and Enterprise reject a + * declaration. * @param {AddColumnsSql[] | Field | Field[] | Schema} newColumnTransforms Either: * - An array of objects with column names and SQL expressions to calculate values * - A single Arrow Field defining one column with its data type (column will be initialized with null values) * - An array of Arrow Fields defining columns with their data types (columns will be initialized with null values) * - An Arrow Schema defining columns with their data types (columns will be initialized with null values) + * - `{ computed }`, declaring columns defined by a SQL expression whose type and inputs are derived from it * @returns {Promise} A promise that resolves to an object * containing the new version number of the table after adding the columns. + * @example + * ```ts + * await table.addColumns({ computed: [{ name: "doubled", valueSql: "x * 2" }] }); + * ``` */ abstract addColumns( - newColumnTransforms: AddColumnsSql[] | Field | Field[] | Schema, + newColumnTransforms: + | AddColumnsSql[] + | Field + | Field[] + | Schema + | { computed: AddColumnsSql[] }, ): Promise; /** @@ -1088,8 +1111,22 @@ export class LocalTable extends Table { // TODO: Support BatchUDF async addColumns( - newColumnTransforms: AddColumnsSql[] | Field | Field[] | Schema, + newColumnTransforms: + | AddColumnsSql[] + | Field + | Field[] + | Schema + | { computed: AddColumnsSql[] }, ): Promise { + // Columns defined by an expression are declared, not materialized here. + if ( + typeof newColumnTransforms === "object" && + !Array.isArray(newColumnTransforms) && + "computed" in newColumnTransforms + ) { + return await this.inner.addComputedColumns(newColumnTransforms.computed); + } + // Handle single Field -> convert to array of Fields if (newColumnTransforms instanceof Field) { newColumnTransforms = [newColumnTransforms]; diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index c4ece20e2..16ca387e6 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -347,6 +347,20 @@ impl Table { Ok(res.into()) } + #[napi(catch_unwind)] + pub async fn add_computed_columns( + &self, + columns: Vec, + ) -> napi::Result { + let table = self.inner_ref()?; + let mut builder = table.add_columns(); + for column in columns { + builder = builder.computed(column.name, column.value_sql); + } + let res = builder.execute().await.default_error()?; + Ok(res.into()) + } + #[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 447bcc88a..84455d74b 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -338,6 +338,9 @@ 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 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 acc2f4c9d..5c98a64f1 100644 --- a/python/python/lancedb/remote/table.py +++ b/python/python/lancedb/remote/table.py @@ -958,7 +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: + def add_columns( + self, + transforms: Dict[str, str] | None = None, + *, + computed: Dict[str, str] | None = None, + ) -> AddColumnsResult: + if computed: + raise NotImplementedError( + "computed columns are supported only on local tables" + ) return LOOP.run(self._table.add_columns(transforms)) def alter_columns( diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index c566fc532..5c9104699 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -1916,7 +1916,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. @@ -1930,11 +1937,38 @@ class Table(ABC): Alternatively, a pyarrow Field or Schema can be provided to add new columns with the specified data types. The new columns will be initialized with null values. + computed: Dict[str, str], optional + A map of column name to a SQL expression defining the column. The + column's type and inputs are derived from the expression, so no + data type is supplied. + + Unlike ``transforms``, the expression is stored rather than + evaluated now: the column is committed with no values, and a + later refresh fills the rows. Declaring one therefore costs the + same on a large table as on an empty one. + + 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. + + Local tables only; LanceDB Cloud and Enterprise raise + ``NotImplementedError``. Cannot be combined with ``transforms``. Returns ------- AddColumnsResult version: the new version number of the table after adding columns. + + Examples + -------- + >>> import lancedb + >>> db = lancedb.connect("./.lancedb") + >>> table = db.create_table("computed_demo", [{"x": 1}, {"x": 2}]) + >>> table.add_columns(computed={"doubled": "x * 2"}) + AddColumnsResult(version=2) + >>> table.to_arrow()["doubled"].to_pylist() + [None, None] """ @abstractmethod @@ -3939,9 +3973,16 @@ 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 alter_columns( self, *alterations: Iterable[Dict[str, str]] @@ -5856,7 +5897,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. @@ -5869,6 +5917,20 @@ class AsyncTable: each row in the table, and can reference existing columns. Alternatively, you can pass a pyarrow field or schema to add new columns with NULLs. + computed: Dict[str, str], optional + A map of column name to a SQL expression defining the column. The + column's type and inputs are derived from the expression. + + Unlike ``transforms``, the expression is stored rather than + evaluated now: the column is committed with no values, and a + later refresh fills the rows. + + 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. + + Local tables only. Cannot be combined with ``transforms``. Returns ------- @@ -5882,6 +5944,14 @@ 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: diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index 2a069c712..6393cd42a 100644 --- a/python/python/tests/test_table.py +++ b/python/python/tests/test_table.py @@ -3854,3 +3854,22 @@ 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_declares_all_null(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] + + # The declaration is durable field metadata. + field = table.schema.field("doubled") + assert field.metadata[b"computed_column.expression"] == b"x * 2" + + +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"}) diff --git a/python/src/table.rs b/python/src/table.rs index cae6b5d9a..a9ff70ad6 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -1510,6 +1510,21 @@ 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 add_columns_with_schema( self_: PyRef<'_, Self>, schema: PyArrowType, diff --git a/rust/lancedb/src/error.rs b/rust/lancedb/src/error.rs index 4a6e6d8d9..6bd1ffa2b 100644 --- a/rust/lancedb/src/error.rs +++ b/rust/lancedb/src/error.rs @@ -71,6 +71,14 @@ pub enum Error { IndexNotFound { name: String }, #[snafu(display("Embedding function '{name}' was not found. : {reason}"))] EmbeddingFunctionNotFound { name: String, reason: String }, + #[snafu(display("Column '{name}' was not found"))] + ColumnNotFound { name: String }, + #[snafu(display("Column '{name}' already exists"))] + ColumnAlreadyExists { name: String }, + #[snafu(display("Column '{name}' is not a computed column"))] + NotAComputedColumn { name: String }, + #[snafu(display("Invalid expression for column '{column}': {message}"))] + InvalidExpression { column: String, message: String }, #[snafu(display("Table '{name}' already exists"))] TableAlreadyExists { name: String }, diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 3816a3a86..3e467b674 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -2700,6 +2700,13 @@ impl BaseTable for RemoteTable { Ok(result) } + // A declaration reaches here as AllNulls, which the remote protocol + // has no representation for. + NewColumnTransform::AllNulls(_) => { + return Err(Error::NotSupported { + message: "computed columns are supported only on local tables".into(), + }); + } _ => { return Err(Error::NotSupported { message: "Only SQL expressions are supported for adding columns".into(), @@ -6449,6 +6456,31 @@ mod tests { assert_eq!(result.version, if old_server { 0 } else { 43 }); } + /// Computed columns are local-only. Both halves say so here rather than + /// reaching the wire and failing somewhere less legible. + #[tokio::test] + async fn test_computed_columns_are_refused() { + let table = Table::new_with_handler("my_table", |request| -> http::Response { + panic!("unexpected request: {}", request.url().path()) + }); + + let declared = Arc::new(Schema::new(vec![Field::new( + "doubled", + DataType::Int32, + true, + )])); + let err = table + .add_columns() + .transform(NewColumnTransform::AllNulls(declared)) + .execute() + .await + .unwrap_err(); + assert!( + matches!(&err, Error::NotSupported { message } if message.contains("local tables")), + "{err:?}" + ); + } + #[tokio::test] async fn test_prewarm_index() { let table = Table::new_with_handler("my_table", |request| { diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 10822cafa..00a51058c 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -68,6 +68,7 @@ pub mod add_columns; mod add_data; pub mod branch_merge; pub mod checkpoint; +pub mod computed_columns; mod create_index; pub mod datafusion; pub(crate) mod dataset; @@ -90,6 +91,9 @@ pub use branch_merge::{ MergeBranchResult, MergeBranchStatus, MergePreview, RowCountSummary, }; pub use chrono::Duration; +pub use computed_columns::{ + ComputedColumn, ComputedColumnKind, computed_column_from_field, computed_columns, +}; pub use delete::DeleteResult; use futures::future::join_all; pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTags}; @@ -741,6 +745,15 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { transforms: NewColumnTransform, read_columns: Option>, ) -> Result; + /// Declare computed columns, each defined by a SQL expression. + async fn add_computed_columns( + &self, + _columns: &[(String, String)], + ) -> Result { + Err(Error::NotSupported { + message: "computed columns are not supported on this table type".into(), + }) + } /// Alter columns in the table. async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result; /// Drop columns from the table. @@ -2628,6 +2641,7 @@ impl NativeTable { namespace_client: Option>, pushdown_operations: HashSet, ) -> Result { + computed_columns::ensure_no_foreign_declarations(batches.arrow_schema().fields())?; // Default params uses format v1. let params = params.unwrap_or(WriteParams { ..Default::default() @@ -3076,6 +3090,13 @@ impl BaseTable for NativeTable { let ds = self.dataset.get().await?; let table_schema = Schema::from(&ds.schema().clone()); + computed_columns::ensure_not_written( + &table_schema, + add.data.schema().fields().iter().map(|f| f.name().as_str()), + )?; + if matches!(add.mode, AddDataMode::Overwrite) { + computed_columns::ensure_no_foreign_declarations(add.data.schema().fields())?; + } let num_partitions = if let Some(parallelism) = add.write_parallelism { parallelism @@ -3236,6 +3257,11 @@ impl BaseTable for NativeTable { params: MergeInsertBuilder, new_data: Box, ) -> Result { + let source_schema = arrow_array::RecordBatchReader::schema(&new_data); + computed_columns::ensure_not_written( + &Schema::from(self.dataset.get().await?.schema()), + source_schema.fields().iter().map(|f| f.name().as_str()), + )?; let result = merge::execute_merge_insert(self, params, new_data).await?; self.bump_freshness(); Ok(result) @@ -3321,6 +3347,12 @@ impl BaseTable for NativeTable { Ok(result) } + async fn add_computed_columns(&self, columns: &[(String, String)]) -> Result { + let result = schema_evolution::execute_declare(self, columns).await?; + self.bump_freshness(); + Ok(result) + } + 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/add_columns.rs b/rust/lancedb/src/table/add_columns.rs index 0d410cd04..e5c4ef8d1 100644 --- a/rust/lancedb/src/table/add_columns.rs +++ b/rust/lancedb/src/table/add_columns.rs @@ -15,6 +15,7 @@ use crate::{Error, Result}; pub struct AddColumnsBuilder { parent: Arc, transform: Option, + computed: Vec<(String, String)>, read_columns: Option>, } @@ -23,6 +24,7 @@ impl std::fmt::Debug for AddColumnsBuilder { f.debug_struct("AddColumnsBuilder") .field("parent", &self.parent) .field("has_transform", &self.transform.is_some()) + .field("computed", &self.computed) .field("read_columns", &self.read_columns) .finish() } @@ -33,19 +35,54 @@ impl AddColumnsBuilder { Self { parent, transform: None, + computed: Vec::new(), read_columns: None, } } - /// Set how the new columns' values are produced. Required. + /// Set how the new columns' values are produced. pub fn transform(mut self, transform: NewColumnTransform) -> Self { self.transform = Some(transform); self } + /// Add a column defined by `expression`, evaluated by a later refresh + /// rather than by this commit. Its type and inputs are derived from the + /// expression. + /// + /// The column is committed with no values, so declaring one costs the same + /// on an empty table as on a large one. Rows get values from a later + /// refresh, which fills every fragment that has none -- including + /// fragments appended since the last refresh. + /// + /// 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. + /// + /// Local tables only: LanceDB Cloud and Enterprise reject a declaration + /// with `NotSupported`. + /// + /// ``` + /// # use lancedb::Table; + /// # async fn declare(table: &Table) -> Result<(), Box> { + /// table + /// .add_columns() + /// .computed("doubled", "x * 2") + /// .execute() + /// .await?; + /// # Ok(()) + /// # } + /// ``` + pub fn computed(mut self, name: impl Into, expression: impl Into) -> Self { + self.computed.push((name.into(), expression.into())); + self + } + /// Limit which existing columns a [`NewColumnTransform::BatchUDF`] mapper - /// receives. Every other transform determines what it reads, so setting - /// this alongside one is an error rather than a silent no-op. + /// receives. Every other transform, and a computed column, determines what + /// it reads, so setting this alongside one is an error rather than a silent + /// no-op. pub fn read_columns(mut self, columns: impl IntoIterator>) -> Self { self.read_columns = Some(columns.into_iter().map(Into::into).collect()); self @@ -56,24 +93,42 @@ impl AddColumnsBuilder { let Self { parent, transform, + computed, read_columns, } = self; - let Some(transform) = transform else { - return Err(Error::InvalidInput { - message: "add_columns requires a transform".into(), - }); - }; - - if read_columns.is_some() && !matches!(transform, NewColumnTransform::BatchUDF(_)) { - return Err(Error::InvalidInput { - message: "read_columns applies only to a BatchUDF transform; \ - every other transform determines what it reads" + match (transform, computed.is_empty()) { + (None, true) => Err(Error::InvalidInput { + message: "add_columns requires a transform or a computed column".into(), + }), + // The two commit through different transforms, so one call covering + // both would be two commits and could half-apply. + (Some(_), false) => Err(Error::InvalidInput { + message: "add_columns cannot mix a transform with computed columns; \ + they cannot be added atomically in one call" .into(), - }); + }), + (Some(transform), true) => { + if read_columns.is_some() && !matches!(transform, NewColumnTransform::BatchUDF(_)) { + return Err(Error::InvalidInput { + message: "read_columns applies only to a BatchUDF transform; \ + every other transform determines what it reads" + .into(), + }); + } + parent.add_columns(transform, read_columns).await + } + (None, false) => { + if read_columns.is_some() { + return Err(Error::InvalidInput { + message: "read_columns applies only to a BatchUDF transform; \ + a computed column's inputs come from its expression" + .into(), + }); + } + parent.add_computed_columns(&computed).await + } } - - parent.add_columns(transform, read_columns).await } } @@ -85,8 +140,8 @@ mod tests { use arrow_schema::{DataType, Field, Schema}; use lance::dataset::{BatchUDF, NewColumnTransform}; - use crate::Table; use crate::connect; + use crate::{Error, Table}; async fn table_with_two_columns(name: &str) -> Table { let conn = connect("memory://").execute().await.unwrap(); @@ -98,10 +153,7 @@ mod tests { async fn test_requires_a_transform() { let table = table_with_two_columns("no_transform").await; let err = table.add_columns().execute().await.unwrap_err(); - assert!( - err.to_string().contains("requires a transform"), - "got: {err}" - ); + assert!(matches!(err, Error::InvalidInput { .. })); } #[tokio::test] @@ -117,7 +169,7 @@ mod tests { .execute() .await .unwrap_err(); - assert!(err.to_string().contains("BatchUDF"), "got: {err}"); + assert!(matches!(err, Error::InvalidInput { .. })); let schema = table.schema().await.unwrap(); assert!( @@ -126,6 +178,47 @@ mod tests { ); } + #[tokio::test] + async fn test_mixing_transform_and_computed_is_rejected() { + let table = table_with_two_columns("mixed_add").await; + let err = table + .add_columns() + .transform(NewColumnTransform::SqlExpressions(vec![( + "eager".into(), + "x * 2".into(), + )])) + .computed("lazy", "x * 3") + .execute() + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidInput { .. })); + + let schema = table.schema().await.unwrap(); + assert!(schema.field_with_name("eager").is_err()); + assert!(schema.field_with_name("lazy").is_err()); + } + + #[tokio::test] + async fn test_read_columns_with_computed_is_rejected() { + let table = table_with_two_columns("read_cols_computed").await; + let err = table + .add_columns() + .computed("doubled", "x * 2") + .read_columns(["x"]) + .execute() + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidInput { .. })); + assert!( + table + .schema() + .await + .unwrap() + .field_with_name("doubled") + .is_err() + ); + } + #[tokio::test] async fn test_read_columns_limits_what_a_batch_udf_sees() { let table = table_with_two_columns("read_cols_udf").await; diff --git a/rust/lancedb/src/table/computed_columns.rs b/rust/lancedb/src/table/computed_columns.rs new file mode 100644 index 000000000..4787b420f --- /dev/null +++ b/rust/lancedb/src/table/computed_columns.rs @@ -0,0 +1,1329 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Computed columns. +//! +//! A computed column is defined by a rule rather than by values supplied at +//! write time. Declaring one commits the column carrying that rule in field +//! metadata but no data, so the cost does not scale with the table; a later +//! refresh fills the rows. +//! +//! The rule is tagged by kind ([`ComputedColumnKind`]) because kinds differ in +//! where the column's type and inputs come from. A SQL expression is +//! self-describing -- both are derived from the expression, so a caller writes +//! neither -- while a kind resolved through a registry cannot be typed without +//! consulting it. Only SQL exists today; the tag is what lets another kind be +//! added without a second reading of the same key. +//! +//! [`computed_columns`] and [`computed_column_from_field`] read declarations +//! back off a schema. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema, SchemaRef}; +use datafusion_common::tree_node::TreeNode; +use lance::dataset::NewColumnTransform; +use lance_datafusion::planner::Planner; + +use crate::{Error, Result}; + +/// Field metadata key marking a column as computed. The value is `"true"`. +pub const COMPUTED_COLUMN_META_KEY: &str = "computed_column"; + +/// Field metadata key naming the kind of rule that defines the column. +pub const KIND_META_KEY: &str = "computed_column.kind"; + +/// Field metadata key holding the SQL expression that defines the column. +pub const EXPRESSION_META_KEY: &str = "computed_column.expression"; + +/// Field metadata key holding the column's inputs, as a JSON array of names. +pub const INPUTS_META_KEY: &str = "computed_column.inputs"; + +/// Value of [`KIND_META_KEY`] for a column defined by a SQL expression. +pub const SQL_KIND: &str = "sql"; + +/// The rule that defines a computed column's values. +/// +/// Non-exhaustive: a kind added later is an additive change, and a caller that +/// only handles the kinds it knows keeps compiling. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub enum ComputedColumnKind { + /// A SQL expression evaluated by DataFusion. It is the whole definition: + /// the column's type and its inputs are both derived from it. + Sql { + /// The expression. + expression: String, + }, + /// A kind this version does not understand, written by a newer one. + /// + /// Reported rather than hidden so a caller can tell a column it cannot + /// refresh apart from one that was never computed. Nothing produces this. + Unrecognized { + /// The kind as it was found in the metadata. + kind: String, + }, +} + +/// A computed column's declaration, as read back from field metadata. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ComputedColumn { + /// Name of the computed column. + pub name: String, + /// The rule that defines it. + pub kind: ComputedColumnKind, + /// Columns the rule reads, recorded at declaration time. + /// + /// Outside the kind because every kind has inputs and the consumers that + /// use them -- refresh planning, dependency ordering -- do not care which + /// kind produced them. Where they come from does differ, and that is + /// settled at declaration: derived from a SQL expression, supplied by the + /// caller for a kind that cannot be parsed. + pub inputs: Vec, +} + +/// Build the field metadata recording a SQL binding. +fn computed_column_metadata(expression: &str, inputs: &[String]) -> HashMap { + HashMap::from([ + (COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()), + (KIND_META_KEY.to_string(), SQL_KIND.to_string()), + (EXPRESSION_META_KEY.to_string(), expression.to_string()), + ( + INPUTS_META_KEY.to_string(), + serde_json::to_string(inputs).unwrap_or_else(|_| "[]".to_string()), + ), + ]) +} + +/// Read a field's computed-column declaration, if it carries one. +/// +/// A field flagged computed but carrying no kind, or a SQL one missing its +/// expression, is not a computed column here: without the rule there is +/// nothing to refresh from, so it is reported as absent rather than as a +/// half-formed declaration. An unrecognized kind is different -- the rule is +/// there and intact, this version just cannot act on it -- and comes back as +/// [`ComputedColumnKind::Unrecognized`]. +pub fn computed_column_from_field(field: &ArrowField) -> Option { + let metadata = field.metadata(); + if metadata.get(COMPUTED_COLUMN_META_KEY).map(String::as_str) != Some("true") { + return None; + } + let kind = match metadata.get(KIND_META_KEY)?.as_str() { + SQL_KIND => ComputedColumnKind::Sql { + expression: metadata.get(EXPRESSION_META_KEY)?.clone(), + }, + other => ComputedColumnKind::Unrecognized { + kind: other.to_string(), + }, + }; + let inputs = metadata + .get(INPUTS_META_KEY) + .and_then(|raw| serde_json::from_str::>(raw).ok()) + .unwrap_or_default(); + Some(ComputedColumn { + name: field.name().clone(), + kind, + inputs, + }) +} + +/// Read every computed-column declaration carried by `schema`, in field order. +/// +/// Introspection is a pure read of the schema the caller already holds, the +/// way a SQL catalog reports a generation expression as another column of +/// `information_schema.columns`. +pub fn computed_columns(schema: &ArrowSchema) -> Vec { + schema + .fields() + .iter() + .filter_map(|field| computed_column_from_field(field)) + .collect() +} + +/// Reject a schema change to a column some declaration reads. +/// +/// A binding is SQL text naming its inputs, so renaming, retyping or dropping +/// one leaves an expression that no longer resolves. Refusing the change keeps +/// a declaration that survived [`plan`] evaluable for as long as it exists. +/// +/// Paths are compared at their root: a declaration reading `metadata` is +/// invalidated by a change to `metadata.age` just as surely. +pub(crate) fn ensure_not_an_input(schema: &SchemaRef, paths: &[&str]) -> Result<()> { + for declaration in computed_columns(schema) { + // The expression, not stored inputs, is the source of truth; an + // expression that no longer parses proves nothing, so refuse. + let inputs = match &declaration.kind { + ComputedColumnKind::Sql { expression } => Planner::new(schema.clone()) + .parse_expr(expression) + .map(|parsed| Planner::column_names_in_expr(&parsed)) + .map_err(|e| Error::InvalidInput { + message: format!( + "computed column '{}' has an unevaluable expression ({e}); drop it \ + before changing the schema", + declaration.name + ), + })?, + _ => declaration.inputs.clone(), + }; + for path in paths { + // Exact target only: the binding travels with the whole column, + // not with a nested field the expression still shapes. + if declaration.name == *path { + continue; + } + if declaration.name == root(path) { + return Err(Error::InvalidInput { + message: format!( + "'{}' is part of computed column '{}'; drop the column and declare \ + it again", + path, declaration.name + ), + }); + } + if inputs.iter().any(|input| root(input) == root(path)) { + return Err(Error::InvalidInput { + message: format!( + "column '{}' is read by computed column '{}'; drop that column first", + path, declaration.name + ), + }); + } + } + } + Ok(()) +} + +/// Reject a write that supplies values for a computed column directly: +/// only refresh materializes one, and refresh never revisits a filled row. +pub(crate) fn ensure_not_written<'a>( + schema: &ArrowSchema, + written: impl IntoIterator, +) -> Result<()> { + let declared: Vec = computed_columns(schema) + .into_iter() + .map(|declaration| declaration.name) + .collect(); + for name in written { + if declared.iter().any(|declared| declared == root(name)) { + return Err(Error::InvalidInput { + message: format!( + "column '{}' is computed; its values come from refresh and cannot be \ + written directly", + root(name) + ), + }); + } + } + Ok(()) +} + +/// Reject a batch holding values for a computed column. Null slots are the +/// declared state, so planner-padded placeholders pass. +pub(crate) fn ensure_batch_writes_no_computed_values( + declared: &[String], + batch: &arrow_array::RecordBatch, +) -> Result<()> { + for name in declared { + if let Some(column) = batch.column_by_name(name) + && column.null_count() != column.len() + { + return Err(Error::InvalidInput { + message: format!( + "column '{name}' is computed; its values come from refresh and cannot \ + be written directly" + ), + }); + } + } + Ok(()) +} + +/// Reject fields carrying declaration metadata that did not come through +/// [`plan`]. One authority for creation, overwrite and raw transforms. +pub(crate) fn ensure_no_foreign_declarations<'a>( + fields: impl IntoIterator>, +) -> Result<()> { + for field in fields { + if field.metadata().keys().any(|k| is_declaration_key(k)) { + return Err(Error::InvalidInput { + message: format!( + "field '{}' carries computed-column metadata; declare computed columns \ + with add_columns().computed()", + field.name() + ), + }); + } + } + Ok(()) +} + +/// True for field-metadata keys that belong to a computed-column declaration. +/// +/// A declaration is immutable through metadata edits: it is validated as a +/// whole at declare time, and rewriting any piece of it -- the flag, the +/// 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.") +} + +/// Reject retyping a computed column itself. +/// +/// A cast keeps the stored expression while changing the type it must yield +/// -- and lance's cast rewrites the field without its metadata, so the +/// declaration silently stops being one. Dropping and redeclaring is the +/// coherent way to change a computed column's type. +pub(crate) fn ensure_not_retyped(schema: &ArrowSchema, paths: &[&str]) -> Result<()> { + for declaration in computed_columns(schema) { + for path in paths { + if declaration.name == root(path) { + return Err(Error::InvalidInput { + message: format!( + "column '{}' is computed; drop it and declare it again to change \ + its type", + declaration.name + ), + }); + } + } + } + Ok(()) +} + +/// The top-level column a possibly nested input path reads. +pub(crate) fn root(path: &str) -> &str { + path.split('.').next().unwrap_or(path) +} + +/// A declaration's expression bound to a schema. +pub(crate) struct BoundExpression { + /// The columns the expression names, as written; nested inputs keep + /// their dotted path. + pub inputs: Vec, + /// The type the expression yields. + pub data_type: DataType, +} + +/// Parse, resolve and compile `expression` against `schema`. +/// +/// Inputs come from the expression as written, before optimization: the +/// simplifier can fold a referenced column out entirely (`true OR x > 0`), +/// and the guard protecting the stored SQL has to see every column the text +/// names, not just the ones the simplified form still reads. +pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result { + let invalid = |message: String| Error::InvalidExpression { + column: column.to_string(), + message, + }; + + let planner = Planner::new(schema.clone()); + let parsed = planner + .parse_expr(expression) + .map_err(|e| invalid(e.to_string()))?; + + // A declaration is evaluated more than once -- staging and writing are + // separate passes, and a refresh years later replays the same text -- so + // a function that can answer differently each time has no coherent value + // to declare. + let mut volatile = None; + parsed + .apply(|expr| { + use datafusion_common::tree_node::TreeNodeRecursion; + if let datafusion_expr::Expr::ScalarFunction(function) = expr + && function.func.signature().volatility != datafusion_expr::Volatility::Immutable + { + volatile = Some(function.func.name().to_string()); + return Ok(TreeNodeRecursion::Stop); + } + Ok(TreeNodeRecursion::Continue) + }) + .map_err(|e| invalid(e.to_string()))?; + if let Some(function) = volatile { + return Err(invalid(format!( + "'{function}' is not deterministic; a computed column's expression must \ + yield the same value every time it is evaluated" + ))); + } + + let mut inputs = Planner::column_names_in_expr(&parsed); + inputs.sort(); + inputs.dedup(); + + // A nested input is recorded by its path but read through its root + // column; Schema::index_of resolves top-level names only. Resolved here + // rather than left to the planner so an unknown column names itself in + // the error instead of surfacing as a plan failure. + let mut indices = Vec::with_capacity(inputs.len()); + for input in &inputs { + let index = schema + .index_of(root(input)) + .map_err(|_| invalid(format!("unknown column '{input}'")))?; + if !indices.contains(&index) { + indices.push(index); + } + } + indices.sort_unstable(); + + // Physical expressions address columns by position, so the planner that + // compiles the expression has to be built on the projected schema + // evaluation will actually read. + let read_schema = Arc::new( + schema + .project(&indices) + .map_err(|e| invalid(e.to_string()))?, + ); + let optimized = planner + .optimize_expr(parsed) + .map_err(|e| invalid(e.to_string()))?; + let physical = Planner::new(read_schema.clone()) + .create_physical_expr(&optimized) + .map_err(|e| invalid(e.to_string()))?; + let data_type = physical + .data_type(read_schema.as_ref()) + .map_err(|e| invalid(e.to_string()))?; + + Ok(BoundExpression { inputs, data_type }) +} + +/// Resolve `(name, expression)` pairs against `schema` into fields carrying +/// their bindings. +/// +/// Everything that can be known statically is checked here rather than at +/// 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. +pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result> { + if columns.is_empty() { + return Err(Error::InvalidInput { + message: "at least one computed column is required".into(), + }); + } + + 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()) { + return Err(Error::ColumnAlreadyExists { name: name.clone() }); + } + + let bound = bind(schema.clone(), name, expression)?; + + // Declared columns start entirely null, so nullability is a property + // of the declaration rather than of what the expression yields. + fields.push( + ArrowField::new(name, bound.data_type, true) + .with_metadata(computed_column_metadata(expression, &bound.inputs)), + ); + declared.push(name); + } + + Ok(fields) +} + +/// Build the transform that declares `columns` against `schema`. +/// +/// An all-null column is how a binding with no values yet is carried into a +/// commit; that it is spelled `AllNulls` is a detail of the commit, not of the +/// column, which is why this is internal and +/// [`AddColumnsBuilder::computed`](super::AddColumnsBuilder::computed) is the +/// public way in. +pub(crate) fn declare( + schema: SchemaRef, + columns: &[(String, String)], +) -> Result { + let fields = plan(schema, columns)?; + Ok(NewColumnTransform::AllNulls(Arc::new(ArrowSchema::new( + fields, + )))) +} + +/// Commit a declaration of a kind this version does not produce, the way a +/// newer lancedb would leave one behind. Bypasses admission, which exists to +/// stop exactly this through the public API. +#[cfg(test)] +pub(super) async fn add_foreign_kind(table: &crate::Table, name: &str, kind: &str) { + let field = ArrowField::new(name, DataType::Int32, true).with_metadata(HashMap::from([ + (COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()), + (KIND_META_KEY.to_string(), kind.to_string()), + (INPUTS_META_KEY.to_string(), r#"["x"]"#.to_string()), + ])); + super::schema_evolution::commit_add_columns( + table.as_native().unwrap(), + NewColumnTransform::AllNulls(Arc::new(ArrowSchema::new(vec![field]))), + None, + ) + .await + .unwrap(); +} + +#[cfg(test)] +mod tests { + use arrow_array::record_batch; + use arrow_schema::DataType; + use futures::TryStreamExt; + use lance::dataset::ColumnAlteration; + + use super::*; + use crate::connect; + use crate::query::{ExecutableQuery, QueryBase, Select}; + use crate::{Error, Table}; + + async fn table_with_ints(name: &str) -> Table { + let conn = connect("memory://").execute().await.unwrap(); + let batch = record_batch!(("x", Int32, [1, 2, 3])).unwrap(); + conn.create_table(name, batch).execute().await.unwrap() + } + + /// Declare `columns` the way a caller would: plan the expressions, then + /// add them through the ordinary column API. + async fn add_computed(table: &Table, columns: &[(String, String)]) -> Result { + let mut builder = table.add_columns(); + for (name, expression) in columns { + builder = builder.computed(name, expression); + } + Ok(builder.execute().await?.version) + } + + async fn declared(table: &Table) -> Vec { + computed_columns(table.schema().await.unwrap().as_ref()) + } + + #[tokio::test] + async fn test_declare_infers_type_and_inputs() { + let table = table_with_ints("declare_infers").await; + let initial = table.version().await.unwrap(); + + let version = add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + assert!(version > initial); + + let schema = table.schema().await.unwrap(); + let field = schema.field_with_name("doubled").unwrap(); + assert_eq!(field.data_type(), &DataType::Int32); + assert!(field.is_nullable()); + + assert_eq!( + declared(&table).await, + vec![ComputedColumn { + name: "doubled".into(), + kind: ComputedColumnKind::Sql { + expression: "x * 2".into() + }, + inputs: vec!["x".into()], + }] + ); + } + + /// The binding reaches the schema only if `AllNulls` carries per-field + /// metadata through the commit. The whole representation rests on it. + #[tokio::test] + async fn test_all_nulls_preserves_field_metadata() { + let table = table_with_ints("metadata_survives").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let schema = table.schema().await.unwrap(); + let metadata = schema.field_with_name("doubled").unwrap().metadata(); + assert_eq!( + metadata.get(COMPUTED_COLUMN_META_KEY).map(String::as_str), + Some("true") + ); + assert_eq!(metadata.get(KIND_META_KEY).map(String::as_str), Some("sql")); + assert_eq!( + metadata.get(EXPRESSION_META_KEY).map(String::as_str), + Some("x * 2") + ); + assert_eq!( + metadata.get(INPUTS_META_KEY).map(String::as_str), + Some(r#"["x"]"#) + ); + } + + #[tokio::test] + async fn test_declared_column_is_all_null() { + let table = table_with_ints("declare_is_null").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let batches = table + .query() + .select(Select::columns(&["doubled"])) + .execute() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + + let total: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(total, 3); + for batch in &batches { + assert_eq!(batch["doubled"].null_count(), batch.num_rows()); + } + } + + #[tokio::test] + async fn test_unknown_column_fails_at_declare_time() { + let table = table_with_ints("unknown_input").await; + let err = add_computed(&table, &[("bad".into(), "missing + 1".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "bad")); + + let schema = table.schema().await.unwrap(); + assert!(schema.field_with_name("bad").is_err()); + } + + #[tokio::test] + async fn test_unparsable_expression_fails_at_declare_time() { + let table = table_with_ints("bad_syntax").await; + let err = add_computed(&table, &[("bad".into(), "x *".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "bad")); + assert!( + table + .schema() + .await + .unwrap() + .field_with_name("bad") + .is_err() + ); + } + + /// A user-defined function is an expression like any other; only its + /// resolution is missing. When a registry-aware planner exists this + /// becomes a supported declaration rather than a new API. + #[tokio::test] + async fn test_unregistered_function_is_rejected_for_now() { + let table = table_with_ints("udf_not_yet").await; + let err = add_computed(&table, &[("vec".into(), "embed(x)".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "vec")); + assert!( + table + .schema() + .await + .unwrap() + .field_with_name("vec") + .is_err() + ); + } + + #[tokio::test] + async fn test_existing_column_name_is_rejected() { + let table = table_with_ints("name_taken").await; + let err = add_computed(&table, &[("x".into(), "x * 2".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::ColumnAlreadyExists { name } if name == "x")); + assert!(declared(&table).await.is_empty()); + } + + #[tokio::test] + async fn test_constant_expression_needs_no_inputs() { + let table = table_with_ints("constant").await; + add_computed(&table, &[("answer".into(), "42".into())]) + .await + .unwrap(); + + let declared = declared(&table).await; + assert_eq!(declared.len(), 1); + assert!(declared[0].inputs.is_empty()); + } + + #[tokio::test] + async fn test_multiple_columns_in_one_commit() { + let table = table_with_ints("multi").await; + let initial = table.version().await.unwrap(); + + add_computed( + &table, + &[ + ("plus".into(), "x + 1".into()), + ("squared".into(), "x * x".into()), + ], + ) + .await + .unwrap(); + + assert_eq!(table.version().await.unwrap(), initial + 1); + let declared = declared(&table).await; + assert_eq!(declared.len(), 2); + assert_eq!(declared[0].name, "plus"); + assert_eq!(declared[1].name, "squared"); + } + + #[tokio::test] + async fn test_duplicate_declaration_in_one_call_is_rejected() { + let table = table_with_ints("dupe").await; + let err = add_computed( + &table, + &[ + ("dup".into(), "x + 1".into()), + ("dup".into(), "x + 2".into()), + ], + ) + .await + .unwrap_err(); + assert!(matches!(err, Error::ColumnAlreadyExists { name } if name == "dup")); + assert!(declared(&table).await.is_empty()); + } + + /// A column added by an ordinary transform is materialized, not bound, so + /// it carries no declaration to report. + #[tokio::test] + async fn test_ordinary_columns_are_not_reported_as_computed() { + let table = table_with_ints("plain").await; + assert!(declared(&table).await.is_empty()); + + table + .add_columns() + .transform(NewColumnTransform::SqlExpressions(vec![( + "eager".into(), + "x * 2".into(), + )])) + .execute() + .await + .unwrap(); + assert!(declared(&table).await.is_empty()); + } + + /// Built-in functions type the column the same way an operator does. + #[tokio::test] + async fn test_builtin_function_inference() { + let conn = connect("memory://").execute().await.unwrap(); + let batch = record_batch!(("name", Utf8, ["ada", "grace"]), ("n", Int32, [-1, 2])).unwrap(); + let table = conn + .create_table("builtins", batch) + .execute() + .await + .unwrap(); + + add_computed( + &table, + &[ + ("shout".into(), "upper(name)".into()), + ("width".into(), "length(name)".into()), + ("magnitude".into(), "abs(n)".into()), + ], + ) + .await + .unwrap(); + + let schema = table.schema().await.unwrap(); + assert_eq!( + schema.field_with_name("shout").unwrap().data_type(), + &DataType::Utf8 + ); + assert_eq!( + schema.field_with_name("magnitude").unwrap().data_type(), + &DataType::Int32 + ); + // length() returns a width-dependent integer type; assert it is one + // rather than pinning which. + assert!( + schema + .field_with_name("width") + .unwrap() + .data_type() + .is_integer() + ); + + let declared = declared(&table).await; + assert_eq!(declared.len(), 3); + assert_eq!(declared[0].inputs, vec!["name".to_string()]); + assert_eq!(declared[2].inputs, vec!["n".to_string()]); + } + + /// The reason the kind is tagged: a declaration written by a newer version + /// has to read back as a computed column this one cannot evaluate, not as + /// an ordinary column. Reported as absent it would be refreshable by + /// nothing and redeclarable over, silently. + #[tokio::test] + async fn test_unrecognized_kind_is_reported_rather_than_hidden() { + let table = table_with_ints("foreign_kind").await; + super::add_foreign_kind(&table, "embedding", "udf").await; + + assert_eq!( + declared(&table).await, + vec![ComputedColumn { + name: "embedding".into(), + kind: ComputedColumnKind::Unrecognized { kind: "udf".into() }, + inputs: vec!["x".into()], + }] + ); + + let err = add_computed(&table, &[("embedding".into(), "x * 2".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::ColumnAlreadyExists { name } if name == "embedding")); + } + + /// A kind is what makes a declaration readable at all, so the flag alone + /// is half-formed in the same way a missing expression is. + #[test] + fn test_flag_without_a_kind_is_not_a_declaration() { + let field = + ArrowField::new("half", DataType::Int32, true).with_metadata(HashMap::from([( + COMPUTED_COLUMN_META_KEY.to_string(), + "true".to_string(), + )])); + assert_eq!(computed_column_from_field(&field), None); + } + + /// A SQL declaration is its expression; without one there is nothing to + /// refresh from. + #[test] + fn test_sql_kind_without_an_expression_is_not_a_declaration() { + let field = ArrowField::new("half", DataType::Int32, true).with_metadata(HashMap::from([ + (COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()), + (KIND_META_KEY.to_string(), SQL_KIND.to_string()), + ])); + assert_eq!(computed_column_from_field(&field), None); + } + + #[tokio::test] + async fn test_inputs_are_deduplicated_and_sorted() { + let conn = connect("memory://").execute().await.unwrap(); + let batch = record_batch!(("b", Int32, [1, 2]), ("a", Int32, [3, 4])).unwrap(); + let table = conn.create_table("dedupe", batch).execute().await.unwrap(); + + add_computed(&table, &[("total".into(), "b + a + b".into())]) + .await + .unwrap(); + + assert_eq!( + declared(&table).await[0].inputs, + vec!["a".to_string(), "b".to_string()] + ); + } + + #[tokio::test] + async fn test_dropping_an_input_is_refused() { + let table = table_with_ints("drop_input").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let err = table.drop_columns(&["x"]).await.unwrap_err(); + assert!( + matches!(&err, Error::InvalidInput { message } if message.contains("doubled")), + "{err:?}" + ); + } + + #[tokio::test] + async fn test_renaming_an_input_is_refused() { + let table = table_with_ints("rename_input").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let err = table + .alter_columns(&[ColumnAlteration::new("x".into()).rename("y".into())]) + .await + .unwrap_err(); + assert!( + matches!(&err, Error::InvalidInput { message } if message.contains("doubled")), + "{err:?}" + ); + } + + /// Nothing resolves against nullability, so it is not a rebinding. + #[tokio::test] + async fn test_altering_an_input_nullability_is_allowed() { + let table = table_with_ints("nullable_input").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + table + .alter_columns(&[ColumnAlteration::new("x".into()).set_nullable(true)]) + .await + .unwrap(); + } + + /// The gate's reproducer: a volatile function evaluates differently in + /// the counting and writing passes, so the declared value is incoherent. + /// Refused at declare time. + #[tokio::test] + async fn test_a_volatile_expression_is_refused() { + let table = table_with_ints("volatile_expr").await; + let err = add_computed(&table, &[("maybe".into(), "random() < 0.5".into())]) + .await + .unwrap_err(); + assert!( + matches!(&err, Error::InvalidExpression { message, .. } + if message.contains("random") && message.contains("deterministic")), + "{err:?}" + ); + } + + /// The gate's reproducer: the simplifier folds `true OR x > 0` to a + /// constant, but the stored SQL still names `x`, so the recorded inputs + /// must too -- otherwise dropping `x` is allowed and refresh breaks. + #[tokio::test] + async fn test_inputs_survive_expression_optimization() { + let table = table_with_ints("optimized_inputs").await; + add_computed(&table, &[("flag".into(), "true OR x > 0".into())]) + .await + .unwrap(); + + assert_eq!(declared(&table).await[0].inputs, vec!["x".to_string()]); + let err = table.drop_columns(&["x"]).await.unwrap_err(); + assert!( + matches!(&err, Error::InvalidInput { message } if message.contains("flag")), + "{err:?}" + ); + } + + /// The gate's reproducer: casting a computed column rewrites the field + /// without its metadata, silently destroying the declaration. + #[tokio::test] + async fn test_retyping_the_computed_column_is_refused() { + use arrow_schema::DataType as ArrowDataType; + + let table = table_with_ints("retype_computed").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let err = table + .alter_columns(&[ColumnAlteration::new("doubled".into()).cast_to(ArrowDataType::Int64)]) + .await + .unwrap_err(); + assert!( + matches!(&err, Error::InvalidInput { message } if message.contains("computed")), + "{err:?}" + ); + + // The declaration survives the refused change. + assert_eq!(declared(&table).await.len(), 1); + } + + /// A declaration cannot be edited, fabricated or erased through field + /// metadata: it is validated as a whole at declare time. + #[tokio::test] + async fn test_declaration_metadata_is_immutable() { + use crate::table::FieldMetadataUpdate; + + let table = table_with_ints("metadata_tamper").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + // Moving the binding. + let err = table + .update_field_metadata(&[ + FieldMetadataUpdate::new("doubled").set(EXPRESSION_META_KEY, "x * 3") + ]) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidInput { .. }), "{err:?}"); + + // Fabricating a declaration on a plain column. + let err = table + .update_field_metadata(&[FieldMetadataUpdate::new("x") + .set(COMPUTED_COLUMN_META_KEY, "true") + .set(KIND_META_KEY, SQL_KIND) + .set(EXPRESSION_META_KEY, "x")]) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidInput { .. }), "{err:?}"); + + // Erasing the declaration wholesale. + let err = table + .update_field_metadata(&[FieldMetadataUpdate::new("doubled") + .set("note", "hi") + .replace()]) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidInput { .. }), "{err:?}"); + + // Ordinary metadata on a computed column still merges, leaving the + // declaration intact. + table + .update_field_metadata(&[FieldMetadataUpdate::new("doubled").set("note", "hi")]) + .await + .unwrap(); + assert_eq!(declared(&table).await.len(), 1); + } + + /// The gate's reproducer: only refresh materializes a declared column; + /// a direct write would store an arbitrary durable value. + #[tokio::test] + async fn test_a_computed_column_cannot_be_written_directly() { + let table = table_with_ints("direct_write").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let batch = record_batch!(("x", Int32, [4]), ("doubled", Int32, [999])).unwrap(); + let err = table.add(batch.clone()).execute().await.unwrap_err(); + assert!( + matches!(&err, Error::InvalidInput { message } if message.contains("refresh")), + "{err:?}" + ); + + let err = table + .update() + .column("doubled", "999") + .execute() + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidInput { .. })); + + let mut merge = table.merge_insert(&["x"]); + merge + .when_matched_update_all(None) + .when_not_matched_insert_all(); + let err = merge + .execute(Box::new(arrow_array::RecordBatchIterator::new( + vec![Ok(batch.clone())], + batch.schema(), + ))) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidInput { .. })); + + // The append that omits the column still works. + let plain = record_batch!(("x", Int32, [4])).unwrap(); + table.add(plain).execute().await.unwrap(); + } + + /// The gate's reproducer: the reciprocal of the declare-under-spec check. + #[tokio::test] + async fn test_installing_an_lsm_spec_over_computed_columns_is_refused() { + use crate::table::LsmWriteSpec; + + let tmp_dir = tempfile::tempdir().unwrap(); + let conn = connect(tmp_dir.path().to_str().unwrap()) + .execute() + .await + .unwrap(); + let schema = Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new( + "x", + DataType::Int32, + false, + )])); + let batch = arrow_array::RecordBatch::try_new( + schema, + vec![Arc::new(arrow_array::Int32Array::from(vec![1, 2])) as _], + ) + .unwrap(); + let table = conn + .create_table("lsm_after", batch) + .execute() + .await + .unwrap(); + table.set_unenforced_primary_key(["x"]).await.unwrap(); + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let err = table + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap_err(); + assert!( + matches!(&err, Error::NotSupported { message } if message.contains("computed")), + "{err:?}" + ); + assert!(table.get_lsm_write_spec().await.unwrap().is_none()); + } + + /// The gate's reproducer: declaration metadata is admitted only through + /// the validated declare path, never smuggled through a raw transform. + #[tokio::test] + async fn test_forged_declaration_metadata_is_rejected() { + let table = table_with_ints("forged_metadata").await; + let field = + ArrowField::new("doubled", DataType::Int32, true).with_metadata(HashMap::from([ + (COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()), + (KIND_META_KEY.to_string(), SQL_KIND.to_string()), + (EXPRESSION_META_KEY.to_string(), "x * 2".to_string()), + (INPUTS_META_KEY.to_string(), "[]".to_string()), + ])); + let err = table + .add_columns() + .transform(NewColumnTransform::AllNulls(Arc::new(ArrowSchema::new( + vec![field], + )))) + .execute() + .await + .unwrap_err(); + assert!( + matches!(&err, Error::InvalidInput { message } if message.contains("computed()")), + "{err:?}" + ); + assert!(declared(&table).await.is_empty()); + } + + /// The gate's reproducer: SQL INSERT is a write path too. + #[tokio::test] + async fn test_sql_insert_cannot_write_a_computed_column() { + use datafusion::prelude::SessionContext; + + let table = table_with_ints("sql_insert").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let ctx = SessionContext::new(); + let provider = + crate::table::datafusion::BaseTableAdapter::try_new(table.base_table().clone()) + .await + .unwrap(); + ctx.register_table("t", Arc::new(provider)).unwrap(); + + let result = async { + ctx.sql("INSERT INTO t (x, doubled) VALUES (4, 999)") + .await? + .collect() + .await + } + .await; + let err = result.unwrap_err().to_string(); + assert!(err.contains("refresh"), "{err}"); + } + + /// The gate's reproducer: an overwrite must not smuggle in a filled + /// declaration. + #[tokio::test] + async fn test_overwrite_cannot_inject_a_declaration() { + use crate::table::AddDataMode; + + let table = table_with_ints("overwrite_inject").await; + let field = + ArrowField::new("doubled", DataType::Int32, true).with_metadata(HashMap::from([ + (COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()), + (KIND_META_KEY.to_string(), SQL_KIND.to_string()), + (EXPRESSION_META_KEY.to_string(), "x * 2".to_string()), + ])); + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("x", DataType::Int32, true), + field, + ])); + let batch = arrow_array::RecordBatch::try_new( + schema, + vec![ + Arc::new(arrow_array::Int32Array::from(vec![1])) as _, + Arc::new(arrow_array::Int32Array::from(vec![999])) as _, + ], + ) + .unwrap(); + + let err = table + .add(batch) + .mode(AddDataMode::Overwrite) + .execute() + .await + .unwrap_err(); + assert!( + matches!(&err, Error::InvalidInput { message } if message.contains("declare")), + "{err:?}" + ); + } + + #[tokio::test] + async fn test_create_table_cannot_inject_a_declaration() { + let conn = connect("memory://").execute().await.unwrap(); + let field = + ArrowField::new("doubled", DataType::Int32, true).with_metadata(HashMap::from([ + (COMPUTED_COLUMN_META_KEY.to_string(), "true".to_string()), + (KIND_META_KEY.to_string(), SQL_KIND.to_string()), + (EXPRESSION_META_KEY.to_string(), "x * 2".to_string()), + ])); + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("x", DataType::Int32, true), + field, + ])); + let batch = arrow_array::RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(arrow_array::Int32Array::from(vec![1])) as _, + Arc::new(arrow_array::Int32Array::from(vec![999])) as _, + ], + ) + .unwrap(); + let err = conn + .create_table("forged_create", batch) + .execute() + .await + .unwrap_err(); + assert!( + matches!(&err, Error::InvalidInput { message } if message.contains("computed()")), + "{err:?}" + ); + } + + #[tokio::test] + async fn test_sql_insert_omitting_computed_is_allowed() { + use datafusion::prelude::SessionContext; + + let table = table_with_ints("sql_insert_omitted").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + let ctx = SessionContext::new(); + let provider = + crate::table::datafusion::BaseTableAdapter::try_new(table.base_table().clone()) + .await + .unwrap(); + ctx.register_table("t", Arc::new(provider)).unwrap(); + ctx.sql("INSERT INTO t (x) VALUES (4)") + .await + .unwrap() + .collect() + .await + .unwrap(); + + table.checkout_latest().await.unwrap(); + assert_eq!(table.count_rows(None).await.unwrap(), 4); + } + + #[tokio::test] + async fn test_a_nested_computed_field_cannot_be_renamed() { + let table = table_with_ints("computed_struct_rename").await; + add_computed(&table, &[("payload".into(), "named_struct('a', x)".into())]) + .await + .unwrap(); + + let err = table + .alter_columns(&[ColumnAlteration::new("payload.a".into()).rename("b".into())]) + .await + .unwrap_err(); + assert!( + matches!(&err, Error::InvalidInput { message } if message.contains("payload")), + "{err:?}" + ); + } + + /// Stale handles must not commit the computed/LSM state in either order. + #[tokio::test] + async fn test_stale_handles_cannot_mix_computed_and_lsm() { + use crate::table::LsmWriteSpec; + + let tmp_dir = tempfile::tempdir().unwrap(); + let uri = tmp_dir.path().to_str().unwrap(); + let schema = Arc::new(ArrowSchema::new(vec![ArrowField::new( + "x", + DataType::Int32, + false, + )])); + let batch = arrow_array::RecordBatch::try_new( + schema.clone(), + vec![Arc::new(arrow_array::Int32Array::from(vec![1])) as _], + ) + .unwrap(); + let conn = connect(uri).execute().await.unwrap(); + let table = conn.create_table("mix", batch).execute().await.unwrap(); + table.set_unenforced_primary_key(["x"]).await.unwrap(); + let stale = conn.open_table("mix").execute().await.unwrap(); + + // Declare on one handle; the stale handle must not install a spec. + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + let err = stale + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap_err(); + assert!(matches!(err, Error::NotSupported { .. }), "install won"); + + // Reverse order on fresh tables. + let batch = arrow_array::RecordBatch::try_new( + schema, + vec![Arc::new(arrow_array::Int32Array::from(vec![1])) as _], + ) + .unwrap(); + let table = conn.create_table("mix2", batch).execute().await.unwrap(); + table.set_unenforced_primary_key(["x"]).await.unwrap(); + let stale = conn.open_table("mix2").execute().await.unwrap(); + table + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap(); + let err = add_computed(&stale, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap_err(); + assert!(matches!(err, Error::NotSupported { .. }), "declare won"); + } + + /// The gate's reproducer: after catch-up activation, an LSM write, and + /// unset, retained SSTable rows survive without a live spec. The catch-up + /// flag is the durable marker; declaration refuses on it. + #[tokio::test] + async fn test_unset_with_retained_lsm_rows_cannot_admit_a_declaration() { + use crate::table::LsmWriteSpec; + use arrow_array::{Int64Array, RecordBatchIterator}; + + let tmp_dir = tempfile::tempdir().unwrap(); + let conn = connect(tmp_dir.path().to_str().unwrap()) + .execute() + .await + .unwrap(); + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", DataType::Int64, false), + ArrowField::new("value", DataType::Int64, false), + ])); + let batch = arrow_array::RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(Int64Array::from(vec![1, 2])) as _, + Arc::new(Int64Array::from(vec![10, 20])) as _, + ], + ) + .unwrap(); + let table = conn + .create_table("t", batch.clone()) + .execute() + .await + .unwrap(); + table.set_unenforced_primary_key(["id"]).await.unwrap(); + table + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap(); + table.require_mem_wal_index_catchup().await.unwrap(); + + let mut merge = table.merge_insert(&["id"]); + merge + .when_matched_update_all(None) + .when_not_matched_insert_all() + .use_lsm(true); + merge + .execute(Box::new(RecordBatchIterator::new(vec![Ok(batch)], schema))) + .await + .unwrap(); + table.unset_lsm_write_spec().await.unwrap(); + + let err = add_computed(&table, &[("doubled".into(), "value * 2".into())]) + .await + .unwrap_err(); + assert!( + matches!(&err, Error::NotSupported { message } if message.contains("LSM")), + "{err:?}" + ); + } + + /// A declaration does not read itself, so it travels with its binding. + #[tokio::test] + async fn test_dropping_the_computed_column_is_allowed() { + let table = table_with_ints("drop_computed").await; + add_computed(&table, &[("doubled".into(), "x * 2".into())]) + .await + .unwrap(); + + table.drop_columns(&["doubled"]).await.unwrap(); + assert!(declared(&table).await.is_empty()); + } +} diff --git a/rust/lancedb/src/table/datafusion/insert.rs b/rust/lancedb/src/table/datafusion/insert.rs index e176c228b..b9bd2396e 100644 --- a/rust/lancedb/src/table/datafusion/insert.rs +++ b/rust/lancedb/src/table/datafusion/insert.rs @@ -17,7 +17,7 @@ use datafusion_physical_plan::stream::RecordBatchStreamAdapter; use datafusion_physical_plan::{ DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties, }; -use futures::TryStreamExt; +use futures::StreamExt; use lance::Dataset; use lance::dataset::transaction::{Operation, Transaction}; use lance::dataset::{CommitBuilder, InsertBuilder, WriteParams, WriteProgressFn}; @@ -194,12 +194,23 @@ impl ExecutionPlan for InsertExec { let output_bytes = MetricBuilder::new(&self.metrics).output_bytes(partition); let input_schema = input_stream.schema(); + let declared: Vec = crate::table::computed_columns::computed_columns( + &arrow_schema::Schema::from(self.dataset.schema()), + ) + .into_iter() + .map(|declaration| declaration.name) + .collect(); let input_stream: SendableRecordBatchStream = Box::pin(InstrumentedRecordBatchStreamAdapter::new( input_schema, - input_stream.map_ok(move |batch| { + input_stream.map(move |batch| { + let batch = batch?; + crate::table::computed_columns::ensure_batch_writes_no_computed_values( + &declared, &batch, + ) + .map_err(|e| datafusion::error::DataFusionError::External(Box::new(e)))?; output_bytes.add(batch.get_array_memory_size()); - batch + Ok(batch) }), partition, &self.metrics, diff --git a/rust/lancedb/src/table/merge/lsm.rs b/rust/lancedb/src/table/merge/lsm.rs index eb2feacbd..5751cd916 100644 --- a/rust/lancedb/src/table/merge/lsm.rs +++ b/rust/lancedb/src/table/merge/lsm.rs @@ -94,7 +94,16 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec) .await? }; + table.checkout_latest().await?; let mut dataset = (*table.dataset.get().await?).clone(); + let schema = arrow_schema::Schema::from(dataset.schema()); + if !crate::table::computed_columns::computed_columns(&schema).is_empty() { + return Err(Error::NotSupported { + message: "an LSM write spec cannot be installed on a table with computed \ + columns: rows in un-compacted tiers are invisible to refresh" + .into(), + }); + } let mut builder = dataset.initialize_mem_wal(); let writer_config_defaults = match spec { LsmWriteSpec::Bucket { diff --git a/rust/lancedb/src/table/schema_evolution.rs b/rust/lancedb/src/table/schema_evolution.rs index ce208111a..7503fd790 100644 --- a/rust/lancedb/src/table/schema_evolution.rs +++ b/rust/lancedb/src/table/schema_evolution.rs @@ -8,12 +8,14 @@ //! - [`alter_columns`](execute_alter_columns): Rename columns, change types, or modify nullability //! - [`drop_columns`](execute_drop_columns): Remove columns from the table +use arrow_schema::Schema as ArrowSchema; use lance::dataset::{ColumnAlteration, NewColumnTransform}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; -use super::NativeTable; -use crate::Result; +use super::computed_columns; +use super::{BaseTable, NativeTable}; +use crate::{Error, Result}; /// The result of an add columns operation. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] @@ -98,6 +100,48 @@ pub(crate) async fn execute_add_columns( table: &NativeTable, transforms: NewColumnTransform, read_columns: Option>, +) -> Result { + // Declarations are admitted only through [`execute_declare`]. + match &transforms { + NewColumnTransform::AllNulls(schema) => { + computed_columns::ensure_no_foreign_declarations(schema.fields())? + } + NewColumnTransform::BatchUDF(udf) => { + computed_columns::ensure_no_foreign_declarations(udf.output_schema.fields())? + } + _ => {} + } + commit_add_columns(table, transforms, read_columns).await +} + +/// Declare validated computed columns. The only admission path for +/// declaration metadata. +pub(crate) async fn execute_declare( + table: &NativeTable, + columns: &[(String, String)], +) -> Result { + // An LSM write spec keeps visible rows in tiers refresh cannot reach; + // checked against latest committed state, not this handle's snapshot. + // The catch-up flag outlives unset and marks retained SSTable rows. + table.checkout_latest().await?; + let catchup = table.dataset.get().await?.manifest().reader_feature_flags + & lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP + != 0; + if catchup || table.get_lsm_write_spec().await?.is_some() { + return Err(Error::NotSupported { + message: "computed columns are not supported on a table with an LSM write \ + spec: rows in un-compacted tiers are invisible to refresh" + .into(), + }); + } + let transform = computed_columns::declare(table.schema().await?, columns)?; + commit_add_columns(table, transform, None).await +} + +pub(crate) async fn commit_add_columns( + table: &NativeTable, + transforms: NewColumnTransform, + read_columns: Option>, ) -> Result { table.dataset.ensure_mutable()?; let mut dataset = (*table.dataset.get().await?).clone(); @@ -116,6 +160,21 @@ pub(crate) async fn execute_alter_columns( ) -> Result { table.dataset.ensure_mutable()?; let mut dataset = (*table.dataset.get().await?).clone(); + // Nullability is not part of what an expression resolves against, so only + // a rename or a retype can invalidate a binding. + let schema = std::sync::Arc::new(ArrowSchema::from(dataset.schema())); + let rebinding = alterations + .iter() + .filter(|alteration| alteration.rename.is_some() || alteration.data_type.is_some()) + .map(|alteration| alteration.path.as_str()) + .collect::>(); + computed_columns::ensure_not_an_input(&schema, &rebinding)?; + let retyped = alterations + .iter() + .filter(|alteration| alteration.data_type.is_some()) + .map(|alteration| alteration.path.as_str()) + .collect::>(); + computed_columns::ensure_not_retyped(schema.as_ref(), &retyped)?; dataset.alter_columns(alterations).await?; let version = dataset.version().version; table.dataset.update(dataset); @@ -131,6 +190,10 @@ pub(crate) async fn execute_drop_columns( ) -> Result { table.dataset.ensure_mutable()?; let mut dataset = (*table.dataset.get().await?).clone(); + computed_columns::ensure_not_an_input( + &std::sync::Arc::new(ArrowSchema::from(dataset.schema())), + columns, + )?; dataset.drop_columns(columns).await?; let version = dataset.version().version; table.dataset.update(dataset); @@ -147,6 +210,44 @@ pub(crate) async fn execute_update_field_metadata( table.dataset.ensure_mutable()?; let mut dataset = (*table.dataset.get().await?).clone(); + // A declaration is validated as a whole at declare time; editing its keys + // here would bypass that, fabricate one on a plain column, or move a + // binding out from under a refresh. A replace on a declared column would + // silently erase it. + let schema = ArrowSchema::from(dataset.schema()); + let declared: Vec = computed_columns::computed_columns(&schema) + .into_iter() + .map(|declaration| declaration.name) + .collect(); + for update in updates { + if update + .metadata + .keys() + .any(|key| computed_columns::is_declaration_key(key)) + { + return Err(Error::InvalidInput { + message: format!( + "metadata keys of a computed-column declaration cannot be edited \ + (path '{}'); drop the column and declare it again", + update.path + ), + }); + } + if update.replace + && declared + .iter() + .any(|name| name == computed_columns::root(&update.path)) + { + return Err(Error::InvalidInput { + message: format!( + "replacing all metadata of computed column '{}' would erase its \ + declaration; drop the column and declare it again", + update.path + ), + }); + } + } + let mut builder = dataset.update_field_metadata(); for update in updates { let entries = update.metadata.iter().map(|(k, v)| (k.clone(), v.clone())); diff --git a/rust/lancedb/src/table/update.rs b/rust/lancedb/src/table/update.rs index 61eb93992..fd9fa6828 100644 --- a/rust/lancedb/src/table/update.rs +++ b/rust/lancedb/src/table/update.rs @@ -82,6 +82,10 @@ pub(crate) async fn execute_update( // 1. Snapshot the current dataset let dataset = table.dataset.get().await?; + super::computed_columns::ensure_not_written( + &arrow_schema::Schema::from(dataset.schema()), + update.columns.iter().map(|(name, _)| name.as_str()), + )?; // 2. Initialize the Lance Core builder let mut builder = LanceUpdateBuilder::new(dataset);