From 9755c57d4111425b81f6fe4494ed58a3bf5cc416 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Fri, 13 Nov 2026 03:20:20 +0800 Subject: [PATCH] fix: prune Function outcomes for replaced fragments (#5005) An input-preserving merge can fail freshness preservation after a successful data write: the replaced fragment loses its signature, but its known/failed-row records remain and fail sidecar validation. This affects tables with persisted Function outcome metadata. Prune both outcome maps with the same live-fragment set used for signatures and row state. Outcomes on untouched fragments remain intact, and inserted rows remain pending fill. The regression performs a real mixed update/insert and reopens the resulting snapshot to check both output columns. Native validation on Linux: the exact regression fails before the fix with `failure evidence names a fragment without a signature`. At `20b10d80408cb455d9e5827f5d7d36a117622dea`, all 24 discovered `table::freshness::tests::function_write_` cases pass, including the new real merge regression. Test lists, counts and logs were retained and SHA-256 verified. [Current-head CI passed both workspace Clippy configurations](https://github.com/lancedb/lancedb/actions/runs/43274052568/job/131377627811). A separate package-only, default-feature Clippy command reported dead-code errors in seven files unchanged from the base; that local failure is retained rather than represented as a pass. Other hosted gates are pending. This does not claim downstream Sophon integration or deployed acceptance. --- rust/lancedb/src/table/freshness.rs | 97 +++++++++++++++++++++++++++++ 1 file changed, 97 insertions(+) diff --git a/rust/lancedb/src/table/freshness.rs b/rust/lancedb/src/table/freshness.rs index 477020fc3..645653a46 100644 --- a/rust/lancedb/src/table/freshness.rs +++ b/rust/lancedb/src/table/freshness.rs @@ -1879,6 +1879,8 @@ pub async fn preserve_function_freshness( pending_fill.retain(|id, _| live.contains(id)); state.signatures.retain(|id, _| live.contains(id)); state.row_state.retain(|id, _| live.contains(id)); + state.failure_coverage.retain(|id, _| live.contains(id)); + state.failed_rows.retain(|id, _| live.contains(id)); state.basis.clear(); let reference = write_sidecar(dataset, &state.normalize()?).await?; let fill_reference = if pending_fill.is_empty() { @@ -3029,6 +3031,101 @@ mod tests { } } + #[tokio::test] + async fn function_write_prunes_outcomes_for_replaced_fragments() { + let table = fresh_function_table().await; + let mut dataset = table + .as_native() + .unwrap() + .dataset + .get() + .await + .unwrap() + .as_ref() + .clone(); + let inputs = fields_for_paths(dataset.schema(), &["title".into(), "body".into()]).unwrap(); + let signatures = signatures_for(&dataset, &[0, 1], &inputs).unwrap(); + let failures = FailureEvidence { + known_rows: BTreeMap::from([ + (0, RoaringBitmap::from_iter([0, 1])), + (1, RoaringBitmap::from_iter([0, 1])), + ]), + failed_rows: BTreeMap::from([(1, RoaringBitmap::from_iter([1]))]), + }; + for column in ["search_text", "search_token_count"] { + record_freshness_with_failures( + &mut dataset, + None, + column, + "fb_01K3TEXT", + &inputs, + signatures.clone(), + &failures, + ) + .await + .unwrap(); + } + + // Replace all of fragment 0, insert one uncomputed row, and leave + // fragment 1 (including its retained failure evidence) untouched. + let source = arrow_array::record_batch!( + ("title", Utf8, ["a", "b", "e"]), + ("spare", Int32, [1, 1, 1]) + ) + .unwrap(); + let mut merge = + MergeInsertBuilder::try_new(Arc::new(dataset), vec!["title".into()]).unwrap(); + merge.when_matched(WhenMatched::UpdateAll); + merge.when_not_matched(WhenNotMatched::InsertAll); + let (merged, stats) = merge + .try_build() + .unwrap() + .execute_reader(Box::new(RecordBatchIterator::new( + [Ok(source.clone())], + source.schema(), + ))) + .await + .unwrap(); + let mut dataset = merged.as_ref().clone(); + assert!(!dataset.get_fragments().iter().any(|f| f.id() == 0)); + assert!(dataset.get_fragments().iter().any(|f| f.id() == 1)); + preserve_function_freshness( + &mut dataset, + FunctionWrite::Merge { + source_schema: &source.schema(), + on: &["title".into()], + num_inserted_rows: stats.num_inserted_rows, + }, + ) + .await + .unwrap(); + + let reopened = dataset + .checkout_version(dataset.version().version) + .await + .unwrap(); + for column in ["search_text", "search_token_count"] { + let StrictFreshness::Certified { + staleness, + failures, + } = strict_freshness_against(&reopened, column, "fb_01K3TEXT", &inputs) + .await + .unwrap() + else { + panic!("the preserved sidecar should retain failure evidence"); + }; + assert_eq!( + failures.known_rows, + BTreeMap::from([(1, RoaringBitmap::from_iter([0, 1]))]) + ); + assert_eq!( + failures.failed_rows, + BTreeMap::from([(1, RoaringBitmap::from_iter([1]))]) + ); + assert_eq!(staleness.pending_fill.len(), 1); + } + } + #[rstest::rstest] #[case::same_definition(false)] #[case::changed_definition(true)]