diff --git a/Cargo.lock b/Cargo.lock index e76049f06..f9128353c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -141,7 +141,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -152,7 +152,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1756,7 +1756,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -3034,7 +3034,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -3257,7 +3257,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -3456,7 +3456,7 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" [[package]] name = "fsst" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-array", "rand 0.9.5", @@ -4561,7 +4561,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -4816,7 +4816,7 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a" [[package]] name = "lance" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arc-swap", "arrow", @@ -4889,7 +4889,7 @@ dependencies = [ [[package]] name = "lance-arrow" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-array", "arrow-buffer", @@ -4911,7 +4911,7 @@ dependencies = [ [[package]] name = "lance-arrow-scalar" version = "58.0.0" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-array", "arrow-buffer", @@ -4925,7 +4925,7 @@ dependencies = [ [[package]] name = "lance-arrow-stats" version = "58.0.0" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-array", "arrow-schema", @@ -4935,7 +4935,7 @@ dependencies = [ [[package]] name = "lance-bitpacking" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrayref", "crunchy", @@ -4946,7 +4946,7 @@ dependencies = [ [[package]] name = "lance-core" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-array", "arrow-buffer", @@ -4984,7 +4984,7 @@ dependencies = [ [[package]] name = "lance-datafusion" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow", "arrow-array", @@ -5014,7 +5014,7 @@ dependencies = [ [[package]] name = "lance-datagen" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow", "arrow-array", @@ -5032,7 +5032,7 @@ dependencies = [ [[package]] name = "lance-derive" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "proc-macro2", "quote", @@ -5042,7 +5042,7 @@ dependencies = [ [[package]] name = "lance-encoding" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-arith", "arrow-array", @@ -5076,7 +5076,7 @@ dependencies = [ [[package]] name = "lance-file" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-arith", "arrow-array", @@ -5108,7 +5108,7 @@ dependencies = [ [[package]] name = "lance-index" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arc-swap", "arrow", @@ -5173,7 +5173,7 @@ dependencies = [ [[package]] name = "lance-index-core" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-array", "arrow-schema", @@ -5196,7 +5196,7 @@ dependencies = [ [[package]] name = "lance-io" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow", "arrow-array", @@ -5233,7 +5233,7 @@ dependencies = [ [[package]] name = "lance-linalg" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-array", "arrow-schema", @@ -5248,7 +5248,7 @@ dependencies = [ [[package]] name = "lance-namespace" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow", "async-trait", @@ -5261,7 +5261,7 @@ dependencies = [ [[package]] name = "lance-namespace-impls" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow", "arrow-ipc", @@ -5315,7 +5315,7 @@ dependencies = [ [[package]] name = "lance-select" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-array", "arrow-buffer", @@ -5330,7 +5330,7 @@ dependencies = [ [[package]] name = "lance-table" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow", "arrow-array", @@ -5371,7 +5371,7 @@ dependencies = [ [[package]] name = "lance-testing" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "arrow-array", "arrow-schema", @@ -5385,7 +5385,7 @@ dependencies = [ [[package]] name = "lance-tokenizer" version = "11.0.0-beta.13" -source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#f73052f90cb3d6673b6348cbec785857b2fb1986" +source = "git+https://github.com/wkalt/lance.git?branch=ticket%2Fent-1961%2Fdelta-deleted-row-ids#09a0514e99221a60ea0c32182081fcaa2944be7f" dependencies = [ "frostem", "icu_segmenter", @@ -6236,7 +6236,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -7628,8 +7628,8 @@ version = "0.14.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" dependencies = [ - "heck 0.4.1", - "itertools 0.11.0", + "heck 0.5.0", + "itertools 0.12.1", "log", "multimap", "petgraph", @@ -7648,7 +7648,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools 0.11.0", + "itertools 0.12.1", "proc-macro2", "quote", "syn 2.0.117", @@ -7919,7 +7919,7 @@ dependencies = [ "once_cell", "socket2 0.6.3", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -8697,7 +8697,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -8768,7 +8768,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -9324,7 +9324,7 @@ version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c1c97747dbf44bb1ca44a561ece23508e99cb592e862f22222dcf42f51d1e451" dependencies = [ - "heck 0.4.1", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.117", @@ -9336,7 +9336,7 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "54254b8531cafa275c5e096f62d48c81435d1015405a91198ddb11e967301d40" dependencies = [ - "heck 0.4.1", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.117", @@ -9359,7 +9359,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -9466,7 +9466,7 @@ dependencies = [ "cfg-if 1.0.4", "libc", "psm", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -9758,7 +9758,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -10735,7 +10735,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] diff --git a/rust/lancedb/src/materialized_view/refresh.rs b/rust/lancedb/src/materialized_view/refresh.rs index 0c8402bc3..77284cd0d 100644 --- a/rust/lancedb/src/materialized_view/refresh.rs +++ b/rust/lancedb/src/materialized_view/refresh.rs @@ -45,6 +45,8 @@ use std::sync::{Arc, Mutex as StdMutex, OnceLock}; use std::time::{SystemTime, UNIX_EPOCH}; use arrow_array::RecordBatch; +use arrow_array::cast::AsArray; +use arrow_array::types::UInt64Type; use arrow_schema::{Schema as ArrowSchema, SchemaRef}; use datafusion::error::DataFusionError; use datafusion::physical_plan::SendableRecordBatchStream; @@ -54,7 +56,7 @@ use lance::Dataset; use lance::dataset::mem_wal::DatasetMemWalExt; use lance::dataset::transaction::{Operation, Transaction}; use lance::dataset::{CommitBuilder, InsertBuilder, WriteDestination, WriteMode, WriteParams}; -use lance_core::ROW_ID; +use lance_core::{ROW_CREATED_AT_VERSION, ROW_ID}; use lance_table::format::Fragment; use serde::{Deserialize, Serialize}; @@ -242,15 +244,16 @@ pub(crate) async fn execute_refresh( ) .await { - Some(new_fragments) => { + Some(increment) => { incremental( view_native, &view_ds, &source_ds, source_version, source_ts, - new_fragments, + increment, definition, + watermark, ) .await } @@ -287,7 +290,7 @@ async fn plan_increment( recorded_ts: Option, full: bool, definition: &MaterializedViewDefinition, -) -> Option> { +) -> Option { if full { return None; } @@ -322,17 +325,23 @@ async fn plan_increment( // Every other fragment new at head is an append: appends and rewrites // are the only operations in the delta that add fragments, and the // rewrite outputs are already-materialized rows rearranged. - return Some( - live.into_iter() + return Some(Increment { + appended: live + .into_iter() .filter(|f| !old_ids.contains(&f.id) && !delta.produced.contains(&f.id)) .collect(), - ); + evict_deleted: delta.deleted_rows, + replace_updated: delta.updated_rows, + }); } - is_pure_append(&old, source_ds, &relevant_field_ids(source_ds, definition)).then(|| { - live.into_iter() + is_pure_append(&old, source_ds, &relevant_field_ids(source_ds, definition)).then(|| Increment { + appended: live + .into_iter() .filter(|f| !old_ids.contains(&f.id)) - .collect() + .collect(), + evict_deleted: false, + replace_updated: false, }) } @@ -347,11 +356,26 @@ const MAX_TRANSACTION_WALK: u64 = 512; /// fragments with placeholder ids (they are assigned at commit), while a /// rewrite's ids are reserved before it commits and are real. Appends are /// therefore derived as what is new at head and not a rewrite output. +/// What an incremental refresh must do to bring the view up to date. +struct Increment { + /// Source fragments whose rows are not in the view yet. + appended: Vec, + /// Source rows left the range, so the view holds rows to evict. + evict_deleted: bool, + /// Source rows changed in the range, so the view holds rows to replace. + replace_updated: bool, +} + struct TxnDelta { /// Fragments consumed by `Rewrite` operations. rewritten: HashSet, /// Fragments produced by `Rewrite` operations. produced: HashSet, + /// The delta removed source rows, so the view holds rows to evict. + deleted_rows: bool, + /// The delta changed source rows in place, so the view holds rows to + /// recompute. + updated_rows: bool, } /// Read the delta from the transaction log, `None` where it holds anything @@ -367,6 +391,8 @@ async fn appends_and_rewrites(cur: &Dataset, from: u64, to: u64) -> Option Option {} + // A delete removes source rows without changing the ones that + // remain, so the view's other rows stay valid: the refresh + // evicts exactly the ids that left. + Operation::Delete { .. } => delta.deleted_rows = true, + // An update changes rows in place, rewriting the fragments that + // hold them. Those outputs carry no new rows -- the same rows + // with new values -- so they are excluded from the append set + // the way a rewrite's outputs are, and the changed rows are + // replaced individually below. + Operation::Update { + removed_fragment_ids, + new_fragments, + updated_fragments, + .. + } => { + delta.updated_rows = true; + delta.rewritten.extend(removed_fragment_ids.iter().copied()); + delta.produced.extend(new_fragments.iter().map(|f| f.id)); + delta + .produced + .extend(updated_fragments.iter().map(|f| f.id)); + } Operation::Rewrite { groups, .. } => { for group in groups { delta @@ -532,9 +580,51 @@ async fn incremental( source_ds: &Dataset, source_version: u64, source_ts: u128, - new_fragments: Vec, + increment: Increment, definition: &MaterializedViewDefinition, + watermark: Option, ) -> Result { + let new_fragments = increment.appended; + let watermark_version = watermark.unwrap_or(0); + // Rows the source dropped since the watermark. The view holds one row + // per source row, keyed by provenance, so evicting them is exact: no + // other view row is affected, which is why a delete no longer forces a + // rebuild. + let mut evicted_view: Option = None; + let mut updated_ids: Vec = Vec::new(); + if (increment.evict_deleted || increment.replace_updated) + && let Some(watermark) = watermark + { + let delta = source_ds + .delta() + .with_begin_version(watermark) + .with_end_version(source_version) + .build()?; + let mut view = view_ds.clone(); + let mut evicted = false; + if increment.evict_deleted { + let mut stream = delta.get_deleted_row_ids().await?; + while let Some(batch) = stream.try_next().await? { + evicted |= evict_by_provenance(&mut view, &row_ids_of(&batch)?).await?; + } + } + if increment.replace_updated { + let mut stream = delta.get_updated_rows().await?; + while let Some(batch) = stream.try_next().await? { + updated_ids.extend(row_ids_of(&batch)?); + } + // Their current values are recomputed below and appended, so the + // rows the view holds for them go now. + evicted |= evict_by_provenance(&mut view, &updated_ids).await?; + } + if evicted { + view_native.dataset.update(view.clone()); + evicted_view = Some(view); + } + } + // The eviction commits, so everything below plans from the view it left + // rather than the snapshot this refresh opened with. + let view_ds = evicted_view.as_ref().unwrap_or(view_ds); // The cap counts rows already materialized, in first-materialized order. let remaining = match definition.limit { Some(limit) => { @@ -550,7 +640,7 @@ async fn incremental( source_version, version: view_ds.version().version, }; - if new_fragments.is_empty() || remaining == Some(0) { + if (new_fragments.is_empty() && updated_ids.is_empty()) || remaining == Some(0) { result.version = stamp_watermark(view_native, view_ds.clone(), source_version, source_ts).await?; return Ok(result); @@ -563,13 +653,41 @@ async fn incremental( let mut stream = compute_stream( source_ds, definition, - Some(new_fragments), - remaining, + RowScope { + fragments: Some(new_fragments), + // An update rewrites whole fragments, so a fragment new at head + // can hold rows the view already has. Their creation version + // does not change, so it -- not fragment identity -- says which + // rows are new. + created_after: increment.replace_updated.then_some(watermark_version), + limit: remaining, + ..Default::default() + }, schema.clone(), rows_written.clone(), ) .await?; + // The updated rows' current values, computed the same way and appended + // in the same commit as the new fragments' rows. + if !updated_ids.is_empty() { + let recomputed = compute_stream( + source_ds, + definition, + RowScope { + row_ids: Some(&updated_ids), + ..Default::default() + }, + schema.clone(), + rows_written.clone(), + ) + .await?; + stream = Box::pin(RecordBatchStreamAdapter::new( + schema.clone(), + stream.chain(recomputed), + )); + } + // Nothing survived the filter: the watermark still has to advance or the // same fragments would be rescanned forever. let Some(first) = stream.try_next().await? else { @@ -619,8 +737,10 @@ async fn rebuild( let stream = compute_stream( source_ds, definition, - None, - definition.limit, + RowScope { + limit: definition.limit, + ..Default::default() + }, schema, rows_written.clone(), ) @@ -763,21 +883,60 @@ fn now_ms() -> u128 { /// `(name, expression)` projection pair, never spliced into SQL text. Each /// view field is filled by the projection of the same name, and /// [`SOURCE_ROW_ID_COLUMN`] by the scan's row id. +/// Which source rows a compute pass reads. +#[derive(Default)] +struct RowScope<'a> { + /// Read only these fragments. + fragments: Option>, + /// Read only these rows. + row_ids: Option<&'a [u64]>, + /// Read only rows created after this version. + created_after: Option, + /// Stop after this many rows. + limit: Option, +} + async fn compute_stream( source: &Dataset, definition: &MaterializedViewDefinition, - fragments: Option>, - limit: Option, + scope: RowScope<'_>, schema: SchemaRef, rows_written: Arc, ) -> Result { + let RowScope { + fragments, + row_ids, + created_after, + limit, + } = scope; let mut scanner = source.scan(); if let Some(fragments) = fragments { scanner.with_fragments(fragments); } scanner.with_row_id(); - if let Some(filter) = &definition.filter { - scanner.filter(filter)?; + // Narrowing to specific rows keeps the definition's own filter, so a row + // that no longer satisfies it simply does not come back -- which is how + // an update that pushes a row out of the view removes it. + let ids_filter = row_ids.map(|ids| { + let list = ids + .iter() + .map(u64::to_string) + .collect::>() + .join(", "); + format!("{ROW_ID} IN ({list})") + }); + let created_filter = + created_after.map(|version| format!("{ROW_CREATED_AT_VERSION} > {version}")); + let clauses: Vec = definition + .filter + .clone() + .map(|f| format!("({f})")) + .into_iter() + .chain(ids_filter) + .chain(created_filter) + .collect(); + if !clauses.is_empty() { + scanner.filter(&clauses.join(" AND "))?; } let transforms: Vec<(&str, &str)> = definition .projections @@ -813,6 +972,41 @@ async fn compute_stream( Ok(Box::pin(RecordBatchStreamAdapter::new(schema, mapped))) } +/// Ids per eviction predicate. A delete names the rows it removes inline, so +/// the batch size is what keeps the SQL bounded when a source drops far more +/// rows than one statement should carry. +const EVICTION_CHUNK: usize = 4096; + +fn row_ids_of(batch: &RecordBatch) -> Result> { + let column = batch.column_by_name(ROW_ID).ok_or_else(|| Error::Runtime { + message: format!("'{ROW_ID}' is missing from a delta batch"), + })?; + let ids = column + .as_primitive_opt::() + .ok_or_else(|| Error::Runtime { + message: "row ids are not UInt64".into(), + })?; + Ok(ids.values().to_vec()) +} + +/// Drop the view rows carrying `ids`, one bounded statement at a time. +/// Repeating a chunk is harmless, so a refresh that fails partway leaves the +/// next one the same work. +async fn evict_by_provenance(view: &mut Dataset, ids: &[u64]) -> Result { + for chunk in ids.chunks(EVICTION_CHUNK) { + let predicate = format!( + "{SOURCE_ROW_ID_COLUMN} IN ({})", + chunk + .iter() + .map(u64::to_string) + .collect::>() + .join(", ") + ); + view.delete(&predicate).await?; + } + Ok(!ids.is_empty()) +} + #[cfg(test)] mod tests { use arrow_array::{Int32Array, record_batch}; @@ -970,27 +1164,50 @@ mod tests { /// Unlike a computed column, a view reflects source mutation: an update /// rebuilds rather than going stale. #[tokio::test] - async fn test_update_forces_a_rebuild() { - let (conn, source) = db_with_source(vec![1, 2]).await; - let view = doubled_view(&conn).await; - view.refresh().execute().await.unwrap(); - - source.update().column("x", "100").execute().await.unwrap(); - let result = view.refresh().execute().await.unwrap(); - assert_eq!(result.mode, RefreshMode::Rebuild); - assert_eq!(read(view.table(), "twice").await, vec![200, 200]); - } - - #[tokio::test] - async fn test_delete_forces_a_rebuild() { + async fn test_update_replaces_the_rows_it_changed() { let (conn, source) = db_with_source(vec![1, 2, 3]).await; let view = doubled_view(&conn).await; view.refresh().execute().await.unwrap(); + // An update changes rows in place, so the view replaces exactly + // those rows and leaves the rest of what it holds alone. + source + .update() + .column("x", "20") + .only_if("x = 2") + .execute() + .await + .unwrap(); + let result = view.refresh().execute().await.unwrap(); + assert_eq!(result.mode, RefreshMode::Incremental); + assert_eq!(result.rows_written, 1, "only the changed row is recomputed"); + assert_eq!(read(view.table(), "twice").await, vec![2, 6, 40]); + } + + #[tokio::test] + async fn test_delete_evicts_the_view_rows_it_removed() { + let (conn, source) = db_with_source(vec![1, 2, 3]).await; + let view = doubled_view(&conn).await; + view.refresh().execute().await.unwrap(); + + // A delete removes source rows without changing the ones that + // remain, so the view evicts exactly those rows and keeps the rest + // rather than recomputing every row it already held. source.delete("x = 2").await.unwrap(); let result = view.refresh().execute().await.unwrap(); - assert_eq!(result.mode, RefreshMode::Rebuild); + assert_eq!(result.mode, RefreshMode::Incremental); assert_eq!(read(view.table(), "twice").await, vec![2, 6]); + + // A delete and an append in one span: both are applied. + source.delete("x = 1").await.unwrap(); + source + .add(record_batch!(("x", Int32, vec![4])).unwrap()) + .execute() + .await + .unwrap(); + let result = view.refresh().execute().await.unwrap(); + assert_eq!(result.mode, RefreshMode::Incremental); + assert_eq!(read(view.table(), "twice").await, vec![6, 8]); } async fn compact(source: &Table) {