From d5fac37ea3da535b3aca96d3583da5a55359fb18 Mon Sep 17 00:00:00 2001 From: kid Date: Wed, 22 Jul 2026 10:26:21 +0800 Subject: [PATCH 1/3] test(mem_wal): cover cross-generation durable-put fencing end to end Batch positions restart at 0 in every MemTable generation, so a durability watch keyed on the generation-local position can be satisfied by the previous generation's flush at the same position (#7760). The lib-level regression test_durable_ack_after_rotation_requires_its_own_wal_append covers the false acknowledgement itself; add the public-surface counterpart from the closed #7759: after a rotation, a peer claims a higher epoch, and writer A's first durable put into the new generation must wait for its own WAL flush and therefore surface the PeerClaimedEpoch fence instead of acking from the stale generation's durability state. --- rust/lance/tests/integration_tests.rs | 1 + rust/lance/tests/mem_wal.rs | 81 +++++++++++++++++++++++++++ 2 files changed, 82 insertions(+) create mode 100644 rust/lance/tests/mem_wal.rs diff --git a/rust/lance/tests/integration_tests.rs b/rust/lance/tests/integration_tests.rs index 7a6d3e71ca4..e6ea4e321ae 100644 --- a/rust/lance/tests/integration_tests.rs +++ b/rust/lance/tests/integration_tests.rs @@ -4,6 +4,7 @@ // NOTE: we only create one integration test binary, to keep compilation overhead down. mod count_pushdown; +mod mem_wal; #[cfg(feature = "slow_tests")] mod query; #[cfg(feature = "slow_tests")] diff --git a/rust/lance/tests/mem_wal.rs b/rust/lance/tests/mem_wal.rs new file mode 100644 index 00000000000..047b7404b15 --- /dev/null +++ b/rust/lance/tests/mem_wal.rs @@ -0,0 +1,81 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The Lance Authors + +//! End-to-end MemWAL tests through the public [`ShardWriter`] surface, as +//! opposed to the lib-level unit tests that can reach internals directly. + +use arrow_array::record_batch; +use lance::dataset::mem_wal::{ShardWriter, ShardWriterConfig}; +use lance_core::FenceReason; +use lance_io::object_store::ObjectStore; +use uuid::Uuid; + +/// A durable put into a post-rotation MemTable must be acknowledged only by +/// its own WAL flush, never by a flush from an older generation. +/// +/// Batch positions restart at 0 in every MemTable generation. A durability +/// watch keyed on that generation-local position is satisfied by the previous +/// generation's flush at the same position — so the first durable put after a +/// rotation used to return success before its own WAL append completed. Here +/// a peer writer has claimed a higher epoch in between, so the false success +/// is observable: the put's own flush is fenced, and the put must surface +/// that fence instead of reporting durability. +#[tokio::test] +async fn durable_put_does_not_alias_across_memtable_generations() { + let temp_dir = tempfile::tempdir().unwrap(); + let base_uri = format!("file://{}", temp_dir.path().display()); + let (store, base_path) = ObjectStore::from_uri(&base_uri).await.unwrap(); + let shard_id = Uuid::new_v4(); + let first_batch = record_batch!(("id", Int32, [1])).unwrap(); + let schema = first_batch.schema(); + let config = ShardWriterConfig { + shard_id, + durable_write: true, + max_wal_buffer_size: 64 * 1024 * 1024, + max_wal_flush_interval: None, + max_memtable_size: 64 * 1024 * 1024, + manifest_scan_batch_size: 2, + ..Default::default() + }; + + let writer_a = ShardWriter::open( + store.clone(), + base_path.clone(), + base_uri.clone(), + config.clone(), + schema.clone(), + vec![], + ) + .await + .unwrap(); + + // Write and fully flush generation N, then rotate to generation N+1. + writer_a.put(vec![first_batch]).await.unwrap(); + let first_generation = writer_a.memtable_stats().await.unwrap().generation; + writer_a.force_seal_active().await.unwrap(); + writer_a.wait_for_flush_drain().await.unwrap(); + assert_eq!( + writer_a.memtable_stats().await.unwrap().generation, + first_generation + 1 + ); + + // A peer claims a higher epoch, fencing writer A's next WAL append. + let writer_b = ShardWriter::open(store, base_path, base_uri, config, schema, vec![]) + .await + .unwrap(); + assert!(writer_b.epoch() > writer_a.epoch()); + + // Generation N+1's first durable put lands at the same local batch + // position (0..1) that generation N already flushed. It must wait for its + // own flush and therefore surface the fence, not ack from the stale + // generation. + let second_batch = record_batch!(("id", Int32, [2])).unwrap(); + let error = writer_a + .put(vec![second_batch]) + .await + .expect_err("generation-2 durable put must wait for its own WAL flush"); + assert_eq!(error.fence_reason(), Some(FenceReason::PeerClaimedEpoch)); + assert!(error.to_string().contains("Writer fenced")); + + writer_b.close().await.unwrap(); +} From 9e23eac1239ed4547595b34ec573087838ef9854 Mon Sep 17 00:00:00 2001 From: kid Date: Wed, 22 Jul 2026 16:37:44 +0800 Subject: [PATCH 2/3] test(mem_wal): enable WAL ticker in fencing regression --- rust/lance/tests/mem_wal.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/rust/lance/tests/mem_wal.rs b/rust/lance/tests/mem_wal.rs index 047b7404b15..a4768e9bb59 100644 --- a/rust/lance/tests/mem_wal.rs +++ b/rust/lance/tests/mem_wal.rs @@ -4,6 +4,8 @@ //! End-to-end MemWAL tests through the public [`ShardWriter`] surface, as //! opposed to the lib-level unit tests that can reach internals directly. +use std::time::Duration; + use arrow_array::record_batch; use lance::dataset::mem_wal::{ShardWriter, ShardWriterConfig}; use lance_core::FenceReason; @@ -32,7 +34,7 @@ async fn durable_put_does_not_alias_across_memtable_generations() { shard_id, durable_write: true, max_wal_buffer_size: 64 * 1024 * 1024, - max_wal_flush_interval: None, + max_wal_flush_interval: Some(Duration::from_millis(10)), max_memtable_size: 64 * 1024 * 1024, manifest_scan_batch_size: 2, ..Default::default() From 19d9629f7b36ac9788a94c95aec7fc49118ea4ad Mon Sep 17 00:00:00 2001 From: kid Date: Fri, 24 Jul 2026 00:20:40 +0800 Subject: [PATCH 3/3] test(mem_wal): add write-read integration coverage --- .../tests/{mem_wal.rs => mem_wal/mod.rs} | 80 +++++++++++++++---- 1 file changed, 64 insertions(+), 16 deletions(-) rename rust/lance/tests/{mem_wal.rs => mem_wal/mod.rs} (61%) diff --git a/rust/lance/tests/mem_wal.rs b/rust/lance/tests/mem_wal/mod.rs similarity index 61% rename from rust/lance/tests/mem_wal.rs rename to rust/lance/tests/mem_wal/mod.rs index a4768e9bb59..c617c2a1589 100644 --- a/rust/lance/tests/mem_wal.rs +++ b/rust/lance/tests/mem_wal/mod.rs @@ -6,12 +6,55 @@ use std::time::Duration; +use arrow_array::cast::AsArray; use arrow_array::record_batch; +use arrow_array::types::Int32Type; use lance::dataset::mem_wal::{ShardWriter, ShardWriterConfig}; use lance_core::FenceReason; use lance_io::object_store::ObjectStore; use uuid::Uuid; +fn durable_writer_config(shard_id: Uuid) -> ShardWriterConfig { + ShardWriterConfig { + shard_id, + durable_write: true, + max_wal_buffer_size: 64 * 1024 * 1024, + max_wal_flush_interval: Some(Duration::from_millis(10)), + max_memtable_size: 64 * 1024 * 1024, + manifest_scan_batch_size: 2, + ..Default::default() + } +} + +#[tokio::test] +async fn durable_put_is_readable_through_public_scan() { + let base_uri = "memory://".to_string(); + let (store, base_path) = ObjectStore::from_uri(&base_uri).await.unwrap(); + let batch = record_batch!(("id", Int32, [1, 2, 3])).unwrap(); + let writer = ShardWriter::open( + store, + base_path, + base_uri, + durable_writer_config(Uuid::new_v4()), + batch.schema(), + vec![], + ) + .await + .unwrap(); + + writer.put(vec![batch]).await.unwrap(); + + let scanned = writer.scan().await.unwrap().try_into_batch().await.unwrap(); + let scanned_ids = scanned["id"].as_primitive::(); + assert_eq!( + scanned_ids.values(), + &[1, 2, 3], + "a durable put must be readable through the public scan API" + ); + + writer.close().await.unwrap(); +} + /// A durable put into a post-rotation MemTable must be acknowledged only by /// its own WAL flush, never by a flush from an older generation. /// @@ -24,21 +67,12 @@ use uuid::Uuid; /// that fence instead of reporting durability. #[tokio::test] async fn durable_put_does_not_alias_across_memtable_generations() { - let temp_dir = tempfile::tempdir().unwrap(); - let base_uri = format!("file://{}", temp_dir.path().display()); + let base_uri = "memory://".to_string(); let (store, base_path) = ObjectStore::from_uri(&base_uri).await.unwrap(); let shard_id = Uuid::new_v4(); let first_batch = record_batch!(("id", Int32, [1])).unwrap(); let schema = first_batch.schema(); - let config = ShardWriterConfig { - shard_id, - durable_write: true, - max_wal_buffer_size: 64 * 1024 * 1024, - max_wal_flush_interval: Some(Duration::from_millis(10)), - max_memtable_size: 64 * 1024 * 1024, - manifest_scan_batch_size: 2, - ..Default::default() - }; + let config = durable_writer_config(shard_id); let writer_a = ShardWriter::open( store.clone(), @@ -56,16 +90,23 @@ async fn durable_put_does_not_alias_across_memtable_generations() { let first_generation = writer_a.memtable_stats().await.unwrap().generation; writer_a.force_seal_active().await.unwrap(); writer_a.wait_for_flush_drain().await.unwrap(); + let current_generation = writer_a.memtable_stats().await.unwrap().generation; assert_eq!( - writer_a.memtable_stats().await.unwrap().generation, - first_generation + 1 + current_generation, + first_generation + 1, + "expected rotation from generation {first_generation}, got {current_generation}" ); // A peer claims a higher epoch, fencing writer A's next WAL append. let writer_b = ShardWriter::open(store, base_path, base_uri, config, schema, vec![]) .await .unwrap(); - assert!(writer_b.epoch() > writer_a.epoch()); + assert!( + writer_b.epoch() > writer_a.epoch(), + "expected peer epoch to increase: writer_a={}, writer_b={}", + writer_a.epoch(), + writer_b.epoch() + ); // Generation N+1's first durable put lands at the same local batch // position (0..1) that generation N already flushed. It must wait for its @@ -76,8 +117,15 @@ async fn durable_put_does_not_alias_across_memtable_generations() { .put(vec![second_batch]) .await .expect_err("generation-2 durable put must wait for its own WAL flush"); - assert_eq!(error.fence_reason(), Some(FenceReason::PeerClaimedEpoch)); - assert!(error.to_string().contains("Writer fenced")); + assert_eq!( + error.fence_reason(), + Some(FenceReason::PeerClaimedEpoch), + "expected a peer-claimed-epoch fence, got: {error}" + ); + assert!( + error.to_string().contains("Writer fenced"), + "expected the writer-fenced error prefix, got: {error}" + ); writer_b.close().await.unwrap(); }