diff --git a/crates/lance-context-core/tests/publication_gc_proof.rs b/crates/lance-context-core/tests/publication_gc_proof.rs new file mode 100644 index 0000000..fa7bb1a --- /dev/null +++ b/crates/lance-context-core/tests/publication_gc_proof.rs @@ -0,0 +1,179 @@ +//! Real Lance manifest/transaction cleanup, with full data readback after reopen. +//! This qualifies storage semantics; it does not implement a distributed GC fence. +use std::{ + collections::{BTreeMap, HashSet}, + sync::Arc, +}; + +use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator, StringArray}; +use arrow_schema::{DataType, Field, Schema}; +use futures::TryStreamExt; +use lance::{ + dataset::{ + cleanup::{cleanup_old_versions, CleanupPolicy}, + WriteMode, WriteParams, + }, + Dataset, +}; + +type TestResult = Result>; + +fn payload(id: i64) -> String { + format!("{id}:{}", "opaque-value".repeat(4096)) +} + +async fn write(uri: &str, id: i64) -> TestResult { + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("payload", DataType::Utf8, false), + ])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(Int64Array::from(vec![id])), + Arc::new(StringArray::from(vec![payload(id)])), + ], + )?; + let reader = RecordBatchIterator::new(vec![Ok(batch)], schema); + Ok(Dataset::write( + reader, + uri, + Some(WriteParams { + mode: if id == 1 { + WriteMode::Create + } else { + WriteMode::Append + }, + ..Default::default() + }), + ) + .await?) +} + +async fn verify(uri: &str, count: i64) -> TestResult { + let dataset = Dataset::open(uri).await?; + assert_eq!(dataset.version().version, count as u64); + let batches: Vec<_> = dataset + .scan() + .try_into_stream() + .await? + .try_collect() + .await?; + let mut rows = BTreeMap::new(); + for batch in batches { + let ids = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let bodies = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + assert!(rows + .insert(ids.value(row), bodies.value(row).to_owned()) + .is_none()); + } + } + assert_eq!(rows, (1..=count).map(|id| (id, payload(id))).collect()); + Ok(dataset) +} + +#[tokio::test] +async fn cleanup_can_remove_exact_transaction_proof_while_latest_data_survives() -> TestResult { + let dir = tempfile::tempdir()?; + let uri = dir.path().join("table.lance").to_str().unwrap().to_owned(); + write(&uri, 1).await?; + let second = write(&uri, 2).await?; + let committed = second + .read_transaction_by_version(2) + .await? + .expect("append transaction"); + assert!(!committed.uuid.is_empty()); + drop(second); + let latest = write(&uri, 3).await?; + let removed = cleanup_old_versions( + &latest, + CleanupPolicy { + versions: Some(HashSet::from([2])), + ..Default::default() + }, + ) + .await?; + assert_eq!(removed.old_versions, 1); + assert!(removed.transaction_files_removed >= 1); + drop(latest); + let reopened = verify(&uri, 3).await?; + let error = reopened + .read_transaction_by_version(2) + .await + .expect_err("collected target still resolvable"); + assert!( + matches!( + &error, + lance::Error::NotFound { .. } + | lance::Error::DatasetNotFound { .. } + | lance::Error::VersionNotFound { .. } + ), + "unexpected lookup failure: {error}" + ); + println!("verified: exact target proof can disappear while all latest rows remain; missing target is not proof of an uncommitted append"); + Ok(()) +} + +#[tokio::test] +async fn numeric_cleanup_floor_preserves_pending_base_target_and_transactions() -> TestResult { + let dir = tempfile::tempdir()?; + let uri = dir.path().join("table.lance").to_str().unwrap().to_owned(); + write(&uri, 1).await?; + let second = write(&uri, 2).await?; + let base = second + .read_transaction_by_version(2) + .await? + .expect("base transaction") + .uuid; + drop(second); + let third = write(&uri, 3).await?; + let target = third + .read_transaction_by_version(3) + .await? + .expect("target transaction") + .uuid; + drop(third); + let latest = write(&uri, 4).await?; + let removed = cleanup_old_versions( + &latest, + CleanupPolicy { + before_version: Some(2), + ..Default::default() + }, + ) + .await?; + assert_eq!( + removed.old_versions, 1, + "floor did not collect the older version" + ); + drop(latest); + let reopened = verify(&uri, 4).await?; + assert!(reopened.read_transaction_by_version(1).await.is_err()); + assert_eq!( + reopened + .read_transaction_by_version(2) + .await? + .expect("base evidence lost") + .uuid, + base + ); + assert_eq!( + reopened + .read_transaction_by_version(3) + .await? + .expect("target evidence lost") + .uuid, + target + ); + println!("verified: numeric cleanup floor removes older history while retaining base/target transactions and full latest data; distributed floor coordination still required"); + Ok(()) +} diff --git a/crates/lance-context-core/tests/publication_gc_stale_commit.rs b/crates/lance-context-core/tests/publication_gc_stale_commit.rs new file mode 100644 index 0000000..790aa8e --- /dev/null +++ b/crates/lance-context-core/tests/publication_gc_stale_commit.rs @@ -0,0 +1,217 @@ +//! Real Lance commit/cleanup interleaving on a local object store. +//! No cloud or distributed-GC claim is made by this experiment. +use std::{collections::BTreeMap, sync::Arc, time::Duration}; + +use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator, StringArray}; +use arrow_schema::{DataType, Field, Schema}; +use futures::TryStreamExt; +use lance::{ + dataset::{ + cleanup::{cleanup_old_versions, CleanupPolicy}, + CommitBuilder, InsertBuilder, WriteMode, WriteParams, + }, + Dataset, +}; +use lance_table::{ + format::{IndexMetadata, Manifest, Transaction}, + io::commit::{ + CommitError, CommitHandler, ConditionalPutCommitHandler, ManifestLocation, + ManifestNamingScheme, ManifestWriter, RenameCommitHandler, + }, +}; +use tokio::sync::Notify; + +type TestResult = Result>; + +fn payload(id: i64) -> String { + format!("{id}:{}", "immutable-value".repeat(4096)) +} + +fn batch(id: i64) -> TestResult { + Ok(RecordBatch::try_new( + Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("payload", DataType::Utf8, false), + ])), + vec![ + Arc::new(Int64Array::from(vec![id])), + Arc::new(StringArray::from(vec![payload(id)])), + ], + )?) +} + +async fn write(uri: &str, id: i64, handler: Arc) -> TestResult { + let data = batch(id)?; + let reader = RecordBatchIterator::new(vec![Ok(data.clone())], data.schema()); + Ok(Dataset::write( + reader, + uri, + Some(WriteParams { + mode: if id == 1 { + WriteMode::Create + } else { + WriteMode::Append + }, + commit_handler: Some(handler), + ..Default::default() + }), + ) + .await?) +} + +async fn verify(dataset: &Dataset, ids: &[i64]) -> TestResult { + let batches: Vec<_> = dataset + .scan() + .try_into_stream() + .await? + .try_collect() + .await?; + let mut rows = BTreeMap::new(); + for batch in batches { + let ids = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + assert!(rows + .insert(ids.value(row), values.value(row).to_owned()) + .is_none()); + } + } + assert_eq!(rows, ids.iter().map(|id| (*id, payload(*id))).collect()); + Ok(()) +} + +#[derive(Debug)] +struct PausedTarget { + inner: Arc, + entered: Arc, + release: Arc, +} + +#[async_trait::async_trait] +#[allow(clippy::too_many_arguments)] +impl CommitHandler for PausedTarget { + async fn commit( + &self, + manifest: &mut Manifest, + indices: Option>, + path: &object_store::path::Path, + store: &lance_io::object_store::ObjectStore, + writer: ManifestWriter, + naming: ManifestNamingScheme, + transaction: Option, + ) -> Result { + // Enforce the fixed target, then suspend before the actual storage write. + // An epoch check in this position has the same delayed-write window. + if manifest.version != 2 { + return Err(CommitError::CommitConflict); + } + self.entered.notify_one(); + self.release.notified().await; + self.inner + .commit(manifest, indices, path, store, writer, naming, transaction) + .await + } +} + +async fn interleave(handler: Arc, collect_target: bool) -> TestResult { + let dir = tempfile::tempdir()?; + let uri = dir.path().join("table.lance").to_str().unwrap().to_owned(); + let base = Arc::new(write(&uri, 1, handler.clone()).await?); + let tx = InsertBuilder::new(base.clone()) + .with_params(&WriteParams { + mode: WriteMode::Append, + ..Default::default() + }) + .execute_uncommitted(vec![batch(99)?]) + .await?; + assert_eq!(tx.read_version, 1); + let delayed_uuid = tx.uuid.clone(); + let entered = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let paused = Arc::new(PausedTarget { + inner: handler.clone(), + entered: entered.clone(), + release: release.clone(), + }); + let worker = tokio::spawn(async move { + CommitBuilder::new(base) + .with_commit_handler(paused) + .with_max_retries(0) + .with_skip_auto_cleanup(true) + .execute(tx) + .await + }); + tokio::time::timeout(Duration::from_secs(30), entered.notified()).await?; + let second = write(&uri, 2, handler.clone()).await?; + let incumbent_uuid = second + .read_transaction_by_version(2) + .await? + .unwrap() + .uuid + .clone(); + assert_ne!(incumbent_uuid, delayed_uuid); + let latest = write(&uri, 3, handler).await?; + if collect_target { + let stats = cleanup_old_versions( + &latest, + CleanupPolicy { + versions: Some(std::collections::HashSet::from([2])), + ..Default::default() + }, + ) + .await?; + assert_eq!(stats.old_versions, 1); + assert!(Dataset::open(&uri) + .await? + .read_transaction_by_version(2) + .await + .is_err()); + } + release.notify_one(); + let result = tokio::time::timeout(Duration::from_secs(30), worker).await??; + let reopened = Dataset::open(&uri).await?; + assert_eq!(reopened.version().version, 3); + verify(&reopened, &[1, 2, 3]).await?; + let actual_uuid = reopened + .read_transaction_by_version(2) + .await? + .unwrap() + .uuid + .clone(); + if collect_target { + let committed = result?; + assert_eq!(committed.version().version, 2); + assert_eq!(actual_uuid, delayed_uuid); + verify(&reopened.checkout_version(2).await?, &[1, 99]).await?; + println!("verified: collected target recreated with delayed UUID; latest version excludes delayed payload, so target UUID alone is insufficient after unconstrained GC"); + } else { + assert!( + result.is_err(), + "occupied fixed target accepted a foreign transaction" + ); + assert_eq!(actual_uuid, incumbent_uuid); + verify(&reopened.checkout_version(2).await?, &[1, 2]).await?; + println!("verified control: retained target rejects delayed foreign commit"); + } + Ok(()) +} + +#[tokio::test] +async fn conditional_put_reuses_a_collected_fixed_target() -> TestResult { + interleave(Arc::new(ConditionalPutCommitHandler), false).await?; + interleave(Arc::new(ConditionalPutCommitHandler), true).await +} + +#[tokio::test] +async fn atomic_rename_reuses_a_collected_fixed_target() -> TestResult { + interleave(Arc::new(RenameCommitHandler), false).await?; + interleave(Arc::new(RenameCommitHandler), true).await +} diff --git a/docs/design/wal-publication-gc.md b/docs/design/wal-publication-gc.md new file mode 100644 index 0000000..e6269b3 --- /dev/null +++ b/docs/design/wal-publication-gc.md @@ -0,0 +1,50 @@ +# Publication recovery and Lance version cleanup + +Refs #378. + +A durable WAL receipt does not by itself make the corresponding payload safe to +retire. The publisher also needs a durable outcome for its immutable source +selection, and the evidence used to establish that outcome must survive cleanup. +The optional publication work does not yet enable this reclamation protocol. + +The Rust integration tests use real Lance 13 APIs, transactions, data files and +full payload reads on a local object store. They establish three storage boundaries: + +- Cleaning an older target version removes its exact transaction proof while all + rows in the latest version can remain readable. A missing target is not proof + that an append never committed. +- A numeric `CleanupPolicy::before_version` floor preserves a pending base and + target, including their transactions, while permitting older history to be + collected. The floor alone does not coordinate concurrent plan admission. +- A delayed fixed-target commit can recreate a collected version slot. The tests + pause the commit immediately before its object-store write, allow another + transaction to publish that target and a successor, collect the target, and + then release the delayed commit. Both conditional PUT and atomic rename accept + the now-absent target. A fresh reader finds the delayed transaction UUID at that + old version, but the latest version excludes its payload. Control cases retain + the target and reject the delayed foreign transaction. + +These are characterization tests for the pinned Lance version, not production +GC coordination or a cloud performance measurement. They intentionally make +changes to these storage semantics visible when upgrading Lance. + +The integration must preserve the pending base and target until the publication +outcome is durable. Cleanup admission and new plan admission must agree on a +persistent lower bound; sampling the current pending plans and then running an +uncoordinated cleaner leaves a race. Every relevant cleanup path must participate, +including automatic cleanup triggered by other writers. + +Staged, uncommitted data files require a separate lifetime protocol. A retained +version floor does not reference those files, and a finite orphan-file grace +period does not protect an arbitrarily delayed publisher. An epoch check before a +storage write also does not revoke a write already in flight. Retiring manifests +or staged files must account for delayed writers as well as active readers and +snapshot transfers. Until these dependencies are resolved, retain the plan and +source data rather than interpreting missing or recreated target evidence as a +publication acknowledgement. + +Run the storage characterization tests with: + +```bash +cargo test -p lance-context-core --test publication_gc_proof --test publication_gc_stale_commit +```