From 13e7bfa94cf840164185bd38be5f6760f546d826 Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Sun, 16 Aug 2026 15:05:30 -0700 Subject: [PATCH] fix: rebuild capped views on removal, and trust only real fragment ids Two ways a refresh could drop rows it never materialized. A cap stops rows being written once the view is full, and the watermark then advances past them, so room freed by a later delete could never be refilled from any delta -- the view sat incomplete under its own cap. A capped view now rebuilds when the source removes or changes rows. Separately, a transaction file carries placeholder ids for the fragments it creates. Merging into an empty source made those collide with the ids the commit went on to assign, so the first rows read as already-materialized and were skipped while the watermark still moved. Only ids of fragments that already existed are treated as real; rewritten rows are excluded by creation version, as they already were. Co-Authored-By: Claude Fable 5 --- rust/lancedb/src/materialized_view/refresh.rs | 91 +++++++++++-------- 1 file changed, 52 insertions(+), 39 deletions(-) diff --git a/rust/lancedb/src/materialized_view/refresh.rs b/rust/lancedb/src/materialized_view/refresh.rs index 9c86e2f6c..fcde43afe 100644 --- a/rust/lancedb/src/materialized_view/refresh.rs +++ b/rust/lancedb/src/materialized_view/refresh.rs @@ -311,6 +311,14 @@ async fn plan_increment( if folded { return None; } + // A cap stops rows being materialized once the view is full, and the + // watermark then advances past them. Eviction can free room those + // rows should fill, but they are no longer in any delta, so a capped + // view cannot reconcile a removal incrementally. Rebuilding it is + // cheap for the same reason it is capped: the scan stops at the cap. + if definition.limit.is_some() && (delta.deleted_rows || delta.updated_rows) { + return None; + } // 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. @@ -409,10 +417,15 @@ async fn appends_and_rewrites(cur: &Dataset, from: u64, to: u64) -> Option { for group in groups { @@ -676,7 +689,6 @@ async fn incremental( definition, RowScope { row_ids: Some(&updated_ids), - limit: remaining, ..Default::default() }, schema.clone(), @@ -688,11 +700,6 @@ async fn incremental( recomputed.chain(stream), )); } - // New and recomputed rows compete for one budget: capping each stream - // separately lets their sum exceed the view's cap. - if let Some(remaining) = remaining { - stream = take_rows(stream, remaining); - } // Nothing survived the filter: the watermark still has to advance or the // same fragments would be rescanned forever. @@ -1071,28 +1078,6 @@ async fn evict_by_provenance( Ok(!ids.is_empty()) } -/// Cap a stream at `limit` rows, slicing the batch that crosses the boundary. -fn take_rows(stream: SendableRecordBatchStream, limit: u64) -> SendableRecordBatchStream { - let schema = stream.schema(); - let capped = stream.scan(0u64, move |taken, batch| { - let out = batch.map(|batch| { - let room = limit.saturating_sub(*taken) as usize; - let batch = if batch.num_rows() > room { - batch.slice(0, room) - } else { - batch - }; - *taken += batch.num_rows() as u64; - batch - }); - std::future::ready(match &out { - Ok(batch) if batch.num_rows() == 0 => None, - _ => Some(out), - }) - }); - Box::pin(RecordBatchStreamAdapter::new(schema, capped)) -} - fn source_row_id_field_id(view_ds: &Dataset) -> Result { view_ds .schema() @@ -1437,6 +1422,32 @@ mod tests { .cloned() } + /// A transaction file carries placeholder ids for the fragments it + /// creates. Into an empty source those collide with the ids the commit + /// assigns, so treating them as already-materialized drops the very + /// first rows while the watermark still advances past them. + #[tokio::test] + async fn test_merge_into_an_empty_source_is_materialized() { + let (conn, source) = db_with_source(vec![]).await; + let view = doubled_view(&conn).await; + view.refresh().execute().await.unwrap(); + assert_eq!(read(view.table(), "twice").await, Vec::::new()); + + let batch = record_batch!(("x", Int32, vec![1, 2])).unwrap(); + let reader = arrow_array::RecordBatchIterator::new(vec![Ok(batch.clone())], batch.schema()); + let mut merge = source.merge_insert(&["x"]); + merge + .when_matched_update_all(None) + .when_not_matched_insert_all(); + merge.execute(Box::new(reader)).await.unwrap(); + + view.refresh().execute().await.unwrap(); + assert_eq!(read(view.table(), "twice").await, vec![2, 4]); + // A second refresh must not double them either. + view.refresh().execute().await.unwrap(); + assert_eq!(read(view.table(), "twice").await, vec![2, 4]); + } + async fn compact(source: &Table) { source .optimize(OptimizeAction::Compact { @@ -1560,11 +1571,12 @@ mod tests { assert_eq!(read(view.table(), "twice").await, vec![2, 4]); } - /// New and recomputed rows draw on one budget. Capping each stream on - /// its own lets their sum exceed the view's limit, so the span has to - /// carry both at once to show it. + /// A cap and incremental reconciliation do not compose: rows skipped at + /// the cap fall behind the watermark, so room freed later cannot be + /// refilled from any delta. A capped view rebuilds instead, which its + /// own cap keeps cheap. #[tokio::test] - async fn test_limit_holds_across_new_and_recomputed_rows() { + async fn test_limited_view_rebuilds_rather_than_reconcile() { let (conn, source) = db_with_source(vec![1, 2]).await; let view = conn .create_materialized_view("capped", "src") @@ -1585,13 +1597,14 @@ mod tests { .await .unwrap(); let result = view.refresh().execute().await.unwrap(); - assert_eq!(result.mode, RefreshMode::Incremental); - assert_eq!( - read(view.table(), "x").await, - vec![2, 11], - "the budget covers both streams, and a row the view held outranks a new one" + assert_eq!(result.mode, RefreshMode::Rebuild); + let held = read(view.table(), "x").await; + assert_eq!(held.len(), 2, "the cap holds: {held:?}"); + let selectable = read(&source, "x").await; + assert!( + held.iter().all(|x| selectable.contains(x)), + "{held:?} is not a subset of {selectable:?}" ); - assert_eq!(result.rows_written, 1); } #[tokio::test]