From def869bb7815ce29ca7cf671a5b17010dee48b15 Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Fri, 14 Aug 2026 14:17:41 -0700 Subject: [PATCH] feat: declare computed columns by SQL expression (#3937) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit add_columns().computed("doubled", "x * 2") stores the expression in field metadata and commits the column empty; a later refresh fills it. Type and inputs are derived from the expression. The declaration stays authoritative for its lifetime: writes that would give the column a value (append, update, merge, SQL insert), schema changes that would break the stored expression or reshape its output, metadata edits, volatile expressions, declaration metadata arriving through any path but the validated declare call, and LSM write specs in either order against latest committed state are all refused. The LSM check also refuses on the mem-wal catch-up feature flag, which outlives unset and marks retained SSTable rows. Simultaneous declare/install interleavings conflict at commit via lance's mem-wal rule (lance#8539). Local tables only. --- Stack created with GitHub Stacks CLI • Give Feedback 💬 --- docs/src/js/classes/Table.md | 27 +- nodejs/__test__/table.test.ts | 19 + nodejs/lancedb/table.ts | 41 +- nodejs/src/table.rs | 14 + python/python/lancedb/_lancedb.pyi | 3 + python/python/lancedb/remote/table.py | 11 +- python/python/lancedb/table.py | 78 +- python/python/tests/test_table.py | 19 + python/src/table.rs | 15 + rust/lancedb/src/error.rs | 8 + rust/lancedb/src/remote/table.rs | 32 + rust/lancedb/src/table.rs | 32 + rust/lancedb/src/table/add_columns.rs | 137 +- rust/lancedb/src/table/computed_columns.rs | 1329 +++++++++++++++++++ rust/lancedb/src/table/datafusion/insert.rs | 17 +- rust/lancedb/src/table/merge/lsm.rs | 9 + rust/lancedb/src/table/schema_evolution.rs | 105 +- rust/lancedb/src/table/update.rs | 4 + 18 files changed, 1865 insertions(+), 35 deletions(-) create mode 100644 rust/lancedb/src/table/computed_columns.rs 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);